在构建AI Agent时你是否也遇到过这样的困境对话轮次一多Agent就“失忆”了忘记了之前的任务目标和上下文或者为了让Agent记住更多内容你不得不将整个对话历史都塞进上下文导致成本飙升、响应变慢这背后其实是传统“回放式记忆”的局限性。今天我们将深入探讨一个更先进的理念——“记忆是重建出来的不是回放出来的”并借助GitHub上的前沿开源项目手把手教你构建一个具备“重建式记忆”能力的智能Agent系统。本文将从一个具体的开源项目my_ai_town出发结合LangGraph、RAG等热门技术为你拆解“重建式记忆”的核心原理、技术实现与工程落地。无论你是AI应用开发者还是对Agent架构感兴趣的研究者都能从中获得一套从理论到实践的完整方案。1. 背景与核心概念从“回放”到“重建”的记忆革命在传统的大模型应用中“记忆”通常被简单等同于“完整的对话历史”。每次与模型交互时我们将之前所有的用户消息和AI回复拼接起来作为新的上下文输入。这种方式可以称为“回放式记忆Playback Memory”。“回放式记忆”的痛点上下文爆炸随着对话进行token数量线性增长很快触及模型上下文长度上限如4K、8K、128K。成本高昂大多数API按token收费冗长的历史意味着每次交互都在为“重复记忆”付费。信息稀释关键信息被淹没在海量历史中模型难以聚焦导致回答质量下降。无关干扰很久以前的、已不相关的对话片段会干扰当前任务的判断。那么什么才是“重建式记忆Reconstructive Memory”呢“重建式记忆”的核心思想我们不需要存储和回放每一句原始对话。相反我们维护一个动态的、结构化的记忆体Memory Store。这个记忆体不是聊天记录的副本而是对历史交互进行摘要、提取、索引和关联后形成的知识网络。当需要“回忆”时Agent根据当前查询从这个知识网络中实时重建Reconstruct出最相关的上下文片段。类比理解就像人类回忆一件事我们并非像录像机一样回放全部细节而是根据几个关键词或感觉重新构建出事件的主要脉络和关键点。“重建式记忆”让AI Agent做到了类似的事情。关键技术栈关联RAG检索增强生成为记忆重建提供了“检索”能力从向量库中快速找到相关记忆片段。Agent Tool CallingAgent利用记忆作为决策依据调用工具执行任务。Workflow如LangGraph定义了记忆的更新、查询和重建的流程与状态管理。上下文管理/压缩如DeepSeek Harness是重建过程中的一种策略用于精简和优化送入模型的上下文。接下来我们将通过一个融合了上述技术的开源项目来具体实现它。2. 环境准备与版本说明我们将以GitHub上的my_ai_town项目为灵感蓝图并结合LangGraph来构建一个具备重建式记忆的Agent系统。请注意my_ai_town本身可能是一个特定场景的应用我们会提取其关于记忆管理的设计思想并用更通用的技术栈实现。核心环境与工具Python: 3.9 或更高版本本文示例基于3.10包管理: pip 或 poetry核心库:langchain: 用于构建AI应用链和Agent的基础框架。langgraph: 用于构建有状态、多环节的Agent工作流核心。langchain-openai: OpenAI模型集成也可替换为其他模型。chromadb/faiss: 用于存储和检索记忆嵌入的向量数据库。pydantic: 用于定义严格的数据模型和状态。开发工具: 任何你喜欢的IDE如VSCode、PyCharm。版本建议requirements.txt示例langchain0.1.0 langgraph0.0.50 langchain-openai0.0.5 chromadb0.4.22 openai1.12.0 pydantic2.5.0 python-dotenv1.0.0重要提示LangChain生态版本迭代较快以上版本为撰写时的稳定版本。实际开发时请查阅官方文档使用兼容的版本组合。你需要一个有效的OpenAI API密钥或其它兼容API的密钥。3. 核心原理与架构拆解我们的目标是构建一个ReconstructiveMemoryAgent。它的核心在于两个部分记忆体和记忆重建引擎。3.1 记忆体设计不止是向量库记忆体不能只是一个简单的聊天记录列表或单一的向量索引。一个良好的记忆体应包含多层次结构短期记忆/工作记忆保存当前对话轮次中的关键信息容量小存取快。长期记忆由历史交互提炼而成存储在向量数据库中容量大。摘要记忆定期或按事件对长期记忆进行概括形成更高层次的“故事线”或“用户画像”。元数据为每条记忆打上时间戳、实体人、地点、事件、情感色彩、重要性权重等标签。在代码中我们用Pydantic模型来定义一条记忆单元from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List from enum import Enum class MemoryType(str, Enum): OBSERVATION “observation” # 观察到的事实 ACTION “action” # Agent执行的动作 USER_FACT “user_fact” # 用户透露的个人信息 GOAL “goal” # 任务目标 class MemoryEntity(BaseModel): “”“单条记忆的模型”“” id: str Field(default_factorylambda: str(uuid.uuid4())) content: str # 记忆的文本内容 type: MemoryType # 记忆类型 embedding: Optional[List[float]] None # 向量嵌入 timestamp: datetime Field(default_factorydatetime.now) entities: List[str] Field(default_factorylist) # 涉及的实体如 [“user”, “project_x”] importance: float Field(default0.5, ge0, le1) # 重要性权重0-1 accessed_count: int 0 # 被访问次数 class Config: arbitrary_types_allowed True3.2 记忆重建引擎如何从记忆体重建上下文这是“重建”理念的核心。当Agent需要回应时它不会读取所有原始记忆而是执行以下步骤查询理解分析当前用户输入提取关键实体、意图和查询嵌入。相关记忆检索从向量长期记忆中根据查询嵌入进行相似性搜索召回Top-K条相关记忆。从短期记忆中直接获取最近几条关键记忆。根据元数据如实体、类型进行过滤。记忆排序与融合结合相关性分数、重要性权重、时间新鲜度、访问频率等因素对检索到的记忆进行综合排序。将排名靠前的记忆片段融合成一段连贯、简洁的“重建上下文”。这里可能用到LLM进行二次摘要或串联。上下文组装将“重建上下文”与系统指令、当前查询一起组装成最终的Prompt送给LLM生成回复。from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma import numpy as np class ReconstructiveMemoryEngine: def __init__(self, vector_store: Chroma, short_term_memory: list, llm): self.vector_store vector_store self.short_term_memory short_term_memory # 短期记忆列表 self.llm llm self.embeddings OpenAIEmbeddings() def reconstruct_context(self, query: str, top_k: int 5) - str: “”“重建与当前查询相关的上下文”“” # 1. 检索相关长期记忆 query_embedding self.embeddings.embed_query(query) # 假设vector_store的检索方法返回 (MemoryEntity, score) 的列表 long_term_memories self.vector_store.similarity_search_with_score_by_embedding(query_embedding, ktop_k) # 2. 获取短期记忆例如最近3条 recent_memories self.short_term_memory[-3:] if self.short_term_memory else [] # 3. 融合与排序 (简化版按相关性分数和重要性加权) all_candidates [] for memory, score in long_term_memories: # 综合得分 相关性得分 * 重要性权重 * 时间衰减因子示例 time_decay np.exp(-0.1 * (datetime.now() - memory.timestamp).days) # 简单衰减 composite_score score * memory.importance * time_decay all_candidates.append((memory, composite_score)) # 加入短期记忆赋予较高权重 for mem in recent_memories: all_candidates.append((mem, 1.5)) # 短期记忆权重更高 # 按综合得分排序 all_candidates.sort(keylambda x: x[1], reverseTrue) # 4. 构建重建后的上下文字符串 reconstructed [] for mem, _ in all_candidates[:top_k3]: # 取排名靠前的 reconstructed.append(f“- [{mem.type}] {mem.content}“) return “\n”.join(reconstructed)4. 完整实战基于LangGraph构建记忆化Agent我们将使用LangGraph来构建Agent的工作流。LangGraph 通过“图”来定义状态流转非常适合管理包含记忆状态的复杂Agent。4.1 定义Agent状态首先定义整个工作流的状态其中包含我们的记忆体。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): “”“Agent的全局状态”“” messages: Annotated[List, add_messages] # 标准的对话消息历史 current_query: str # 当前用户输入 short_term_memory: List[MemoryEntity] # 短期记忆 reconstructed_context: str # 重建后的上下文 response: str # Agent的最终响应4.2 构建工作流节点我们将工作流分解为几个节点每个节点负责一项具体工作。节点1接收并更新查询from langgraph.graph import StateGraph workflow StateGraph(AgentState) def receive_query(state: AgentState) - AgentState: “”“接收用户输入并更新状态”“” # 假设最新一条消息是用户输入 last_message state[“messages”][-1] state[“current_query”] last_message.content return state节点2记忆重建def reconstruct_memory(state: AgentState) - AgentState: “”“调用记忆重建引擎生成相关上下文”“” # 初始化引擎实际应用中应全局共享 engine get_memory_engine() # 假设的获取引擎的函数 reconstructed_ctx engine.reconstruct_context(state[“current_query”]) state[“reconstructed_context”] reconstructed_ctx return state节点3调用LLM生成回复from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI def generate_response(state: AgentState) - AgentState: “”“结合重建的上下文生成Agent回复”“” prompt ChatPromptTemplate.from_messages([ (“system”, “““你是一个有帮助的AI助手。以下是与当前对话相关的背景记忆并非完整历史 {reconstructed_context} 请根据以上记忆和当前对话回应用户的需求。如果记忆不相关请忽略。”””), (“placeholder”, “{messages}“), # LangGraph会自动填充消息历史 ]) model ChatOpenAI(model“gpt-4-turbo-preview”, temperature0) chain prompt | model # 调用模型 response chain.invoke({ “reconstructed_context”: state[“reconstructed_context”], “messages”: state[“messages”] }) state[“response”] response.content return state节点4记忆固化def consolidate_memory(state: AgentState) - AgentState: “”“将本轮交互中有价值的信息存入长期记忆”“” current_query state[“current_query”] agent_response state[“response”] # 1. 创建观察记忆用户输入 observation_memory MemoryEntity( contentf“User said: {current_query}“, typeMemoryType.OBSERVATION, entities[“user”] # 简单示例实际应从查询中提取实体 ) # 2. 创建行动记忆Agent回复 action_memory MemoryEntity( contentf“Assistant responded: {agent_response}“, typeMemoryType.ACTION, entities[“assistant”] ) # 3. 使用LLM判断信息是否值得长期存储并提取关键事实高级功能 # 此处简化直接存储 memory_engine get_memory_engine() memory_engine.add_memory(observation_memory) memory_engine.add_memory(action_memory) # 4. 更新短期记忆将本轮关键记忆加入 state[“short_term_memory”].extend([observation_memory, action_memory]) # 限制短期记忆长度防止无限增长 if len(state[“short_term_memory”]) 10: state[“short_term_memory”] state[“short_term_memory”][-10:] return state4.3 组装并运行工作流将节点添加到图中并定义执行顺序。# 添加节点 workflow.add_node(“receive”, receive_query) workflow.add_node(“reconstruct”, reconstruct_memory) workflow.add_node(“generate”, generate_response) workflow.add_node(“consolidate”, consolidate_memory) # 定义边执行顺序 workflow.set_entry_point(“receive”) workflow.add_edge(“receive”, “reconstruct”) workflow.add_edge(“reconstruct”, “generate”) workflow.add_edge(“generate”, “consolidate”) workflow.add_edge(“consolidate”, END) # END是LangGraph预定义的特殊节点 # 编译图 app workflow.compile() # 运行Agent initial_state { “messages”: [(“user”, “你好请帮我记住我最喜欢的编程语言是Python。”)], “short_term_memory”: [], “reconstructed_context”: “”, “response”: “” } final_state app.invoke(initial_state) print(final_state[“response”]) # 输出可能”好的我已经记住您最喜欢的编程语言是Python了。” # 进行多轮对话 next_state { “messages”: final_state[“messages”] [(“user”, “我刚刚最喜欢的语言是什么”)], “short_term_memory”: final_state[“short_term_memory”], “reconstructed_context”: “”, “response”: “” } final_state2 app.invoke(next_state) print(final_state2[“response”]) # 输出可能”您刚刚告诉我您最喜欢的编程语言是Python。” # 注意这里Agent是从记忆体重建出这个事实而不是回放了整个对话历史。5. 高级优化与最佳实践基础的“重建”已经实现但要投入生产环境还需要考虑以下优化点5.1 记忆的摘要与压缩长期记忆不能无限增长。需要定期对记忆进行摘要。定期摘要每N轮对话后或用LLM对近期记忆生成一个摘要将摘要作为一条新的高级记忆存入并可选地归档原始记忆。重要性衰减引入遗忘机制根据时间、访问频率动态调整记忆的重要性权重过低的可被清理或归档。5.2 结构化记忆与图数据库对于复杂场景可以考虑使用图数据库如Neo4j来存储记忆。节点记忆片段、实体人、物、概念。边记忆之间的关系如“属于”、“导致”、“发生于”。优势能更自然地表达记忆间的复杂关联支持更复杂的图谱查询重建上下文时不仅能找相似还能找关联。5.3 上下文窗口的智能管理即使重建了上下文最终送入模型的Prompt长度也需控制。动态裁剪如果重建的上下文过长使用LLM进行二次压缩总结。分层注入将最关键的记忆放在Prompt最前面次要的放在后面或作为参考信息。5.4 测试与评估如何评估“重建式记忆”的好坏准确性Agent回忆的事实是否准确相关性重建的上下文是否与当前查询强相关效率相比回放全部历史token使用量是否显著下降响应速度是否提升连贯性在多轮对话中Agent的表现是否一致、自然6. 常见问题与排查思路问题现象可能原因排查与解决思路Agent“忘记”了重要信息1. 记忆未被成功存储。2. 记忆检索时相关性得分低未被召回。3. 记忆重要性权重设置过低被过滤。1. 检查记忆固化节点的代码确认存储逻辑无误查看向量库中是否有对应记录。2. 检查检索时的相似度算法和top_k值尝试调整嵌入模型或检索策略。3. 检查并调整记忆的重要性评估逻辑。重建的上下文不相关导致回答跑偏1. 查询嵌入质量差。2. 记忆的嵌入未能正确表征其语义。3. 元数据实体、类型过滤过严或过松。1. 尝试不同的嵌入模型如text-embedding-3-small。2. 确保存储记忆时content字段是清晰、无歧义的文本。3. 优化实体识别和打标流程调整过滤阈值。响应速度变慢1. 向量库中记忆条目过多检索慢。2. 重建逻辑过于复杂多次调用LLM。1. 对向量库建立索引或进行分片。定期清理不重要的记忆。2. 优化重建流程对于简单查询可以优先从短期记忆或缓存中获取。Token使用量未明显下降1. 重建的上下文仍然过长。2. 系统指令或其他固定Prompt部分过长。1. 在重建引擎后增加一个“上下文压缩”节点使用LLM对重建结果进行摘要。2. 精简系统Prompt移除不必要的指令。7. 总结从项目到架构的思考通过本文的实践我们超越了简单的“聊天记录保存”实现了一个具备动态重建能力的Agent记忆系统。关键收获在于理念转变记忆不是负担而是需要精心管理的资产。从“存储-回放”到“提取-重建”的转变是构建长效、高效Agent的关键。架构清晰利用LangGraph等工具可以将记忆的感知存储、认知重建、行动响应流程清晰定义使系统更易维护和扩展。工程化考量记忆系统涉及嵌入模型、向量数据库、摘要算法、缓存策略等多个组件需要综合考虑性能、成本和准确性。你可以在此基础上继续深化集成更多工具让Agent不仅能记忆还能通过工具调用操作外部系统形成“记忆-决策-行动”的闭环。实现多模态记忆除了文本能否存储和重建图像、音频相关的记忆探索更先进的记忆模型如神经图灵机NTM、可微分神经计算机DNC等思想在LLM Agent中的应用。“记忆是重建出来的”这一哲学不仅适用于AI也启发我们设计更智能、更人性化的交互系统。希望本文提供的代码和思路能成为你构建下一代AI Agent的坚实起点。