1. 从“健忘”到“长情”为什么AI应用需要持久记忆如果你尝试过和早期的聊天机器人对话或者用过一些基础的AI助手一个最直观的感受可能就是它记性太差了。你刚告诉它“我叫张三喜欢打篮球”下一句问“我有什么爱好”它可能就答不上来了。这种“金鱼式”的七秒记忆极大地限制了AI应用的实用性和用户体验。它让每一次对话都像是初次见面无法构建连贯的、有深度的交互。这就是“持久记忆”要解决的核心痛点。它不是一个炫技的功能而是让AI应用从“玩具”走向“工具”的关键一步。想象一下一个客服机器人能记住用户上次反馈的问题和解决进度一个个人助理能记住你的日程偏好和饮食习惯一个创作工具能记住你整个项目的风格设定和人物关系。这种连续性才是智能的体现。在之前的教程中我们构建的AI应用大多是“无状态”的。每次调用模型我们传入当前的用户输入和历史对话记录通常以消息列表的形式模型基于这些信息生成回复。一旦对话结束这些上下文就消失了。下次再开启新对话一切归零。要实现持久记忆本质上就是要解决两个问题1. 记住什么2. 怎么记住“记住什么”涉及记忆的粒度。是记住整个对话的原始文本还是提炼出关键实体如人名、地点、事件和关系或者是总结出用户的长期偏好和意图不同的应用场景需要不同颗粒度的记忆。“怎么记住”则涉及技术实现。简单的方法是把所有历史记录都存进数据库下次对话时全部加载进来。但这样效率低下且很快就会超出模型能处理的上下文长度限制Token限制。因此我们需要更智能的记忆管理机制能存储、能检索、能更新、能遗忘清理不重要的信息。LangGraph作为LangChain框架中用于构建有状态、多环节工作流的利器为我们实现这种智能的、可持久化的记忆系统提供了完美的架构。它不再将AI应用视为一次性的函数调用而是一个可以长期运行、内部状态不断演进的“智能体”。状态State在LangGraph中是一个核心概念而记忆本质上就是状态的一部分。通过将记忆结构化为状态对象并利用LangGraph的持久化存储后端我们就能轻松实现跨越多次对话会话的记忆留存。接下来的内容我将带你深入LangGraph的内部从零开始构建一个具备持久记忆功能的AI对话助手。我们会从最基础的状态定义讲起逐步深入到复杂的记忆检索与更新策略并分享我在实际项目中踩过的坑和总结的最佳实践。目标是让你在15天内不仅能“学会”更能“掌握”如何为你的AI应用注入“长情”的灵魂。2. 理解LangGraph的状态State模型记忆的容器在动手写代码之前我们必须先吃透LangGraph的核心抽象——状态State。这是实现持久记忆的基石。你可以把State想象成你应用的“大脑”它记录了当前智能体所知道的一切。而记忆就是存储在这个大脑中的长期知识。2.1 State的本质一个可演化的字典在LangGraph中State通常是一个Python字典dict或者Pydantic模型。这个字典的键Key定义了智能体有哪些“认知维度”值Value则是这些维度当前的具体内容。一个典型的对话智能体State可能长这样{ messages: [HumanMessage(content你好我叫李雷。), AIMessage(content你好李雷很高兴认识你)], user_name: 李雷, conversation_summary: 用户进行了自我介绍。, user_interests: [科技, 音乐], # ... 其他自定义字段 }这里messages键存储了原始的对话消息流这是LangChain生态的标准做法。而user_name,conversation_summary,user_interests就是我们为持久记忆自定义的字段。LangGraph的工作流Graph由多个节点Node组成每个节点都是一个函数它们接收当前的State进行一些操作比如调用大模型、查询数据库然后返回一个更新后的State字典。LangGraph框架会自动将这个返回的字典与旧的State合并推动状态向前演进。关键理解State的更新是“增量式”的。你不需要在节点函数中返回完整的State只需要返回你想要修改的那部分。例如一个节点只更新了user_interests那么它只需返回{user_interests: [科技, 音乐, 摄影]}框架会自动将其合并到总State中。这非常符合记忆“逐步积累”的特性。2.2 定义State从简单到复杂在代码中我们如何定义State呢LangGraph推荐使用Pydantic的BaseModel因为它能提供清晰的类型提示和验证。我们从最简单的只记录消息的State开始from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph.message import add_messages import operator class BasicState(TypedDict): messages: Annotated[List, add_messages]这里用到了一个高级特性Annotated和add_messages。Annotated[List, add_messages]告诉LangGraphmessages字段是一个列表但它的合并逻辑不是简单的覆盖而是由add_messages函数决定。add_messages是LangGraph提供的一个归约器Reducer它会将新旧消息列表智能地拼接在一起确保对话流是连贯的。这是实现基础对话记忆的关键。但对于持久记忆这还不够。我们需要扩展这个State加入自定义的记忆字段。让我们定义一个更实用的MemoryStatefrom pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class MemoryState(BaseModel): # 核心对话流 messages: List Field(default_factorylist, description对话消息历史) # 持久记忆字段 user_profile: dict Field(default_factorydict, description用户档案如姓名、偏好等) conversation_history_summary: str Field(default, description本轮对话的摘要) long_term_memory: List[str] Field(default_factorylist, description长期记忆存储提炼的关键事实) memory_retrieved: List[str] Field(default_factorylist, description本次交互中检索到的相关记忆) last_updated: Optional[datetime] Field(defaultNone, description记忆最后更新时间)在这个State定义中我们区分了几种不同类型的记忆messages: 原始对话记录是短期、高保真的上下文。user_profile: 结构化档案适合存储稳定信息如用户名、邮箱、系统设置。long_term_memory: 非结构化或半结构化的长期记忆池。我们可能把从对话中提取的关键事实例如“用户有一只叫‘豆包’的猫”以文本片段形式存入这里。conversation_history_summary: 对话摘要。这是一种压缩记忆的技术用于在上下文窗口有限时用一段简短的文字概括之前的对话替代冗长的原始消息。memory_retrieved: 这是一个“工作记忆”。在每次对话轮次中我们从long_term_memory里检索出与当前问题相关的片段临时放在这里供大模型参考。这避免了把全部记忆都塞进上下文。2.3 状态持久化让记忆跨越会话定义了State结构如何让它持久化呢这就是LangGraph的Checkpointer发挥作用的地方。Checkpointer是一个存储后端负责将State序列化后保存到数据库、文件系统或内存中。每次工作流运行时都有一个唯一的thread_id标识会话。通过thread_id我们可以加载或保存对应会话的完整State。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 创建一个SQLite持久化存储 conn sqlite3.connect(:memory:) # 实际项目可改为文件路径如 ./memory.db checkpointer SqliteSaver(conn) # 在创建图时传入 checkpointer graph workflow.compile(checkpointercheckpointer)配置好Checkpointer后LangGraph会在每个节点执行后自动保存State。当你下次用相同的thread_id调用这个图时它会从上次中断的地方恢复整个State包括所有自定义的记忆字段。这就实现了真正的持久记忆——即使服务器重启应用关闭记忆依然存在。实操心得Checkpointer的选择SqliteSaver轻量简单适合原型开发和中小型应用。生产环境可以考虑更强大的后端如PostgresSaver或MongoDBSaver。MemorySaver仅用于测试进程退出后记忆会丢失。选择时需考虑数据的查询需求如果你需要能根据记忆内容进行复杂检索例如“找出所有提到‘项目截止日期’的记忆”那么支持索引的关系型数据库Postgres或文档数据库MongoDB是更好的选择。3. 构建记忆的完整生命周期存储、检索与更新有了State这个容器接下来我们要设计记忆是如何被写入、读取和管理的。一个健壮的记忆系统不应该只是简单的“追加日志”而应该像人脑一样有选择地记住重要的东西并在需要时快速回想起来。3.1 记忆的存储从对话中提炼关键信息不是所有对话内容都值得存入长期记忆。我们需要一个“记忆加工”节点。这个节点的任务是分析最新的对话内容判断是否有需要长期保存的新事实并将其格式化后存入long_term_memory。假设我们的智能体是一个个人生活助手。我们可以设计一个update_memory节点函数from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def extract_memory_facts(state: MemoryState) - dict: 分析最新对话提取需要长期记忆的事实。 messages state.messages if len(messages) 2: return {} # 没有足够的新对话 # 获取最近一轮的对话用户最新消息和AI的回复 latest_interaction messages[-2:] # 假设最后两条是最近的一问一答 # 或者更稳健的方法找到最后一条HumanMessage及其后的AIMessage # 构建提示词让LLM进行信息提取 prompt f 请分析以下对话内容提取出关于用户的、需要被长期记住的客观事实或明确偏好。 只提取新出现的、重要的信息。如果对话只是寒暄或没有新事实则输出“无”。 对话内容 {latest_interaction} 请以简洁的陈述句列出事实每条事实用‘- ’开头。例如 - 用户的名字叫李雷。 - 用户有一只名叫豆包的猫。 - 用户对人工智能编程感兴趣。 response llm.invoke(prompt) new_facts_text response.content.strip() new_facts [] if new_facts_text and new_facts_text ! 无: # 按行分割清理空行和‘- ’前缀 new_facts [line.strip(- ).strip() for line in new_facts_text.split(\n) if line.strip()] # 返回要更新的State部分 if new_facts: updated_memory state.long_term_memory new_facts return {long_term_memory: updated_memory, last_updated: datetime.now()} else: return {last_updated: datetime.now()} # 即使没新事实也更新时间戳这个函数的核心是使用一个大语言模型LLM作为“记忆筛选器”。我们将最近的对话片段交给它让它判断并提取关键事实。这样做的好处是灵活且语义理解能力强但代价是增加了LLM调用开销和延迟。在实际项目中这是一个需要权衡的点。对于高频、对延迟敏感的应用可以设计更简单的基于规则如关键词匹配的提取器。3.2 记忆的检索在需要时精准回想当用户提出一个新问题时我们需要从庞大的long_term_memory中找出相关的记忆来辅助回答。这就是检索Retrieval。最常用的方法是向量检索。向量化记忆将每一条long_term_memory中的文本片段例如“用户有一只名叫豆包的猫”通过嵌入模型Embedding Model转换为一个高维向量Vector。向量化问题同样将用户的当前问题转换为向量。相似度计算计算问题向量与所有记忆向量之间的余弦相似度。返回最相关的记忆选取相似度最高的Top-K条记忆作为本次推理的上下文。我们需要在State初始化或更新记忆时就为每条记忆生成向量并存储。这里引入一个向量数据库如Chroma、FAISS来管理它们会更高效。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document # 初始化嵌入模型和向量库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def retrieve_related_memory(state: MemoryState) - dict: 从长期记忆中检索与当前对话相关的片段。 if not state.long_term_memory: return {memory_retrieved: []} # 获取最新的用户问题 latest_user_message None for msg in reversed(state.messages): if isinstance(msg, HumanMessage): latest_user_message msg.content break if not latest_user_message: return {memory_retrieved: []} # 使用向量库进行相似度检索 # 注意我们需要确保long_term_memory的内容已经添加到vectorstore中。 # 这通常在update_memory节点中同步完成。 docs vectorstore.similarity_search(latest_user_message, k3) # 检索最相关的3条 retrieved_texts [doc.page_content for doc in docs] return {memory_retrieved: retrieved_texts}这里有一个关键的设计决策记忆的向量化存储和检索是独立于LangGraph State的。State里的long_term_memory存储原始文本用于管理和展示向量数据库存储对应的向量用于快速检索。两者通过索引或ID关联。在update_memory节点中每当新增一条记忆我们既要把它追加到State的列表里也要将其添加到向量数据库中。3.3 记忆的更新与整合避免冗余和冲突记忆不是只增不减的。随着时间的推移记忆库可能变得臃肿包含重复或矛盾的信息例如用户先说“喜欢蓝色”后来说“最喜欢绿色”。我们需要一个“记忆整理”机制。一种策略是在每次存储新记忆前先进行去重和冲突检测。这可以放在update_memory节点中在调用LLM提取事实后、保存前进行。def consolidate_memory(current_memory: List[str], new_facts: List[str]) - List[str]: 整合新旧记忆去重并解决简单冲突。 consolidated current_memory.copy() for new_fact in new_facts: # 简单的基于嵌入相似度的去重此处简化实际需调用嵌入模型计算 # 假设我们有一个函数 is_similar 来判断两条记忆是否语义相似 is_duplicate False for old_fact in consolidated: if is_similar(new_fact, old_fact): # 这是一个需要实现的函数 is_duplicate True # 可以选择用新的表述替换旧的或者直接忽略 break if not is_duplicate: consolidated.append(new_fact) # 可选定期清理非常陈旧的或不重要的记忆 # 这需要为每条记忆附加元数据如创建时间、访问频率、重要性评分 return consolidated更高级的策略是定期运行一个后台整理任务使用LLM对记忆库进行总结、去重和结构化。例如将几十条关于用户喜好的零散记忆合并成一条结构化的记录{favorite_colors: [绿色, 蓝色], pets: [{name: 豆包, type: 猫}]}。这能极大地提升检索效率和记忆质量。4. 将记忆整合进LangGraph工作流设计对话智能体现在我们将上述所有组件组装到一个完整的LangGraph工作流中。我们的目标是创建一个具备持久记忆的对话智能体它的工作流程如下接收用户输入。检索相关记忆根据用户输入从持久化存储中召回相关的长期记忆。生成回复结合当前对话上下文messages和检索到的记忆memory_retrieved调用大模型生成回复。更新记忆分析本轮对话提取新事实并存入长期记忆。保存状态将更新后的所有状态包括新的messages和long_term_memory持久化。4.1 定义工作流节点我们将创建三个主要的节点函数和一个条件边决定是否更新记忆。from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 初始化模型和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def retrieve_node(state: MemoryState): 检索节点从长期记忆中查找相关信息。 retrieved retrieve_related_memory(state) # 调用前面定义的检索函数 return retrieved def generate_response_node(state: MemoryState): 生成回复节点结合记忆和上下文调用LLM生成回复。 # 1. 准备系统提示词注入检索到的记忆 system_prompt f 你是一个有帮助的、且拥有记忆的AI助手。 以下是关于当前用户的已知信息长期记忆请在你的回答中自然、恰当地运用这些信息 {chr(10).join(state.memory_retrieved) if state.memory_retrieved else 暂无相关长期记忆。} 请基于对话历史和以上记忆进行回复。如果记忆信息与当前问题无关则忽略它。 # 2. 构建完整的消息列表系统提示 历史对话 # LangGraph的 add_messages Reducer已经帮我们管理了state.messages # 这里我们需要在调用模型前将系统提示插入到消息列表的最前面。 # 注意为了避免在每次循环中都重复添加系统消息一种更好的方法是在图初始化时设置。 # 这里我们采用动态构建的方式。 messages_for_llm [SystemMessage(contentsystem_prompt)] state.messages # 3. 调用LLM response llm.invoke(messages_for_llm) # 4. 将AI的回复添加到消息流中 # 注意我们不需要手动修改state.messagesLangGraph的add_messages机制会在节点返回后处理。 # 我们只需要返回一个包含新AIMessage的字典。 return {messages: [AIMessage(contentresponse.content)]} def update_memory_node(state: MemoryState): 更新记忆节点分析对话提取并存储新记忆。 # 调用前面定义的记忆提取函数 memory_updates extract_memory_facts(state) # 如果提取到了新事实还需要同步到向量数据库 if long_term_memory in memory_updates and memory_updates[long_term_memory] ! state.long_term_memory: new_facts memory_updates[long_term_memory][len(state.long_term_memory):] # 获取新增的部分 for fact in new_facts: # 将新记忆文本添加到向量库 vectorstore.add_texts(texts[fact], metadatas[{source: conversation}]) return memory_updates def should_update_memory(state: MemoryState) - str: 条件判断函数决定是否走更新记忆的路径。 # 简单的策略只有在用户发送了消息并且AI已经回复后才考虑更新记忆。 # 更复杂的策略可以分析消息内容例如只有包含实质性信息的对话才触发记忆更新。 if len(state.messages) 0 and isinstance(state.messages[-1], AIMessage): # 检查上一条用户消息是否可能包含新信息这里简化处理总是返回True return update return end4.2 构建并编译图# 创建状态图指定我们自定义的State类型这里使用TypedDict简化示例 from typing import TypedDict, List from langgraph.graph import add_messages class GraphState(TypedDict): messages: Annotated[List, add_messages] memory_retrieved: List[str] long_term_memory: List[str] # 初始化图 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_response_node) workflow.add_node(update_memory, update_memory_node) # 设置入口点 workflow.set_entry_point(retrieve) # 定义边 workflow.add_edge(retrieve, generate) workflow.add_conditional_edges( generate, should_update_memory, # 条件判断函数 { update: update_memory, # 如果返回update则前往更新记忆节点 end: END # 否则结束 } ) workflow.add_edge(update_memory, END) # 更新记忆后结束本轮 # 编译图并启用持久化检查点 graph workflow.compile(checkpointercheckpointer)4.3 运行与测试现在我们可以运行这个具备记忆的智能体了。# 初始化一个会话线程 thread_id user_123_session_1 config {configurable: {thread_id: thread_id}} # 第一轮对话 initial_state {messages: [HumanMessage(content你好我是王小明。我喜欢打篮球和编程。)]} result graph.invoke(initial_state, configconfig) print(AI:, result[messages][-1].content) # 输出AI的欢迎回复其中应能体现“王小明”、“篮球”、“编程”等记忆。 # 第二轮对话在相同thread_id下记忆会被保留 new_state {messages: [HumanMessage(content你还记得我的爱好是什么吗)]} result2 graph.invoke(new_state, configconfig) # 注意这里传入的是增量state print(AI:, result2[messages][-1].content) # AI应该能回答出篮球和编程。 # 同时在result2的state中你应该能看到memory_retrieved字段里包含了之前存储的相关记忆。通过这个流程智能体在第二次对话时retrieve_node会从向量数据库中检索到与“爱好”相关的记忆“我喜欢打篮球和编程”并将其注入到生成回复的上下文中从而实现“记得”的效果。5. 实战中的挑战与优化策略在实际项目中直接使用上述基础架构可能会遇到不少问题。下面分享几个我踩过的坑以及对应的优化思路。5.1 记忆的准确性与幻觉问题问题LLM在提取记忆事实时可能产生幻觉将不存在的信息或错误的理解存入记忆。例如用户说“我昨天看了电影《沙丘》”LLM可能错误提取为“用户喜欢科幻电影《沙丘》”。一旦这个错误记忆被存入它会在未来的检索中被当作事实使用导致后续对话出现连环错误。解决方案提高提取提示词的质量在extract_memory_facts的提示词中严格要求“只提取用户明确陈述的客观事实”并使用少样本示例Few-shot引导LLM。引入置信度评分让LLM为提取的每条事实输出一个置信度分数例如0-1。只存储高置信度如0.8的事实。低置信度的可以暂时放入一个“待确认”缓冲区在后续对话中通过主动提问来确认例如“你刚才提到喜欢《沙丘》是指你喜欢这部电影吗”。定期人工审核与清理对于关键应用设计后台界面允许管理员查看和修正AI自动生成的记忆。5.2 记忆检索的精准度与效率问题1检索不准。简单的向量相似度检索可能召回不相关的记忆。比如用户问“推荐晚餐吃什么”可能召回“用户有一只猫”这条记忆因为“吃”和“猫粮”在向量空间可能接近。问题2记忆爆炸。长期运行后long_term_memory列表和向量数据库会变得非常大导致检索变慢且无关记忆干扰度增加。优化策略分层记忆结构不要把所有记忆都混在一个池子里。可以按主题、类型或时间分桶。例如分为facts_personal个人信息、facts_preferences偏好、facts_conversational对话事件。检索时先根据当前问题判断应该检索哪个或哪几个记忆桶。记忆摘要与压缩定期例如每100条对话后运行一个记忆总结任务。使用LLM将大量零散记忆总结成几条高度凝练的陈述。例如将“喜欢咖啡”、“讨厌茶”、“每天喝三杯咖啡”总结为“用户有重度咖啡依赖不喝茶”。原始细节可以归档日常检索主要使用摘要这样可以大幅减少检索量并提升相关性。元数据过滤为每条记忆附加丰富的元数据如category类别、created_at创建时间、access_count访问次数、importance_score重要性分数可由LLM初步生成。检索时除了向量相似度还可以结合元数据进行过滤和排序。例如优先检索category匹配且importance_score高的记忆。5.3 状态管理的复杂性与调试问题随着自定义State字段增多节点函数需要处理的逻辑越来越复杂。State的结构变更如新增一个记忆字段可能影响多个节点。调试一个持久化的、多轮次的工作流状态也变得困难。最佳实践保持State的扁平与清晰避免过深的嵌套结构。使用明确的字段名。可以为不同的功能模块使用子状态但需权衡清晰度和复杂度。为State变化添加日志在每个节点函数的开始和结束打印或记录State关键字段的变化。LangGraph本身也提供了一些调试工具。这对于追踪记忆是如何被创建、检索和使用的至关重要。版本化记忆模式如果记忆结构MemoryState的字段需要升级旧版本Checkpointer保存的状态可能无法加载。在设计之初就要考虑向后兼容或者提供状态迁移脚本。一种方法是使用灵活的存储格式如JSON并在加载时进行模式验证和填充默认值。5.4 成本与延迟控制问题每次对话都进行记忆提取和向量检索意味着额外的LLM调用和嵌入模型调用增加了成本和响应延迟。优化策略懒更新与批处理不必每轮对话都触发update_memory_node。可以设定一个阈值例如当对话轮次积累到5轮或者检测到对话包含明显的事实性陈述时才触发记忆更新。should_update_memory函数可以实现更精细的逻辑。缓存嵌入向量对于long_term_memory中的文本其嵌入向量是固定的。确保它们被计算一次后存储在向量数据库中避免重复计算。异步处理对于记忆更新、总结整理等非实时关键任务可以将其放入后台异步队列执行不阻塞主对话流程。主流程只进行快速的记忆检索。实现持久记忆是构建高级AI应用的必经之路而LangGraph提供了强大且灵活的状态管理框架来支撑这一目标。从理解State模型开始到设计记忆的存储、检索、更新生命周期再到将其整合进一个健壮的工作流每一步都需要仔细权衡。记住没有“最好”的记忆系统只有最适合你应用场景的设计。从简单的键值对记忆开始逐步迭代加入向量检索、记忆摘要、冲突解决等高级特性你的AI应用将真正拥有“理解过去、服务现在”的能力。 LangGraph实战：为AI应用构建持久记忆系统的完整指南
1. 从“健忘”到“长情”为什么AI应用需要持久记忆如果你尝试过和早期的聊天机器人对话或者用过一些基础的AI助手一个最直观的感受可能就是它记性太差了。你刚告诉它“我叫张三喜欢打篮球”下一句问“我有什么爱好”它可能就答不上来了。这种“金鱼式”的七秒记忆极大地限制了AI应用的实用性和用户体验。它让每一次对话都像是初次见面无法构建连贯的、有深度的交互。这就是“持久记忆”要解决的核心痛点。它不是一个炫技的功能而是让AI应用从“玩具”走向“工具”的关键一步。想象一下一个客服机器人能记住用户上次反馈的问题和解决进度一个个人助理能记住你的日程偏好和饮食习惯一个创作工具能记住你整个项目的风格设定和人物关系。这种连续性才是智能的体现。在之前的教程中我们构建的AI应用大多是“无状态”的。每次调用模型我们传入当前的用户输入和历史对话记录通常以消息列表的形式模型基于这些信息生成回复。一旦对话结束这些上下文就消失了。下次再开启新对话一切归零。要实现持久记忆本质上就是要解决两个问题1. 记住什么2. 怎么记住“记住什么”涉及记忆的粒度。是记住整个对话的原始文本还是提炼出关键实体如人名、地点、事件和关系或者是总结出用户的长期偏好和意图不同的应用场景需要不同颗粒度的记忆。“怎么记住”则涉及技术实现。简单的方法是把所有历史记录都存进数据库下次对话时全部加载进来。但这样效率低下且很快就会超出模型能处理的上下文长度限制Token限制。因此我们需要更智能的记忆管理机制能存储、能检索、能更新、能遗忘清理不重要的信息。LangGraph作为LangChain框架中用于构建有状态、多环节工作流的利器为我们实现这种智能的、可持久化的记忆系统提供了完美的架构。它不再将AI应用视为一次性的函数调用而是一个可以长期运行、内部状态不断演进的“智能体”。状态State在LangGraph中是一个核心概念而记忆本质上就是状态的一部分。通过将记忆结构化为状态对象并利用LangGraph的持久化存储后端我们就能轻松实现跨越多次对话会话的记忆留存。接下来的内容我将带你深入LangGraph的内部从零开始构建一个具备持久记忆功能的AI对话助手。我们会从最基础的状态定义讲起逐步深入到复杂的记忆检索与更新策略并分享我在实际项目中踩过的坑和总结的最佳实践。目标是让你在15天内不仅能“学会”更能“掌握”如何为你的AI应用注入“长情”的灵魂。2. 理解LangGraph的状态State模型记忆的容器在动手写代码之前我们必须先吃透LangGraph的核心抽象——状态State。这是实现持久记忆的基石。你可以把State想象成你应用的“大脑”它记录了当前智能体所知道的一切。而记忆就是存储在这个大脑中的长期知识。2.1 State的本质一个可演化的字典在LangGraph中State通常是一个Python字典dict或者Pydantic模型。这个字典的键Key定义了智能体有哪些“认知维度”值Value则是这些维度当前的具体内容。一个典型的对话智能体State可能长这样{ messages: [HumanMessage(content你好我叫李雷。), AIMessage(content你好李雷很高兴认识你)], user_name: 李雷, conversation_summary: 用户进行了自我介绍。, user_interests: [科技, 音乐], # ... 其他自定义字段 }这里messages键存储了原始的对话消息流这是LangChain生态的标准做法。而user_name,conversation_summary,user_interests就是我们为持久记忆自定义的字段。LangGraph的工作流Graph由多个节点Node组成每个节点都是一个函数它们接收当前的State进行一些操作比如调用大模型、查询数据库然后返回一个更新后的State字典。LangGraph框架会自动将这个返回的字典与旧的State合并推动状态向前演进。关键理解State的更新是“增量式”的。你不需要在节点函数中返回完整的State只需要返回你想要修改的那部分。例如一个节点只更新了user_interests那么它只需返回{user_interests: [科技, 音乐, 摄影]}框架会自动将其合并到总State中。这非常符合记忆“逐步积累”的特性。2.2 定义State从简单到复杂在代码中我们如何定义State呢LangGraph推荐使用Pydantic的BaseModel因为它能提供清晰的类型提示和验证。我们从最简单的只记录消息的State开始from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph.message import add_messages import operator class BasicState(TypedDict): messages: Annotated[List, add_messages]这里用到了一个高级特性Annotated和add_messages。Annotated[List, add_messages]告诉LangGraphmessages字段是一个列表但它的合并逻辑不是简单的覆盖而是由add_messages函数决定。add_messages是LangGraph提供的一个归约器Reducer它会将新旧消息列表智能地拼接在一起确保对话流是连贯的。这是实现基础对话记忆的关键。但对于持久记忆这还不够。我们需要扩展这个State加入自定义的记忆字段。让我们定义一个更实用的MemoryStatefrom pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class MemoryState(BaseModel): # 核心对话流 messages: List Field(default_factorylist, description对话消息历史) # 持久记忆字段 user_profile: dict Field(default_factorydict, description用户档案如姓名、偏好等) conversation_history_summary: str Field(default, description本轮对话的摘要) long_term_memory: List[str] Field(default_factorylist, description长期记忆存储提炼的关键事实) memory_retrieved: List[str] Field(default_factorylist, description本次交互中检索到的相关记忆) last_updated: Optional[datetime] Field(defaultNone, description记忆最后更新时间)在这个State定义中我们区分了几种不同类型的记忆messages: 原始对话记录是短期、高保真的上下文。user_profile: 结构化档案适合存储稳定信息如用户名、邮箱、系统设置。long_term_memory: 非结构化或半结构化的长期记忆池。我们可能把从对话中提取的关键事实例如“用户有一只叫‘豆包’的猫”以文本片段形式存入这里。conversation_history_summary: 对话摘要。这是一种压缩记忆的技术用于在上下文窗口有限时用一段简短的文字概括之前的对话替代冗长的原始消息。memory_retrieved: 这是一个“工作记忆”。在每次对话轮次中我们从long_term_memory里检索出与当前问题相关的片段临时放在这里供大模型参考。这避免了把全部记忆都塞进上下文。2.3 状态持久化让记忆跨越会话定义了State结构如何让它持久化呢这就是LangGraph的Checkpointer发挥作用的地方。Checkpointer是一个存储后端负责将State序列化后保存到数据库、文件系统或内存中。每次工作流运行时都有一个唯一的thread_id标识会话。通过thread_id我们可以加载或保存对应会话的完整State。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 创建一个SQLite持久化存储 conn sqlite3.connect(:memory:) # 实际项目可改为文件路径如 ./memory.db checkpointer SqliteSaver(conn) # 在创建图时传入 checkpointer graph workflow.compile(checkpointercheckpointer)配置好Checkpointer后LangGraph会在每个节点执行后自动保存State。当你下次用相同的thread_id调用这个图时它会从上次中断的地方恢复整个State包括所有自定义的记忆字段。这就实现了真正的持久记忆——即使服务器重启应用关闭记忆依然存在。实操心得Checkpointer的选择SqliteSaver轻量简单适合原型开发和中小型应用。生产环境可以考虑更强大的后端如PostgresSaver或MongoDBSaver。MemorySaver仅用于测试进程退出后记忆会丢失。选择时需考虑数据的查询需求如果你需要能根据记忆内容进行复杂检索例如“找出所有提到‘项目截止日期’的记忆”那么支持索引的关系型数据库Postgres或文档数据库MongoDB是更好的选择。3. 构建记忆的完整生命周期存储、检索与更新有了State这个容器接下来我们要设计记忆是如何被写入、读取和管理的。一个健壮的记忆系统不应该只是简单的“追加日志”而应该像人脑一样有选择地记住重要的东西并在需要时快速回想起来。3.1 记忆的存储从对话中提炼关键信息不是所有对话内容都值得存入长期记忆。我们需要一个“记忆加工”节点。这个节点的任务是分析最新的对话内容判断是否有需要长期保存的新事实并将其格式化后存入long_term_memory。假设我们的智能体是一个个人生活助手。我们可以设计一个update_memory节点函数from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def extract_memory_facts(state: MemoryState) - dict: 分析最新对话提取需要长期记忆的事实。 messages state.messages if len(messages) 2: return {} # 没有足够的新对话 # 获取最近一轮的对话用户最新消息和AI的回复 latest_interaction messages[-2:] # 假设最后两条是最近的一问一答 # 或者更稳健的方法找到最后一条HumanMessage及其后的AIMessage # 构建提示词让LLM进行信息提取 prompt f 请分析以下对话内容提取出关于用户的、需要被长期记住的客观事实或明确偏好。 只提取新出现的、重要的信息。如果对话只是寒暄或没有新事实则输出“无”。 对话内容 {latest_interaction} 请以简洁的陈述句列出事实每条事实用‘- ’开头。例如 - 用户的名字叫李雷。 - 用户有一只名叫豆包的猫。 - 用户对人工智能编程感兴趣。 response llm.invoke(prompt) new_facts_text response.content.strip() new_facts [] if new_facts_text and new_facts_text ! 无: # 按行分割清理空行和‘- ’前缀 new_facts [line.strip(- ).strip() for line in new_facts_text.split(\n) if line.strip()] # 返回要更新的State部分 if new_facts: updated_memory state.long_term_memory new_facts return {long_term_memory: updated_memory, last_updated: datetime.now()} else: return {last_updated: datetime.now()} # 即使没新事实也更新时间戳这个函数的核心是使用一个大语言模型LLM作为“记忆筛选器”。我们将最近的对话片段交给它让它判断并提取关键事实。这样做的好处是灵活且语义理解能力强但代价是增加了LLM调用开销和延迟。在实际项目中这是一个需要权衡的点。对于高频、对延迟敏感的应用可以设计更简单的基于规则如关键词匹配的提取器。3.2 记忆的检索在需要时精准回想当用户提出一个新问题时我们需要从庞大的long_term_memory中找出相关的记忆来辅助回答。这就是检索Retrieval。最常用的方法是向量检索。向量化记忆将每一条long_term_memory中的文本片段例如“用户有一只名叫豆包的猫”通过嵌入模型Embedding Model转换为一个高维向量Vector。向量化问题同样将用户的当前问题转换为向量。相似度计算计算问题向量与所有记忆向量之间的余弦相似度。返回最相关的记忆选取相似度最高的Top-K条记忆作为本次推理的上下文。我们需要在State初始化或更新记忆时就为每条记忆生成向量并存储。这里引入一个向量数据库如Chroma、FAISS来管理它们会更高效。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document # 初始化嵌入模型和向量库 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def retrieve_related_memory(state: MemoryState) - dict: 从长期记忆中检索与当前对话相关的片段。 if not state.long_term_memory: return {memory_retrieved: []} # 获取最新的用户问题 latest_user_message None for msg in reversed(state.messages): if isinstance(msg, HumanMessage): latest_user_message msg.content break if not latest_user_message: return {memory_retrieved: []} # 使用向量库进行相似度检索 # 注意我们需要确保long_term_memory的内容已经添加到vectorstore中。 # 这通常在update_memory节点中同步完成。 docs vectorstore.similarity_search(latest_user_message, k3) # 检索最相关的3条 retrieved_texts [doc.page_content for doc in docs] return {memory_retrieved: retrieved_texts}这里有一个关键的设计决策记忆的向量化存储和检索是独立于LangGraph State的。State里的long_term_memory存储原始文本用于管理和展示向量数据库存储对应的向量用于快速检索。两者通过索引或ID关联。在update_memory节点中每当新增一条记忆我们既要把它追加到State的列表里也要将其添加到向量数据库中。3.3 记忆的更新与整合避免冗余和冲突记忆不是只增不减的。随着时间的推移记忆库可能变得臃肿包含重复或矛盾的信息例如用户先说“喜欢蓝色”后来说“最喜欢绿色”。我们需要一个“记忆整理”机制。一种策略是在每次存储新记忆前先进行去重和冲突检测。这可以放在update_memory节点中在调用LLM提取事实后、保存前进行。def consolidate_memory(current_memory: List[str], new_facts: List[str]) - List[str]: 整合新旧记忆去重并解决简单冲突。 consolidated current_memory.copy() for new_fact in new_facts: # 简单的基于嵌入相似度的去重此处简化实际需调用嵌入模型计算 # 假设我们有一个函数 is_similar 来判断两条记忆是否语义相似 is_duplicate False for old_fact in consolidated: if is_similar(new_fact, old_fact): # 这是一个需要实现的函数 is_duplicate True # 可以选择用新的表述替换旧的或者直接忽略 break if not is_duplicate: consolidated.append(new_fact) # 可选定期清理非常陈旧的或不重要的记忆 # 这需要为每条记忆附加元数据如创建时间、访问频率、重要性评分 return consolidated更高级的策略是定期运行一个后台整理任务使用LLM对记忆库进行总结、去重和结构化。例如将几十条关于用户喜好的零散记忆合并成一条结构化的记录{favorite_colors: [绿色, 蓝色], pets: [{name: 豆包, type: 猫}]}。这能极大地提升检索效率和记忆质量。4. 将记忆整合进LangGraph工作流设计对话智能体现在我们将上述所有组件组装到一个完整的LangGraph工作流中。我们的目标是创建一个具备持久记忆的对话智能体它的工作流程如下接收用户输入。检索相关记忆根据用户输入从持久化存储中召回相关的长期记忆。生成回复结合当前对话上下文messages和检索到的记忆memory_retrieved调用大模型生成回复。更新记忆分析本轮对话提取新事实并存入长期记忆。保存状态将更新后的所有状态包括新的messages和long_term_memory持久化。4.1 定义工作流节点我们将创建三个主要的节点函数和一个条件边决定是否更新记忆。from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 初始化模型和工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) embeddings OpenAIEmbeddings() vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def retrieve_node(state: MemoryState): 检索节点从长期记忆中查找相关信息。 retrieved retrieve_related_memory(state) # 调用前面定义的检索函数 return retrieved def generate_response_node(state: MemoryState): 生成回复节点结合记忆和上下文调用LLM生成回复。 # 1. 准备系统提示词注入检索到的记忆 system_prompt f 你是一个有帮助的、且拥有记忆的AI助手。 以下是关于当前用户的已知信息长期记忆请在你的回答中自然、恰当地运用这些信息 {chr(10).join(state.memory_retrieved) if state.memory_retrieved else 暂无相关长期记忆。} 请基于对话历史和以上记忆进行回复。如果记忆信息与当前问题无关则忽略它。 # 2. 构建完整的消息列表系统提示 历史对话 # LangGraph的 add_messages Reducer已经帮我们管理了state.messages # 这里我们需要在调用模型前将系统提示插入到消息列表的最前面。 # 注意为了避免在每次循环中都重复添加系统消息一种更好的方法是在图初始化时设置。 # 这里我们采用动态构建的方式。 messages_for_llm [SystemMessage(contentsystem_prompt)] state.messages # 3. 调用LLM response llm.invoke(messages_for_llm) # 4. 将AI的回复添加到消息流中 # 注意我们不需要手动修改state.messagesLangGraph的add_messages机制会在节点返回后处理。 # 我们只需要返回一个包含新AIMessage的字典。 return {messages: [AIMessage(contentresponse.content)]} def update_memory_node(state: MemoryState): 更新记忆节点分析对话提取并存储新记忆。 # 调用前面定义的记忆提取函数 memory_updates extract_memory_facts(state) # 如果提取到了新事实还需要同步到向量数据库 if long_term_memory in memory_updates and memory_updates[long_term_memory] ! state.long_term_memory: new_facts memory_updates[long_term_memory][len(state.long_term_memory):] # 获取新增的部分 for fact in new_facts: # 将新记忆文本添加到向量库 vectorstore.add_texts(texts[fact], metadatas[{source: conversation}]) return memory_updates def should_update_memory(state: MemoryState) - str: 条件判断函数决定是否走更新记忆的路径。 # 简单的策略只有在用户发送了消息并且AI已经回复后才考虑更新记忆。 # 更复杂的策略可以分析消息内容例如只有包含实质性信息的对话才触发记忆更新。 if len(state.messages) 0 and isinstance(state.messages[-1], AIMessage): # 检查上一条用户消息是否可能包含新信息这里简化处理总是返回True return update return end4.2 构建并编译图# 创建状态图指定我们自定义的State类型这里使用TypedDict简化示例 from typing import TypedDict, List from langgraph.graph import add_messages class GraphState(TypedDict): messages: Annotated[List, add_messages] memory_retrieved: List[str] long_term_memory: List[str] # 初始化图 workflow StateGraph(GraphState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_response_node) workflow.add_node(update_memory, update_memory_node) # 设置入口点 workflow.set_entry_point(retrieve) # 定义边 workflow.add_edge(retrieve, generate) workflow.add_conditional_edges( generate, should_update_memory, # 条件判断函数 { update: update_memory, # 如果返回update则前往更新记忆节点 end: END # 否则结束 } ) workflow.add_edge(update_memory, END) # 更新记忆后结束本轮 # 编译图并启用持久化检查点 graph workflow.compile(checkpointercheckpointer)4.3 运行与测试现在我们可以运行这个具备记忆的智能体了。# 初始化一个会话线程 thread_id user_123_session_1 config {configurable: {thread_id: thread_id}} # 第一轮对话 initial_state {messages: [HumanMessage(content你好我是王小明。我喜欢打篮球和编程。)]} result graph.invoke(initial_state, configconfig) print(AI:, result[messages][-1].content) # 输出AI的欢迎回复其中应能体现“王小明”、“篮球”、“编程”等记忆。 # 第二轮对话在相同thread_id下记忆会被保留 new_state {messages: [HumanMessage(content你还记得我的爱好是什么吗)]} result2 graph.invoke(new_state, configconfig) # 注意这里传入的是增量state print(AI:, result2[messages][-1].content) # AI应该能回答出篮球和编程。 # 同时在result2的state中你应该能看到memory_retrieved字段里包含了之前存储的相关记忆。通过这个流程智能体在第二次对话时retrieve_node会从向量数据库中检索到与“爱好”相关的记忆“我喜欢打篮球和编程”并将其注入到生成回复的上下文中从而实现“记得”的效果。5. 实战中的挑战与优化策略在实际项目中直接使用上述基础架构可能会遇到不少问题。下面分享几个我踩过的坑以及对应的优化思路。5.1 记忆的准确性与幻觉问题问题LLM在提取记忆事实时可能产生幻觉将不存在的信息或错误的理解存入记忆。例如用户说“我昨天看了电影《沙丘》”LLM可能错误提取为“用户喜欢科幻电影《沙丘》”。一旦这个错误记忆被存入它会在未来的检索中被当作事实使用导致后续对话出现连环错误。解决方案提高提取提示词的质量在extract_memory_facts的提示词中严格要求“只提取用户明确陈述的客观事实”并使用少样本示例Few-shot引导LLM。引入置信度评分让LLM为提取的每条事实输出一个置信度分数例如0-1。只存储高置信度如0.8的事实。低置信度的可以暂时放入一个“待确认”缓冲区在后续对话中通过主动提问来确认例如“你刚才提到喜欢《沙丘》是指你喜欢这部电影吗”。定期人工审核与清理对于关键应用设计后台界面允许管理员查看和修正AI自动生成的记忆。5.2 记忆检索的精准度与效率问题1检索不准。简单的向量相似度检索可能召回不相关的记忆。比如用户问“推荐晚餐吃什么”可能召回“用户有一只猫”这条记忆因为“吃”和“猫粮”在向量空间可能接近。问题2记忆爆炸。长期运行后long_term_memory列表和向量数据库会变得非常大导致检索变慢且无关记忆干扰度增加。优化策略分层记忆结构不要把所有记忆都混在一个池子里。可以按主题、类型或时间分桶。例如分为facts_personal个人信息、facts_preferences偏好、facts_conversational对话事件。检索时先根据当前问题判断应该检索哪个或哪几个记忆桶。记忆摘要与压缩定期例如每100条对话后运行一个记忆总结任务。使用LLM将大量零散记忆总结成几条高度凝练的陈述。例如将“喜欢咖啡”、“讨厌茶”、“每天喝三杯咖啡”总结为“用户有重度咖啡依赖不喝茶”。原始细节可以归档日常检索主要使用摘要这样可以大幅减少检索量并提升相关性。元数据过滤为每条记忆附加丰富的元数据如category类别、created_at创建时间、access_count访问次数、importance_score重要性分数可由LLM初步生成。检索时除了向量相似度还可以结合元数据进行过滤和排序。例如优先检索category匹配且importance_score高的记忆。5.3 状态管理的复杂性与调试问题随着自定义State字段增多节点函数需要处理的逻辑越来越复杂。State的结构变更如新增一个记忆字段可能影响多个节点。调试一个持久化的、多轮次的工作流状态也变得困难。最佳实践保持State的扁平与清晰避免过深的嵌套结构。使用明确的字段名。可以为不同的功能模块使用子状态但需权衡清晰度和复杂度。为State变化添加日志在每个节点函数的开始和结束打印或记录State关键字段的变化。LangGraph本身也提供了一些调试工具。这对于追踪记忆是如何被创建、检索和使用的至关重要。版本化记忆模式如果记忆结构MemoryState的字段需要升级旧版本Checkpointer保存的状态可能无法加载。在设计之初就要考虑向后兼容或者提供状态迁移脚本。一种方法是使用灵活的存储格式如JSON并在加载时进行模式验证和填充默认值。5.4 成本与延迟控制问题每次对话都进行记忆提取和向量检索意味着额外的LLM调用和嵌入模型调用增加了成本和响应延迟。优化策略懒更新与批处理不必每轮对话都触发update_memory_node。可以设定一个阈值例如当对话轮次积累到5轮或者检测到对话包含明显的事实性陈述时才触发记忆更新。should_update_memory函数可以实现更精细的逻辑。缓存嵌入向量对于long_term_memory中的文本其嵌入向量是固定的。确保它们被计算一次后存储在向量数据库中避免重复计算。异步处理对于记忆更新、总结整理等非实时关键任务可以将其放入后台异步队列执行不阻塞主对话流程。主流程只进行快速的记忆检索。实现持久记忆是构建高级AI应用的必经之路而LangGraph提供了强大且灵活的状态管理框架来支撑这一目标。从理解State模型开始到设计记忆的存储、检索、更新生命周期再到将其整合进一个健壮的工作流每一步都需要仔细权衡。记住没有“最好”的记忆系统只有最适合你应用场景的设计。从简单的键值对记忆开始逐步迭代加入向量检索、记忆摘要、冲突解决等高级特性你的AI应用将真正拥有“理解过去、服务现在”的能力。