1. 项目缘起当AI对话变成“金鱼记忆”不知道你有没有过这样的体验你花了好几个小时和一个AI助手深入探讨一个复杂的项目从技术选型聊到架构设计甚至分享了你个人的工作习惯和偏好。第二天你兴致勃勃地打开对话想继续昨天的思路结果AI助手的第一句回复是“你好我是你的AI助手有什么可以帮你的吗” 那一刻你感觉之前所有的深入交流都像是对着空气说话一切又得从头再来。这就是典型的“跨会话失忆”问题。目前绝大多数基于大语言模型的AI应用无论是ChatGPT的Web界面还是我们自行搭建的Agent系统其对话记忆通常被严格限制在单次会话的上下文窗口内。一旦会话结束记忆便随之清零。这对于需要长期协作、持续跟踪复杂任务或希望AI能真正“了解”用户个性的场景来说是致命的短板。我最近在折腾一个多智能体协作的模拟小镇项目时就深刻体会到了这种痛苦。我希望小镇里的“居民”AI Agent不仅能完成即时任务还能记住彼此的互动、形成偏好、甚至发展出简单的“人际关系”。这显然不是单次对话能承载的。于是我开始寻找一种能让AI拥有“长期记忆”的解决方案目标很明确让AI能记住“我是谁”记住“我们聊过什么”并在未来的每一次互动中自然地运用这些记忆。经过一番探索和实战我最终基于TRAE Work这套框架成功实现了一个稳定、可扩展的跨会话长期记忆系统。整个过程并没有想象中复杂核心思路清晰工具链成熟。今天我就把这次从零到一的完整实践过程拆解给你你会发现给AI装上“记忆硬盘”其实分分钟就能搞定。2. 核心武器库TRAE Work与LangGraph的强强联合在动手之前我们得先搞清楚要用什么工具以及为什么是它们。我的技术选型主要围绕两个核心TRAE Work和LangGraph。TRAE Work是什么你可以把它理解为一个为AI应用开发量身定制的“脚手架”或“低代码平台”。它提供了一套标准化的模块和接口让你能快速搭建起AI智能体Agent所需的各种基础组件比如工具调用、记忆管理、状态持久化等。它的优势在于“开箱即用”和“标准化”避免了我们从零开始造轮子特别是处理Agent之间复杂的编排和状态流转时TRAE Work能省下大量精力。LangGraph则是LangChain生态系统中的明星组件专为构建有状态的、多步骤的AI应用而设计。它用“图”Graph的概念来建模应用流程节点代表执行步骤如调用LLM、运行工具边代表步骤之间的流转条件。最关键的是LangGraph内置了对状态管理的强力支持。它维护一个中心化的“状态”State对象这个状态可以在图的执行过程中被读取和修改并且——这是实现长期记忆的关键——这个状态可以被持久化到数据库里。那么它们俩是如何协作实现长期记忆的呢TRAE Work作为组织者它定义了整个AI应用的结构包括有哪些Agent每个Agent能做什么工具以及它们之间如何协作。它提供了一个清晰的框架来集成LangGraph。LangGraph作为记忆引擎我们将需要长期记忆的数据比如用户信息、历史对话摘要、项目上下文设计为LangGraph状态State的一部分。每次对话或任务执行都是一个LangGraph图的运行过程这个过程会更新状态。持久化层作为记忆硬盘LangGraph支持将状态序列化后存储到外部数据库如PostgreSQL, Redis。当一次会话结束时状态被保存当下一次会话开始时我们可以根据会话IDSession ID或用户ID从数据库中加载之前的状态从而实现记忆的跨会话延续。这个组合的优势在于记忆不再是散落在对话历史中的文本片段而是变成了一个结构化的、可编程的状态对象。我们可以精准地在其中存储和检索信息比如用户的职业偏好user.preference.job、上次聊到的项目进度project.last_milestone而不仅仅是模糊的聊天记录。3. 实战第一步定义你的记忆数据结构纸上谈兵终觉浅我们直接进入实战。第一步也是最关键的一步是设计你的记忆模型。你要想清楚你希望AI记住什么这些信息应该如何组织在我的AI小镇项目中我希望每个“居民”Agent都能记住两件事自己的身份和状态名字、性格简介、当前情绪值、拥有的物品清单。与他人的互动记忆和另一个居民最近一次有意义的交互内容摘要以及对对方的好感度变化。基于此我设计了如下的记忆状态结构使用Pydantic模型定义清晰且类型安全from typing import Dict, List, Optional from pydantic import BaseModel, Field class InteractionMemory(BaseModel): 一次交互的记忆摘要 with_agent: str Field(description交互对象的名称) timestamp: str Field(description交互发生的时间) summary: str Field(description交互内容的简短摘要) sentiment_delta: float Field(description本次交互导致的好感度变化, ge-1.0, le1.0) class AgentProfile(BaseModel): 智能体的长期档案 name: str Field(description智能体名称) biography: str Field(description背景故事或性格描述) current_mood: float Field(default0.5, description当前情绪值0-1之间, ge0.0, le1.0) inventory: List[str] Field(default_factorylist, description拥有的物品列表) # 长期记忆与其他Agent的交互历史 memories: Dict[str, List[InteractionMemory]] Field( default_factorydict, description键为其他Agent名值为与该Agent的交互记忆列表 ) class TownState(BaseModel): 小镇的全局共享状态也作为LangGraph的State agents: Dict[str, AgentProfile] Field(default_factorydict, description所有居民档案) current_date: str Field(description小镇的当前虚拟日期) # 可以扩展其他全局记忆如小镇公告、公共事件等这个设计有几个巧思分层结构TownState是顶层的图状态包含所有Agent。这样在LangGraph的任何节点都能访问到全局记忆。记忆摘要化我们不存储完整的对话历史那会很快撑爆上下文而是存储由LLM生成的summary。这大大压缩了记忆体积并提取了核心信息。结构化情感sentiment_delta是一个量化的数值便于后续计算和影响决策比如好感度低的Agent之间更可能发生争吵。注意记忆结构的设计直接影响后续实现的复杂度和使用效果。建议从最核心的1-2个字段开始逐步迭代。切忌一开始就设计一个庞杂无比的“记忆宇宙”那会让你在编码和调试时陷入泥潭。4. 构建记忆化的工作流LangGraph图编排有了记忆数据结构接下来就要把它嵌入到AI的工作流中。我们使用LangGraph来构建一个具备记忆能力的对话流程。假设我们有一个简单的“居民聊天”工作流两个Agent相遇根据彼此的长期记忆生成对话。以下是这个图的简化版实现from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator # 1. 定义Graph的状态继承自我们之前设计的TownState class ConversationState(TownState): current_speaker: str current_listener: str conversation_topic: Optional[str] None # 这是一个“计算属性”通过修饰器实现用于在图中动态获取当前说话者的记忆 recent_memory: Annotated[Optional[InteractionMemory], operator.add] None # 2. 创建图构建器 workflow StateGraph(ConversationState) # 3. 定义节点函数 def retrieve_memory(state: ConversationState): 节点检索长期记忆 speaker_profile state.agents[state.current_speaker] listener_memories speaker_profile.memories.get(state.current_listener, []) if listener_memories: # 取最近的一次记忆 latest_memory sorted(listener_memories, keylambda x: x.timestamp)[-1] state.recent_memory latest_memory else: state.recent_memory None return state def generate_dialogue(state: ConversationState): 节点基于记忆生成对话 # 准备LLM的提示词注入记忆 prompt_template f 你是{state.current_speaker}正在和{state.current_listener}聊天。 以下是你们上一次交互的摘要{state.recent_memory.summary if state.recent_memory else 这是你们第一次相遇。} 上次交互后你对TA的好感度变化了{state.recent_memory.sentiment_delta if state.recent_memory else 0}。 请生成一句符合你性格的、基于上述记忆的对话内容。 # 这里调用LLM (例如通过ChatOpenAI) # llm_response chat_model.invoke(prompt_template) # 为简化示例我们模拟一个回复 if state.recent_memory and state.recent_memory.sentiment_delta 0.2: dialogue f微笑嗨{state.current_listener}上次聊得很愉快你提到的那个想法我觉得很棒 elif state.recent_memory and state.recent_memory.sentiment_delta -0.2: dialogue f冷淡{state.current_listener}... 又见面了。 else: dialogue f你好{state.current_listener}今天天气不错。 # 更新状态比如可以记录本次生成的对话到临时上下文 state.conversation_topic dialogue return state def update_long_term_memory(state: ConversationState): 节点将本次交互摘要存入长期记忆 if not state.recent_memory and state.conversation_topic: # 如果是第一次交互 new_memory InteractionMemory( with_agentstate.current_listener, timestampstate.current_date, summaryf初次见面对话主题关于{state.conversation_topic[:50]}..., # 摘要化 sentiment_delta0.1 # 初次见面默认微小正好感 ) speaker_profile state.agents[state.current_speaker] if state.current_listener not in speaker_profile.memories: speaker_profile.memories[state.current_listener] [] speaker_profile.memories[state.current_listener].append(new_memory) # 同样也可以为listener创建关于speaker的记忆 # 如果已有记忆这里可以设计逻辑来更新或追加记忆例如合并摘要、重新计算情感等。 # 更复杂的实现可以在这里调用LLM对新旧记忆进行总结归纳。 return state # 4. 将节点添加到图中 workflow.add_node(“retrieve_memory”, retrieve_memory) workflow.add_node(“generate_dialogue”, generate_dialogue) workflow.add_node(“update_memory”, update_long_term_memory) # 5. 设置边定义执行顺序 workflow.set_entry_point(“retrieve_memory”) workflow.add_edge(“retrieve_memory”, “generate_dialogue”) workflow.add_edge(“generate_dialogue”, “update_memory”) workflow.add_edge(“update_memory”, END) # 6. 编译图 app workflow.compile()这个图定义了三个核心步骤检索记忆 - 生成对话 - 更新记忆。它形成了一个完整的“记忆循环”。每次对话不仅读取历史还会产生新的记忆。ConversationState中的agents字典即所有居民的AgentProfile就是被持久化的长期记忆载体。5. 记忆的存与取集成TRAE Work与持久化存储现在我们有了一个能在内存中运作的、有状态的图。但如何让这个状态在会话间存活呢这就需要持久化。我们将把LangGraph的状态存储到数据库中而TRAE Work可以帮助我们优雅地管理这个过程。我选择PostgreSQL作为持久化存储因为它可靠且通过pgvector扩展可以方便地支持未来可能的基于向量相似度的记忆检索。LangGraph社区提供了LangGraph Postgres Storage集成包。以下是集成到TRAE Work项目中的关键步骤1. 环境配置与依赖安装在TRAE Work项目的配置文件中添加数据库连接信息。# config.yaml 或环境变量 DATABASE_URL: “postgresql://user:passwordlocalhost:5432/ai_town_memory”安装必要的包pip install langgraph-postgres psycopg2-binary2. 创建可持久化的图实例修改图的编译方式使其绑定一个持久化存储后端。from langgraph.postgres import PostgresStore import asyncpg from langgraph.graph import StateGraph async def create_persistable_graph(): # 1. 连接到Postgres conn await asyncpg.connect(os.getenv(“DATABASE_URL”)) store await PostgresStore.create(conn) # 2. 使用我们之前定义的workflow workflow StateGraph(ConversationState) # ... (添加节点的代码与之前相同) ... # 3. 编译时指定存储 app workflow.compile( storestore, # 配置检查点每次运行后自动保存状态 checkpointerstore.as_checkpointer(serde“json”) ) return app, store3. 在TRAE Work中管理会话TRAE Work通常以API服务的形式运行。我们需要将“会话”Session的概念与LangGraph的“线程”Thread或“检查点”Checkpoint关联起来。一个简单的做法是使用user_id或session_id作为线程ID。from trae_work.sdk import Agent, Tool # 假设TRAE Work的SDK如此导入 class MemoryEnabledChatAgent(Agent): def __init__(self, name: str, graph_app): super().__init__(name) self.graph_app graph_app # 一个内存中的映射将TRAE Work会话ID映射到LangGraph线程ID self.session_to_thread {} async def on_message(self, session_id: str, user_input: str): # 获取或创建LangGraph线程 thread_id self.session_to_thread.get(session_id) config {“configurable”: {“thread_id”: thread_id}} if thread_id else {} # 准备初始状态例如从请求中解析出current_speaker等 initial_state { “current_speaker”: self.name, “current_listener”: “User”, # 或者从输入中解析 “conversation_topic”: user_input, “agents”: {...} # 这里应该从数据库加载TownState的初始数据或从config传入 } # 执行图 events self.graph_app.astream(initial_state, config) final_state None async for event in events: # 处理事件流例如获取生成的对话 if “generate_dialogue” in event: final_state event[“generate_dialogue”] if final_state: # 更新会话映射如果是新线程graph_app执行后会创建新的thread_id new_thread_id final_state[“configurable”][“thread_id”] self.session_to_thread[session_id] new_thread_id # 返回AI的回复 return final_state.get(“conversation_topic”, “I have no response.”) return “Processing completed.”4. 启动与恢复当用户开始一个新会话时session_id是全新的thread_id不存在LangGraph会创建一个新的检查点并执行。当用户再次发起请求可能是同一个会话的后续消息我们通过session_id找到对应的thread_idLangGraph会从上次保存的检查点恢复整个TownState包括所有更新后的记忆然后继续执行。这就实现了跨消息、甚至跨天、跨周的长期记忆。踩坑实录状态序列化的坑。在初期我直接尝试Pickle序列化整个Pydantic状态对象遇到了循环引用和自定义类型无法序列化的问题。LangGraph Postgres Store默认使用JSON序列化。关键点确保你的State中所有字段都是JSON可序列化的基本类型、列表、字典、嵌套的Pydantic模型。避免在State中直接存放复杂的数据库连接对象或函数。如果需要将其设为transient瞬态字段不参与持久化。6. 优化与进阶让记忆更智能、更高效基础版本跑通后我们面临两个现实问题1. 记忆越来越多每次全量加载和检索效率低下。2. 简单的最近一次记忆检索可能遗漏更早但更相关的信息。优化一记忆摘要与滚动压缩我们不能让memories列表无限增长。一个常见的策略是定期进行记忆摘要。例如每存储10次交互记忆后触发一个LLM调用将这10条记忆压缩成1-2条高度概括的摘要并计算一个平均情感值然后替换掉原来的详细列表。这类似于人类记忆的“模糊化”和“要点化”过程。可以在update_long_term_memory节点中添加检查逻辑def should_compress(memory_list: List[InteractionMemory]) - bool: return len(memory_list) 10 def compress_memories(memory_list: List[InteractionMemory]) - InteractionMemory: # 调用LLM生成摘要 summaries “\n”.join([f”- {m.summary} (好感变化: {m.sentiment_delta})” for m in memory_list]) compression_prompt f”请将以下多次交互摘要融合成一条概括性的长期记忆描述并估算一个整体的好感度变化趋势值-1到1之间\n{summaries}” # llm_response call_llm(compression_prompt) # 解析llm_response生成新的CompressedMemory # ... return compressed_memory优化二基于向量的相关性检索当记忆条数较多时我们需要根据当前对话的“上下文”或“意图”去记忆中寻找最相关的内容而不是仅仅取最近的一条。这需要引入向量数据库。存储时在创建InteractionMemory的summary字段时同时用文本嵌入模型如text-embedding-3-small为其生成向量嵌入embedding并将(embedding, memory)存入向量数据库如PGVector, Chroma并关联speaker_id和listener_id。检索时将当前用户的问题或对话上下文也转化为向量然后在向量数据库中执行相似度搜索查询与当前说话者相关的、最相关的N条记忆。# 伪代码示例 from langchain_community.vectorstores import PGVector from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(model“text-embedding-3-small”) # 存储记忆 vector_store.add_texts( texts[memory.summary], metadatas[{“speaker”: speaker_id, “listener”: listener_id, “memory_id”: memory.id}] ) # 检索相关记忆 query “当前对话的上下文” relevant_docs vector_store.similarity_search_with_score( query, filter{“speaker”: current_speaker_id}, k3 )这样AI就能根据“聊天的内容”回忆起“相关的事情”而不是“最近的事情”记忆的运用变得更加智能。7. 避坑指南与性能考量在实践过程中我遇到了不少坑这里总结一下希望能帮你绕过去状态爆炸这是最大的风险。无限制地存储原始对话或过多细节会导致状态对象巨大序列化/反序列化慢数据库压力大。务必坚持“摘要化”原则并设定记忆条数的上限或自动压缩策略。LLM调用成本与延迟无论是生成对话、摘要记忆还是压缩记忆都需要调用LLM。这会产生API成本并增加响应延迟。策略对非实时任务如记忆压缩使用异步队列如Celery后台处理。对于对话生成可以考虑使用更小、更快的模型来处理基于记忆的上下文构建而只让大模型做最终的内容润色。记忆冲突与一致性在多Agent并发修改同一份全局状态如两个Agent同时更新对彼此的好感度时可能产生竞态条件。策略LangGraph的状态更新在单次图运行中是顺序的但对于并发的用户请求需要通过数据库的事务机制或乐观锁例如使用状态版本号来保证一致性。TRAE Work的框架通常提供了会话隔离机制要理解其原理并正确配置。冷启动问题新用户或新Agent没有任何记忆如何引导对话策略准备一个高质量的默认AgentProfile模板包含基础的身份设定。在retrieve_memory节点如果发现没有记忆可以转向一个“引导对话”的节点主动询问用户信息来创建初始记忆。调试困难当对话行为不符合预期时是记忆检索错了还是LLM生成错了策略在开发阶段务必在关键节点如检索后、更新前将状态内容打印到日志或保存下来。LangGraph也提供了可视化工具来查看图的执行路径和状态变化善用它们。性能方面主要关注点数据库IO每次会话恢复都需要加载整个状态。对于非常大的状态可以考虑只加载活跃Agent的档案或者采用更细粒度的存储策略如每个Agent的档案单独存一张表。上下文长度虽然我们存储的是摘要但当一个Agent与很多人都有交互记忆时把所有相关记忆的摘要都拼接到提示词中仍可能超长。解决方案使用向量检索只选取最相关的几条或者在拼接前对摘要进行进一步的概括和裁剪。8. 效果评估与未来展望实现之后效果是立竿见影的。在我的AI小镇里居民们开始展现出“连续性”。一个昨天因为分享食物而获得好感的居民今天见面时会更友好曾经争吵过的两位再次相遇时气氛会略显尴尬。虽然只是简单的数值和文本摘要驱动但整个模拟世界的“真实感”和“沉浸感”得到了质的提升。这套系统的价值远不止于游戏。想象一下这些场景个性化客服助手记住用户上次咨询的产品型号、遇到的故障细节以及解决进度无需用户重复描述。持续学习的学习伴侣记住学生各个知识点的掌握程度、易错题型制定个性化的复习计划。项目协作AI记住项目的完整上下文、之前的决策理由、各个成员的分工和进度成为团队永不遗忘的“第二大脑”。当然目前的实现还是一个相对基础的版本。未来的优化方向有很多记忆的主动遗忘与强化引入类似“艾宾浩斯遗忘曲线”的机制不重要的记忆随时间衰减重要的记忆高情感波动、多次提及被强化。记忆的推理与连接让AI不仅能存储和检索记忆还能在不同记忆之间建立联系进行简单的推理。例如“A和B是朋友B和C吵过架那么A对C可能持谨慎态度”。多模态记忆结合图像、音频识别让AI不仅能记住文字对话还能记住用户上传的图片风格、语音语调偏好等。这次基于TRAE Work和LangGraph的实践让我深刻感受到为AI赋予长期记忆不再是实验室里的概念而是有成熟工具链支持、可以快速落地的工程实践。其核心在于将记忆视为可持久化的、结构化的应用程序状态并用工作流引擎LangGraph来管理它的读写生命周期。如果你也在为AI的“金鱼记忆”而烦恼不妨就从定义一个最简单的user_profile.md文件开始迈出让AI“记住”你的第一步。你会发现一旦记忆的齿轮开始转动你与AI的协作将进入一个全新的、更富有深度的阶段。