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社区通常讨论两种主要模式：

1. 对话缓冲记忆 (ConversationBufferMemory)：这是最直接的方式，就是把所有历史对话都保存下来。在LangGraph中，你可以简单地在State里维护一个列表。它的优点是信息完整，缺点是上下文窗口有限，成本随对话长度线性增长。它更适合短对话或需要精确引用历史的场景。

2. 对话摘要记忆 (ConversationSummaryMemory)：这种方式不是保存所有原始对话，而是动态地维护一个不断更新的对话摘要。每次新交互后，系统会调用大模型，将新对话和旧摘要融合，生成一个新的、更精炼的摘要。它的优点是能极大地压缩信息，支持超长对话，缺点是有信息损耗，且增加了LLM调用成本。

在LangGraph中实现摘要记忆，意味着你需要设计一个专门的“记忆更新节点”。这个节点的职责是：接收当前的State（包含新对话和旧摘要），调用LLM生成新摘要，然后更新State中的摘要字段。这正体现了LangGraph将复杂逻辑分解为可编排节点的优势。

2.3 基于向量检索的记忆：让记忆更精准

对于更复杂的应用，比如知识库助手，单纯的线性历史或摘要还不够。用户可能会在很久之后突然问到一个之前讨论过的细节。这时，我们需要的是基于检索的记忆。

其原理是：将历史对话中的每一段（或经过处理的片段）转换为向量嵌入，存入向量数据库（如Chroma、Weaviate）。当新的用户输入到来时，我们将其也转换为向量，并从数据库中检索出语义上最相关的若干段历史，作为“记忆”注入本次对话的上下文。这种方式实现了“按需取用”的记忆，非常高效。

在LangGraph中实现这种记忆，通常需要两个节点：一个“记忆写入节点”负责将重要的对话片段向量化并存储；一个“记忆读取节点”负责根据当前输入检索相关记忆。State中可能需要一个retrieved_memories字段来存放检索结果。

3. 实战构建：一个带摘要记忆的LangGraph智能体

理论讲完了，我们开始动手。我们的目标是构建一个具有对话摘要记忆的聊天智能体。这个智能体不仅能回答当前问题，还能记住对话的要点，并在后续对话中自然地引用。

3.1 环境准备与State定义

首先，确保你已安装必要库：langgraph,langchain-openai（或其他LLM），langchain。

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 构建工作流节点

我们的工作流将包含三个核心节点：

1. summarize_or_store节点：决定是更新摘要，还是仅仅存储对话。
2. generate_response节点：调用LLM生成回复，并传入当前摘要作为上下文。
3. 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(model="gpt-4o-mini", temperature=0) 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(content=system_prompt)] + messages # 调用LLM response = llm.invoke(full_messages) # 将AI的回复作为一条新消息添加到状态中 # 注意：在LangGraph中，我们通常返回一个包含更新后messages字段的字典。 # 但由于我们使用了add_messages归约器，我们直接返回AIMessage，它会被自动追加。 return {“messages”: AIMessage(content=response.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_function=embeddings, 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, k=2) # 检索最相关的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 {} # 此节点可以不修改State

4.3 重构工作流集成长期记忆

现在，我们将长期记忆的读写节点整合到主工作流中。新的流程可能是：

1. memory_retrieval: 检索相关长期记忆。
2. summarize_or_store: 处理摘要记忆。
3. generate_response: 生成回复，此时它的系统提示需要同时包含摘要和检索到的长期记忆。
4. memory_writing: 判断并存储有价值的对话到长期记忆。
5. 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(content=system_prompt)] + messages response = llm.invoke(full_messages) return {“messages”: AIMessage(content=response.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(checkpointer=checkpointer) # 第一次调用，会创建检查点 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提供的状态管理范式和可编排的工作流，让这件事变得结构清晰、模块化。从简单的对话历史缓冲，到动态摘要，再到基于向量的长期记忆检索，你可以根据应用场景的复杂度，像搭积木一样构建合适的记忆层。记住，没有一种记忆策略是万能的，最好的系统往往是多种策略的混合体，并在真实用户数据的反馈中持续迭代。