1. 先搞清楚“给小模型Agent注入大模型记忆”到底解决了什么如果你在折腾LLM Agent尤其是用本地部署的小模型比如7B、13B参数来构建最头疼的问题之一可能就是“记忆”能力。小模型本身的知识储备和上下文理解有限让它处理多轮对话、长期任务或者需要历史信息辅助的决策时经常表现得像个“金鱼”说完就忘。最近有个研究思路挺有意思不是去微调小模型也不是给它接个臃肿的向量数据库而是尝试把大模型的“记忆”能力直接“注入”给小模型Agent。简单说就是让小模型在需要回忆或参考历史时能去“借用”一下大模型的记忆库。根据一些论文和社区实践这种方法在特定任务上能让小模型Agent的性能提升相当可观有的评测里甚至能涨20多个点。这方法最吸引人的地方在于“无需训练”。你不用去准备海量的训练数据也不用耗费大量的GPU时间去微调模型基本上属于一种“即插即用”的增强策略。它特别适合那些已经有一个基于小模型跑起来的Agent原型但受限于模型能力在复杂、长程任务中表现不佳的场景。所以这篇文章就是来拆解这个思路的它具体是怎么做的真的能直接拿来用吗在你自己本地环境里部署时关键的步骤和容易踩的坑有哪些我会结合常见的Agent框架比如LangChain、LangGraph和开源方案把从原理到落地的过程捋一遍。2. 核心思路记忆不是存储而是“检索-增强”的能力在深入操作之前得先破除一个误区。这里说的“注入记忆”并不是把小模型的参数换成大模型的也不是把大模型的权重灌给小模型。它的核心是一种架构层面的增强更准确地说是一种混合推理框架。传统的Agent记忆不管是放在向量库里的还是简单存在列表里的都是“死”的。小模型自己去检索、去理解这些记忆然后做出决策瓶颈在于小模型自身的理解能力。而这个新思路把过程拆成了两步记忆检索与“富化”当Agent需要用到记忆比如回顾之前的对话历史、任务步骤结果时不是让小模型直接处理原始记忆片段而是让一个大模型或大模型API先对这些记忆进行理解、总结、提炼甚至推理。大模型在这里扮演了一个“记忆加工厂”的角色把原始的、可能杂乱的信息加工成更精炼、更高层次、更易于理解的“记忆摘要”或“关键洞察”。决策与执行然后把这个经过大模型加工后的、高质量的“记忆摘要”连同当前的任务指令一起交给小模型Agent去做最终的决策和工具调用。这样一来小模型不需要自己去吃力地理解长篇累牍的历史记录它拿到的是已经被“预处理”好的、信息密度更高的提示。它的任务从“理解一切并决策”变成了“基于高质量摘要进行决策”难度大大降低。为什么说“无需训练”因为整个过程中小模型和大模型的参数都是冻结的。你不需要用数据去调整它们的权重。你只是在设计一个工作流Workflow决定在哪个环节、如何调用大模型来处理记忆然后把处理结果喂给小模型。这本质上是一种提示工程Prompt Engineering和流程编排的优化。性能提升的关键点在哪提升主要来自于两个方面一是信息质量的提升大模型提炼的记忆摘要更准确、更相关二是减轻了小模型的认知负荷让它能更专注于当前最需要的推理和动作生成。在一些需要复杂记忆回溯的任务如多步骤游戏、长期对话管理、基于历史的分析上效果尤其明显。3. 动手之前你的环境与方案选型在开始折腾代码之前先盘清楚你的家当和要选的路。这决定了后续所有步骤的复杂度。3.1 硬件与模型资源盘点这是最实际的一步别等到跑起来了才发现资源不够。小模型本地Agent核心模型你打算用哪个小模型常见的如 Llama 3.1 8B、Qwen2.5 7B、Gemma 2 9B 等。确保它已经在你本地部署好了并且有一个稳定的推理接口如 OpenAI-compatible API。Ollama、vLLM、LM Studio 都是不错的本地部署工具。资源运行这个小模型需要多少显存8B模型在4-bit量化下通常需要6-8GB显存。确保你的GPU或系统内存如果靠CPU推理扛得住。大模型记忆处理器选择一云端API最方便。比如 OpenAI 的 GPT-4/GPT-3.5-Turbo、Anthropic 的 Claude、国内平台的 API。优势是能力强、省心劣势是会产生费用且需要稳定的网络。选择二本地大模型更可控。如果你有一张24GB显存的卡可以部署一个70B级别的模型如Qwen1.5 72B的量化版。优势是数据不出本地、无使用成本劣势是对硬件要求高且推理速度可能慢于API。选择三同一个模型不同角色最省资源。有些研究也尝试用同一个模型比如你的13B小模型同时扮演“记忆处理器”和“行动者”但通过不同的系统提示词Prompt来区分角色。这属于极限优化效果通常不如使用真正的大模型。我的建议是初次实验优先使用云端大模型API。这能排除掉因本地大模型部署、性能问题带来的干扰让你先把整个工作流跑通。等流程验证无误后再考虑是否要迁移到本地大模型以降低成本或满足隐私要求。3.2 软件与框架准备你需要一个能编排Agent工作流的框架。目前主流的有两个选择LangGraph这是当前构建复杂、有状态Agent的首选。它用“图”的概念来定义工作流节点Node是函数边Edge是流转条件天然适合处理带循环、分支和记忆的状态机。强烈推荐用于实现这种带记忆增强的复杂Agent。LangChain更早也更全面的框架。它的AgentExecutor和Memory模块也能实现但在表达复杂、定制化的记忆处理流程时代码可能不如LangGraph清晰直观。此外你还需要Python环境3.9。必要的包langgraph,langchain,openai(或其他大模型SDK)以及小模型客户端的SDK。一个简单的记忆存储初期可以用Python的list或dict在内存里存后期可以换成数据库或向量库如Chroma、Weaviate。4. 实战构建基于LangGraph的混合记忆Agent下面我们以LangGraph为核心构建一个最简单的“小模型决策 大模型处理记忆”的Agent。假设我们的场景是一个旅行规划助手它能记住用户之前提过的偏好比如“不喜欢爬山”、“对海鲜过敏”并在后续规划中运用这些信息。4.1 第一步定义状态与记忆结构在LangGraph里所有节点共享一个“状态”。我们首先要定义这个状态里有什么。from typing import TypedDict, List, Annotated import operator from datetime import datetime class AgentState(TypedDict): # 用户当前输入的问题 user_input: str # 完整的原始对话历史供大模型处理用 raw_conversation_history: List[str] # 经过大模型提炼后的“增强记忆摘要” enhanced_memory_summary: str # 小模型基于摘要和当前输入生成的最终回答 final_response: str # 记录当前步骤方便控制流程 step: str这里的关键是区分了raw_conversation_history和enhanced_memory_summary。原始历史我们只存不用直接给小模型增强摘要才是给小模型的“营养餐”。4.2 第二步创建两个模型客户端我们需要初始化两个模型的客户端一个给大模型记忆处理器一个给小模型行动代理。from langchain_openai import ChatOpenAI # 假设你的小模型也提供了OpenAI兼容的API from openai import OpenAI # 1. 大模型客户端 (例如 GPT-4) big_llm ChatOpenAI( modelgpt-4-turbo, temperature0.1, # 处理记忆需要稳定性温度调低 api_keyyour-openai-api-key ) # 2. 小模型客户端 (例如本地部署的Qwen2.5 7B) # 假设它在 http://localhost:8000/v1 提供了兼容OpenAI的API small_llm_client OpenAI( base_urlhttp://localhost:8000/v1, api_keyno-key-required # 根据你的本地服务设置 ) def call_small_model(prompt: str) - str: 调用本地小模型的简单函数 try: response small_llm_client.chat.completions.create( modelqwen2.5-7b-instruct, # 你的模型名 messages[{role: user, content: prompt}], temperature0.7, max_tokens500 ) return response.choices[0].message.content except Exception as e: return fError calling small model: {e}4.3 第三步编写核心节点函数这是整个系统的灵魂我们将工作流拆成三个主要节点。节点A更新原始记忆这个节点很简单就是把最新的用户对话追加到历史记录里。def update_raw_memory(state: AgentState): 将当前用户输入存入原始对话历史 new_history state.get(raw_conversation_history, []) # 简单存储实际可以加上时间戳、角色等 new_history.append(fUser: {state[user_input]}) return {raw_conversation_history: new_history}节点B调用大模型生成增强记忆摘要这是“注入记忆”的关键一步。我们让大模型回顾整个原始历史并提炼出对当前任务有用的关键信息。def enhance_memory_with_big_model(state: AgentState): 使用大模型处理原始历史生成记忆摘要 raw_history state.get(raw_conversation_history, []) current_query state[user_input] if not raw_history: # 如果没有历史摘要为空 return {enhanced_memory_summary: } # 构建给大模型的提示词 memory_prompt f 你是一个专业的记忆辅助助手。请分析以下的对话历史并提炼出与用户最新查询相关的关键信息、用户偏好和事实细节。 最新查询是{current_query} 对话历史 {chr(10).join(raw_history[-10:])} # 只取最近10轮防止上下文过长 请生成一个简洁、结构化的摘要专注于对回答最新查询有帮助的信息。摘要不要包含无关细节。 摘要格式 - 关键用户偏好 - 相关历史事实 - 当前对话目标 try: # 调用大模型 response big_llm.invoke(memory_prompt) summary response.content except Exception as e: summary fMemory enhancement failed: {e} return {enhanced_memory_summary: summary}节点C小模型基于摘要进行决策和回答小模型终于登场了。它收到的提示词里包含了经过大模型加工的“增强记忆”任务变得简单。def generate_response_with_small_model(state: AgentState): 小模型利用增强记忆摘要来生成最终回答 enhanced_memory state.get(enhanced_memory_summary, ) current_query state[user_input] # 构建给小模型的提示词 agent_prompt f 你是一个旅行规划助手。请根据以下信息回答用户的问题。 【增强记忆摘要来自系统分析】 {enhanced_memory} 【用户当前问题】 {current_query} 请基于上述记忆摘要如果存在专业、友好地回答用户的问题。如果记忆摘要为空或与问题无关则根据你的通用知识回答。 直接给出回答不要提及“根据记忆摘要”这类话。 final_answer call_small_model(agent_prompt) return {final_response: final_answer}4.4 第四步用LangGraph组装工作流现在我们把三个节点连起来形成一个完整的图。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(update_memory, update_raw_memory) workflow.add_node(enhance_memory, enhance_memory_with_big_model) workflow.add_node(generate_response, generate_response_with_small_model) # 设置边定义执行顺序 workflow.set_entry_point(update_memory) workflow.add_edge(update_memory, enhance_memory) workflow.add_edge(enhance_memory, generate_response) workflow.add_edge(generate_response, END) # 编译图 app workflow.compile()4.5 第五步运行与测试现在可以运行这个混合Agent了。# 初始化状态 initial_state { user_input: 我想去一个温暖的海边城市度假有什么推荐吗, raw_conversation_history: [], enhanced_memory_summary: , final_response: , step: start } # 执行第一轮 result1 app.invoke(initial_state) print(用户第一问:, result1[user_input]) print(增强记忆摘要:, result1[enhanced_memory_summary]) print(小模型回答:, result1[final_response]) print(- * 50) # 模拟第二轮对话用户提供了更多偏好 new_state { user_input: 对了我海鲜过敏而且不喜欢太商业化的地方。, raw_conversation_history: result1[raw_conversation_history], # 继承历史 enhanced_memory_summary: , final_response: , step: start } result2 app.invoke(new_state) print(用户第二问:, result2[user_input]) print(增强记忆摘要应包含‘海鲜过敏’和‘不喜欢商业化’:, result2[enhanced_memory_summary]) print(小模型回答应体现对过敏的考虑:, result2[final_response]) print(- * 50) # 第三轮基于前两轮记忆进行规划 new_state { user_input: 基于我之前说的帮我规划一个5天的行程吧。, raw_conversation_history: result2[raw_conversation_history], enhanced_memory_summary: , final_response: , step: start } result3 app.invoke(new_state) print(用户第三问:, result3[user_input]) print(增强记忆摘要应综合所有偏好:, result3[enhanced_memory_summary]) print(小模型最终行程规划应避开海鲜推荐非商业化景点:, result3[final_response])通过这个流程你可以清晰地看到小模型在生成最终回答时并没有直接面对冗长的“User: 我想去海边... User: 我海鲜过敏...”历史而是拿到了一个由大模型生成的、结构化的摘要“- 关键用户偏好海鲜过敏不喜欢商业化。 - 相关历史事实想去温暖海边城市度假。”。这极大地降低了小模型的认知负担。5. 效果验证、调优与常见坑点跑通流程只是第一步。要让这个“记忆注入”真正有效你需要系统地验证和调优。5.1 如何验证性能是否真的“涨点”你不能只靠感觉。需要设计一些可量化的评估方式任务完成度设计一个多轮交互的标准化任务例如“规划一个包含住宿、餐饮、交通的旅行方案需满足用户逐步提出的3个约束条件”。对比纯小模型Agent和“记忆注入”版Agent统计任务完全成功的比例。记忆相关性在最终回答中人工或用一个裁判模型判断回答是否准确包含了历史对话中提到的关键约束如“海鲜过敏”。计算相关约束被正确采纳的比例。响应质量评分使用一个裁判大模型如GPT-4对两个Agent在相同多轮对话下的最终回答进行评分1-10分比较平均分。人工评估最可靠但也最耗时。让多人对两组对话的连贯性、有用性、记忆准确性进行盲评。注意论文里“涨27个点”可能是在特定基准测试如WebShop、ALFWorld上的结果。你的实际提升幅度取决于任务复杂度、小模型基础能力、大模型的质量以及提示词设计。能观察到稳定、显著的提升就是成功。5.2 关键调优旋钮如果效果不理想别急着否定方案先检查这几个地方给大模型的“记忆加工”提示词这是最重要的部分。你让大模型总结什么是总结全部事实还是只总结用户偏好是总结最近3轮还是全部历史提示词里要明确指令。例如“请从对话历史中提取出用户明确声明的禁忌项如过敏、不喜欢和强偏好项如喜欢、必须。”原始记忆的存储与截断不能无限制存储所有历史。通常保留最近10-20轮对话或者当历史太长时触发大模型进行一次“总结性压缩”把更早的对话压缩成一段永久的背景知识。小模型的提示词设计如何把“增强记忆摘要”呈现给小模型是直接拼接还是用特定的格式如XML标签明确告诉小模型“以下是系统为你整理的相关背景信息请务必在回答中考虑这些信息[摘要]”。大模型的选择与成本如果使用GPT-4效果最好但成本高。可以尝试用GPT-3.5-Turbo来处理记忆性价比更高。本地大模型则要平衡效果和速度。错误处理与降级如果大模型API调用失败或超时工作流不能崩溃。应该有一个降级策略比如直接跳过记忆增强步骤让小模型使用原始历史或一个空的摘要继续工作并在日志中记录错误。5.3 一定会遇到的坑与排查清单当你发现Agent行为异常、记忆没用上或者回答质量下降时按这个顺序查检查记忆摘要内容首先打印出enhanced_memory_summary。看看大模型到底产出了什么。是不是空的是不是跑题了如果摘要不对后面全错。可能原因给大模型的提示词不清晰原始历史格式混乱大模型本身能力或温度参数问题。检查给小模型的完整提示词在call_small_model函数里把最终拼好的agent_prompt打印出来。确认增强记忆摘要被正确地放在了提示词中。检查小模型输出小模型的回答是否完全忽略了摘要可能是提示词里指令不够强硬或者小模型能力太弱无法遵循复杂指令。检查状态流转在LangGraph中确保每个节点都正确修改了状态。特别是raw_conversation_history是否随着对话轮次正确增长。资源与延迟延迟每次调用都涉及一次大模型API请求这会显著增加单轮响应时间。考虑是否需要对记忆增强进行异步调用或缓存。成本如果使用付费API频繁调用大模型处理长历史成本会快速上升。需要设计策略比如只在检测到对话主题切换或用户明确提及历史时才触发记忆增强而不是每轮都触发。记忆冲突与过时信息用户可能改变主意。早期的“喜欢爬山”可能被后期的“膝盖不好不爬山了”覆盖。你的记忆增强逻辑是否能处理这种冲突更高级的实现需要让大模型能识别和解决信息冲突。6. 从Demo到生产进阶考量与模式扩展上面的例子是一个最简单的线性流程。在实际生产中你需要考虑更多。6.1 更复杂的记忆处理策略条件触发不是每轮对话都进行昂贵的记忆增强。可以在图中增加一个“判断节点”只有当用户输入包含“之前”、“记得”、“根据我上次说的”等关键词或者对话轮次超过一定数量时才走enhance_memory分支否则直接跳过让小模型基于原始历史或上一个摘要回答。记忆分层将记忆分为“工作记忆”当前会话和“长期记忆”跨会话。大模型可以负责将工作记忆中重要的部分提炼、归档到长期记忆如向量数据库中。当需要时再从长期记忆中检索相关片段并用大模型进行融合增强。记忆反思让Agent定期或在任务完成后主动调用大模型对过去一段时间的经历进行“反思”总结学到了什么、犯了什么错误并将反思结论作为更高级别的记忆存储起来指导未来行为。这就是所谓的“Reflection”机制。6.2 与现有框架集成我们的示例是手搓的。你可以将这个模式集成到更成熟的框架中在LangChain中你可以自定义一个BaseMemory类在其load_memory_variables方法中不是直接返回原始历史而是调用大模型处理后再返回。然后将这个自定义Memory类赋给AgentExecutor。在LangGraph中如示例所示你可以将记忆增强作为一个独立的节点或子图灵活地插入到工作流的任何位置。LangGraph的状态管理让这变得非常自然。6.3 成本、延迟与性能的权衡这是生产部署的核心矛盾。缓存对于相同的原始历史生成的记忆摘要很可能相同。可以缓存(raw_history_hash, current_query) - enhanced_summary的结果避免重复调用大模型。摘要复用如果连续几轮对话主题没变可以复用上一轮的增强记忆摘要而不是每轮重新生成。小模型的能力边界这个模式的前提是小模型在得到“好提示”后能做出正确决策。如果小模型本身能力太差比如指令遵循能力弱那么再好的记忆摘要也无济于事。你需要为你的小模型找到能力的“甜点区”。降级方案必须设计优雅的降级。当大模型服务不可用时系统应能自动切换到一个轻量级记忆方案如简单的关键词匹配或最近几条原始历史保证Agent基本可用。最终给小型Agent注入大模型记忆本质上是一种“资源错配”的巧妙利用用大模型强大的理解能力去弥补小模型在记忆处理和上下文关联上的短板而让小模型专注于它相对擅长的指令遵循和工具调用。它不是一个“银弹”但在多轮交互、需要长期记忆的复杂任务场景下是一个经过验证的、能显著提升表现且无需训练的高性价比策略。在你自己的项目里先从线性流程跑通再逐步加入条件判断、缓存和错误处理就能把它变成一个稳定可用的增强模块。 
长河君 · 2026/8/24 5:46:18 · 阅读 1.2k 无需训练！用大模型增强小模型Agent记忆的实战指南
1. 先搞清楚“给小模型Agent注入大模型记忆”到底解决了什么如果你在折腾LLM Agent尤其是用本地部署的小模型比如7B、13B参数来构建最头疼的问题之一可能就是“记忆”能力。小模型本身的知识储备和上下文理解有限让它处理多轮对话、长期任务或者需要历史信息辅助的决策时经常表现得像个“金鱼”说完就忘。最近有个研究思路挺有意思不是去微调小模型也不是给它接个臃肿的向量数据库而是尝试把大模型的“记忆”能力直接“注入”给小模型Agent。简单说就是让小模型在需要回忆或参考历史时能去“借用”一下大模型的记忆库。根据一些论文和社区实践这种方法在特定任务上能让小模型Agent的性能提升相当可观有的评测里甚至能涨20多个点。这方法最吸引人的地方在于“无需训练”。你不用去准备海量的训练数据也不用耗费大量的GPU时间去微调模型基本上属于一种“即插即用”的增强策略。它特别适合那些已经有一个基于小模型跑起来的Agent原型但受限于模型能力在复杂、长程任务中表现不佳的场景。所以这篇文章就是来拆解这个思路的它具体是怎么做的真的能直接拿来用吗在你自己本地环境里部署时关键的步骤和容易踩的坑有哪些我会结合常见的Agent框架比如LangChain、LangGraph和开源方案把从原理到落地的过程捋一遍。2. 核心思路记忆不是存储而是“检索-增强”的能力在深入操作之前得先破除一个误区。这里说的“注入记忆”并不是把小模型的参数换成大模型的也不是把大模型的权重灌给小模型。它的核心是一种架构层面的增强更准确地说是一种混合推理框架。传统的Agent记忆不管是放在向量库里的还是简单存在列表里的都是“死”的。小模型自己去检索、去理解这些记忆然后做出决策瓶颈在于小模型自身的理解能力。而这个新思路把过程拆成了两步记忆检索与“富化”当Agent需要用到记忆比如回顾之前的对话历史、任务步骤结果时不是让小模型直接处理原始记忆片段而是让一个大模型或大模型API先对这些记忆进行理解、总结、提炼甚至推理。大模型在这里扮演了一个“记忆加工厂”的角色把原始的、可能杂乱的信息加工成更精炼、更高层次、更易于理解的“记忆摘要”或“关键洞察”。决策与执行然后把这个经过大模型加工后的、高质量的“记忆摘要”连同当前的任务指令一起交给小模型Agent去做最终的决策和工具调用。这样一来小模型不需要自己去吃力地理解长篇累牍的历史记录它拿到的是已经被“预处理”好的、信息密度更高的提示。它的任务从“理解一切并决策”变成了“基于高质量摘要进行决策”难度大大降低。为什么说“无需训练”因为整个过程中小模型和大模型的参数都是冻结的。你不需要用数据去调整它们的权重。你只是在设计一个工作流Workflow决定在哪个环节、如何调用大模型来处理记忆然后把处理结果喂给小模型。这本质上是一种提示工程Prompt Engineering和流程编排的优化。性能提升的关键点在哪提升主要来自于两个方面一是信息质量的提升大模型提炼的记忆摘要更准确、更相关二是减轻了小模型的认知负荷让它能更专注于当前最需要的推理和动作生成。在一些需要复杂记忆回溯的任务如多步骤游戏、长期对话管理、基于历史的分析上效果尤其明显。3. 动手之前你的环境与方案选型在开始折腾代码之前先盘清楚你的家当和要选的路。这决定了后续所有步骤的复杂度。3.1 硬件与模型资源盘点这是最实际的一步别等到跑起来了才发现资源不够。小模型本地Agent核心模型你打算用哪个小模型常见的如 Llama 3.1 8B、Qwen2.5 7B、Gemma 2 9B 等。确保它已经在你本地部署好了并且有一个稳定的推理接口如 OpenAI-compatible API。Ollama、vLLM、LM Studio 都是不错的本地部署工具。资源运行这个小模型需要多少显存8B模型在4-bit量化下通常需要6-8GB显存。确保你的GPU或系统内存如果靠CPU推理扛得住。大模型记忆处理器选择一云端API最方便。比如 OpenAI 的 GPT-4/GPT-3.5-Turbo、Anthropic 的 Claude、国内平台的 API。优势是能力强、省心劣势是会产生费用且需要稳定的网络。选择二本地大模型更可控。如果你有一张24GB显存的卡可以部署一个70B级别的模型如Qwen1.5 72B的量化版。优势是数据不出本地、无使用成本劣势是对硬件要求高且推理速度可能慢于API。选择三同一个模型不同角色最省资源。有些研究也尝试用同一个模型比如你的13B小模型同时扮演“记忆处理器”和“行动者”但通过不同的系统提示词Prompt来区分角色。这属于极限优化效果通常不如使用真正的大模型。我的建议是初次实验优先使用云端大模型API。这能排除掉因本地大模型部署、性能问题带来的干扰让你先把整个工作流跑通。等流程验证无误后再考虑是否要迁移到本地大模型以降低成本或满足隐私要求。3.2 软件与框架准备你需要一个能编排Agent工作流的框架。目前主流的有两个选择LangGraph这是当前构建复杂、有状态Agent的首选。它用“图”的概念来定义工作流节点Node是函数边Edge是流转条件天然适合处理带循环、分支和记忆的状态机。强烈推荐用于实现这种带记忆增强的复杂Agent。LangChain更早也更全面的框架。它的AgentExecutor和Memory模块也能实现但在表达复杂、定制化的记忆处理流程时代码可能不如LangGraph清晰直观。此外你还需要Python环境3.9。必要的包langgraph,langchain,openai(或其他大模型SDK)以及小模型客户端的SDK。一个简单的记忆存储初期可以用Python的list或dict在内存里存后期可以换成数据库或向量库如Chroma、Weaviate。4. 实战构建基于LangGraph的混合记忆Agent下面我们以LangGraph为核心构建一个最简单的“小模型决策 大模型处理记忆”的Agent。假设我们的场景是一个旅行规划助手它能记住用户之前提过的偏好比如“不喜欢爬山”、“对海鲜过敏”并在后续规划中运用这些信息。4.1 第一步定义状态与记忆结构在LangGraph里所有节点共享一个“状态”。我们首先要定义这个状态里有什么。from typing import TypedDict, List, Annotated import operator from datetime import datetime class AgentState(TypedDict): # 用户当前输入的问题 user_input: str # 完整的原始对话历史供大模型处理用 raw_conversation_history: List[str] # 经过大模型提炼后的“增强记忆摘要” enhanced_memory_summary: str # 小模型基于摘要和当前输入生成的最终回答 final_response: str # 记录当前步骤方便控制流程 step: str这里的关键是区分了raw_conversation_history和enhanced_memory_summary。原始历史我们只存不用直接给小模型增强摘要才是给小模型的“营养餐”。4.2 第二步创建两个模型客户端我们需要初始化两个模型的客户端一个给大模型记忆处理器一个给小模型行动代理。from langchain_openai import ChatOpenAI # 假设你的小模型也提供了OpenAI兼容的API from openai import OpenAI # 1. 大模型客户端 (例如 GPT-4) big_llm ChatOpenAI( modelgpt-4-turbo, temperature0.1, # 处理记忆需要稳定性温度调低 api_keyyour-openai-api-key ) # 2. 小模型客户端 (例如本地部署的Qwen2.5 7B) # 假设它在 http://localhost:8000/v1 提供了兼容OpenAI的API small_llm_client OpenAI( base_urlhttp://localhost:8000/v1, api_keyno-key-required # 根据你的本地服务设置 ) def call_small_model(prompt: str) - str: 调用本地小模型的简单函数 try: response small_llm_client.chat.completions.create( modelqwen2.5-7b-instruct, # 你的模型名 messages[{role: user, content: prompt}], temperature0.7, max_tokens500 ) return response.choices[0].message.content except Exception as e: return fError calling small model: {e}4.3 第三步编写核心节点函数这是整个系统的灵魂我们将工作流拆成三个主要节点。节点A更新原始记忆这个节点很简单就是把最新的用户对话追加到历史记录里。def update_raw_memory(state: AgentState): 将当前用户输入存入原始对话历史 new_history state.get(raw_conversation_history, []) # 简单存储实际可以加上时间戳、角色等 new_history.append(fUser: {state[user_input]}) return {raw_conversation_history: new_history}节点B调用大模型生成增强记忆摘要这是“注入记忆”的关键一步。我们让大模型回顾整个原始历史并提炼出对当前任务有用的关键信息。def enhance_memory_with_big_model(state: AgentState): 使用大模型处理原始历史生成记忆摘要 raw_history state.get(raw_conversation_history, []) current_query state[user_input] if not raw_history: # 如果没有历史摘要为空 return {enhanced_memory_summary: } # 构建给大模型的提示词 memory_prompt f 你是一个专业的记忆辅助助手。请分析以下的对话历史并提炼出与用户最新查询相关的关键信息、用户偏好和事实细节。 最新查询是{current_query} 对话历史 {chr(10).join(raw_history[-10:])} # 只取最近10轮防止上下文过长 请生成一个简洁、结构化的摘要专注于对回答最新查询有帮助的信息。摘要不要包含无关细节。 摘要格式 - 关键用户偏好 - 相关历史事实 - 当前对话目标 try: # 调用大模型 response big_llm.invoke(memory_prompt) summary response.content except Exception as e: summary fMemory enhancement failed: {e} return {enhanced_memory_summary: summary}节点C小模型基于摘要进行决策和回答小模型终于登场了。它收到的提示词里包含了经过大模型加工的“增强记忆”任务变得简单。def generate_response_with_small_model(state: AgentState): 小模型利用增强记忆摘要来生成最终回答 enhanced_memory state.get(enhanced_memory_summary, ) current_query state[user_input] # 构建给小模型的提示词 agent_prompt f 你是一个旅行规划助手。请根据以下信息回答用户的问题。 【增强记忆摘要来自系统分析】 {enhanced_memory} 【用户当前问题】 {current_query} 请基于上述记忆摘要如果存在专业、友好地回答用户的问题。如果记忆摘要为空或与问题无关则根据你的通用知识回答。 直接给出回答不要提及“根据记忆摘要”这类话。 final_answer call_small_model(agent_prompt) return {final_response: final_answer}4.4 第四步用LangGraph组装工作流现在我们把三个节点连起来形成一个完整的图。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(update_memory, update_raw_memory) workflow.add_node(enhance_memory, enhance_memory_with_big_model) workflow.add_node(generate_response, generate_response_with_small_model) # 设置边定义执行顺序 workflow.set_entry_point(update_memory) workflow.add_edge(update_memory, enhance_memory) workflow.add_edge(enhance_memory, generate_response) workflow.add_edge(generate_response, END) # 编译图 app workflow.compile()4.5 第五步运行与测试现在可以运行这个混合Agent了。# 初始化状态 initial_state { user_input: 我想去一个温暖的海边城市度假有什么推荐吗, raw_conversation_history: [], enhanced_memory_summary: , final_response: , step: start } # 执行第一轮 result1 app.invoke(initial_state) print(用户第一问:, result1[user_input]) print(增强记忆摘要:, result1[enhanced_memory_summary]) print(小模型回答:, result1[final_response]) print(- * 50) # 模拟第二轮对话用户提供了更多偏好 new_state { user_input: 对了我海鲜过敏而且不喜欢太商业化的地方。, raw_conversation_history: result1[raw_conversation_history], # 继承历史 enhanced_memory_summary: , final_response: , step: start } result2 app.invoke(new_state) print(用户第二问:, result2[user_input]) print(增强记忆摘要应包含‘海鲜过敏’和‘不喜欢商业化’:, result2[enhanced_memory_summary]) print(小模型回答应体现对过敏的考虑:, result2[final_response]) print(- * 50) # 第三轮基于前两轮记忆进行规划 new_state { user_input: 基于我之前说的帮我规划一个5天的行程吧。, raw_conversation_history: result2[raw_conversation_history], enhanced_memory_summary: , final_response: , step: start } result3 app.invoke(new_state) print(用户第三问:, result3[user_input]) print(增强记忆摘要应综合所有偏好:, result3[enhanced_memory_summary]) print(小模型最终行程规划应避开海鲜推荐非商业化景点:, result3[final_response])通过这个流程你可以清晰地看到小模型在生成最终回答时并没有直接面对冗长的“User: 我想去海边... User: 我海鲜过敏...”历史而是拿到了一个由大模型生成的、结构化的摘要“- 关键用户偏好海鲜过敏不喜欢商业化。 - 相关历史事实想去温暖海边城市度假。”。这极大地降低了小模型的认知负担。5. 效果验证、调优与常见坑点跑通流程只是第一步。要让这个“记忆注入”真正有效你需要系统地验证和调优。5.1 如何验证性能是否真的“涨点”你不能只靠感觉。需要设计一些可量化的评估方式任务完成度设计一个多轮交互的标准化任务例如“规划一个包含住宿、餐饮、交通的旅行方案需满足用户逐步提出的3个约束条件”。对比纯小模型Agent和“记忆注入”版Agent统计任务完全成功的比例。记忆相关性在最终回答中人工或用一个裁判模型判断回答是否准确包含了历史对话中提到的关键约束如“海鲜过敏”。计算相关约束被正确采纳的比例。响应质量评分使用一个裁判大模型如GPT-4对两个Agent在相同多轮对话下的最终回答进行评分1-10分比较平均分。人工评估最可靠但也最耗时。让多人对两组对话的连贯性、有用性、记忆准确性进行盲评。注意论文里“涨27个点”可能是在特定基准测试如WebShop、ALFWorld上的结果。你的实际提升幅度取决于任务复杂度、小模型基础能力、大模型的质量以及提示词设计。能观察到稳定、显著的提升就是成功。5.2 关键调优旋钮如果效果不理想别急着否定方案先检查这几个地方给大模型的“记忆加工”提示词这是最重要的部分。你让大模型总结什么是总结全部事实还是只总结用户偏好是总结最近3轮还是全部历史提示词里要明确指令。例如“请从对话历史中提取出用户明确声明的禁忌项如过敏、不喜欢和强偏好项如喜欢、必须。”原始记忆的存储与截断不能无限制存储所有历史。通常保留最近10-20轮对话或者当历史太长时触发大模型进行一次“总结性压缩”把更早的对话压缩成一段永久的背景知识。小模型的提示词设计如何把“增强记忆摘要”呈现给小模型是直接拼接还是用特定的格式如XML标签明确告诉小模型“以下是系统为你整理的相关背景信息请务必在回答中考虑这些信息[摘要]”。大模型的选择与成本如果使用GPT-4效果最好但成本高。可以尝试用GPT-3.5-Turbo来处理记忆性价比更高。本地大模型则要平衡效果和速度。错误处理与降级如果大模型API调用失败或超时工作流不能崩溃。应该有一个降级策略比如直接跳过记忆增强步骤让小模型使用原始历史或一个空的摘要继续工作并在日志中记录错误。5.3 一定会遇到的坑与排查清单当你发现Agent行为异常、记忆没用上或者回答质量下降时按这个顺序查检查记忆摘要内容首先打印出enhanced_memory_summary。看看大模型到底产出了什么。是不是空的是不是跑题了如果摘要不对后面全错。可能原因给大模型的提示词不清晰原始历史格式混乱大模型本身能力或温度参数问题。检查给小模型的完整提示词在call_small_model函数里把最终拼好的agent_prompt打印出来。确认增强记忆摘要被正确地放在了提示词中。检查小模型输出小模型的回答是否完全忽略了摘要可能是提示词里指令不够强硬或者小模型能力太弱无法遵循复杂指令。检查状态流转在LangGraph中确保每个节点都正确修改了状态。特别是raw_conversation_history是否随着对话轮次正确增长。资源与延迟延迟每次调用都涉及一次大模型API请求这会显著增加单轮响应时间。考虑是否需要对记忆增强进行异步调用或缓存。成本如果使用付费API频繁调用大模型处理长历史成本会快速上升。需要设计策略比如只在检测到对话主题切换或用户明确提及历史时才触发记忆增强而不是每轮都触发。记忆冲突与过时信息用户可能改变主意。早期的“喜欢爬山”可能被后期的“膝盖不好不爬山了”覆盖。你的记忆增强逻辑是否能处理这种冲突更高级的实现需要让大模型能识别和解决信息冲突。6. 从Demo到生产进阶考量与模式扩展上面的例子是一个最简单的线性流程。在实际生产中你需要考虑更多。6.1 更复杂的记忆处理策略条件触发不是每轮对话都进行昂贵的记忆增强。可以在图中增加一个“判断节点”只有当用户输入包含“之前”、“记得”、“根据我上次说的”等关键词或者对话轮次超过一定数量时才走enhance_memory分支否则直接跳过让小模型基于原始历史或上一个摘要回答。记忆分层将记忆分为“工作记忆”当前会话和“长期记忆”跨会话。大模型可以负责将工作记忆中重要的部分提炼、归档到长期记忆如向量数据库中。当需要时再从长期记忆中检索相关片段并用大模型进行融合增强。记忆反思让Agent定期或在任务完成后主动调用大模型对过去一段时间的经历进行“反思”总结学到了什么、犯了什么错误并将反思结论作为更高级别的记忆存储起来指导未来行为。这就是所谓的“Reflection”机制。6.2 与现有框架集成我们的示例是手搓的。你可以将这个模式集成到更成熟的框架中在LangChain中你可以自定义一个BaseMemory类在其load_memory_variables方法中不是直接返回原始历史而是调用大模型处理后再返回。然后将这个自定义Memory类赋给AgentExecutor。在LangGraph中如示例所示你可以将记忆增强作为一个独立的节点或子图灵活地插入到工作流的任何位置。LangGraph的状态管理让这变得非常自然。6.3 成本、延迟与性能的权衡这是生产部署的核心矛盾。缓存对于相同的原始历史生成的记忆摘要很可能相同。可以缓存(raw_history_hash, current_query) - enhanced_summary的结果避免重复调用大模型。摘要复用如果连续几轮对话主题没变可以复用上一轮的增强记忆摘要而不是每轮重新生成。小模型的能力边界这个模式的前提是小模型在得到“好提示”后能做出正确决策。如果小模型本身能力太差比如指令遵循能力弱那么再好的记忆摘要也无济于事。你需要为你的小模型找到能力的“甜点区”。降级方案必须设计优雅的降级。当大模型服务不可用时系统应能自动切换到一个轻量级记忆方案如简单的关键词匹配或最近几条原始历史保证Agent基本可用。最终给小型Agent注入大模型记忆本质上是一种“资源错配”的巧妙利用用大模型强大的理解能力去弥补小模型在记忆处理和上下文关联上的短板而让小模型专注于它相对擅长的指令遵循和工具调用。它不是一个“银弹”但在多轮交互、需要长期记忆的复杂任务场景下是一个经过验证的、能显著提升表现且无需训练的高性价比策略。在你自己的项目里先从线性流程跑通再逐步加入条件判断、缓存和错误处理就能把它变成一个稳定可用的增强模块。