AI工程师的上下文管理术：让长对话不失忆的工程实践

LLM最大的局限之一是有限的上下文窗口。GPT-4o有128K tokenGemini 1.5 Pro有100万token——听起来很大但实际生产中长对话积累、知识库检索内容、工具调用结果……很快就能填满。更根本的问题是不是塞满上下文就好而是如何让有限的上下文空间承载最有价值的信息。上下文管理的四个核心问题1.记忆问题LLM默认无状态每次请求独立无法记住上一次对话2.容量问题即使有百万token窗口塞满了速度变慢、成本飙升3.相关性问题不是所有历史信息都有价值旧信息可能干扰新回复4.一致性问题长对话中模型可能忘记早先约定的规则或用户偏好## 对话历史的管理策略### 策略1滑动窗口Sliding Window最简单的方案只保留最近N轮对话超出则丢弃最旧的pythonfrom collections import dequefrom typing import Optionalclass SlidingWindowMemory: 滑动窗口对话记忆 def __init__(self, max_messages: int 20, system_prompt: str ): self.max_messages max_messages self.system_prompt system_prompt self.messages deque(maxlenmax_messages) def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) def get_messages(self) - list[dict]: 获取完整消息列表含系统提示 messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.extend(list(self.messages)) return messages def chat(self, user_input: str, client, model: str gpt-4o) - str: self.add_message(user, user_input) response client.chat.completions.create( modelmodel, messagesself.get_messages(), ) assistant_reply response.choices[0].message.content self.add_message(assistant, assistant_reply) return assistant_reply优点简单、可预测、成本可控缺点丢失历史信息用户提到你之前说的XXX可能无法响应### 策略2摘要压缩Summary Compression当对话超过阈值时将旧对话压缩为摘要pythonfrom openai import OpenAIclass SummaryCompressedMemory: 带摘要压缩的对话记忆 def __init__(self, client: OpenAI, max_recent_messages: int 10, compression_threshold: int 20, system_prompt: str ): self.client client self.max_recent max_recent_messages self.compression_threshold compression_threshold self.system_prompt system_prompt self.summary # 历史摘要 self.recent_messages [] # 最近的完整消息 def _compress(self): 将旧消息压缩为摘要 # 取出最旧的一半消息进行压缩 to_compress self.recent_messages[:len(self.recent_messages)//2] remaining self.recent_messages[len(self.recent_messages)//2:] conversation_text \n.join([ f{m[role].upper()}: {m[content]} for m in to_compress ]) compress_prompt f 已有历史摘要 {self.summary} 新增对话 {conversation_text} 请将上述内容合并为一段简洁的对话摘要保留关键信息、用户偏好和重要决策 response self.client.chat.completions.create( modelgpt-4o-mini, # 用便宜的小模型做压缩 messages[{role: user, content: compress_prompt}] ) self.summary response.choices[0].message.content self.recent_messages remaining def add_message(self, role: str, content: str): self.recent_messages.append({role: role, content: content}) # 超过阈值时触发压缩 if len(self.recent_messages) self.compression_threshold: self._compress() def get_messages(self) - list[dict]: messages [] # 系统提示 system_content self.system_prompt if self.summary: system_content f\n\n## 对话历史摘要\n{self.summary} if system_content: messages.append({role: system, content: system_content}) # 最近的完整消息 messages.extend(self.recent_messages[-self.max_recent:]) return messages### 策略3重要性过滤Importance Filtering不是所有消息都同等重要根据重要性动态选择保留哪些pythonclass ImportanceFilteredMemory: 基于重要性评分的记忆管理 IMPORTANCE_PROMPT 评估以下对话消息的重要性0-10分考虑 - 包含用户关键偏好或需求3 - 包含关键决策或约定3 - 包含具体数据或代码2 - 是普通闲聊-2 - 已被后续消息覆盖-2 消息{message} 只输出一个0到10的整数 def __init__(self, client, max_total_tokens: int 4000): self.client client self.max_tokens max_total_tokens self.messages_with_scores [] def _estimate_tokens(self, text: str) - int: return len(text) // 3 # 粗略估算中文约1字1token def _score_message(self, message: dict) - float: 对消息重要性打分 response self.client.chat.completions.create( modelgpt-4o-mini, messages[{ role: user, content: self.IMPORTANCE_PROMPT.format( messagef{message[role]}: {message[content]} ) }] ) try: return float(response.choices[0].message.content.strip()) except: return 5.0 def add_message(self, role: str, content: str, auto_score: bool False): message {role: role, content: content} score self._score_message(message) if auto_score else 5.0 self.messages_with_scores.append({ message: message, score: score, tokens: self._estimate_tokens(content) }) def get_messages(self, budget_tokens: int None) - list[dict]: 在token预算内优先选择高分消息 budget budget_tokens or self.max_tokens # 最近消息强制保留保证对话连贯性 recent self.messages_with_scores[-4:] recent_tokens sum(m[tokens] for m in recent) remaining_budget budget - recent_tokens # 其余消息按分数排序在预算内选择 older self.messages_with_scores[:-4] older_sorted sorted(older, keylambda x: x[score], reverseTrue) selected [] for item in older_sorted: if remaining_budget item[tokens]: selected.append(item) remaining_budget - item[tokens] # 按原始顺序重排 all_selected_indices set(id(m) for m in selected) ordered [m[message] for m in self.messages_with_scores if id(m) in all_selected_indices or m in recent] return ordered## 外部记忆用向量数据库存储长期记忆对于需要跨会话记忆的场景必须使用外部记忆存储pythonfrom langchain_openai import OpenAIEmbeddingsfrom langchain_community.vectorstores import FAISSfrom datetime import datetimeimport jsonclass VectorMemoryStore: 基于向量数据库的长期记忆 def __init__(self, user_id: str, persist_dir: str ./memory_store): self.user_id user_id self.persist_dir persist_dir self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) self._load_or_create() def _load_or_create(self): import os store_path f{self.persist_dir}/{self.user_id} if os.path.exists(store_path): self.vectorstore FAISS.load_local(store_path, self.embeddings) else: os.makedirs(store_path, exist_okTrue) self.vectorstore FAISS.from_texts( [初始化记忆存储], self.embeddings, metadatas[{type: init}] ) def save_memory(self, content: str, memory_type: str conversation): 保存一条记忆 self.vectorstore.add_texts( texts[content], metadatas[{ user_id: self.user_id, type: memory_type, timestamp: datetime.now().isoformat(), }] ) # 持久化 self.vectorstore.save_local(f{self.persist_dir}/{self.user_id}) def retrieve_relevant(self, query: str, k: int 5) - list[str]: 检索相关记忆 docs self.vectorstore.similarity_search(query, kk) return [doc.page_content for doc in docs if doc.metadata.get(type) ! init] def build_memory_context(self, current_query: str) - str: 构建记忆上下文注入到系统提示 memories self.retrieve_relevant(current_query) if not memories: return memory_text \n.join([f- {m} for m in memories]) return f## 相关历史记忆以下是与当前对话相关的历史信息{memory_text}这些信息来自过去的对话可能对当前回复有参考价值## 实体记忆跟踪对话中的关键实体pythonclass EntityMemory: 跟踪对话中的关键实体人名、项目名、偏好等 def __init__(self, client: OpenAI): self.client client self.entities {} # {entity_name: description} def extract_and_update(self, user_message: str, assistant_message: str): 从对话中提取实体信息 prompt f 从以下对话中提取关键实体信息人名、项目名、用户偏好、重要事实等。 如果已有实体有新信息更新它如果是全新实体添加它。 已知实体{json.dumps(self.entities, ensure_asciiFalse)} 用户{user_message} 助手{assistant_message} 以JSON格式输出需要更新的实体不需要变化的不要输出 {{实体名: 实体描述, ...}} 如果没有新实体信息输出空对象{{}} response self.client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], response_format{type: json_object} ) updates json.loads(response.choices[0].message.content) self.entities.update(updates) def get_context(self) - str: if not self.entities: return entity_text \n.join([f- {k}: {v} for k, v in self.entities.items()]) return f## 已知实体信息\n{entity_text}## 综合应用完整的记忆管理系统将上述策略组合为生产可用的记忆系统pythonclass ProductionMemorySystem: 生产级对话记忆系统 def __init__(self, client: OpenAI, user_id: str, system_prompt: str ): self.client client self.base_system system_prompt # 短期记忆滑动窗口 self.recent_history deque(maxlen20) # 中期记忆摘要压缩 self.session_summary # 长期记忆向量存储 self.vector_memory VectorMemoryStore(user_id) # 实体记忆 self.entity_memory EntityMemory(client) def build_messages(self, user_input: str) - list[dict]: 构建完整的消息列表 # 1. 从长期记忆检索相关信息 long_term_context self.vector_memory.build_memory_context(user_input) # 2. 获取实体信息 entity_context self.entity_memory.get_context() # 3. 构建系统提示 system_parts [self.base_system] if long_term_context: system_parts.append(long_term_context) if entity_context: system_parts.append(entity_context) if self.session_summary: system_parts.append(f## 本次会话摘要\n{self.session_summary}) system_message \n\n.join(filter(None, system_parts)) messages [] if system_message: messages.append({role: system, content: system_message}) messages.extend(list(self.recent_history)) messages.append({role: user, content: user_input}) return messages def chat(self, user_input: str) - str: messages self.build_messages(user_input) response self.client.chat.completions.create( modelgpt-4o, messagesmessages, ) assistant_reply response.choices[0].message.content # 更新各层记忆 self.recent_history.append({role: user, content: user_input}) self.recent_history.append({role: assistant, content: assistant_reply}) self.entity_memory.extract_and_update(user_input, assistant_reply) # 定期保存重要信息到长期记忆 self._maybe_save_to_long_term(user_input, assistant_reply) return assistant_reply def _maybe_save_to_long_term(self, user_msg: str, assistant_msg: str): 判断是否需要保存到长期记忆 # 简单启发式用户消息超过50字或包含关键词 keywords [记住, 以后, 我的偏好, 一直, 总是] should_save (len(user_msg) 50 or any(kw in user_msg for kw in keywords)) if should_save: memory_content f用户说{user_msg}\n助手回复{assistant_msg[:200]}... self.vector_memory.save_memory(memory_content)## 工程实践小结上下文管理没有银弹根据场景选择合适策略| 场景 | 推荐方案 ||------|----------|| 简单对话机器人 | 滑动窗口20轮|| 长期助手跨会话| 向量记忆 摘要压缩 || 需要记住用户信息 | 实体记忆 || 复杂多任务Agent | 综合记忆系统 || 成本敏感场景 | 重要性过滤 小模型压缩 |核心原则不是把所有信息塞进上下文而是把最相关的信息放在最合适的位置。好的上下文管理是让模型用有限的注意力处理最重要的信息。