1. 项目概述为什么你的AI Agent总是“健忘”最近在折腾各种AI Agent项目从自动化客服到代码助手发现一个几乎所有人都会踩的坑Agent的“失忆症”。你精心设计的智能体在对话进行到第三轮、第四轮时突然就忘了之前说过什么或者把用户几分钟前才确认的关键信息给弄混了。这感觉就像和一个短期记忆只有七秒的人合作非常影响体验和效率。这个问题本质上就是上下文管理没做好。所谓上下文你可以把它理解成AI Agent的“工作记忆”和“长期档案”。它不仅仅是当前这一轮对话的输入更是包含了整个会话历史、用户偏好、任务目标、已执行操作结果等一系列信息的集合。一个管理不善的上下文就像一间堆满杂乱文件的办公室AI这个“员工”在里面根本找不到需要的东西自然表现不佳。更糟的是现在主流的基于Transformer架构的大模型对输入长度即上下文窗口有硬性限制。你不可能把无限长的对话历史都塞进去这就引出了上下文管理的核心挑战如何在有限的“内存”里存放最有价值的信息并让AI能快速准确地检索到它们这个项目就是一次针对AI Agent上下文管理的深度实战。我们不谈空洞的理论直接上手解决几个最实际的问题当对话越来越长如何避免关键信息被“挤出”上下文窗口如何让Agent记住跨会话的重要用户信息在多轮复杂任务中如何确保后续步骤能准确引用之前的中间结果通过一套组合策略和具体的代码实现我们可以显著提升Agent的“记忆力”让它真正成为一个靠谱的、可持续协作的智能伙伴。2. 上下文管理的核心挑战与设计思路在动手写代码之前我们必须先搞清楚敌人是谁。AI Agent的上下文管理难点主要集中在三个方面长度限制、信息密度和检索效率。2.1 长度限制模型的“记忆天花板”几乎所有大语言模型都有一个上下文窗口上限比如早期的4K、8K到现在常见的32K、128K甚至更长。但无论上限多高它总是有限的。对于一个持续运行的Agent比如一个全天在线的客服机器人几天甚至几小时的对话记录很容易就超过这个限制。粗暴地截断最新的N个Token是最简单的方法但风险极高——你可能会把任务最开始的指令或者用户的核心诉求给丢掉导致Agent彻底跑偏。2.2 信息密度别让废话淹没了重点即使你的对话长度还在窗口内上下文的质量也至关重要。一段冗长的、包含大量寒暄、重复和无关细节的对话历史会严重稀释关键信息的浓度。模型需要花费更多的“注意力”去处理这些噪音从而影响其对核心指令和事实的判断。这就好比让你从一篇啰嗦的报告里找关键结论效率肯定低下。2.3 检索效率知道从哪里找到“记忆”当我们将部分历史存储到外部数据库如向量库进行长期记忆时如何快速、准确地找回当前对话所需的相关片段就成了另一个挑战。传统的基于语义相似度的向量检索在对话场景下可能不够精准。比如用户问“刚才我们讨论的那个方案的价格是多少”向量检索可能会返回一堆讨论“方案”或“价格”的片段但未必能精准定位到“刚才讨论的那个特定方案”的价格陈述。基于这些挑战我们的设计思路需要是多层级的、智能化的而不是单一的解决方案。核心思路分层记忆系统我们可以借鉴计算机存储体系结构的思想为Agent构建一个分层记忆系统工作记忆Working Memory相当于CPU的缓存。存放当前轮次对话的绝对核心信息以及为完成当前子任务必须立即使用的信息。这部分信息会直接、完整地送入模型的上下文窗口。会话记忆Session Memory相当于内存。存储当前这次完整会话从用户连接到断开的所有历史。当工作记忆不够用时从这里进行智能检索和摘要提取补充进上下文。长期记忆Long-term Memory相当于硬盘。存储跨会话的用户画像、偏好、历史决策、知识库等。这部分信息通常存储在向量数据库或关系型数据库中需要时才被检索激活。我们的实战将围绕如何实现和维护这三层记忆展开重点在于工作记忆和会话记忆之间的动态调度以及如何高效利用长期记忆。3. 实战架构构建一个健壮的上下文管理器理论说再多不如一行代码。接下来我们构建一个名为ContextManager的类它将作为Agent的“记忆中枢”。这里以Python为例使用LangChain的部分概念但会剥离框架细节聚焦于核心逻辑你可以轻松适配到其他框架。3.1 基础数据结构定义首先我们需要定义记忆的基本单元和存储结构。from typing import Dict, List, Any, Optional from datetime import datetime import hashlib class MemoryFragment: 记忆片段上下文管理的基本单元 def __init__(self, content: str, role: str “user”, # ‘user’, ‘assistant’, ‘system’ timestamp: datetime None, metadata: Dict[str, Any] None): self.content content # 文本内容 self.role role # 发言角色 self.timestamp timestamp or datetime.now() self.metadata metadata or {} # 可存放重要性评分、实体标签等 # 生成唯一ID便于追踪 self.id hashlib.md5(f“{self.timestamp}_{self.content}”.encode()).hexdigest()[:8] def to_dict(self) - Dict[str, Any]: return { “id”: self.id, “content”: self.content, “role”: self.role, “timestamp”: self.timestamp.isoformat(), “metadata”: self.metadata }这个MemoryFragment类封装了对话中的一个回合。metadata字段非常关键后续我们可以用它来标记这个片段的重要性比如用户明确说“记住这个”或者用NER工具提取出的人名、地名、产品名等实体便于结构化检索。接下来是上下文管理器的主体框架class ContextManager: def __init__(self, llm, embedding_model, vector_store, max_working_tokens: int 2000): 初始化上下文管理器。 :param llm: 大语言模型实例用于生成摘要和判断。 :param embedding_model: 嵌入模型用于向量化文本。 :param vector_store: 向量数据库客户端用于存储和检索长期记忆。 :param max_working_tokens: 工作记忆的最大Token预算。 self.llm llm self.embedding_model embedding_model self.vector_store vector_store self.max_working_tokens max_working_tokens # 记忆存储 self.working_memory: List[MemoryFragment] [] # 工作记忆当前焦点 self.session_memory: List[MemoryFragment] [] # 会话记忆完整历史 self.long_term_memory_ids set() # 标记已存入长期记忆的片段ID # 关键实体/事实缓存简易版 self.entity_cache: Dict[str, List[str]] {} # 实体名 - [相关陈述1, 相关陈述2] def add_interaction(self, user_input: str, agent_response: str): 添加一轮完整的用户-Agent交互到记忆中 user_fragment MemoryFragment(user_input, role“user”) agent_fragment MemoryFragment(agent_response, role“assistant”) self.session_memory.extend([user_fragment, agent_fragment]) # 可选立即进行重要性分析和实体提取 self._analyze_fragment(user_fragment) self._analyze_fragment(agent_fragment) # 添加后触发工作记忆的整理和刷新 self._refresh_working_memory()这个骨架包含了核心的存储结构。add_interaction方法在每轮对话后调用将对话存入会话记忆并触发工作记忆的刷新。3.2 工作记忆的动态刷新策略_refresh_working_memory是核心算法之一它决定了当下哪些信息能进入模型的“视线”。def _refresh_working_memory(self): 动态刷新工作记忆确保不超过Token限制且包含最关键信息 # 策略1永远保留系统指令和会话最初的目标 core_fragments [] for frag in self.session_memory: if frag.role “system” or “objective” in frag.metadata.get(“tags”, []): core_fragments.append(frag) # 策略2优先加入最近发生的几轮对话最近性 recent_fragments self.session_memory[-6:] # 例如最近3轮交互 # 策略3从会话记忆中检索与当前话题最相关的片段相关性 # 这里用当前最后一条用户输入作为查询 if self.session_memory: latest_query self.session_memory[-1].content relevant_fragments self._retrieve_relevant_from_session(latest_query, top_k3) # 合并所有候选片段并去重按ID candidate_dict {} for frag in core_fragments recent_fragments relevant_fragments: candidate_dict[frag.id] frag candidates list(candidate_dict.values()) # 策略4按重要性评分排序如果metadata里有 candidates.sort(keylambda x: x.metadata.get(“importance_score”, 0), reverseTrue) # 开始组装工作记忆并计算Token数这里用简易字符数/4模拟 new_working_memory [] current_tokens 0 for frag in candidates: frag_tokens len(frag.content) // 4 if current_tokens frag_tokens self.max_working_tokens: # 如果放不下尝试生成摘要 summarized self._summarize_if_necessary(frag, frag_tokens, current_tokens) if summarized: # 摘要的Token数会少很多 sum_tokens len(summarized) // 4 if current_tokens sum_tokens self.max_working_tokens: summary_frag MemoryFragment(summarized, rolefrag.role, metadata{“summary_of”: frag.id}) new_working_memory.append(summary_frag) current_tokens sum_tokens # 即使不能摘要也跳过保证不超限 continue new_working_memory.append(frag) current_tokens frag_tokens self.working_memory new_working_memory这个刷新策略融合了多个维度核心保留系统指令和初始目标永不丢弃。最近性最近的对话通常最重要。相关性检索与当前查询语义相关的历史片段。重要性通过评分可基于规则或模型预测区分信息价值。摘要降级当重要但冗长的片段放不下时用LLM生成摘要放入而不是直接丢弃。_retrieve_relevant_from_session和_summarize_if_necessary是两个需要实现的关键辅助函数它们分别处理语义检索和文本摘要。3.3 从会话记忆中实现智能检索简单的向量检索在对话场景下可能“误伤”很高。我们需要结合对话的结构特征。def _retrieve_relevant_from_session(self, query: str, top_k: int 3) - List[MemoryFragment]: 从会话记忆中检索相关片段结合语义和对话结构 relevant [] # 方法A基于窗口的邻近检索解决指代问题 # 如果查询中包含“上面说的”、“刚才提到的”优先查找邻近的上文。 if any(word in query for word in [“上面”, “刚才”, “之前”, “前述”]): # 取当前轮次之前的最近N个片段 start_idx max(0, len(self.session_memory) - 10) # 向前看10个片段 relevant.extend(self.session_memory[start_idx:-1]) # 不包括自己 # 方法B基于实体的精确匹配 # 从查询中提取实体简易关键词匹配生产环境可用NER模型 query_keywords set(query.replace(“?”, “”).replace(“。”, “”).split()) for frag in self.session_memory: frag_content frag.content frag_words set(frag_content.replace(“?”, “”).replace(“。”, “”).split()) # 如果存在实体交集且不是最近的避免重复则加入候选 if query_keywords frag_words and frag not in relevant: relevant.append(frag) # 方法C语义向量检索作为兜底 if len(relevant) top_k: # 使用embedding_model计算查询向量并从session_memory中找最相似的 # 这里省略具体的向量化与相似度计算代码假设有一个函数 semantic_search semantic_results self._semantic_search(query, self.session_memory, top_ktop_k) for frag in semantic_results: if frag not in relevant: relevant.append(frag) # 去重并返回Top K seen_ids set() final_results [] for frag in relevant: if frag.id not in seen_ids: seen_ids.add(frag.id) final_results.append(frag) if len(final_results) top_k: break return final_results这个检索函数体现了混合策略先尝试用规则邻近性、实体匹配解决对话中常见的指代和精确查询再用语义检索作为泛化能力的补充。这比单纯用向量检索的准确率要高得多。3.4 摘要生成与信息压缩当我们需要保留某个片段的信息但又受限于Token时摘要是最佳选择。def _summarize_if_necessary(self, fragment: MemoryFragment, original_tokens: int, current_used_tokens: int) - Optional[str]: 在必要时为记忆片段生成摘要 available_tokens self.max_working_tokens - current_used_tokens # 如果可用Token连一个简短的摘要都放不下或者原片段本身就很短则不摘要 if available_tokens 50 or original_tokens 100: return None # 根据剩余空间决定摘要的详细程度 target_length available_tokens * 4 # Token转字符数的粗略估计 prompt f“”” 请将以下文本压缩成不超过{target_length}个字符的摘要保留核心事实、决策和用户要求。 文本内容 {fragment.content} 摘要 “”” try: # 调用LLM生成摘要 response self.llm.invoke(prompt) summary response.content.strip() # 确保摘要确实比原文短 if len(summary) len(fragment.content) * 0.7: # 摘要至少压缩30% return summary except Exception as e: print(f“生成摘要时出错: {e}”) return None注意频繁调用LLM生成摘要是有成本的。在实际应用中可以对片段进行重要性预判只对高重要性片段启用摘要或者缓存摘要结果避免重复计算。4. 长期记忆的集成与向量检索优化会话记忆终究会随着会话结束而清空。对于需要跨会话记忆的信息如用户说“我住在北京”、“我不喜欢吃香菜”我们必须将其存入长期记忆。4.1 信息筛选与存入长期记忆不是所有信息都值得长期记忆。我们需要一个筛选策略。def _analyze_fragment(self, fragment: MemoryFragment): 分析记忆片段决定是否存入长期记忆并提取实体 content fragment.content # 规则1用户明确要求记住的信息 if “记住” in content or “记一下” in content: fragment.metadata[“importance_score”] 10 self._save_to_long_term(fragment) # 规则2使用小型分类模型或启发式规则判断是否为事实性陈述 # 例如包含“是”、“叫”、“在”、“喜欢”等陈述性动词且主语是“我”或用户提及的实体。 if self._is_factual_statement(content): fragment.metadata[“tags”] fragment.metadata.get(“tags”, []) [“fact”] # 提取实体和事实存入结构化缓存 extracted self._extract_entity_and_fact(content) if extracted: entity, fact extracted self.entity_cache.setdefault(entity, []).append(fact) # 如果同一个实体的事实积累到一定数量打包存入长期记忆 if len(self.entity_cache[entity]) 3: self._consolidate_entity_facts(entity) # 规则3对话中确定的任务结果或关键决策 if “确定” in content or “就这样” in content or “final” in content.lower(): fragment.metadata[“importance_score”] 8 def _save_to_long_term(self, fragment: MemoryFragment): 将片段存入向量数据库作为长期记忆 if fragment.id in self.long_term_memory_ids: return # 生成向量嵌入 embedding self.embedding_model.embed_query(fragment.content) # 存储到向量库元数据中包含ID、角色、时间戳等 self.vector_store.add_texts( texts[fragment.content], embeddings[embedding], metadatas[fragment.to_dict()] ) self.long_term_memory_ids.add(fragment.id)_is_factual_statement和_extract_entity_and_fact是两个需要根据实际情况实现的函数可以用规则也可以用更精细的NLP模型。核心思想是自动识别并结构化那些值得长期保存的用户信息。4.2 长期记忆的检索超越简单相似度从向量库检索长期记忆时直接使用当前用户查询进行搜索可能找不到最相关的历史事实。我们需要对查询进行“重写”或“扩展”。def retrieve_from_long_term(self, query: str) - List[str]: 从长期记忆中检索相关信息使用查询重写提升效果 # 步骤1查询重写。将当前对话式查询改写成更利于检索的陈述句。 rewrite_prompt f“”” 原查询“{query}” 这是一个在对话中提出的问题。请将它重写为一个或多个独立的、包含核心事实的陈述句以便于从知识库中查找相关背景信息。 例如如果原查询是“刚才我提到的那个方案怎么样”可以重写为“用户之前讨论过一个方案”。 重写后的陈述 1. “”” rewritten_statements [] try: response self.llm.invoke(rewrite_prompt) # 解析LLM返回的多个陈述句 lines response.content.strip().split(‘\n’) for line in lines: if line.strip() and not line.strip().startswith((‘例如’, ‘重写’)): rewritten_statements.append(line.strip().lstrip(‘1. ‘).lstrip(‘2. ‘).lstrip(‘- ‘)) except: rewritten_statements [query] # 失败则回退到原查询 # 步骤2对每个重写后的陈述进行向量检索 all_results [] for stmt in rewritten_statements: embedding self.embedding_model.embed_query(stmt) docs self.vector_store.similarity_search_by_vector(embedding, k2) all_results.extend([doc.page_content for doc in docs]) # 步骤3去重并返回 seen set() unique_results [] for res in all_results: if res not in seen: seen.add(res) unique_results.append(res) return unique_results[:5] # 返回最多5条最相关的查询重写是一个被低估但极其有效的技巧。它能将“你之前说的那个东西多少钱”这类指代模糊的查询转化为“用户之前询问了某产品的价格”这样的可检索语句极大提升了长期记忆检索的命中率。5. 组装与调用让Agent拥有“记忆”能力现在我们将上下文管理器集成到Agent的主循环中。class MyAgent: def __init__(self, context_manager: ContextManager, ...): self.context_manager context_manager # ... 其他初始化 def get_context_for_prompt(self) - str: 组装最终的提示词上下文 # 1. 系统指令永远在最前 system_prompt “你是专业的助手请根据对话历史谨慎回答。\n” # 2. 从工作记忆中组装核心上下文 working_context “\n”.join([f“{frag.role}: {frag.content}” for frag in self.context_manager.working_memory]) # 3. 从长期记忆中检索相关背景信息 if self.context_manager.session_memory: latest_query self.context_manager.session_memory[-1].content long_term_info self.context_manager.retrieve_from_long_term(latest_query) if long_term_info: long_term_context “\n以下是你已知的跨会话背景信息\n” “\n”.join([f“- {info}” for info in long_term_info]) else: long_term_context “” else: long_term_context “” # 4. 当前查询 current_query self.context_manager.session_memory[-1].content if self.context_manager.session_memory else “” final_prompt f“{system_prompt}\n### 对话历史 ###\n{working_context}\n{long_term_context}\n### 当前问题 ###\n用户: {current_query}\n助手:” return final_prompt def chat_round(self, user_input: str) - str: 处理一轮对话 # 1. 先将用户输入添加到上下文管理器此时还未生成Agent回复 self.context_manager.add_interaction(user_input, “”) # 先占位回复为空 # 注意实际add_interaction应在获得回复后这里为演示流程简化。 # 2. 获取整合了所有记忆的提示词 prompt self.get_context_for_prompt() # 3. 调用LLM获得回复 agent_response self.llm.invoke(prompt) # 4. 将真正的Agent回复添加到记忆 self.context_manager.add_interaction(user_input, agent_response) return agent_response这个get_context_for_prompt方法清晰地展示了三层记忆的融合系统指令打底工作记忆提供最近的对话焦点长期记忆提供跨会话的背景知识最后加上当前问题。这样组装出来的提示词信息密度高且直接相关能极大提升模型回复的准确性和一致性。6. 高级技巧与避坑指南在实际部署中你会遇到更多细节问题。以下是一些来自实战的经验和避坑点。6.1 重要性评分的自动化计算之前我们手动给片段打importance_score这不可持续。可以训练一个轻量级模型或设计一套规则来自动评分规则引擎包含数字、日期、具体名称的陈述句 (2分)用户使用“重要”、“记住”、“关键”等词汇 (3分)对话中反复提及的实体所在片段 (1分/次)Agent成功完成任务后的总结性片段 (2分)纯问候语、感叹词等 (-1分)微调小型分类模型收集一批人工标注了重要性的对话数据微调一个像BERT-Tiny这样的模型在线预测。虽然需要前期投入但长期来看更精准。6.2 处理超长文档和复杂任务当Agent需要处理长文档如PDF、长文章或执行步骤繁多的任务时需要特殊的上下文管理策略文档分块与索引将长文档切成有重叠的块分别嵌入并存入向量库。当用户提问时只检索最相关的几个块放入上下文。任务状态跟踪对于多步任务如“订机票-选座位-订酒店”在上下文中显式维护一个任务状态对象。task_state { “current_step”: “selecting_seat”, “completed_steps”: [“flight_booked”], “extracted_info”: {“flight_number”: “CA123”, “date”: “2023-10-01”} }将这个状态对象的文本摘要作为系统指令的一部分或作为一个特殊的记忆片段确保Agent始终“记得”任务进展到哪一步了。6.3 避免“记忆幻觉”与信息冲突LLM有时会混淆记忆甚至“捏造”上下文里不存在的内容。 mitigation策略提供精确引用在组装上下文时为每个来自记忆的片段加上来源标识如[来自历史对话2023-09-15]。并指示模型“如果答案基于历史信息请注明依据”。关键信息确认对于从长期记忆中检索出的、非常关键的事实如用户地址、电话号码在Agent使用前可以设计一个确认环节“根据我的记录您住在北京海淀区对吗”。定期记忆复盘在对话间歇或结束时可以触发一个总结环节让LLM输出本轮对话达成的关键结论和待办事项并以此更新长期记忆覆盖可能存在的旧错误信息。6.4 性能与成本优化摘要缓存对相同的记忆片段只摘要一次将结果缓存起来。下次需要时直接使用缓存摘要。向量检索的预过滤在向量检索前先用时间、实体标签等元数据做一层过滤缩小搜索范围提升速度并降低成本。分层存储将会话记忆存储在内存或Redis中长期记忆才用向量数据库。对于不活跃的长期记忆可以归档到更便宜的冷存储中。7. 效果评估与迭代如何知道你的上下文管理策略是有效的不能只靠感觉需要设计评估指标。人工评估金标准构造一批多轮对话测试集让评估人员判断Agent在后续轮次中事实一致性是否错误地改变了之前确认过的事实如用户说喜欢咖啡后面却回答用户喜欢茶指代解析准确性是否能正确理解“上面说的那个”、“它”、“他”所指代的内容任务连贯性在多步任务中后续步骤是否正确使用了前序步骤的结果自动评估指标关键实体留存率在对话中标记关键实体如产品名、参数。计算在后续N轮对话的上下文中这些实体仍然存在的比例。检索相关性对于从长期记忆中检索出的信息计算其与当前查询的真实相关性可以用交叉编码器模型打分。Token使用效率计算送入模型的上下文窗口中与最终回复真正相关的Token占比。避免大量无关历史占用宝贵窗口。根据评估结果持续调整你的刷新策略、检索策略和重要性评分规则。上下文管理没有一劳永逸的银弹它是一个需要结合具体应用场景不断调优的子系统。实现一个不“失忆”的AI Agent核心在于认识到上下文管理是一个独立的、重要的工程问题。它需要你精心设计数据的存储、淘汰、检索和组装策略。通过构建分层记忆系统结合规则与模型驱动的智能调度并紧密集成到Agent的推理循环中你能显著提升智能体的协作感和可靠性。这其中的每一个环节从重要性判断到查询重写都充满了值得深入优化的细节。