1. 从“臃肿”到“精干”一次Agent上下文管理的实战复盘最近在折腾一个基于LangChain的智能客服Agent项目遇到了一个典型问题随着对话轮次增加上下文Context像滚雪球一样越来越大。每次调用大模型都得把过去十几轮甚至几十轮的对话历史一股脑塞进去Token消耗直线上升成本看着心疼。更糟的是我发现Agent的“记忆力”反而变差了——它开始混淆不同用户的诉求甚至把几天前的旧指令当作当前任务来执行出现了典型的“记忆乱窜”现象。这让我意识到单纯地堆叠历史对话并不是构建长期记忆的好方法反而是一种资源浪费和性能负担。于是我启动了一个代号为“Agent Harness”的优化项目目标很明确给Agent的上下文“瘦身”。不是简单地截断或丢弃而是通过更智能的结构化与压缩让有限的Token承载更精准、更持久的记忆。经过几轮迭代最终实现了Token用量下降约40%同时任务完成准确率和上下文关联度显著提升的效果。这个过程涉及对上下文本质的重新思考、多种工程化手段的尝试以及记忆架构的设计。如果你也在为Agent的上下文膨胀和记忆混乱而头疼希望这篇实战复盘能给你带来一些直接的思路和可操作的代码。2. 诊断为什么你的Agent上下文会“虚胖”且“健忘”在动手优化之前我们必须先搞清楚问题出在哪里。Agent的上下文通常指的是为了完成当前任务需要提供给大模型LLM的所有相关信息。在对话系统中这几乎就等于完整的对话历史。这种设计简单粗暴但隐藏着几个致命缺陷。2.1 Token消耗的“元凶”冗余与低信息密度首先最直观的问题是冗余。想象一下人类对话我们不会在每次开口前都把之前的对话全文复述一遍。但在大多数Agent实现中正是这么做的。用户问“今天的天气怎么样”Agent回答“北京晴15-25度”。五分钟后用户又问“那明天呢”标准的做法是把第一轮问答完整地再次输入。这里面“今天的天气怎么样”和“北京晴15-25度”对于回答明天天气这个问题信息价值极低但它们却占据了宝贵的Token。这种逐轮追加的模式使得上下文长度呈线性甚至更快的速度增长。其次是信息的低密度化。自然语言本身就有大量的冗余比如语气词、重复性描述、客套话等。在长上下文中这些低信息密度的内容会不断累积稀释了真正关键信息如用户意图、实体、决策点的浓度。大模型需要花费额外的“注意力”去处理这些噪音影响了核心任务的表现。2.2 记忆“乱窜”的根源缺乏结构化与隔离记忆混乱是另一个棘手问题。其根源在于平铺直叙的对话历史是一种“非结构化”的记忆。当Agent需要回答“我上次提到的那个项目进度如何”时它需要在冗长的文本中寻找“项目”这个关键词并关联到“上次”这个模糊的时间点。这个过程极易出错可能关联到错误用户的项目或者错误时间点的项目。更深层的原因是缺乏记忆隔离。一个服务于多用户的Agent其上下文在逻辑上应该是多个独立“会话线程”的集合。但如果简单地将所有对话拼接不同用户的信息就会在Token序列中相互“污染”。模型可能会把用户A对产品的抱怨错误地应用到用户B的咨询中。这就是为什么你的WorkBuddy一个假设的AI助手的记忆会“乱窜”——因为没有在底层为不同用户、不同会话建立清晰的隔离边界。2.3 重新定义“上下文”从聊天记录到知识图谱因此优化的第一步是转变思维上下文不等于聊天记录。上下文应该是为完成当前任务所必需的、经过提炼的知识状态。这引出了两个关键概念工作记忆Working Memory处理当前任务直接需要的、高度活跃的信息。它应该是精简、聚焦的。长期记忆Long-term Memory存储历史经验、用户画像、领域知识等。它需要被结构化存储并能按需精准检索而非全量加载。我们的目标就是建立一套机制将原始的、冗长的对话流动态地转化为“工作记忆”“按需检索的长期记忆”的组合。这正是“上下文工程”的核心。3. 瘦身方案一无损压缩与智能摘要直接对原始文本进行压缩是减少Token最立竿见影的方法。这里有几个层次的操作。3.1 对话轮次合并与关键信息提取不要逐条存储用户和AI的发言。我们可以按“话题”或“意图”对对话进行分组和摘要。# 示例一个简单的对话摘要函数 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI def summarize_conversation_turn(user_input, ai_response, llm): 将一轮对话压缩为一条结构化记录。 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个高效的对话记录员。请将以下一轮对话提炼成一条简洁的记录包含用户核心意图、AI回复的要点、以及涉及的关键实体如人名、地点、任务名。以JSON格式输出。”), (“human”, “用户: {user_input}\nAI: {ai_response}”) ]) chain prompt | llm result chain.invoke({“user_input”: user_input, “ai_response”: ai_response}) # 假设LLM返回了合法的JSON字符串 import json summary_record json.loads(result.content) return summary_record # 使用示例 llm ChatOpenAI(model“gpt-3.5-turbo”) record summarize_conversation_turn(“帮我订一张明天北京飞上海的机票”, “好的已为您查询。明天上午9点国航CA1501航班有余票经济舱价格1200元。请问需要预订吗”, llm) print(record) # 输出可能类似{“intent”: “预订机票”, “user_query”: “明天北京飞上海”, “ai_action”: “查询航班CA1501并报价”, “entities”: {“date”: “明天”, “departure”: “北京”, “arrival”: “上海”, “flight”: “CA1501”, “price”: “1200”}}通过这种方式原来需要几十个Token的一轮对话被压缩成了一个包含关键信息的JSON对象可能只需要十几个Token。历史对话不再是一堆文本而是一个结构化的记录列表。注意摘要的粒度需要权衡。过于粗略会丢失细节如具体航班号过于详细则压缩效果不佳。通常针对任务类型设计不同的摘要模板。3.2 利用大模型自身的上下文窗口管理工具一些先进的大模型API或框架提供了原生的上下文管理功能。例如Anthropic的Claude Code或类似工具可能支持“上下文分层”或“标记重要片段”。虽然我们不能直接使用未经确认的工具但思路可以借鉴在发送请求前对长上下文进行分析识别出与当前问题最相关的段落并为其添加高权重标记同时将不相关的段落移至低优先级或排除在外。这需要利用LLM本身做一个轻量的预处理。# 概念性代码相关性筛选 def filter_relevant_context(full_history, current_query, llm): 从完整历史中筛选出与当前问题最相关的部分。 prompt f“”” 给定以下对话历史和一个当前问题请从历史中选出最直接相关的1-3轮对话。仅输出所选对话的索引号从0开始用逗号分隔。 对话历史 {full_history} 当前问题{current_query} “”” response llm.invoke(prompt) relevant_indices [int(i.strip()) for i in response.content.split(‘,’)] relevant_context “\n”.join([full_history[i] for i in relevant_indices]) return relevant_context这个方法可以将万字符的上下文动态缩减到千字符以内显著节省Token。4. 瘦身方案二结构化记忆与向量检索如果说压缩是“节流”那么建立结构化的长期记忆系统就是“开源”。它允许我们存储海量信息但只在需要时精准提取彻底告别全量加载。4.1 构建记忆图谱从文本到向量我们不再保存原始对话文本而是将其核心信息转换为向量Embeddings存入向量数据库如Chroma, Pinecone, Weaviate。这就是构建Agent的“长期记忆库”。记忆单元设计每条记忆不应是一整段对话而是一个个细粒度的“事实”或“事件”。例如“用户偏好靠窗座位”、“项目A的截止日期是2024-06-01”、“用户曾反馈过登录页面加载慢”。每个单元包含内容事实的文本描述。元数据来源会话ID、用户ID、时间戳、类型偏好、事实、任务等、重要性分数。向量内容的嵌入表示。存入向量库当一轮对话被摘要和结构化后将其拆解成多个记忆单元生成向量并存入数据库。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directory“./mem_db”) def create_memory_units(summary_record, session_id, user_id): units [] # 示例从摘要记录中创建多个记忆单元 if “entities” in summary_record: for key, value in summary_record[“entities”].items(): content f“用户提到了{key}: {value}” metadata {“session_id”: session_id, “user_id”: user_id, “type”: “entity”, “source”: “conversation”} # 这里可以更精细地处理比如“预订机票”作为一个意图单元 units.append({“content”: content, “metadata”: metadata}) # 将意图也作为一个单元 intent_content f“用户意图是: {summary_record.get(‘intent’, ‘N/A’)}” units.append({“content”: intent_content, “metadata”: {“session_id”: session_id, “user_id”: user_id, “type”: “intent”}}) return units def store_memories(units, vectorstore): texts [u[“content”] for u in units] metadatas [u[“metadata”] for u in units] vectorstore.add_texts(textstexts, metadatasmetadatas)4.2 精准检索让记忆“随用随取”当新的用户查询到来时我们不再拼接全部历史而是将当前查询转换为向量。在向量数据库中严格限定检索范围例如通过元数据过滤session_id和user_id搜索最相关的K条记忆。将这些检索到的记忆作为“补充上下文”或“长期记忆片段”与当前的“工作记忆”即最近一两轮对话一起构成最终的提示词。def retrieve_relevant_memories(query, vectorstore, session_id, user_id, k5): # 关键使用元数据过滤器实现记忆隔离 filter_dict {“session_id”: session_id, “user_id”: user_id} relevant_docs vectorstore.similarity_search(query, kk, filterfilter_dict) # 将检索到的记忆文本组合起来 memory_context “\n”.join([doc.page_content for doc in relevant_docs]) return memory_context这种方法完美解决了“记忆乱窜”。因为检索时通过session_id和user_id进行了强制隔离用户A的记忆绝不会出现在用户B的上下文中。同时由于只检索相关的几条记忆上下文的长度是可控的通常只有几百个Token而不是几千个。4.3 关于向量库中词义统一的思考在构建记忆单元时一个常见问题是像“上下文理解”和“语境推测”这类意思相近的词需要统一吗我的经验是不需要强制统一但可以增强关联。大模型的嵌入模型如OpenAI的text-embedding-3本身对同义词和近义词就有很好的语义捕捉能力。“上下文理解”和“语境推测”的向量在空间上应该是接近的。当你用其中一个查询时很可能也会检索到另一个。如果你追求极致的精确度可以做一些后处理同义词扩展在检索时不仅用原始查询也用其同义词进行查询合并结果。知识图谱关联在元数据中记录这两个概念是相关的但这会引入额外的复杂度。对于大多数应用相信嵌入模型的能力就够了。过度工程化反而可能引入新的问题。5. 瘦身方案三记忆的分层、衰减与更新人的记忆是会遗忘的Agent的记忆也应该有生命周期。一个只增不减的记忆库最终会变得臃肿且充满过时信息。5.1 设计三层记忆架构我借鉴了认知科学中的一些概念为Agent设计了一个简单的三层记忆架构感官记忆/工作记忆保存当前正在处理的对话轮次最近1-3轮。信息鲜活但容量极小随着对话推进快速被覆盖或转移到下一层。短期记忆存放近期活跃的、与当前会话高度相关的记忆单元通过向量检索得到。这些是当前任务的“背景板”会话结束后其活性会逐渐降低。长期记忆向量数据库中存储的所有记忆单元。容量大但信息处于“休眠”状态需要时通过检索激活。在每次交互中Agent的上下文主要由“工作记忆”和“短期记忆”即检索结果构成长期记忆库是背后的支撑。5.2 实现记忆衰减与重要性加权不是所有记忆都同等重要。我们可以为每个记忆单元引入“重要性”分数和“最后访问时间”。重要性可以在创建时由LLM初步判断例如用户明确说“这很重要”也可以根据后续被检索到的频率动态更新。高频被检索的记忆重要性提高。最后访问时间每次该记忆被检索到就更新这个时间戳。我们可以定期运行一个“记忆整理”后台任务def cleanup_memories(vectorstore, decay_factor0.9, retention_threshold0.1): # 这是一个概念性函数实际实现可能需要遍历所有记忆 # 模拟衰减逻辑新分数 旧重要性分数 * decay_factor # 如果分数低于retention_threshold则考虑归档或删除 pass通过这种机制长期不用的、不重要的记忆会逐渐“淡出”从而控制记忆库的总规模和质量。对于关键记忆如用户的核心偏好可以通过手动标记或高频访问保持其高重要性避免被衰减。5.3 记忆的更新与冲突解决当新信息与旧记忆冲突时怎么办例如用户之前说“我喜欢蓝色”现在又说“我其实更喜欢绿色”。时间戳优先最简单的策略是相信最新的信息。在存储新记忆“喜欢绿色”的同时可以降低旧记忆“喜欢蓝色”的重要性分数或在元数据中标记其为“已过时”。事实融合对于更复杂的情况可以引入一个“记忆融合”步骤。当检测到冲突时触发一个LLM调用分析这两条记忆生成一条更准确的新记忆例如“用户对颜色的偏好可能从蓝色转变为了绿色或视上下文而定”然后替换或补充旧记忆。6. 工程集成打造Agent Harness框架将上述所有策略组合起来就需要一个轻量的框架来“驾驭”HarnessAgent的上下文和记忆。这个框架负责在每次调用LLM前智能地组装上下文。6.1 上下文组装流水线我设计了一个处理流水线它的工作流程如下输入当前用户查询 用户ID 会话ID。检索长期记忆用当前查询去向量库检索严格过滤user_id和session_id得到N条相关记忆。获取工作记忆从缓存或数据库中取出最近2-3轮原始对话或它们的摘要。生成摘要/压缩可选如果工作记忆较长可以对其进行一轮压缩摘要。组装最终提示将系统指令、检索到的记忆、压缩后的工作记忆、当前查询按顺序组合成最终的提示词。调用LLM发送组装好的提示词获得回复。更新记忆将本轮交互的核心信息创建为新的记忆单元存入向量库。同时更新工作记忆缓存。class AgentHarness: def __init__(self, llm, vectorstore, memory_manager): self.llm llm self.vectorstore vectorstore self.memory_manager memory_manager # 负责记忆的存储、检索、衰减 def process_query(self, user_query, user_id, session_id): # 1. 检索相关长期记忆 long_term_memories self.memory_manager.retrieve(user_query, user_id, session_id) # 2. 获取工作记忆最近对话 working_memory self.memory_manager.get_working_memory(session_id) # 3. (可选)压缩工作记忆 if len(working_memory) 3: # 假设超过3轮则压缩 working_memory self._compress_working_memory(working_memory) # 4. 组装提示词 prompt self._assemble_prompt(system_msg, long_term_memories, working_memory, user_query) # 5. 调用LLM response self.llm.invoke(prompt) # 6. 更新记忆 self.memory_manager.store_turn(user_id, session_id, user_query, response.content) return response.content6.2 关键配置参数与调优在实际使用中有几个参数对效果影响巨大需要仔细调优检索数量 (k)每次从向量库取多少条记忆太少可能信息不全太多则引入噪音。通常从3-5条开始测试。工作记忆长度保留最近几轮原始对话对于需要紧密跟随上下文的任务如代码调试可能需要3-5轮对于开放式聊天1-2轮可能就够了。记忆衰减因子决定旧记忆被遗忘的速度。需要根据业务场景调整。摘要的粒度摘要得太粗会丢失细节太细则压缩效果差。可能需要为不同类型的对话设计不同的摘要模板。我的经验是建立一个简单的评估体系在测试集上同时监控Token消耗、任务完成准确率、以及人工评估的“记忆相关性”。通过A/B测试找到最适合你场景的参数组合。7. 实战效果与避坑指南经过上述改造我的智能客服Agent焕然一新。最明显的效果是平均每次API调用的Token输入量下降了约40%因为上下文从完整的“史记”变成了精炼的“摘要”加“相关史料”。成本压力骤减。更惊喜的是效果提升。由于记忆检索是精准的、隔离的Agent混淆用户和任务的情况基本消失。当用户问起“我上次咨询的那个问题”时Agent能准确找到对应会话中的相关记忆而不是胡乱抓取一个相似话题。任务完成的准确率提升了约15%。当然踩坑是必不可少的摘要失真早期使用的摘要Prompt不够好导致关键信息如订单号、具体时间丢失。对策设计Prompt时明确列出必须保留的信息类型。最好能用一批样本进行测试检查摘要的保真度。向量检索不准当用户使用模糊指代如“那个”、“这东西”时直接检索效果很差。对策在检索前先用LLM对当前查询进行一次“查询重写”将其扩展成更完整、包含更多实体信息的描述再用这个描述去检索。记忆更新延迟存入向量库的记忆并非立即可用于下一次检索取决于向量化的批次和索引更新策略。对策对于当前会话中刚产生的、极其重要的信息可以同时放在工作记忆和一份“临时记忆缓存”中确保下一轮对话能立刻用到。系统复杂度增加引入了向量数据库、记忆管理模块系统架构变复杂了。对策做好模块化设计并将记忆相关的操作封装成独立服务便于维护和升级。这次“瘦身”实践让我深刻体会到对于AI Agent而言“更多”并不总是意味着“更好”。尤其是在上下文窗口有限、Token成本高昂的现实约束下如何高效、智能地组织和使用信息比盲目地堆砌信息更重要。通过结构化记忆、精准检索和动态管理我们完全可以让Agent在更小的“脑容量”下表现出更强大的“记忆力”和“专注力”。这不仅仅是成本的优化更是Agent智能体健壮性和实用性的关键一步。
1. 从“臃肿”到“精干”一次Agent上下文管理的实战复盘最近在折腾一个基于LangChain的智能客服Agent项目遇到了一个典型问题随着对话轮次增加上下文Context像滚雪球一样越来越大。每次调用大模型都得把过去十几轮甚至几十轮的对话历史一股脑塞进去Token消耗直线上升成本看着心疼。更糟的是我发现Agent的“记忆力”反而变差了——它开始混淆不同用户的诉求甚至把几天前的旧指令当作当前任务来执行出现了典型的“记忆乱窜”现象。这让我意识到单纯地堆叠历史对话并不是构建长期记忆的好方法反而是一种资源浪费和性能负担。于是我启动了一个代号为“Agent Harness”的优化项目目标很明确给Agent的上下文“瘦身”。不是简单地截断或丢弃而是通过更智能的结构化与压缩让有限的Token承载更精准、更持久的记忆。经过几轮迭代最终实现了Token用量下降约40%同时任务完成准确率和上下文关联度显著提升的效果。这个过程涉及对上下文本质的重新思考、多种工程化手段的尝试以及记忆架构的设计。如果你也在为Agent的上下文膨胀和记忆混乱而头疼希望这篇实战复盘能给你带来一些直接的思路和可操作的代码。2. 诊断为什么你的Agent上下文会“虚胖”且“健忘”在动手优化之前我们必须先搞清楚问题出在哪里。Agent的上下文通常指的是为了完成当前任务需要提供给大模型LLM的所有相关信息。在对话系统中这几乎就等于完整的对话历史。这种设计简单粗暴但隐藏着几个致命缺陷。2.1 Token消耗的“元凶”冗余与低信息密度首先最直观的问题是冗余。想象一下人类对话我们不会在每次开口前都把之前的对话全文复述一遍。但在大多数Agent实现中正是这么做的。用户问“今天的天气怎么样”Agent回答“北京晴15-25度”。五分钟后用户又问“那明天呢”标准的做法是把第一轮问答完整地再次输入。这里面“今天的天气怎么样”和“北京晴15-25度”对于回答明天天气这个问题信息价值极低但它们却占据了宝贵的Token。这种逐轮追加的模式使得上下文长度呈线性甚至更快的速度增长。其次是信息的低密度化。自然语言本身就有大量的冗余比如语气词、重复性描述、客套话等。在长上下文中这些低信息密度的内容会不断累积稀释了真正关键信息如用户意图、实体、决策点的浓度。大模型需要花费额外的“注意力”去处理这些噪音影响了核心任务的表现。2.2 记忆“乱窜”的根源缺乏结构化与隔离记忆混乱是另一个棘手问题。其根源在于平铺直叙的对话历史是一种“非结构化”的记忆。当Agent需要回答“我上次提到的那个项目进度如何”时它需要在冗长的文本中寻找“项目”这个关键词并关联到“上次”这个模糊的时间点。这个过程极易出错可能关联到错误用户的项目或者错误时间点的项目。更深层的原因是缺乏记忆隔离。一个服务于多用户的Agent其上下文在逻辑上应该是多个独立“会话线程”的集合。但如果简单地将所有对话拼接不同用户的信息就会在Token序列中相互“污染”。模型可能会把用户A对产品的抱怨错误地应用到用户B的咨询中。这就是为什么你的WorkBuddy一个假设的AI助手的记忆会“乱窜”——因为没有在底层为不同用户、不同会话建立清晰的隔离边界。2.3 重新定义“上下文”从聊天记录到知识图谱因此优化的第一步是转变思维上下文不等于聊天记录。上下文应该是为完成当前任务所必需的、经过提炼的知识状态。这引出了两个关键概念工作记忆Working Memory处理当前任务直接需要的、高度活跃的信息。它应该是精简、聚焦的。长期记忆Long-term Memory存储历史经验、用户画像、领域知识等。它需要被结构化存储并能按需精准检索而非全量加载。我们的目标就是建立一套机制将原始的、冗长的对话流动态地转化为“工作记忆”“按需检索的长期记忆”的组合。这正是“上下文工程”的核心。3. 瘦身方案一无损压缩与智能摘要直接对原始文本进行压缩是减少Token最立竿见影的方法。这里有几个层次的操作。3.1 对话轮次合并与关键信息提取不要逐条存储用户和AI的发言。我们可以按“话题”或“意图”对对话进行分组和摘要。# 示例一个简单的对话摘要函数 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI def summarize_conversation_turn(user_input, ai_response, llm): 将一轮对话压缩为一条结构化记录。 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个高效的对话记录员。请将以下一轮对话提炼成一条简洁的记录包含用户核心意图、AI回复的要点、以及涉及的关键实体如人名、地点、任务名。以JSON格式输出。”), (“human”, “用户: {user_input}\nAI: {ai_response}”) ]) chain prompt | llm result chain.invoke({“user_input”: user_input, “ai_response”: ai_response}) # 假设LLM返回了合法的JSON字符串 import json summary_record json.loads(result.content) return summary_record # 使用示例 llm ChatOpenAI(model“gpt-3.5-turbo”) record summarize_conversation_turn(“帮我订一张明天北京飞上海的机票”, “好的已为您查询。明天上午9点国航CA1501航班有余票经济舱价格1200元。请问需要预订吗”, llm) print(record) # 输出可能类似{“intent”: “预订机票”, “user_query”: “明天北京飞上海”, “ai_action”: “查询航班CA1501并报价”, “entities”: {“date”: “明天”, “departure”: “北京”, “arrival”: “上海”, “flight”: “CA1501”, “price”: “1200”}}通过这种方式原来需要几十个Token的一轮对话被压缩成了一个包含关键信息的JSON对象可能只需要十几个Token。历史对话不再是一堆文本而是一个结构化的记录列表。注意摘要的粒度需要权衡。过于粗略会丢失细节如具体航班号过于详细则压缩效果不佳。通常针对任务类型设计不同的摘要模板。3.2 利用大模型自身的上下文窗口管理工具一些先进的大模型API或框架提供了原生的上下文管理功能。例如Anthropic的Claude Code或类似工具可能支持“上下文分层”或“标记重要片段”。虽然我们不能直接使用未经确认的工具但思路可以借鉴在发送请求前对长上下文进行分析识别出与当前问题最相关的段落并为其添加高权重标记同时将不相关的段落移至低优先级或排除在外。这需要利用LLM本身做一个轻量的预处理。# 概念性代码相关性筛选 def filter_relevant_context(full_history, current_query, llm): 从完整历史中筛选出与当前问题最相关的部分。 prompt f“”” 给定以下对话历史和一个当前问题请从历史中选出最直接相关的1-3轮对话。仅输出所选对话的索引号从0开始用逗号分隔。 对话历史 {full_history} 当前问题{current_query} “”” response llm.invoke(prompt) relevant_indices [int(i.strip()) for i in response.content.split(‘,’)] relevant_context “\n”.join([full_history[i] for i in relevant_indices]) return relevant_context这个方法可以将万字符的上下文动态缩减到千字符以内显著节省Token。4. 瘦身方案二结构化记忆与向量检索如果说压缩是“节流”那么建立结构化的长期记忆系统就是“开源”。它允许我们存储海量信息但只在需要时精准提取彻底告别全量加载。4.1 构建记忆图谱从文本到向量我们不再保存原始对话文本而是将其核心信息转换为向量Embeddings存入向量数据库如Chroma, Pinecone, Weaviate。这就是构建Agent的“长期记忆库”。记忆单元设计每条记忆不应是一整段对话而是一个个细粒度的“事实”或“事件”。例如“用户偏好靠窗座位”、“项目A的截止日期是2024-06-01”、“用户曾反馈过登录页面加载慢”。每个单元包含内容事实的文本描述。元数据来源会话ID、用户ID、时间戳、类型偏好、事实、任务等、重要性分数。向量内容的嵌入表示。存入向量库当一轮对话被摘要和结构化后将其拆解成多个记忆单元生成向量并存入数据库。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directory“./mem_db”) def create_memory_units(summary_record, session_id, user_id): units [] # 示例从摘要记录中创建多个记忆单元 if “entities” in summary_record: for key, value in summary_record[“entities”].items(): content f“用户提到了{key}: {value}” metadata {“session_id”: session_id, “user_id”: user_id, “type”: “entity”, “source”: “conversation”} # 这里可以更精细地处理比如“预订机票”作为一个意图单元 units.append({“content”: content, “metadata”: metadata}) # 将意图也作为一个单元 intent_content f“用户意图是: {summary_record.get(‘intent’, ‘N/A’)}” units.append({“content”: intent_content, “metadata”: {“session_id”: session_id, “user_id”: user_id, “type”: “intent”}}) return units def store_memories(units, vectorstore): texts [u[“content”] for u in units] metadatas [u[“metadata”] for u in units] vectorstore.add_texts(textstexts, metadatasmetadatas)4.2 精准检索让记忆“随用随取”当新的用户查询到来时我们不再拼接全部历史而是将当前查询转换为向量。在向量数据库中严格限定检索范围例如通过元数据过滤session_id和user_id搜索最相关的K条记忆。将这些检索到的记忆作为“补充上下文”或“长期记忆片段”与当前的“工作记忆”即最近一两轮对话一起构成最终的提示词。def retrieve_relevant_memories(query, vectorstore, session_id, user_id, k5): # 关键使用元数据过滤器实现记忆隔离 filter_dict {“session_id”: session_id, “user_id”: user_id} relevant_docs vectorstore.similarity_search(query, kk, filterfilter_dict) # 将检索到的记忆文本组合起来 memory_context “\n”.join([doc.page_content for doc in relevant_docs]) return memory_context这种方法完美解决了“记忆乱窜”。因为检索时通过session_id和user_id进行了强制隔离用户A的记忆绝不会出现在用户B的上下文中。同时由于只检索相关的几条记忆上下文的长度是可控的通常只有几百个Token而不是几千个。4.3 关于向量库中词义统一的思考在构建记忆单元时一个常见问题是像“上下文理解”和“语境推测”这类意思相近的词需要统一吗我的经验是不需要强制统一但可以增强关联。大模型的嵌入模型如OpenAI的text-embedding-3本身对同义词和近义词就有很好的语义捕捉能力。“上下文理解”和“语境推测”的向量在空间上应该是接近的。当你用其中一个查询时很可能也会检索到另一个。如果你追求极致的精确度可以做一些后处理同义词扩展在检索时不仅用原始查询也用其同义词进行查询合并结果。知识图谱关联在元数据中记录这两个概念是相关的但这会引入额外的复杂度。对于大多数应用相信嵌入模型的能力就够了。过度工程化反而可能引入新的问题。5. 瘦身方案三记忆的分层、衰减与更新人的记忆是会遗忘的Agent的记忆也应该有生命周期。一个只增不减的记忆库最终会变得臃肿且充满过时信息。5.1 设计三层记忆架构我借鉴了认知科学中的一些概念为Agent设计了一个简单的三层记忆架构感官记忆/工作记忆保存当前正在处理的对话轮次最近1-3轮。信息鲜活但容量极小随着对话推进快速被覆盖或转移到下一层。短期记忆存放近期活跃的、与当前会话高度相关的记忆单元通过向量检索得到。这些是当前任务的“背景板”会话结束后其活性会逐渐降低。长期记忆向量数据库中存储的所有记忆单元。容量大但信息处于“休眠”状态需要时通过检索激活。在每次交互中Agent的上下文主要由“工作记忆”和“短期记忆”即检索结果构成长期记忆库是背后的支撑。5.2 实现记忆衰减与重要性加权不是所有记忆都同等重要。我们可以为每个记忆单元引入“重要性”分数和“最后访问时间”。重要性可以在创建时由LLM初步判断例如用户明确说“这很重要”也可以根据后续被检索到的频率动态更新。高频被检索的记忆重要性提高。最后访问时间每次该记忆被检索到就更新这个时间戳。我们可以定期运行一个“记忆整理”后台任务def cleanup_memories(vectorstore, decay_factor0.9, retention_threshold0.1): # 这是一个概念性函数实际实现可能需要遍历所有记忆 # 模拟衰减逻辑新分数 旧重要性分数 * decay_factor # 如果分数低于retention_threshold则考虑归档或删除 pass通过这种机制长期不用的、不重要的记忆会逐渐“淡出”从而控制记忆库的总规模和质量。对于关键记忆如用户的核心偏好可以通过手动标记或高频访问保持其高重要性避免被衰减。5.3 记忆的更新与冲突解决当新信息与旧记忆冲突时怎么办例如用户之前说“我喜欢蓝色”现在又说“我其实更喜欢绿色”。时间戳优先最简单的策略是相信最新的信息。在存储新记忆“喜欢绿色”的同时可以降低旧记忆“喜欢蓝色”的重要性分数或在元数据中标记其为“已过时”。事实融合对于更复杂的情况可以引入一个“记忆融合”步骤。当检测到冲突时触发一个LLM调用分析这两条记忆生成一条更准确的新记忆例如“用户对颜色的偏好可能从蓝色转变为了绿色或视上下文而定”然后替换或补充旧记忆。6. 工程集成打造Agent Harness框架将上述所有策略组合起来就需要一个轻量的框架来“驾驭”HarnessAgent的上下文和记忆。这个框架负责在每次调用LLM前智能地组装上下文。6.1 上下文组装流水线我设计了一个处理流水线它的工作流程如下输入当前用户查询 用户ID 会话ID。检索长期记忆用当前查询去向量库检索严格过滤user_id和session_id得到N条相关记忆。获取工作记忆从缓存或数据库中取出最近2-3轮原始对话或它们的摘要。生成摘要/压缩可选如果工作记忆较长可以对其进行一轮压缩摘要。组装最终提示将系统指令、检索到的记忆、压缩后的工作记忆、当前查询按顺序组合成最终的提示词。调用LLM发送组装好的提示词获得回复。更新记忆将本轮交互的核心信息创建为新的记忆单元存入向量库。同时更新工作记忆缓存。class AgentHarness: def __init__(self, llm, vectorstore, memory_manager): self.llm llm self.vectorstore vectorstore self.memory_manager memory_manager # 负责记忆的存储、检索、衰减 def process_query(self, user_query, user_id, session_id): # 1. 检索相关长期记忆 long_term_memories self.memory_manager.retrieve(user_query, user_id, session_id) # 2. 获取工作记忆最近对话 working_memory self.memory_manager.get_working_memory(session_id) # 3. (可选)压缩工作记忆 if len(working_memory) 3: # 假设超过3轮则压缩 working_memory self._compress_working_memory(working_memory) # 4. 组装提示词 prompt self._assemble_prompt(system_msg, long_term_memories, working_memory, user_query) # 5. 调用LLM response self.llm.invoke(prompt) # 6. 更新记忆 self.memory_manager.store_turn(user_id, session_id, user_query, response.content) return response.content6.2 关键配置参数与调优在实际使用中有几个参数对效果影响巨大需要仔细调优检索数量 (k)每次从向量库取多少条记忆太少可能信息不全太多则引入噪音。通常从3-5条开始测试。工作记忆长度保留最近几轮原始对话对于需要紧密跟随上下文的任务如代码调试可能需要3-5轮对于开放式聊天1-2轮可能就够了。记忆衰减因子决定旧记忆被遗忘的速度。需要根据业务场景调整。摘要的粒度摘要得太粗会丢失细节太细则压缩效果差。可能需要为不同类型的对话设计不同的摘要模板。我的经验是建立一个简单的评估体系在测试集上同时监控Token消耗、任务完成准确率、以及人工评估的“记忆相关性”。通过A/B测试找到最适合你场景的参数组合。7. 实战效果与避坑指南经过上述改造我的智能客服Agent焕然一新。最明显的效果是平均每次API调用的Token输入量下降了约40%因为上下文从完整的“史记”变成了精炼的“摘要”加“相关史料”。成本压力骤减。更惊喜的是效果提升。由于记忆检索是精准的、隔离的Agent混淆用户和任务的情况基本消失。当用户问起“我上次咨询的那个问题”时Agent能准确找到对应会话中的相关记忆而不是胡乱抓取一个相似话题。任务完成的准确率提升了约15%。当然踩坑是必不可少的摘要失真早期使用的摘要Prompt不够好导致关键信息如订单号、具体时间丢失。对策设计Prompt时明确列出必须保留的信息类型。最好能用一批样本进行测试检查摘要的保真度。向量检索不准当用户使用模糊指代如“那个”、“这东西”时直接检索效果很差。对策在检索前先用LLM对当前查询进行一次“查询重写”将其扩展成更完整、包含更多实体信息的描述再用这个描述去检索。记忆更新延迟存入向量库的记忆并非立即可用于下一次检索取决于向量化的批次和索引更新策略。对策对于当前会话中刚产生的、极其重要的信息可以同时放在工作记忆和一份“临时记忆缓存”中确保下一轮对话能立刻用到。系统复杂度增加引入了向量数据库、记忆管理模块系统架构变复杂了。对策做好模块化设计并将记忆相关的操作封装成独立服务便于维护和升级。这次“瘦身”实践让我深刻体会到对于AI Agent而言“更多”并不总是意味着“更好”。尤其是在上下文窗口有限、Token成本高昂的现实约束下如何高效、智能地组织和使用信息比盲目地堆砌信息更重要。通过结构化记忆、精准检索和动态管理我们完全可以让Agent在更小的“脑容量”下表现出更强大的“记忆力”和“专注力”。这不仅仅是成本的优化更是Agent智能体健壮性和实用性的关键一步。
