1. 项目概述为什么我们需要会话记忆如果你正在开发一个AI应用无论是客服机器人、个人助理还是创意写作工具你肯定遇到过这样的场景用户问“我昨天提到的那个项目进展如何了”而你的AI助手一脸茫然地回答“抱歉我不记得您之前提到过什么项目”。这种对话的割裂感是早期AI应用最让人头疼的问题之一。会话记忆简单来说就是让AI能够记住并理解跨越多次交互的上下文信息让对话像真人聊天一样连贯、有深度。在“15天学会AI应用开发”这个系列里我们一路从基础概念、API调用走到了构建复杂工作流的阶段。今天这第十七篇我们要啃下一块硬骨头使用LangGraph来实现真正可用的会话记忆功能。这不仅仅是把聊天记录堆在一起传给大模型那么简单。你需要考虑记忆的存储、检索、更新以及如何避免“记忆爆炸”上下文过长导致成本飙升和效果下降。市面上很多教程只讲LangChain的ConversationBufferMemory但那只是一个简单的开始离生产级应用还有距离。而LangGraph作为LangChain生态中用于构建有状态、多步骤工作流的框架为我们提供了更强大、更灵活的武器来设计和实现记忆系统。我见过太多项目在记忆功能上栽跟头。有的把所有对话都塞进上下文很快令牌数就超标了有的记忆检索不准总是答非所问还有的无法处理多轮对话中的主题切换。所以这篇文章我会带你从零开始在LangGraph中构建一个包含短期工作记忆和长期知识记忆的双层系统并分享我在实际部署中踩过的坑和调优技巧。无论你是想做一个能记住用户偏好的智能体还是需要追踪复杂任务状态的自动化流程这里的内容都能给你直接的参考。2. LangGraph中的状态管理与记忆设计原理在动手写代码之前我们必须先理解LangGraph是如何管理“状态”的因为记忆的本质就是一种特殊的状态。LangGraph的核心抽象是StateGraph它维护着一个状态对象这个对象会随着工作流在各个节点Node间流转而被不断修改。2.1 理解State记忆的容器在LangGraph中State通常是一个TypedDict或Pydantic模型它定义了工作流中所有需要被记住的信息的结构。对于会话记忆我们的State至少需要包含以下几个部分from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator class GraphState(TypedDict): # 当前用户输入 input: str # 模型的最终回复 output: str # 核心对话历史记录 conversation_history: List[str] # 可选从历史中提取的摘要或关键实体用于长期记忆 summary: str # 可选当前会话的元数据如用户ID、会话开始时间 session_id: str这里的关键是conversation_history。一个简单的实现就是用一个列表来存储交替出现的用户消息和AI回复。但是直接存储原始字符串会遇到两个问题1) 令牌数增长过快2) 当历史很长时大模型可能无法有效关注到最关键的历史信息。2.2 记忆的两种模式缓冲与摘要这就是为什么我们需要设计更智能的记忆策略。LangChain社区通常讨论两种主要模式对话缓冲记忆 (ConversationBufferMemory)这是最直接的方式就是把所有历史对话都保存下来。在LangGraph中你可以简单地在State里维护一个列表。它的优点是信息完整缺点是上下文窗口有限成本随对话长度线性增长。它更适合短对话或需要精确引用历史的场景。对话摘要记忆 (ConversationSummaryMemory)这种方式不是保存所有原始对话而是动态地维护一个不断更新的对话摘要。每次新交互后系统会调用大模型将新对话和旧摘要融合生成一个新的、更精炼的摘要。它的优点是能极大地压缩信息支持超长对话缺点是有信息损耗且增加了LLM调用成本。在LangGraph中实现摘要记忆意味着你需要设计一个专门的“记忆更新节点”。这个节点的职责是接收当前的State包含新对话和旧摘要调用LLM生成新摘要然后更新State中的摘要字段。这正体现了LangGraph将复杂逻辑分解为可编排节点的优势。2.3 基于向量检索的记忆让记忆更精准对于更复杂的应用比如知识库助手单纯的线性历史或摘要还不够。用户可能会在很久之后突然问到一个之前讨论过的细节。这时我们需要的是基于检索的记忆。其原理是将历史对话中的每一段或经过处理的片段转换为向量嵌入存入向量数据库如Chroma、Weaviate。当新的用户输入到来时我们将其也转换为向量并从数据库中检索出语义上最相关的若干段历史作为“记忆”注入本次对话的上下文。这种方式实现了“按需取用”的记忆非常高效。在LangGraph中实现这种记忆通常需要两个节点一个“记忆写入节点”负责将重要的对话片段向量化并存储一个“记忆读取节点”负责根据当前输入检索相关记忆。State中可能需要一个retrieved_memories字段来存放检索结果。3. 实战构建一个带摘要记忆的LangGraph智能体理论讲完了我们开始动手。我们的目标是构建一个具有对话摘要记忆的聊天智能体。这个智能体不仅能回答当前问题还能记住对话的要点并在后续对话中自然地引用。3.1 环境准备与State定义首先确保你已安装必要库langgraph,langchain-openai或其他LLMlangchain。pip install langgraph langchain-openai接下来我们定义一个更完善的State。除了基本的输入输出和历史我们增加了summary字段来存储动态摘要并引入num_turns来跟踪对话轮次以便在某些条件下触发摘要更新。from typing import TypedDict, List, Optional from langgraph.graph.message import add_messages from typing_extensions import TypedDict class AgentState(TypedDict): # 消息列表LangGraph内置的add_messages函数能很好地处理它 messages: Annotated[List, add_messages] # 动态更新的对话摘要 summary: str # 当前对话轮次用于控制摘要更新频率 turn_count: int这里我们使用了Annotated[List, add_messages]。这是LangGraph的一个高级特性add_messages是一个归约器它定义了当多个节点试图修改messages字段时如何将这些修改合并通常是追加。这比手动操作列表更安全、更声明式。3.2 构建工作流节点我们的工作流将包含三个核心节点summarize_or_store节点决定是更新摘要还是仅仅存储对话。generate_response节点调用LLM生成回复并传入当前摘要作为上下文。update_turn_count节点简单递增轮次计数器。让我们先实现最关键的summarize_or_store节点。它的逻辑是每隔N轮对话或者当历史消息的预估令牌数超过某个阈值时触发摘要更新。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import SystemMessage, HumanMessage, AIMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0) def summarize_or_store(state: AgentState): messages state[“messages”] old_summary state.get(“summary”, “”) turn_count state[“turn_count”] # 判断是否需要更新摘要例如每3轮或者这是第5轮后的第一轮 need_summarize (turn_count 0 and turn_count % 3 0) or turn_count 5 new_messages [] if need_summarize: # 准备给LLM的提示词让它基于旧摘要和新对话生成新摘要 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个高效的对话摘要助手。请根据之前的对话摘要和最新的几轮对话生成一个全新的、连贯的对话摘要。摘要应抓住核心议题、关键决定和待办事项。”), (“human”, “旧摘要{old_summary}\n\n最近的对话历史{recent_chat}”) ]) # 提取最近3轮对话用于生成摘要 recent_chat “\n”.join([f”{msg.type}: {msg.content}” for msg in messages[-6:]]) # 假设每轮包含用户和AI两条消息 chain prompt | llm new_summary chain.invoke({“old_summary”: old_summary, “recent_chat”: recent_chat}).content # 更新状态中的摘要 # 注意我们不清空messages但摘要更新后后续对话可以基于更精炼的摘要进行 return {“summary”: new_summary} else: # 不需要更新摘要只需确保新消息被添加到state中这通常由归约器自动处理此处无需额外操作 # 但我们可以选择将最新的用户消息单独存储或处理这里我们直接返回空字典因为messages的更新由其他节点负责。 return {} def generate_response(state: AgentState): messages state[“messages”] current_summary state.get(“summary”, “”) # 构建系统提示将当前摘要作为上下文注入 system_prompt f”””你是一个有帮助的助手。以下是当前对话的摘要帮助你理解之前的对话背景 {current_summary} 请基于以上背景和最新对话给出友好、专业的回复。如果摘要为空则仅根据最新对话回复。””” # 准备完整的消息列表系统提示 完整对话历史 full_messages [SystemMessage(contentsystem_prompt)] messages # 调用LLM response llm.invoke(full_messages) # 将AI的回复作为一条新消息添加到状态中 # 注意在LangGraph中我们通常返回一个包含更新后messages字段的字典。 # 但由于我们使用了add_messages归约器我们直接返回AIMessage它会被自动追加。 return {“messages”: AIMessage(contentresponse.content)} def update_turn_count(state: AgentState): # 简单地增加对话轮次计数 return {“turn_count”: state[“turn_count”] 1}3.3 编排图与条件边定义了节点函数后我们需要用StateGraph把它们连接起来并设定执行逻辑。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“summarize_or_store”, summarize_or_store) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“update_turn_count”, update_turn_count) # 设置入口点 workflow.set_entry_point(“summarize_or_store”) # 添加边定义节点执行顺序 workflow.add_edge(“summarize_or_store”, “generate_response”) workflow.add_edge(“generate_response”, “update_turn_count”) workflow.add_edge(“update_turn_count”, END) # 编译图 app workflow.compile()这个图是线性的summarize_or_store-generate_response-update_turn_count- 结束。但实际应用中我们可能需要更复杂的条件逻辑比如根据摘要是否被更新来决定下一步做什么。这就需要用到条件边。假设我们想在更新摘要后给用户一个提示比如“已更新对话摘要”我们可以修改图from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver workflow StateGraph(AgentState) workflow.add_node(“summarize_or_store”, summarize_or_store) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“update_turn_count”, update_turn_count) workflow.add_node(“notify_summary_updated”, lambda state: {“messages”: AIMessage(content“系统提示我已更新了对我们对话要点的理解。”)}) workflow.set_entry_point(“summarize_or_store”) # 定义条件判断函数 def should_notify(state: AgentState): # 我们可以根据state中的某个标志位判断这里为了简单假设summarize_or_store节点会设置一个flag。 # 实际上我们需要在summarize_or_store的返回值里增加一个字段例如 “summary_updated”: True # 这里仅作示例假设我们通过检查summary是否改变来判断实际中需要更精确的比较 return “notify_summary_updated” # 从 summarize_or_store 出来后根据条件决定下一个节点 workflow.add_conditional_edges( “summarize_or_store”, should_notify, { “notify_summary_updated”: “notify_summary_updated”, # 如果不需要通知直接去生成回复 “generate_response”: “generate_response”, } ) workflow.add_edge(“notify_summary_updated”, “generate_response”) workflow.add_edge(“generate_response”, “update_turn_count”) workflow.add_edge(“update_turn_count”, END) app workflow.compile()3.4 运行与测试现在让我们运行这个智能体进行一段多轮对话。# 初始化状态 initial_state { “messages”: [HumanMessage(content“你好我想计划一次去杭州的旅行。”)], “summary”: “”, “turn_count”: 1 } # 执行第一轮 config {“configurable”: {“thread_id”: “user_123”}} # thread_id用于会话隔离 result app.invoke(initial_state, config) print(“AI回复:”, result[“messages”][-1].content) print(“当前摘要:”, result.get(“summary”)) print(“当前轮次:”, result[“turn_count”]) # 模拟用户后续输入继续对话 next_state { “messages”: [HumanMessage(content“我比较喜欢自然风光有什么推荐吗”)], “summary”: result.get(“summary”, “”), “turn_count”: result[“turn_count”] } result2 app.invoke(next_state, config) print(“\n第二轮AI回复:”, result2[“messages”][-1].content)通过观察summary字段的变化你可以看到系统是如何在后台逐步凝练对话要点的。当turn_count达到触发条件如3的倍数时summary会被更新从而影响后续对话的上下文。4. 进阶集成向量数据库实现长期记忆摘要记忆解决了长上下文问题但对于“大海捞针”式的信息检索——比如用户在几百轮对话后突然问“我们最开始说的那个预算数字是多少”——摘要可能已经丢失了这个细节。这时就需要长期记忆而向量检索是当前最有效的实现方式。4.1 设计长期记忆存储节点我们需要扩展State并新增节点来处理向量记忆。class AgentStateWithMemory(TypedDict): messages: Annotated[List, add_messages] summary: str turn_count: int # 新增本次对话检索到的相关记忆片段 relevant_memories: List[str] # 新增一个唯一会话标识用于在向量库中隔离不同对话的记忆 session_id: str然后我们创建一个memory_retrieval节点。这个节点会在每次对话前从向量数据库中检索与当前用户问题最相关的历史片段。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings import hashlib # 初始化向量存储使用内存版Chroma示例 embeddings OpenAIEmbeddings(model“text-embedding-3-small”) # 注意生产环境应使用持久化存储且考虑多会话隔离 vectorstore Chroma(embedding_functionembeddings, persist_directory“./chroma_db”) def memory_retrieval(state: AgentStateWithMemory): user_input state[“messages”][-1].content # 获取最新用户消息 session_id state[“session_id”] # 构建检索查询。可以加入会话ID作为过滤器实现记忆隔离。 # 这里简化处理仅基于语义检索。 docs vectorstore.similarity_search(user_input, k2) # 检索最相关的2条记忆 retrieved_texts [doc.page_content for doc in docs] return {“relevant_memories”: retrieved_texts}4.2 设计记忆写入节点我们还需要一个memory_writing节点负责判断哪些对话片段值得存入长期记忆并将其向量化存储。判断逻辑可以是1) AI回复中包含具体事实或承诺2) 用户表达了明确的偏好或需求。def is_worth_remembering(message_pair): 启发式规则判断对话片段是否值得记忆。这是一个简化示例。 user_msg, ai_msg message_pair # 示例规则如果AI的回复中包含数字、具体名词或承诺性词语则值得记忆 keywords [“预算”, “价格”, “答应”, “保证”, “推荐”, “地址”, “时间”] if any(keyword in ai_msg.content for keyword in keywords): return True return False def memory_writing(state: AgentStateWithMemory): messages state[“messages”] session_id state[“session_id”] # 获取最新的完整一轮对话用户消息AI回复 if len(messages) 2: latest_user_msg messages[-2] if isinstance(messages[-2], HumanMessage) else None latest_ai_msg messages[-1] if isinstance(messages[-1], AIMessage) else None if latest_user_msg and latest_ai_msg and is_worth_remembering((latest_user_msg, latest_ai_msg)): # 构建要存储的文本。可以将会话ID作为元数据存入便于过滤。 text_to_store f”User: {latest_user_msg.content}\nAssistant: {latest_ai_msg.content}” # 生成一个唯一ID例如基于内容和时间戳的哈希 doc_id hashlib.md5(f”{session_id}_{text_to_store}”.encode()).hexdigest() # 存入向量数据库 vectorstore.add_texts( texts[text_to_store], metadatas[{“session_id”: session_id, “type”: “dialogue_pair”}], ids[doc_id] ) print(f”已存入长期记忆: {text_to_store[:50]}...”) return {} # 此节点可以不修改State4.3 重构工作流集成长期记忆现在我们将长期记忆的读写节点整合到主工作流中。新的流程可能是memory_retrieval: 检索相关长期记忆。summarize_or_store: 处理摘要记忆。generate_response: 生成回复此时它的系统提示需要同时包含摘要和检索到的长期记忆。memory_writing: 判断并存储有价值的对话到长期记忆。update_turn_count。# 更新generate_response节点使其能利用长期记忆 def generate_response_with_memory(state: AgentStateWithMemory): messages state[“messages”] current_summary state.get(“summary”, “”) relevant_memories state.get(“relevant_memories”, []) memory_context “\n”.join([f”- {mem}” for mem in relevant_memories]) if relevant_memories else “无相关长期记忆。” system_prompt f”””你是一个有帮助的助手拥有两种记忆 1. **对话摘要**概括近期对话{current_summary} 2. **相关长期记忆**从整个会话历史中检索出的相关片段 {memory_context} 请综合以上所有背景信息和最新对话给出准确、连贯的回复。如果长期记忆与当前问题直接相关请优先依据长期记忆回答。””” full_messages [SystemMessage(contentsystem_prompt)] messages response llm.invoke(full_messages) return {“messages”: AIMessage(contentresponse.content)} # 重新构建图 workflow_advanced StateGraph(AgentStateWithMemory) workflow_advanced.add_node(“retrieve_memories”, memory_retrieval) workflow_advanced.add_node(“summarize”, summarize_or_store) workflow_advanced.add_node(“generate”, generate_response_with_memory) workflow_advanced.add_node(“store_memories”, memory_writing) workflow_advanced.add_node(“update_count”, update_turn_count) workflow_advanced.set_entry_point(“retrieve_memories”) workflow_advanced.add_edge(“retrieve_memories”, “summarize”) workflow_advanced.add_edge(“summarize”, “generate”) workflow_advanced.add_edge(“generate”, “store_memories”) workflow_advanced.add_edge(“store_memories”, “update_count”) workflow_advanced.add_edge(“update_count”, END) app_advanced workflow_advanced.compile()这个工作流实现了记忆的完整闭环检索 - 摘要 - 生成利用双重记忆- 存储。它能让你的AI应用在长时间、多话题的对话中依然保持出色的连贯性和准确性。5. 生产环境部署的注意事项与调优技巧将上述原型部署到生产环境你还会遇到一系列挑战。以下是我在实际项目中总结的经验5.1 记忆的隔离与安全性会话隔离绝对不能将用户A的记忆泄露给用户B。在上面的示例中我们通过session_id在元数据中过滤。在生产中你需要确保向量数据库的查询严格限定在当前用户的session_id或user_id范围内。一个更安全的做法是为每个用户或会话创建独立的向量集合Collection。记忆清理记忆不是越多越好。需要设计记忆的过期和清理策略。例如为每条记忆打上时间戳定期清理超过一定时间如30天的记忆或者当向量库中文档数量超过阈值时淘汰最不常被检索到的记忆。隐私与合规如果对话内容涉及敏感信息在存储到长期记忆向量库前必须进行脱敏处理。或者可以考虑只存储对话的向量嵌入而不存储原始文本但这会增加检索后还原的复杂性。5.2 性能与成本优化摘要更新的触发策略不要每轮对话都更新摘要这成本太高。除了按轮次还可以基于令牌数估算。例如使用tiktoken库估算conversation_history的令牌数超过某个阈值如2000 tokens再触发摘要更新。向量检索的优化索引选择对于海量记忆使用HNSW等高性能索引。检索前过滤在计算相似度前先用session_id、timestamp等元数据过滤掉大部分不相关的文档能极大提升检索速度。混合检索结合关键词搜索BM25和向量检索可以提高召回率和准确性尤其是对于特定名称、数字的记忆。分级记忆策略不是所有信息都值得存入昂贵的向量数据库。可以采用分级策略工作记忆内存最近3-5轮对话的原始记录用于保证即时连贯性。摘要记忆数据库周期生成的对话摘要用于维持中长期话题脉络。长期事实记忆向量库只有被判定为重要事实、用户偏好、承诺等信息才存入。5.3 调试与监控可视化记忆内容在开发阶段提供一个简单的管理界面可以查看和搜索当前会话的摘要内容和向量库中存储的记忆片段。这有助于你理解智能体“记住”了什么以及为什么这样回答。记录记忆触发日志记录每次摘要更新、向量存储和检索的事件包括触发原因、消耗的令牌数、检索到的内容。这些日志是优化触发策略和检索参数如top-k值的宝贵依据。评估记忆效果设计一些测试用例比如在对话中途插入“我们最开始说了什么”之类的问题来定量评估记忆系统的准确性。可以用人工评估或基于LLM的自动评估来打分。5.4 一个常见的陷阱与解决方案问题记忆冲突或混淆当检索到的长期记忆片段与最新的对话摘要或上下文矛盾时AI可能会产生混乱的回复。解决方案在系统提示词中加入记忆优先级和冲突解决指令。例如“你拥有以下记忆背景。请注意1.最新对话和对话摘要的优先级最高代表用户的当前意图和近期共识。2.长期记忆来自更早的历史仅供参考。如果长期记忆与最新信息有冲突请以最新信息和对话摘要为准并可以礼貌地指出‘根据我们最新的讨论...’。”通过这样的提示你可以引导LLM成为一个更理智的“记忆管理者”而不是被混杂的信息淹没。6. 从会话记忆到智能体状态持久化我们目前讨论的记忆主要集中在对话内容本身。但在更复杂的智能体应用中“状态”远不止对话历史。它可能包括工具调用结果例如查询天气API返回的数据。任务执行进度一个多步骤任务如订机票、写报告当前进行到哪一步。用户的个性化配置主题偏好、语言风格等。LangGraph的检查点Checkpoint机制正是为这种广义的状态持久化而生的。它允许你在工作流执行到任何节点后将整个State完整地保存下来。即使进程重启你也可以从上一个检查点恢复智能体能够无缝继续之前的工作。from langgraph.checkpoint import MemorySaver # 在编译图时加入检查点存储器 checkpointer MemorySaver() app_persistent workflow_advanced.compile(checkpointercheckpointer) # 第一次调用会创建检查点 config {“configurable”: {“thread_id”: “trip_planning_456”}} result1 app_persistent.invoke(initial_state, config) # 模拟应用重启后... # 我们可以获取最新的检查点状态并在此基础上继续 from langgraph.checkpoint import BaseCheckpointSaver loaded_state checkpointer.get_tuple(config) # 获取最新状态 # 然后继续invoke智能体会从上次中断的地方继续执行逻辑。对于需要长时间运行、涉及复杂状态变迁的智能体比如自动化客服工单处理、多步骤研究分析结合检查点机制的持久化记忆是必不可少的。它确保了智能体的可靠性和用户体验的连续性。实现一个健壮、高效的会话记忆系统是AI应用从“玩具”走向“工具”的关键一步。LangGraph提供的状态管理范式和可编排的工作流让这件事变得结构清晰、模块化。从简单的对话历史缓冲到动态摘要再到基于向量的长期记忆检索你可以根据应用场景的复杂度像搭积木一样构建合适的记忆层。记住没有一种记忆策略是万能的最好的系统往往是多种策略的混合体并在真实用户数据的反馈中持续迭代。
1. 项目概述为什么我们需要会话记忆如果你正在开发一个AI应用无论是客服机器人、个人助理还是创意写作工具你肯定遇到过这样的场景用户问“我昨天提到的那个项目进展如何了”而你的AI助手一脸茫然地回答“抱歉我不记得您之前提到过什么项目”。这种对话的割裂感是早期AI应用最让人头疼的问题之一。会话记忆简单来说就是让AI能够记住并理解跨越多次交互的上下文信息让对话像真人聊天一样连贯、有深度。在“15天学会AI应用开发”这个系列里我们一路从基础概念、API调用走到了构建复杂工作流的阶段。今天这第十七篇我们要啃下一块硬骨头使用LangGraph来实现真正可用的会话记忆功能。这不仅仅是把聊天记录堆在一起传给大模型那么简单。你需要考虑记忆的存储、检索、更新以及如何避免“记忆爆炸”上下文过长导致成本飙升和效果下降。市面上很多教程只讲LangChain的ConversationBufferMemory但那只是一个简单的开始离生产级应用还有距离。而LangGraph作为LangChain生态中用于构建有状态、多步骤工作流的框架为我们提供了更强大、更灵活的武器来设计和实现记忆系统。我见过太多项目在记忆功能上栽跟头。有的把所有对话都塞进上下文很快令牌数就超标了有的记忆检索不准总是答非所问还有的无法处理多轮对话中的主题切换。所以这篇文章我会带你从零开始在LangGraph中构建一个包含短期工作记忆和长期知识记忆的双层系统并分享我在实际部署中踩过的坑和调优技巧。无论你是想做一个能记住用户偏好的智能体还是需要追踪复杂任务状态的自动化流程这里的内容都能给你直接的参考。2. LangGraph中的状态管理与记忆设计原理在动手写代码之前我们必须先理解LangGraph是如何管理“状态”的因为记忆的本质就是一种特殊的状态。LangGraph的核心抽象是StateGraph它维护着一个状态对象这个对象会随着工作流在各个节点Node间流转而被不断修改。2.1 理解State记忆的容器在LangGraph中State通常是一个TypedDict或Pydantic模型它定义了工作流中所有需要被记住的信息的结构。对于会话记忆我们的State至少需要包含以下几个部分from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator class GraphState(TypedDict): # 当前用户输入 input: str # 模型的最终回复 output: str # 核心对话历史记录 conversation_history: List[str] # 可选从历史中提取的摘要或关键实体用于长期记忆 summary: str # 可选当前会话的元数据如用户ID、会话开始时间 session_id: str这里的关键是conversation_history。一个简单的实现就是用一个列表来存储交替出现的用户消息和AI回复。但是直接存储原始字符串会遇到两个问题1) 令牌数增长过快2) 当历史很长时大模型可能无法有效关注到最关键的历史信息。2.2 记忆的两种模式缓冲与摘要这就是为什么我们需要设计更智能的记忆策略。LangChain社区通常讨论两种主要模式对话缓冲记忆 (ConversationBufferMemory)这是最直接的方式就是把所有历史对话都保存下来。在LangGraph中你可以简单地在State里维护一个列表。它的优点是信息完整缺点是上下文窗口有限成本随对话长度线性增长。它更适合短对话或需要精确引用历史的场景。对话摘要记忆 (ConversationSummaryMemory)这种方式不是保存所有原始对话而是动态地维护一个不断更新的对话摘要。每次新交互后系统会调用大模型将新对话和旧摘要融合生成一个新的、更精炼的摘要。它的优点是能极大地压缩信息支持超长对话缺点是有信息损耗且增加了LLM调用成本。在LangGraph中实现摘要记忆意味着你需要设计一个专门的“记忆更新节点”。这个节点的职责是接收当前的State包含新对话和旧摘要调用LLM生成新摘要然后更新State中的摘要字段。这正体现了LangGraph将复杂逻辑分解为可编排节点的优势。2.3 基于向量检索的记忆让记忆更精准对于更复杂的应用比如知识库助手单纯的线性历史或摘要还不够。用户可能会在很久之后突然问到一个之前讨论过的细节。这时我们需要的是基于检索的记忆。其原理是将历史对话中的每一段或经过处理的片段转换为向量嵌入存入向量数据库如Chroma、Weaviate。当新的用户输入到来时我们将其也转换为向量并从数据库中检索出语义上最相关的若干段历史作为“记忆”注入本次对话的上下文。这种方式实现了“按需取用”的记忆非常高效。在LangGraph中实现这种记忆通常需要两个节点一个“记忆写入节点”负责将重要的对话片段向量化并存储一个“记忆读取节点”负责根据当前输入检索相关记忆。State中可能需要一个retrieved_memories字段来存放检索结果。3. 实战构建一个带摘要记忆的LangGraph智能体理论讲完了我们开始动手。我们的目标是构建一个具有对话摘要记忆的聊天智能体。这个智能体不仅能回答当前问题还能记住对话的要点并在后续对话中自然地引用。3.1 环境准备与State定义首先确保你已安装必要库langgraph,langchain-openai或其他LLMlangchain。pip install langgraph langchain-openai接下来我们定义一个更完善的State。除了基本的输入输出和历史我们增加了summary字段来存储动态摘要并引入num_turns来跟踪对话轮次以便在某些条件下触发摘要更新。from typing import TypedDict, List, Optional from langgraph.graph.message import add_messages from typing_extensions import TypedDict class AgentState(TypedDict): # 消息列表LangGraph内置的add_messages函数能很好地处理它 messages: Annotated[List, add_messages] # 动态更新的对话摘要 summary: str # 当前对话轮次用于控制摘要更新频率 turn_count: int这里我们使用了Annotated[List, add_messages]。这是LangGraph的一个高级特性add_messages是一个归约器它定义了当多个节点试图修改messages字段时如何将这些修改合并通常是追加。这比手动操作列表更安全、更声明式。3.2 构建工作流节点我们的工作流将包含三个核心节点summarize_or_store节点决定是更新摘要还是仅仅存储对话。generate_response节点调用LLM生成回复并传入当前摘要作为上下文。update_turn_count节点简单递增轮次计数器。让我们先实现最关键的summarize_or_store节点。它的逻辑是每隔N轮对话或者当历史消息的预估令牌数超过某个阈值时触发摘要更新。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.messages import SystemMessage, HumanMessage, AIMessage llm ChatOpenAI(modelgpt-4o-mini, temperature0) def summarize_or_store(state: AgentState): messages state[“messages”] old_summary state.get(“summary”, “”) turn_count state[“turn_count”] # 判断是否需要更新摘要例如每3轮或者这是第5轮后的第一轮 need_summarize (turn_count 0 and turn_count % 3 0) or turn_count 5 new_messages [] if need_summarize: # 准备给LLM的提示词让它基于旧摘要和新对话生成新摘要 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个高效的对话摘要助手。请根据之前的对话摘要和最新的几轮对话生成一个全新的、连贯的对话摘要。摘要应抓住核心议题、关键决定和待办事项。”), (“human”, “旧摘要{old_summary}\n\n最近的对话历史{recent_chat}”) ]) # 提取最近3轮对话用于生成摘要 recent_chat “\n”.join([f”{msg.type}: {msg.content}” for msg in messages[-6:]]) # 假设每轮包含用户和AI两条消息 chain prompt | llm new_summary chain.invoke({“old_summary”: old_summary, “recent_chat”: recent_chat}).content # 更新状态中的摘要 # 注意我们不清空messages但摘要更新后后续对话可以基于更精炼的摘要进行 return {“summary”: new_summary} else: # 不需要更新摘要只需确保新消息被添加到state中这通常由归约器自动处理此处无需额外操作 # 但我们可以选择将最新的用户消息单独存储或处理这里我们直接返回空字典因为messages的更新由其他节点负责。 return {} def generate_response(state: AgentState): messages state[“messages”] current_summary state.get(“summary”, “”) # 构建系统提示将当前摘要作为上下文注入 system_prompt f”””你是一个有帮助的助手。以下是当前对话的摘要帮助你理解之前的对话背景 {current_summary} 请基于以上背景和最新对话给出友好、专业的回复。如果摘要为空则仅根据最新对话回复。””” # 准备完整的消息列表系统提示 完整对话历史 full_messages [SystemMessage(contentsystem_prompt)] messages # 调用LLM response llm.invoke(full_messages) # 将AI的回复作为一条新消息添加到状态中 # 注意在LangGraph中我们通常返回一个包含更新后messages字段的字典。 # 但由于我们使用了add_messages归约器我们直接返回AIMessage它会被自动追加。 return {“messages”: AIMessage(contentresponse.content)} def update_turn_count(state: AgentState): # 简单地增加对话轮次计数 return {“turn_count”: state[“turn_count”] 1}3.3 编排图与条件边定义了节点函数后我们需要用StateGraph把它们连接起来并设定执行逻辑。from langgraph.graph import StateGraph, END # 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“summarize_or_store”, summarize_or_store) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“update_turn_count”, update_turn_count) # 设置入口点 workflow.set_entry_point(“summarize_or_store”) # 添加边定义节点执行顺序 workflow.add_edge(“summarize_or_store”, “generate_response”) workflow.add_edge(“generate_response”, “update_turn_count”) workflow.add_edge(“update_turn_count”, END) # 编译图 app workflow.compile()这个图是线性的summarize_or_store-generate_response-update_turn_count- 结束。但实际应用中我们可能需要更复杂的条件逻辑比如根据摘要是否被更新来决定下一步做什么。这就需要用到条件边。假设我们想在更新摘要后给用户一个提示比如“已更新对话摘要”我们可以修改图from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver workflow StateGraph(AgentState) workflow.add_node(“summarize_or_store”, summarize_or_store) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“update_turn_count”, update_turn_count) workflow.add_node(“notify_summary_updated”, lambda state: {“messages”: AIMessage(content“系统提示我已更新了对我们对话要点的理解。”)}) workflow.set_entry_point(“summarize_or_store”) # 定义条件判断函数 def should_notify(state: AgentState): # 我们可以根据state中的某个标志位判断这里为了简单假设summarize_or_store节点会设置一个flag。 # 实际上我们需要在summarize_or_store的返回值里增加一个字段例如 “summary_updated”: True # 这里仅作示例假设我们通过检查summary是否改变来判断实际中需要更精确的比较 return “notify_summary_updated” # 从 summarize_or_store 出来后根据条件决定下一个节点 workflow.add_conditional_edges( “summarize_or_store”, should_notify, { “notify_summary_updated”: “notify_summary_updated”, # 如果不需要通知直接去生成回复 “generate_response”: “generate_response”, } ) workflow.add_edge(“notify_summary_updated”, “generate_response”) workflow.add_edge(“generate_response”, “update_turn_count”) workflow.add_edge(“update_turn_count”, END) app workflow.compile()3.4 运行与测试现在让我们运行这个智能体进行一段多轮对话。# 初始化状态 initial_state { “messages”: [HumanMessage(content“你好我想计划一次去杭州的旅行。”)], “summary”: “”, “turn_count”: 1 } # 执行第一轮 config {“configurable”: {“thread_id”: “user_123”}} # thread_id用于会话隔离 result app.invoke(initial_state, config) print(“AI回复:”, result[“messages”][-1].content) print(“当前摘要:”, result.get(“summary”)) print(“当前轮次:”, result[“turn_count”]) # 模拟用户后续输入继续对话 next_state { “messages”: [HumanMessage(content“我比较喜欢自然风光有什么推荐吗”)], “summary”: result.get(“summary”, “”), “turn_count”: result[“turn_count”] } result2 app.invoke(next_state, config) print(“\n第二轮AI回复:”, result2[“messages”][-1].content)通过观察summary字段的变化你可以看到系统是如何在后台逐步凝练对话要点的。当turn_count达到触发条件如3的倍数时summary会被更新从而影响后续对话的上下文。4. 进阶集成向量数据库实现长期记忆摘要记忆解决了长上下文问题但对于“大海捞针”式的信息检索——比如用户在几百轮对话后突然问“我们最开始说的那个预算数字是多少”——摘要可能已经丢失了这个细节。这时就需要长期记忆而向量检索是当前最有效的实现方式。4.1 设计长期记忆存储节点我们需要扩展State并新增节点来处理向量记忆。class AgentStateWithMemory(TypedDict): messages: Annotated[List, add_messages] summary: str turn_count: int # 新增本次对话检索到的相关记忆片段 relevant_memories: List[str] # 新增一个唯一会话标识用于在向量库中隔离不同对话的记忆 session_id: str然后我们创建一个memory_retrieval节点。这个节点会在每次对话前从向量数据库中检索与当前用户问题最相关的历史片段。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings import hashlib # 初始化向量存储使用内存版Chroma示例 embeddings OpenAIEmbeddings(model“text-embedding-3-small”) # 注意生产环境应使用持久化存储且考虑多会话隔离 vectorstore Chroma(embedding_functionembeddings, persist_directory“./chroma_db”) def memory_retrieval(state: AgentStateWithMemory): user_input state[“messages”][-1].content # 获取最新用户消息 session_id state[“session_id”] # 构建检索查询。可以加入会话ID作为过滤器实现记忆隔离。 # 这里简化处理仅基于语义检索。 docs vectorstore.similarity_search(user_input, k2) # 检索最相关的2条记忆 retrieved_texts [doc.page_content for doc in docs] return {“relevant_memories”: retrieved_texts}4.2 设计记忆写入节点我们还需要一个memory_writing节点负责判断哪些对话片段值得存入长期记忆并将其向量化存储。判断逻辑可以是1) AI回复中包含具体事实或承诺2) 用户表达了明确的偏好或需求。def is_worth_remembering(message_pair): 启发式规则判断对话片段是否值得记忆。这是一个简化示例。 user_msg, ai_msg message_pair # 示例规则如果AI的回复中包含数字、具体名词或承诺性词语则值得记忆 keywords [“预算”, “价格”, “答应”, “保证”, “推荐”, “地址”, “时间”] if any(keyword in ai_msg.content for keyword in keywords): return True return False def memory_writing(state: AgentStateWithMemory): messages state[“messages”] session_id state[“session_id”] # 获取最新的完整一轮对话用户消息AI回复 if len(messages) 2: latest_user_msg messages[-2] if isinstance(messages[-2], HumanMessage) else None latest_ai_msg messages[-1] if isinstance(messages[-1], AIMessage) else None if latest_user_msg and latest_ai_msg and is_worth_remembering((latest_user_msg, latest_ai_msg)): # 构建要存储的文本。可以将会话ID作为元数据存入便于过滤。 text_to_store f”User: {latest_user_msg.content}\nAssistant: {latest_ai_msg.content}” # 生成一个唯一ID例如基于内容和时间戳的哈希 doc_id hashlib.md5(f”{session_id}_{text_to_store}”.encode()).hexdigest() # 存入向量数据库 vectorstore.add_texts( texts[text_to_store], metadatas[{“session_id”: session_id, “type”: “dialogue_pair”}], ids[doc_id] ) print(f”已存入长期记忆: {text_to_store[:50]}...”) return {} # 此节点可以不修改State4.3 重构工作流集成长期记忆现在我们将长期记忆的读写节点整合到主工作流中。新的流程可能是memory_retrieval: 检索相关长期记忆。summarize_or_store: 处理摘要记忆。generate_response: 生成回复此时它的系统提示需要同时包含摘要和检索到的长期记忆。memory_writing: 判断并存储有价值的对话到长期记忆。update_turn_count。# 更新generate_response节点使其能利用长期记忆 def generate_response_with_memory(state: AgentStateWithMemory): messages state[“messages”] current_summary state.get(“summary”, “”) relevant_memories state.get(“relevant_memories”, []) memory_context “\n”.join([f”- {mem}” for mem in relevant_memories]) if relevant_memories else “无相关长期记忆。” system_prompt f”””你是一个有帮助的助手拥有两种记忆 1. **对话摘要**概括近期对话{current_summary} 2. **相关长期记忆**从整个会话历史中检索出的相关片段 {memory_context} 请综合以上所有背景信息和最新对话给出准确、连贯的回复。如果长期记忆与当前问题直接相关请优先依据长期记忆回答。””” full_messages [SystemMessage(contentsystem_prompt)] messages response llm.invoke(full_messages) return {“messages”: AIMessage(contentresponse.content)} # 重新构建图 workflow_advanced StateGraph(AgentStateWithMemory) workflow_advanced.add_node(“retrieve_memories”, memory_retrieval) workflow_advanced.add_node(“summarize”, summarize_or_store) workflow_advanced.add_node(“generate”, generate_response_with_memory) workflow_advanced.add_node(“store_memories”, memory_writing) workflow_advanced.add_node(“update_count”, update_turn_count) workflow_advanced.set_entry_point(“retrieve_memories”) workflow_advanced.add_edge(“retrieve_memories”, “summarize”) workflow_advanced.add_edge(“summarize”, “generate”) workflow_advanced.add_edge(“generate”, “store_memories”) workflow_advanced.add_edge(“store_memories”, “update_count”) workflow_advanced.add_edge(“update_count”, END) app_advanced workflow_advanced.compile()这个工作流实现了记忆的完整闭环检索 - 摘要 - 生成利用双重记忆- 存储。它能让你的AI应用在长时间、多话题的对话中依然保持出色的连贯性和准确性。5. 生产环境部署的注意事项与调优技巧将上述原型部署到生产环境你还会遇到一系列挑战。以下是我在实际项目中总结的经验5.1 记忆的隔离与安全性会话隔离绝对不能将用户A的记忆泄露给用户B。在上面的示例中我们通过session_id在元数据中过滤。在生产中你需要确保向量数据库的查询严格限定在当前用户的session_id或user_id范围内。一个更安全的做法是为每个用户或会话创建独立的向量集合Collection。记忆清理记忆不是越多越好。需要设计记忆的过期和清理策略。例如为每条记忆打上时间戳定期清理超过一定时间如30天的记忆或者当向量库中文档数量超过阈值时淘汰最不常被检索到的记忆。隐私与合规如果对话内容涉及敏感信息在存储到长期记忆向量库前必须进行脱敏处理。或者可以考虑只存储对话的向量嵌入而不存储原始文本但这会增加检索后还原的复杂性。5.2 性能与成本优化摘要更新的触发策略不要每轮对话都更新摘要这成本太高。除了按轮次还可以基于令牌数估算。例如使用tiktoken库估算conversation_history的令牌数超过某个阈值如2000 tokens再触发摘要更新。向量检索的优化索引选择对于海量记忆使用HNSW等高性能索引。检索前过滤在计算相似度前先用session_id、timestamp等元数据过滤掉大部分不相关的文档能极大提升检索速度。混合检索结合关键词搜索BM25和向量检索可以提高召回率和准确性尤其是对于特定名称、数字的记忆。分级记忆策略不是所有信息都值得存入昂贵的向量数据库。可以采用分级策略工作记忆内存最近3-5轮对话的原始记录用于保证即时连贯性。摘要记忆数据库周期生成的对话摘要用于维持中长期话题脉络。长期事实记忆向量库只有被判定为重要事实、用户偏好、承诺等信息才存入。5.3 调试与监控可视化记忆内容在开发阶段提供一个简单的管理界面可以查看和搜索当前会话的摘要内容和向量库中存储的记忆片段。这有助于你理解智能体“记住”了什么以及为什么这样回答。记录记忆触发日志记录每次摘要更新、向量存储和检索的事件包括触发原因、消耗的令牌数、检索到的内容。这些日志是优化触发策略和检索参数如top-k值的宝贵依据。评估记忆效果设计一些测试用例比如在对话中途插入“我们最开始说了什么”之类的问题来定量评估记忆系统的准确性。可以用人工评估或基于LLM的自动评估来打分。5.4 一个常见的陷阱与解决方案问题记忆冲突或混淆当检索到的长期记忆片段与最新的对话摘要或上下文矛盾时AI可能会产生混乱的回复。解决方案在系统提示词中加入记忆优先级和冲突解决指令。例如“你拥有以下记忆背景。请注意1.最新对话和对话摘要的优先级最高代表用户的当前意图和近期共识。2.长期记忆来自更早的历史仅供参考。如果长期记忆与最新信息有冲突请以最新信息和对话摘要为准并可以礼貌地指出‘根据我们最新的讨论...’。”通过这样的提示你可以引导LLM成为一个更理智的“记忆管理者”而不是被混杂的信息淹没。6. 从会话记忆到智能体状态持久化我们目前讨论的记忆主要集中在对话内容本身。但在更复杂的智能体应用中“状态”远不止对话历史。它可能包括工具调用结果例如查询天气API返回的数据。任务执行进度一个多步骤任务如订机票、写报告当前进行到哪一步。用户的个性化配置主题偏好、语言风格等。LangGraph的检查点Checkpoint机制正是为这种广义的状态持久化而生的。它允许你在工作流执行到任何节点后将整个State完整地保存下来。即使进程重启你也可以从上一个检查点恢复智能体能够无缝继续之前的工作。from langgraph.checkpoint import MemorySaver # 在编译图时加入检查点存储器 checkpointer MemorySaver() app_persistent workflow_advanced.compile(checkpointercheckpointer) # 第一次调用会创建检查点 config {“configurable”: {“thread_id”: “trip_planning_456”}} result1 app_persistent.invoke(initial_state, config) # 模拟应用重启后... # 我们可以获取最新的检查点状态并在此基础上继续 from langgraph.checkpoint import BaseCheckpointSaver loaded_state checkpointer.get_tuple(config) # 获取最新状态 # 然后继续invoke智能体会从上次中断的地方继续执行逻辑。对于需要长时间运行、涉及复杂状态变迁的智能体比如自动化客服工单处理、多步骤研究分析结合检查点机制的持久化记忆是必不可少的。它确保了智能体的可靠性和用户体验的连续性。实现一个健壮、高效的会话记忆系统是AI应用从“玩具”走向“工具”的关键一步。LangGraph提供的状态管理范式和可编排的工作流让这件事变得结构清晰、模块化。从简单的对话历史缓冲到动态摘要再到基于向量的长期记忆检索你可以根据应用场景的复杂度像搭积木一样构建合适的记忆层。记住没有一种记忆策略是万能的最好的系统往往是多种策略的混合体并在真实用户数据的反馈中持续迭代。
)
