1. 项目概述为什么我们需要会话记忆如果你正在开发一个AI应用无论是客服机器人、智能助手还是创意写作工具你肯定遇到过这样的场景用户问“我昨天提到的那个项目进展如何了”而你的AI助手一脸茫然地回答“抱歉我不记得您之前提到过什么项目。” 这种对话的割裂感会让用户体验大打折扣也让AI显得“很笨”。这就是“会话记忆”要解决的核心问题——让AI能够记住并理解对话的上下文。在传统的聊天机器人开发中实现记忆功能往往意味着你需要手动管理一个对话历史列表每次调用大模型时把整个历史记录一股脑地塞进提示词Prompt里。这种方法简单粗暴但问题也显而易见随着对话轮次增加上下文会越来越长不仅消耗大量Token意味着更高的成本和更慢的响应还可能因为超出模型的最大上下文长度而被截断导致记忆丢失。而LangGraph的出现为我们提供了一种更优雅、更强大的解决方案。它不是一个独立的大模型而是构建在LangChain之上的一个框架专门用于创建有状态的、多步骤的工作流。你可以把它想象成一个流程图设计工具但每个节点都是一个可以执行特定任务如调用LLM、查询数据库的“智能体”节点之间的连线定义了数据也就是“状态”的流动规则。正是这种对“状态”的精细化管理能力让实现复杂、可控的会话记忆变得可能。在接下来的内容里我不会只教你“如何用几行代码开启记忆功能”那太浅了。我会带你深入LangGraph的内部拆解它的三要素State、Node、Edge并手把手教你设计一个既能记住关键信息又能自动清理冗余记忆的智能会话系统。无论你是刚学完前端想转型AI开发还是正在面试AI应用开发岗位理解并掌握这套方法都将是你技术栈中一个亮眼的加分项。2. LangGraph核心三要素深度解析要玩转LangGraph实现高级记忆功能你必须吃透它的三个核心概念State状态、Node节点和Edge边。很多人看了官方文档还是云里雾里觉得抽象接下来我用最直白的方式给你讲明白。2.1 State记忆的载体与蓝图State是LangGraph的灵魂它定义了你整个工作流中需要记住和传递的所有信息。你可以把它理解为一个Python的TypedDict或者Pydantic的BaseModel。它不仅仅是一个存储对话历史的“垃圾桶”更是一份精心设计的“数据蓝图”。一个设计良好的State应该只包含必要的信息。比如对于一个基础的聊天应用你的State可能长这样from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 核心消息历史。Annotated是LangGraph的魔法add_messages是一个归约函数。 messages: Annotated[List[dict], add_messages] # 业务相关从对话中提取出的结构化信息如用户偏好、任务目标等。 user_profile: dict current_task: str # 系统相关控制流程的标记比如是否需要查询知识库。 requires_search: bool这里的关键是messages字段。我们使用了Annotated[List[dict], add_messages]。add_messages是一个归约函数它的作用是当新消息到来时它不是简单地替换掉旧的messages列表而是将新消息追加到列表末尾。这是实现累积式记忆的基础。没有这个归约操作状态就无法在节点间正确传递和更新。实操心得在设计State时一定要遵循“最小必要”原则。不要一股脑地把所有中间计算结果都塞进去。只放那些需要在不同节点间共享并且对后续步骤有决定作用的数据。过多的状态字段会让图变得难以理解和调试。比如requires_search这样的控制标志就非常有用它可以让一个节点决定下一步是调用LLM还是先执行搜索。2.2 Node执行具体任务的单元Node就是图中的一个个“工作站点”。每个Node都是一个普通的Python函数或可调用对象它接收当前的整个State作为输入然后返回一个包含要更新字段的字典。一个典型的LLM调用节点可能是这样的from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4o”) def call_llm(state: State): # 1. 从状态中获取消息历史 messages state[“messages”] # 2. 调用LLM传入历史消息作为上下文 response llm.invoke(messages) # 3. 将LLM的回复作为一条新消息准备更新到状态中 new_message {“role”: “assistant”, “content”: response.content} # 4. 返回一个字典LangGraph会用这个字典来更新State return {“messages”: [new_message]}这个函数干了三件事读取状态、执行逻辑调用AI、返回更新。LangGraph会把你返回的字典{“messages”: [new_message]}与当前的State按照State定义中的规则比如add_messages进行合并从而产生新的State。你可以创建各种功能的节点查询数据库的节点、调用外部API的节点、进行条件判断的节点等等。一个复杂的AI应用就是由这些各司其职的节点协作完成的。2.3 Edge决定流程走向的规则Edge定义了节点执行完毕后接下来该去哪个节点。这是LangGraph实现复杂逻辑和循环的关键。边分为两种普通边Conditional Edge和条件边Conditional Edge。普通边是固定的连接比如“节点A执行完无条件进入节点B”。条件边则允许你根据当前State的内容动态决定下一步。这通常通过一个路由函数来实现。def route_after_search(state: State): # 根据State中的某个标志位决定下一步 if state.get(“requires_search”): # 如果需要搜索则前往“search_knowledge_base”节点 return “search_knowledge_base” else: # 否则直接去调用LLM生成回复 return “call_llm”在构建图时你会这样使用它from langgraph.graph import StateGraph, END workflow StateGraph(State) # 添加节点 workflow.add_node(“call_llm”, call_llm) workflow.add_node(“search_knowledge_base”, search_function) # 设置条件边 workflow.add_conditional_edges( “start_node”, # 从哪个节点出发 route_after_search, # 路由函数 {“search_knowledge_base”: “search_knowledge_base”, “call_llm”: “call_llm”} # 可能的目的地映射 ) # 设置普通边 workflow.add_edge(“search_knowledge_base”, “call_llm”)通过State、Node、Edge的灵活组合你就能构建出像“先判断用户意图 - 如需则查询知识库 - 结合查询结果生成回复”这样的智能工作流。记忆State在这个流程中被每个节点读取和修改从而实现了有状态的、上下文感知的对话。3. 构建带记忆的会话工作流从零到一理解了核心概念我们现在来动手搭建一个真正的、带记忆的聊天应用。我们将构建一个相对完整的流程它不仅能记住对话历史还能根据对话内容决定是否要调用一个模拟的“知识库”来获取更精准的信息。3.1 环境准备与State设计首先安装必要的库。我建议使用虚拟环境来管理依赖。pip install langgraph langchain-openai接下来我们设计一个更健壮的State。除了基本的消息历史我们增加一些字段来支持更复杂的逻辑。from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class ChatState(TypedDict): “”“会话状态定义”“” # 核心对话历史使用add_messages自动追加 messages: Annotated[List[dict], add_messages] # 从用户最新消息中解析出的意图 detected_intent: Optional[str] # 是否需要查询知识库 needs_knowledge_lookup: bool # 从知识库查询到的结果如果有 knowledge_result: Optional[str] # 一个简单的对话轮次计数器可用于触发记忆总结 turn_count: Annotated[int, operator.add] # 使用加法归约每次1注意turn_count字段我们使用了operator.add作为归约函数。这意味着每次有节点返回{“turn_count”: 1}时这个1会被加到现有的turn_count值上完美实现了计数功能。3.2 创建功能节点我们将创建四个节点构成一个完整的工作流意图识别节点、知识库查询节点、LLM回复生成节点和记忆管理节点。节点1意图识别节点这个节点负责分析用户的最新消息判断其意图并决定是否需要查询知识库。def detect_intent(state: ChatState): “”“分析用户意图判断是否需要外部知识”“” # 获取最新的用户消息 messages state[“messages”] last_message messages[-1][“content”] if messages else “” # 这里为了演示使用简单的关键词匹配。在实际项目中你可以用一个小型分类模型或更复杂的NLU逻辑。 needs_lookup False intent “general_chat” keywords_need_knowledge [“什么是” “如何” “教程” “解释一下” “公司的政策”] for kw in keywords_need_knowledge: if kw in last_message: needs_lookup True intent “knowledge_query” break # 返回要更新的状态部分 return { “detected_intent”: intent, “needs_knowledge_lookup”: needs_lookup }节点2知识库查询节点这是一个模拟节点在实际应用中这里会连接你的数据库、向量搜索引擎或API。def query_knowledge_base(state: ChatState): “”“模拟查询知识库”“” query state[“messages”][-1][“content”] # 模拟根据查询返回结果 mock_knowledge { “什么是LangGraph”: “LangGraph是用于构建有状态多智能体应用的框架。”, “如何退款”: “请登录官网在‘我的订单’页面申请退款。”, “公司的政策”: “公司规定每周五可以远程办公。” } result mock_knowledge.get(query, “未找到相关信息。”) return {“knowledge_result”: result}节点3生成回复节点这是核心节点它综合对话历史、用户意图和知识库结果生成最终的回复。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.7) def generate_response(state: ChatState): “”“生成AI助手的回复”“” messages state[“messages”] knowledge state.get(“knowledge_result”) intent state.get(“detected_intent”) # 构建系统提示词将知识作为上下文注入 system_prompt “““你是一个有帮助的助手。请根据对话历史和以下知识如果有的话来回答用户问题。 知识上下文{knowledge} 当前用户意图被识别为{intent} ”””.format(knowledgeknowledge if knowledge else “无”, intentintent) # 准备完整的消息列表系统提示 历史对话 full_messages [{“role”: “system”, “content”: system_prompt}] messages # 调用LLM response llm.invoke(full_messages) # 将助手的回复作为一条新消息返回用于更新状态 return {“messages”: [{“role”: “assistant”, “content”: response.content}]}节点4记忆管理节点高级功能这是一个可选但非常重要的节点。当对话轮次太多时我们可以触发这个节点来对历史记忆进行“总结”或“压缩”防止上下文过长。def manage_memory(state: ChatState): “”“每5轮对话后尝试压缩早期记忆”“” turn_count state.get(“turn_count”, 0) if turn_count % 5 0 and turn_count 0: # 每5轮触发一次 old_messages state[“messages”] if len(old_messages) 10: # 如果消息太多 # 提取需要长期记住的关键信息这里简化处理实际可用LLM总结 # 例如让LLM总结前10轮对话的要点 summary_prompt f“请用一段话总结以下对话的核心内容{str(old_messages[:10])}” # 调用LLM生成总结... # summary llm.invoke(summary_prompt) # 模拟一个总结 summary “[系统已将早期对话总结为用户咨询了关于AI开发和学习路径的问题。]” # 用总结替换掉早期的详细消息保留近期对话 new_messages [{“role”: “system”, “content”: summary}] old_messages[10:] return {“messages”: new_messages} # 如果不需要压缩则返回空字典不改变状态 return {}3.3 组装工作流图现在我们把所有节点用边连接起来形成一个有向图。from langgraph.graph import StateGraph, START, END # 1. 创建图并指定状态类型 workflow StateGraph(ChatState) # 2. 添加所有节点 workflow.add_node(“detect_intent”, detect_intent) workflow.add_node(“query_knowledge”, query_knowledge_base) workflow.add_node(“generate_response”, generate_response) workflow.add_node(“manage_memory”, manage_memory) # 3. 设置入口 workflow.set_entry_point(“detect_intent”) # 4. 设置条件边意图识别后根据是否需要查询知识库来路由 def route_after_intent(state: ChatState): if state.get(“needs_knowledge_lookup”): return “query_knowledge” else: return “generate_response” workflow.add_conditional_edges( “detect_intent”, route_after_intent, {“query_knowledge”: “query_knowledge”, “generate_response”: “generate_response”} ) # 5. 设置普通边知识查询后必然去生成回复 workflow.add_edge(“query_knowledge”, “generate_response”) # 6. 设置普通边生成回复后进入记忆管理节点 workflow.add_edge(“generate_response”, “manage_memory”) # 7. 记忆管理节点执行完后一轮对话结束回到入口等待下一条用户消息。 # 这里我们让记忆管理节点连接回detect_intent同时增加一个计数器更新。 # 但注意我们需要在manage_memory节点里也更新turn_count或者单独加一个节点。 # 更清晰的做法在generate_response节点后先更新计数器再管理记忆。 # 让我们调整一下增加一个专门更新计数的节点。 def increment_turn(state: ChatState): “”“增加对话轮次计数”“” return {“turn_count”: 1} workflow.add_node(“increment_turn”, increment_turn) # 调整边生成回复 - 增加计数 - 管理记忆 - 结束本轮回到入口前实际上需要等待新输入这由外部驱动 workflow.add_edge(“generate_response”, “increment_turn”) workflow.add_edge(“increment_turn”, “manage_memory”) workflow.add_edge(“manage_memory”, END) # 先设为END实际流控由外部调用决定。 # 8. 编译图 app workflow.compile()3.4 运行与测试图编译好后我们就可以运行它了。LangGraph应用app的invoke方法需要一个初始状态。# 初始化状态包含第一条用户消息 initial_state: ChatState { “messages”: [{“role”: “user”, “content”: “什么是LangGraph”}], “detected_intent”: None, “needs_knowledge_lookup”: False, “knowledge_result”: None, “turn_count”: 0 } # 执行图 try: final_state app.invoke(initial_state) print(“最终回复”, final_state[“messages”][-1][“content”]) print(“当前状态”, {k: v for k, v in final_state.items() if k ! ‘messages’}) print(“消息历史长度”, len(final_state[“messages”])) except Exception as e: print(“执行出错”, e)执行上述代码你会看到工作流依次经过检测到意图为“knowledge_query” - 触发知识库查询 - 结合查询结果生成回复 - 增加计数 - 管理记忆可能不触发- 结束。最终状态里包含了完整的对话历史和更新后的各种标志。注意事项这里的图以END结束意味着一次invoke只处理单轮交互。在实际的聊天服务器中比如用FastAPI你会维护一个持久化的app实例并且每次用户发送新消息时都将当前的总状态包含所有历史作为输入调用app.invoke。新的用户消息会通过add_messages的归约功能自动追加到messages列表中从而实现跨轮次的记忆。这正是LangGraph管理会话状态的精妙之处。4. 高级记忆模式与优化策略基础的记忆功能实现了但要让AI真正显得“聪明”我们还需要更高级的策略。直接塞入全部历史是最差的方法下面介绍几种实用的记忆优化模式。4.1 记忆窗口与滑动窗口这是最简单有效的策略只保留最近N轮对话。在LangGraph中你可以在State更新时截断messages列表。def apply_sliding_window(state: ChatState, window_size: int 10): “”“应用滑动窗口只保留最近N条消息”“” messages state[“messages”] if len(messages) window_size: # 保留最后的window_size条消息可以根据需要保留一条系统总结在最前面 state[“messages”] messages[-window_size:] return state你可以将这个函数作为一个独立的节点插入到图中例如在manage_memory节点中调用或者在每次invoke之前作为预处理步骤。4.2 记忆总结与压缩当对话很长时丢弃早期信息可能丢失重要上下文比如用户最初设定的目标。更好的方法是将遥远的记忆压缩成一段简短的摘要。这就是前面manage_memory节点演示的思路。更高级的实现可以提取实体和关键事实使用LLM或NER模型从早期对话中提取人名、地点、任务目标等关键信息将其转化为结构化的数据存入State的独立字段如user_profile,task_goal而不是放在冗长的messages里。增量式总结不要每次都从头总结所有历史。可以维护一个“记忆摘要”字段每经过几轮对话就让LLM基于“旧的摘要”和“新的几轮对话”生成一个“更新的摘要”。class AdvancedState(TypedDict): messages: Annotated[List[dict], add_messages] # 近期详细对话 memory_summary: str # 长期记忆摘要 # ... 其他字段 def summarize_memory_incrementally(state: AdvancedState): “”“增量更新记忆摘要”“” new_messages state[“messages”][-3:] # 假设只总结最新的3轮对话 old_summary state.get(“memory_summary”, “”) prompt f“““ 旧的记忆摘要{old_summary} 最近发生的对话{new_messages} 请将最近对话中的重要信息整合到记忆摘要中生成一个更新的、简洁的摘要。只输出摘要文本。 ””” # new_summary llm.invoke(prompt) new_summary f“{old_summary} 最近用户询问了关于API密钥的问题。” # 模拟 return {“memory_summary”: new_summary, “messages”: []} # 清空已总结的messages然后在生成回复时将memory_summary和近期的messages一起作为上下文提供给LLM。4.3 基于向量数据库的长期记忆对于需要记忆海量、非结构化信息如用户上传的文档、过往的邮件内容的场景滑动窗口和总结都力有未逮。这时就需要引入向量数据库。其核心思想是将对话历史中的每一段文本或用户提供的文档转换成向量Embedding。将这些向量和原始文本一起存入向量数据库如Chroma, Weaviate, Pinecone。当需要回忆时将当前用户的问题也转换成向量在向量数据库中搜索与之最相关的几段历史文本。将这些搜索到的“相关记忆”作为上下文注入到当前的Prompt中。这实现了“按需记忆”LLM不需要看到全部历史只需要看到与当前问题最相关的片段。这在LangGraph中可以作为一个独立的“记忆检索”节点来实现。from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings # 初始化向量库示例持久化需配置persist_directory vectorstore Chroma(embedding_functionOpenAIEmbeddings(), collection_name“chat_memory”) def store_memory(state: AdvancedState): “”“将上一轮对话的重要信息存入向量库”“” last_interaction state[“messages”][-2:] # 假设存储最近一轮的问答对 if last_interaction: text_to_store f“User: {last_interaction[0][‘content’]}\nAssistant: {last_interaction[1][‘content’]}” # 生成向量并存储可以关联一个会话ID vectorstore.add_texts([text_to_store], metadatas[{“session_id”: “user_123”}]) return {} # 不修改核心状态 def retrieve_memory(state: AdvancedState): “”“根据当前问题检索相关记忆”“” current_question state[“messages”][-1][“content”] # 从向量库中搜索最相关的3条记忆 docs vectorstore.similarity_search(current_question, k3, filter{“session_id”: “user_123”}) retrieved_memories “\n”.join([doc.page_content for doc in docs]) return {“retrieved_context”: retrieved_memories}然后在generate_response节点中将state[“retrieved_context”]也加入到系统提示词里。这样AI就拥有了一个庞大且精准的“长期记忆库”。实操心得向量检索记忆非常强大但要注意“记忆污染”问题。如果检索到了不相关或过时的信息反而会干扰AI的判断。因此设计好的元数据过滤如按时间、会话ID、主题过滤和检索后的重排序Rerank机制至关重要。对于简单会话滑动窗口总结通常就足够了对于知识密集型助手向量数据库是必选项。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种奇怪的问题。下面是我踩过坑后总结的一些实战经验。5.1 状态更新不生效这是新手最常见的问题。请务必检查归约函数是否正确使用对于列表追加必须用Annotated[List, add_messages]。对于数值累加用Annotated[int, operator.add]。如果你希望直接替换某个值则不需要Annotated。节点返回值格式节点函数必须返回一个字典字典的键必须是State中定义的字段名。返回{“messages”: [new_message]}LangGraph会通过归约函数将其合并到State的messages列表里。如果你返回{“messages”: “hello”}一个字符串就会因为类型不匹配而出错。编译后修改了State定义如果你修改了State类的定义比如增删字段必须重新编译图重新运行workflow.compile()。否则图会使用旧的状态结构导致错误。5.2 如何调试复杂的图当你的图有多个分支和循环时跟踪执行流会很困难。使用stream方法app.stream(initial_state)会返回一个生成器每执行一个节点就产出一次当前状态。你可以遍历它打印出每个节点执行后的状态快照清晰看到数据是如何流动和变化的。for step, output in app.stream(initial_state): print(f“步骤后状态: {step}”) print(output)可视化你的图LangGraph提供了内置的可视化功能。app.get_graph().draw_mermaid_png()可以生成图像需要安装pygraphviz。虽然指令中禁止在输出里用Mermaid但你在本地调试时这是个神器。打印节点输入输出在每个节点的函数开头和结尾添加print语句输出接收到的state和返回的更新值。这是最直接的调试方法。5.3 处理流式输出用户希望看到AI一个字一个字地回复而不是等待全部生成完。LangGraph原生支持流式输出关键在于generate_response节点中使用支持流式的LLM如ChatOpenAI的stream方法并将图配置为支持流式。from langchain_core.runnables import RunnableConfig def generate_response_streaming(state: ChatState): messages state[“messages”] # ... 准备prompt ... # 使用stream方法 stream llm.stream(full_messages) collected_chunks [] for chunk in stream: content chunk.content if content is not None: collected_chunks.append(content) # 关键使用yield来流式返回部分结果 yield {“messages”: [{“role”: “assistant”, “content”: content}]} # 流结束后可以做一些最终处理 full_reply “”.join(collected_chunks) # 注意在流式节点中通常最终状态由框架自动处理你可能不需要返回最终字典。 # 在图中需要将节点标记为支持流式 workflow.add_node(“generate_response”, generate_response_streaming)然后使用app.stream(initial_state, stream_mode“values”)来调用你就可以在客户端逐步收到AI的回复了。这对提升用户体验至关重要。5.4 性能与成本优化Token管理记忆是最大的Token消耗源。务必实施记忆压缩策略总结、滑动窗口。在调用LLM前可以计算一下messages的大致Token数使用tiktoken等库如果超过阈值则触发记忆总结节点。异步执行如果图中有多个可以并行执行的节点比如同时查询两个不同的数据库LangGraph支持异步节点。使用async def定义节点函数并在添加节点时注明可以显著提升吞吐量。缓存对于纯函数式、输入相同的节点如某些数据转换节点可以考虑使用functools.lru_cache进行缓存避免重复计算。但注意对于调用LLM或查询可变数据库的节点不能缓存。5.5 与FastAPI等Web框架集成在生产环境中你的LangGraph应用通常作为一个后台服务。集成模式如下from fastapi import FastAPI, HTTPException from pydantic import BaseModel import asyncio app_fastapi FastAPI() # 假设你的LangGraph应用实例为 agent_app class ChatRequest(BaseModel): message: str session_id: str # 内存中存储不同会话的状态生产环境应使用Redis等 session_states {} app_fastapi.post(“/chat”) async def chat_endpoint(request: ChatRequest): session_id request.session_id # 获取或初始化该会话的状态 current_state session_states.get(session_id) if current_state is None: current_state {“messages”: [], “turn_count”: 0, “session_id”: session_id} # 确保状态符合你的State定义 # 这里需要将字典转换为你的State类型可能需要一些处理 # 将用户新消息添加到状态中 # 注意这里模拟了add_messages的归约。更严谨的做法是调用一个专门添加消息的节点。 new_user_message {“role”: “user”, “content”: request.message} current_state[“messages”] current_state.get(“messages”, []) [new_user_message] try: # 调用LangGraph应用处理当前状态 final_state await agent_app.ainvoke(current_state) # 更新会话状态 session_states[session_id] final_state # 提取最新的助手回复 last_message final_state[“messages”][-1] if last_message[“role”] ! “assistant”: raise HTTPException(status_code500, detail“No assistant response generated”) return {“reply”: last_message[“content”], “session_id”: session_id} except Exception as e: raise HTTPException(status_code500, detailstr(e))这个模式清晰地将LangGraph作为对话引擎Web框架处理HTTP请求、会话管理和状态持久化。