1. 项目概述为什么精确记忆是AI应用开发的关键一步如果你正在学习AI应用开发尤其是基于大语言模型LLM构建智能体Agent那么“记忆”这个概念你一定绕不过去。简单来说一个没有记忆的AI应用就像每次和你聊天都是第一次见面的陌生人对话无法连贯任务无法延续。而“15天学会AI应用开发”这个系列进行到第十八篇聚焦于“使用LangGraph实现精确记忆功能”这标志着你从搭建基础流程迈入了构建具备“长期思考”和“上下文感知”能力的高级应用阶段。LangGraph作为LangChain生态系统中的一个新星它解决的核心问题就是为AI应用提供复杂、有状态的工作流编排。与LangChain更侧重于链式Chain和工具Tool的组装不同LangGraph引入了“图”Graph和“状态”State的概念让开发者能够以更直观、更强大的方式描述智能体的决策循环和状态变迁。而“精确记忆”正是通过精心设计的状态管理来实现的。这不仅仅是记住上一条用户消息而是能根据对话历史、工具调用结果、中间思考过程动态地构建、更新和检索一个结构化的记忆体。在当前的AI应用开发生态中无论是构建客服助手、数据分析代理还是游戏NPC对记忆功能的需求都极为迫切。用户不希望每次提问都要重复背景信息智能体也需要基于历史交互做出更精准的决策。因此掌握LangGraph的精确记忆实现是你从“能跑通Demo”到“能做出可用产品”的关键分水岭。接下来的内容我将以一个虚拟的“旅行规划助手”为例带你从零开始拆解如何利用LangGraph构建一个具备精确记忆能力的智能体涵盖核心概念、状态设计、图构建以及实战中的避坑技巧。2. LangGraph核心三要素与精确记忆的关联在深入代码之前我们必须先理解LangGraph赖以运行的三个核心概念State状态、Node节点和Edge边。精确记忆功能的实现本质上就是对State的精心设计和操作。2.1 State记忆的载体与蓝图State是一个定义了智能体在运行过程中所有需要记住的信息的Pydantic模型。它就是你为AI应用设计的“记忆结构蓝图”。一个设计良好的State应该包含对话的所有上下文。以一个旅行规划助手为例一个基础的State可能长这样from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史LangGraph内置的“消息”处理自动管理对话轮次 messages: Annotated[List, add_messages] # 用户明确提出的目的地 destination: str # 用户偏好的旅行日期范围 travel_dates: List[str] # 收集到的用户兴趣点如美食、博物馆、徒步 user_interests: List[str] # 规划生成的每日行程草稿 itinerary_draft: List[str] # 当前对话所处的阶段如收集信息、生成行程、修改行程 current_phase: str这里的关键点在于Annotated[List, add_messages]。add_messages是LangGraph提供的一个归约器Reducer它不是一个简单的列表追加。它的作用是当新消息产生时它会智能地与已有的messages列表进行合并确保对话历史是连贯且去重的。这是实现“记忆”的基础设施。实操心得在设计State时一定要区分“原始信息”和“衍生信息”。例如destination是用户直接说的原始信息而itinerary_draft是根据多种信息推导出的衍生信息。明确这种区分有助于你在不同的Node中清晰地知道该读取或修改State的哪一部分。2.2 Node记忆的处理器与更新器Node是图中的节点代表一个具体的功能单元。每个Node接收当前的State执行一些操作如调用LLM、运行工具函数然后返回一个更新后的State片段。Node是“记忆”被读取、处理和写入的地方。继续旅行助手的例子我们可能有以下几个Nodecollect_info_node: 读取messages中的最新用户输入解析出destination、travel_dates等更新到State中。generate_itinerary_node: 读取destination、travel_dates、user_interests调用LLM生成itinerary_draft。refine_itinerary_node: 根据用户对itinerary_draft的反馈存在于messages中修改itinerary_draft。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o) def collect_info_node(state: State): 从最新对话中提取结构化信息 # 1. 获取最新的用户消息 last_message state[“messages”][-1] user_input last_message.content # 2. 这里可以调用一个LLM或一个解析函数来提取信息 # 例如使用一个简单的提示词让LLM提取 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个信息提取助手。从用户输入中提取目的地、日期和兴趣。”), (“human”, “用户说{input}”) ]) chain prompt | llm # 假设LLM返回一个结构化的JSON字符串这里简化为解析逻辑 # 实际项目中你可能使用PydanticOutputParser或Function Calling extracted_info parse_user_input(user_input) # 假设的解析函数 # 3. 返回要更新的State片段 return { “destination”: extracted_info.get(“destination”, state.get(“destination”, “”)), “travel_dates”: extracted_info.get(“dates”, state.get(“travel_dates”, [])), “user_interests”: extracted_info.get(“interests”, state.get(“user_interests”, [])) }2.3 Edge记忆流转的决策路径Edge决定了State在Node之间如何流转。它分为两种条件边Conditional Edge: 根据State的某个条件如current_phase的值决定下一步走哪个Node。这实现了智能体的“决策”能力。固定边Fixed Edge: 无条件地指向下一个Node。精确记忆的“精确”二字很大程度上体现在Edge的逻辑上。智能体需要根据当前记忆State的内容决定下一步做什么。在我们的旅行助手图中Edge的逻辑可能是开始 -collect_info_node。collect_info_node执行后检查State中的destination和travel_dates是否已填满。如果未填满通过条件边回到collect_info_node继续询问current_phase保持为“收集信息”。如果已填满通过条件边前往generate_itinerary_nodecurrent_phase变为“生成行程”。generate_itinerary_node执行后固定边指向一个“等待用户反馈”的节点。用户反馈后根据反馈内容是确认还是修改决定是结束对话还是进入refine_itinerary_node。这种基于State内容的条件路由使得智能体的行为完全由其“记忆”驱动实现了精确的、上下文相关的流程控制。3. 构建具备精确记忆的旅行规划助手图理解了核心概念后我们开始动手构建。我们将分步创建一个能够记住用户偏好、并根据历史交互动态调整行程的旅行规划助手。3.1 定义完整的状态模型首先我们定义一个更健壮的State。使用TypedDict并利用Annotated进行注解是推荐做法。from typing import TypedDict, List, Optional, Annotated from langgraph.graph.message import add_messages import operator class TravelPlannerState(TypedDict): 旅行规划助手的状态定义 # 核心对话历史由add_messages自动管理 messages: Annotated[List, add_messages] # 用户信息 destination: Optional[str] travel_dates: Annotated[List[str], operator.add] # 使用operator.add归约器合并列表 user_interests: Annotated[List[str], operator.add] budget_level: Optional[str] # “经济”“舒适”“豪华” # 助手生成的信息 itinerary_draft: Optional[List[dict]] # 每天行程的字典列表 confirmed_activities: Annotated[List[dict], operator.add] # 用户确认的活动 # 控制流状态 info_collection_complete: bool itinerary_generated: bool awaiting_feedback: bool这里我们引入了新的归约器operator.add用于列表字段。这意味着当多个Node返回对同一个列表字段如user_interests的更新时这些列表会被合并而不是覆盖。这对于逐步收集信息非常有用。3.2 实现关键功能节点接下来我们实现三个核心节点。节点一信息收集与解析节点这个节点负责从自由文本对话中提取结构化信息。在实际项目中强烈建议使用LLM的Function Calling或结构化输出功能来实现。from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI # 定义我们希望提取的信息结构 class TravelInfo(BaseModel): destination: Optional[str] Field(None, description“旅行目的地如城市名”) travel_dates: List[str] Field(default_factorylist, description“旅行日期格式为YYYY-MM-DD”) interests: List[str] Field(default_factorylist, description“用户兴趣关键词如美食、历史、购物”) budget: Optional[str] Field(None, description“预算水平经济/舒适/豪华”) llm ChatOpenAI(model“gpt-4o”, temperature0).with_structured_output(TravelInfo) def collect_and_parse_node(state: TravelPlannerState): 使用LLM结构化输出提取旅行信息 # 获取最近的对话上下文例如最后3轮对话 recent_messages state[“messages”][-6:] if len(state[“messages”]) 6 else state[“messages”] conversation_context “\n”.join([f“{msg.type}: {msg.content}” for msg in recent_messages]) # 构造提示词 system_prompt “““你是一个旅行信息提取专家。请从对话历史中提取用户提到的旅行相关信息。 如果用户没有明确提及某项信息请将其留空null。 请确保提取的信息准确不要臆造。””” human_prompt f“对话历史\n{conversation_context}” # 调用LLM进行结构化提取 extracted_info: TravelInfo llm.invoke([ (“system”, system_prompt), (“human”, human_prompt) ]) # 准备状态更新 update {} if extracted_info.destination: update[“destination”] extracted_info.destination if extracted_info.travel_dates: # operator.add归约器会合并列表 update[“travel_dates”] extracted_info.travel_dates if extracted_info.interests: update[“user_interests”] extracted_info.interests if extracted_info.budget: update[“budget_level”] extracted_info.budget # 判断信息是否收集完整这里以目的地和日期为例 required_info_present (state.get(“destination”) or extracted_info.destination) and len(state.get(“travel_dates”, [])) len(extracted_info.travel_dates) 1 update[“info_collection_complete”] required_info_present return update注意事项直接使用LLM从对话中提取信息可能存在“幻觉”或过度推断。一个更稳健的做法是结合“主动询问”节点。当信息不足时让助手主动提出明确问题而不是依赖LLM去猜测。这可以通过在State中设置一个missing_info字段并在条件边中判断来实现。节点二行程生成节点当核心信息收集完成后触发此节点生成初步行程。def generate_itinerary_node(state: TravelPlannerState): 基于用户信息生成旅行行程草案 if not state[“info_collection_complete”]: # 如果信息不全不生成行程返回空更新 return {“itinerary_draft”: None} # 从状态中获取所需信息 dest state[“destination”] dates state[“travel_dates”] interests state[“user_interests”] budget state.get(“budget_level”, “舒适”) # 构造给LLM的提示词 prompt f“““你是一个专业的旅行规划师。请为以下需求生成一个初步的每日行程安排。 目的地{dest} 旅行日期{‘, ‘.join(dates)} 用户兴趣{‘, ‘.join(interests)} 预算水平{budget} 请以JSON格式返回一个列表列表中的每个元素代表一天包含date、morning、afternoon、evening、notes字段。 行程要合理符合目的地特点和用户兴趣。””” # 调用LLM生成行程这里简化了调用和解析过程 # 实际应用中应使用结构化输出确保返回格式正确 response llm.invoke(prompt) # 假设我们有一个函数能解析LLM返回的JSON字符串 itinerary parse_itinerary_response(response.content) return { “itinerary_draft”: itinerary, “itinerary_generated”: True, “awaiting_feedback”: True # 生成后等待用户反馈 }节点三反馈处理与行程精炼节点这是体现“精确记忆”和持续学习的关键。助手需要根据用户对草案的反馈结合之前所有的记忆原始需求、历史修改来调整行程。def refine_itinerary_node(state: TravelPlannerState): 处理用户反馈并精炼行程 # 1. 获取用户最新的反馈消息 last_message state[“messages”][-1] user_feedback last_message.content # 2. 获取当前的行程草案和历史确认的活动 current_draft state.get(“itinerary_draft”, []) confirmed state.get(“confirmed_activities”, []) # 3. 构建包含完整上下文的提示词给LLM # 这是“精确记忆”的体现LLM的上下文包含了所有历史信息 refinement_prompt f“““你正在优化一个旅行行程。以下是所有相关信息 - 原始需求目的地{state[‘destination’]}日期{state[‘travel_dates’]}兴趣{state[‘user_interests’]}。 - 当前行程草案{current_draft} - 用户已确认的活动请务必保留{confirmed} - 用户的最新反馈{user_feedback} 请根据用户反馈修改行程草案。输出新的完整行程草案JSON格式同前。 如果用户确认了某部分请将其加入confirmed_activities列表。 ””” response llm.invoke(refinement_prompt) new_itinerary parse_itinerary_response(response.content) # 4. 更新状态 update {“itinerary_draft”: new_itinerary, “awaiting_feedback”: True} # 5. 进阶可以在这里添加逻辑从LLM回复中解析出用户确认的活动并更新confirmed_activities # 例如如果用户说“我特别喜欢第二天的博物馆安排”则可以将该活动加入确认列表。 # 这需要更复杂的解析可能用到另一个LLM调用或规则。 return update3.3 编排节点与条件边构建完整工作流现在我们将所有节点用边连接起来形成一个完整的、有状态的图。from langgraph.graph import StateGraph, END from langgraph.graph import START # 1. 创建图构建器 workflow StateGraph(TravelPlannerState) # 2. 添加节点 workflow.add_node(“collect_info”, collect_and_parse_node) workflow.add_node(“generate_itinerary”, generate_itinerary_node) workflow.add_node(“refine_itinerary”, refine_itinerary_node) workflow.add_node(“ask_for_missing_info”, ask_for_missing_info_node) # 假设有一个询问缺失信息的节点 # 3. 设置入口点 workflow.set_entry_point(“collect_info”) # 4. 添加边定义逻辑流 # 从信息收集节点出来后判断信息是否完整 def route_after_collect(state: TravelPlannerState): if state.get(“info_collection_complete”): return “generate_itinerary” else: # 信息不完整进入主动询问节点 return “ask_for_missing_info” workflow.add_conditional_edges( “collect_info”, route_after_collect, { “generate_itinerary”: “generate_itinerary”, “ask_for_missing_info”: “ask_for_missing_info”, } ) # 询问缺失信息节点后总是回到信息收集节点以接收用户回答 workflow.add_edge(“ask_for_missing_info”, “collect_info”) # 行程生成节点后进入等待反馈状态实际上是通过状态标志并由外部驱动进入精炼节点 # 这里我们简化让生成节点直接连接到END实际交互中应由外部事件触发。 workflow.add_edge(“generate_itinerary”, END) # 5. 编译图 app workflow.compile() # 6. 可视化图可选需要graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(“Graph compiled successfully.”)这个图定义了一个基本流程收集信息 - 判断完整性 - 若完整则生成行程结束若不完整则主动询问 - 再次收集。refine_itinerary_node在这个简单流程中尚未被连接它通常会在生成行程后由另一轮用户交互触发例如通过一个新的图调用并设置awaiting_feedback为True的初始状态。4. 运行、调试与持久化让记忆长期保存4.1 运行与交互LangGraph应用app的调用核心是它的invoke方法你需要传入初始状态。# 初始化状态 initial_state: TravelPlannerState { “messages”: [{“role”: “user”, “content”: “我想下个月去杭州玩三天喜欢美食和自然风光。”}], “destination”: None, “travel_dates”: [], “user_interests”: [], “budget_level”: None, “itinerary_draft”: None, “confirmed_activities”: [], “info_collection_complete”: False, “itinerary_generated”: False, “awaiting_feedback”: False, } # 运行图 final_state app.invoke(initial_state) print(“目的地:”, final_state[“destination”]) print(“行程草案:”, final_state[“itinerary_draft”])对于多轮交互你需要在每次用户输入后将新消息添加到messages中然后再次调用app.invoke。关键点在于每次调用传入的State应该是上一次调用的最终State。这样LangGraph的归约器如add_messages就会正确地将新旧消息合并实现对话记忆的延续。# 模拟第二轮交互 new_state_for_round2 final_state.copy() # 添加新的用户消息 new_state_for_round2[“messages”].append({“role”: “user”, “content”: “我的预算比较宽松可以按舒适档来。”}) # 再次调用图图会根据当前状态包含完整历史决定执行路径 final_state_round2 app.invoke(new_state_for_round2)4.2 状态持久化实现长期记忆上述代码中的State存在于内存中应用重启就消失了。要实现真正的“长期记忆”必须将State持久化到数据库。核心思路是为每个对话会话Session分配一个唯一IDthread_id将每次invoke后的最终State以该ID为键存储起来。下次对话时根据thread_id加载历史State。import json # 假设使用一个简单的字典模拟数据库 memory_store {} def run_agent_with_memory(thread_id: str, user_input: str): # 1. 尝试从记忆库加载历史状态 saved_state memory_store.get(thread_id) if saved_state: # 注意从数据库加载的可能是JSON字符串需要反序列化为State字典 current_state json.loads(saved_state) else: # 2. 如果是新会话初始化状态 current_state: TravelPlannerState { “messages”: [], “destination”: None, # ... 其他字段初始化 } # 3. 将用户新输入添加到消息历史 current_state[“messages”].append({“role”: “user”, “content”: user_input}) # 4. 调用LangGraph应用 new_state app.invoke(current_state) # 5. 将最新状态保存回记忆库 memory_store[thread_id] json.dumps(new_state) # 6. 从最新状态中提取助手的回复通常是messages中的最后一条assistant消息 assistant_messages [msg for msg in new_state[“messages”] if msg[“role”] “assistant”] last_assistant_msg assistant_messages[-1] if assistant_messages else None return last_assistant_msg[“content”] if last_assistant_msg else “” # 使用示例 thread_id “user_123_session_1” print(run_agent_with_memory(thread_id, “我想规划去杭州的旅行。”)) print(run_agent_with_memory(thread_id, “时间是下个月15号到18号。”)) # 这次调用能记住上一轮说的“杭州”实操心得在实际生产环境中State可能包含复杂的嵌套对象如Pydantic模型、datetime等直接json.dumps可能会失败。你需要自定义JSON序列化/反序列化器或者使用支持复杂对象存储的数据库如MongoDB。此外频繁保存整个State可能效率低下可以考虑增量更新。4.3 调试与问题排查LangGraph提供了强大的可视化工具来调试图执行流程。使用app.get_graph().draw_mermaid_png()可以生成流程图。但更实用的是在开发时使用**流式输出Streaming**来观察每一步的状态变化。# 使用stream()方法逐步观察执行过程 inputs {“messages”: [{“role”: “user”, “content”: “去杭州玩”}]} for output in app.stream(inputs, stream_mode“values”): # output是一个元组 (node_name, state_update) node, state_update output print(f“\n--- 节点 [{node}] 执行完毕 ---“) print(“状态更新:”, state_update) # 你可以在这里检查特定字段例如 if “destination” in state_update: print(f“解析出的目的地: {state_update[‘destination’]}”)通过流式输出你可以清晰地看到每个节点对State做了哪些修改这对于排查“为什么某个字段没被正确更新”或“流程为什么没有按预期路由”等问题至关重要。5. 进阶技巧与常见陷阱在掌握了基础构建方法后一些进阶技巧和避坑经验能让你构建的应用更加鲁棒和高效。5.1 记忆的优化压缩与摘要随着对话轮次增加messages历史会越来越长这会导致两个问题1) 消耗大量Token增加成本和延迟2) 可能超过LLM的上下文窗口限制。解决方案是记忆压缩。一种常见策略是定期对messages历史进行摘要Summarize而不是完整保存。你可以在State中增加一个summary字段并设计一个summarize_node。当对话轮次达到一定数量或者检测到话题切换时触发该节点。def summarize_messages_node(state: TravelPlannerState): 将冗长的消息历史压缩成摘要 long_memory state[“messages”] if len(long_memory) 10: # 假设阈值是10条消息 return {} # 无需摘要 # 提取需要永久记忆的关键信息如已确认的行程、用户明确偏好 core_info f“目的地{state[‘destination’]}。已确认活动{state[‘confirmed_activities’]}。” # 让LLM对过去的对话细节进行摘要 prompt f“““请将以下对话历史压缩成一个简洁的摘要保留关于旅行需求、决策和用户偏好的关键事实。 核心信息{core_info} 详细对话历史 {long_memory} 摘要””” summary llm.invoke(prompt).content # 更新状态用摘要替换旧的历史消息只保留最近几条以保证连贯性 return { “messages”: state[“messages”][-4:] [{“role”: “system”, “content”: f“历史摘要{summary}”}], “conversation_summary”: summary }5.2 子图管理复杂记忆模块当一个智能体的功能非常复杂时将所有逻辑放在一个主图中会难以维护。LangGraph的**子图Subgraph**功能允许你将一部分功能连同其专属的记忆状态模块化。例如你可以将“餐厅推荐”这个功能抽离成一个子图。这个子图有自己的状态如cuisine_preference,price_range,past_recommendations主图在需要时调用它。子图内部可以有自己的记忆和决策循环结束后将结果返回给主图。这实现了记忆的模块化和封装。from langgraph.graph import StateGraph as SubStateGraph # 定义餐厅推荐子图的状态 class RestaurantSubState(TypedDict): query: str past_recommendations: List[str] result: Optional[str] # 构建子图... restaurant_subgraph_builder SubStateGraph(RestaurantSubState) # ... 添加子图的节点和边 restaurant_subgraph restaurant_subgraph_builder.compile() # 在主图的某个节点中调用子图 def handle_restaurant_request_node(state: TravelPlannerState): if “找餐厅” in state[“messages”][-1].content: # 准备子图输入 sub_input {“query”: “杭州西湖边杭帮菜”, “past_recommendations”: state.get(“mentioned_restaurants”, [])} # 调用子图 sub_result restaurant_subgraph.invoke(sub_input) # 将子图结果整合到主状态 return { “messages”: [{“role”: “assistant”, “content”: f“推荐餐厅{sub_result[‘result’]}”}], “mentioned_restaurants”: state.get(“mentioned_restaurants”, []) [sub_result[‘result’]] } return {}5.3 常见陷阱与解决方案状态污染多个节点无意中修改了同一个状态字段导致意外覆盖。解决方案精心设计State结构明确每个字段的“所有者”哪个节点主要负责更新。使用Annotated和归约器来定义字段的合并策略如operator.add用于列表合并而非覆盖。条件边逻辑过于复杂路由函数route_after_collect变得冗长且难以维护。解决方案将路由逻辑也封装成独立的、可测试的小函数或者使用多个简单的条件边串联。对于极其复杂的路由可以考虑引入一个专门的“路由节点”该节点调用LLM或规则引擎来决定下一个节点。LLM调用不稳定在collect_and_parse_node中LLM可能无法稳定输出结构化信息。解决方案优先使用Function Calling/Structured Output这是最可靠的方式。设置重试机制在LangChain调用LLM时配置max_retries。后置验证与清洗在Node中添加逻辑对LLM的产出进行格式验证如果失败可以返回一个请求澄清的状态更新引导用户再次输入。图编译或执行错误app.compile()或app.invoke()时报错。排查步骤检查State类型确保所有Node返回的字典键名都能对应到State的字段且类型匹配。使用stream调试如前所述用流式输出观察哪个节点出错。简化测试先构建一个最小可运行图两个节点一条边逐步添加功能。记忆持久化性能每次交互都保存/加载整个State在State很大时成为瓶颈。解决方案增量更新只保存State中变化的部分需要更精细的状态管理。使用更快的存储考虑Redis等内存数据库。定期快照不一定每次调用都持久化可以定期或在关键节点后持久化。精确记忆功能的实现是将AI应用从“单次问答机”升级为“持续协作伙伴”的灵魂。通过LangGraph的状态管理你能够清晰地定义、维护和利用智能体的记忆。这个过程需要仔细设计状态结构、合理划分节点职责、谨慎规划流程路由。