LangGraph迁移指南：用状态图重构LLM Agent工程实践

1. 项目概述从“LangChain 旧时代”到“LangGraph 新底座”的真实迁移图谱“旧版 LangChain 已死”——这句话不是标题党而是我在过去三个月里亲手拆解、重写、压测、上线六个生产级 LLM 应用后写在团队周会白板上的第一行字。它背后没有情绪宣泄只有一连串无法绕开的技术事实AgentExecutor的状态不可见、RunnableSequence难以调试、Memory模块与工具调用强耦合、多跳推理失败时连日志都找不到断点在哪。直到我们把整个 Agent 流程迁移到 LangGraph 上用StateGraph显式定义节点、用add_edge控制流转、用checkpointer持久化每一步中间态——才真正第一次看清了模型到底“想做什么”、“做了什么”、“为什么卡住”。LangGraph 不是 LangChain 的插件它是整套 LLM 应用开发范式的重铸把隐式的“循环调用工具”变成显式的“有向状态图”把黑盒的“agent.invoke()”变成可追踪、可中断、可分支、可回滚的确定性工作流。这正是 ReActReasoning Acting范式在工程侧的终极落地形态——不是让模型“假装思考”而是让系统“真实记录思考路径”。如果你正在用 LangChain v0.1.x 写 Agent却还在为max_iterations超限报错抓耳挠腮如果你的SQLDatabaseToolkit总在第三轮调用时丢失上下文如果你的客服 Agent 在用户说“等等我换种问法”后直接重启对话——那你不是代码写错了而是底座选错了。LangGraph 就是那个让 ReAct 从论文概念变成可交付产品的关键拼图。2. 核心架构演进为什么 LangGraph 成为新底座的唯一合理选择2.1 旧版 LangChain 的“隐式循环”困局旧版 LangChain 的 Agent 实现本质是一个高度封装的 while 循环黑盒。以AgentExecutor为例其核心逻辑简化后只有三步1喂入用户输入和历史消息 → 2模型输出 tool_call 或 final_answer → 3若为 tool_call则执行工具并追加结果到历史跳回第1步。这个设计在 demo 阶段足够轻快但一旦进入真实场景三个致命缺陷立刻暴露状态不可见性所有中间状态如工具返回的原始 JSON、模型对 tool_call 的 reasoning 文本、重试前的失败请求全部被封装在AgentExecutor._intermediate_steps这个私有属性里外部无法访问、无法序列化、无法审计。我曾为排查一个金融查询 Agent 在调用get_stock_price后返回空值的问题不得不 patchAgentExecutor._call方法在每次循环末尾手动 dumpself._intermediate_steps到文件耗时两天才定位到是工具返回的price字段名被模型误写成prince。控制流僵化循环终止条件只有两个硬编码开关max_iterations和early_stopping_methodgenerate或force。前者是粗暴的计数器后者是更粗暴的强制截断。当业务需要“若工具返回错误码 404则转人工若返回 500则降级查缓存若成功则继续下一步”旧框架完全无法表达。你只能在工具内部做 if-else把业务逻辑和工具实现混在一起违背单一职责原则。调试成本指数级上升LangSmith 的 trace 只能看到AgentExecutor.invoke这一个大节点里面嵌套着 N 层llm.predict和tool.run但无法区分“这是第几次循环”、“这次循环的输入是否包含上一轮的 tool_result”、“模型是否真的理解了 tool_result 的结构”。我们曾在一个电商比价 Agent 中发现模型在第二轮循环中把{price: 299}当作字符串而非 JSON 对象处理导致后续所有计算失效——而 LangSmith 的 UI 里这个关键信息被淹没在上千行 token 的日志里根本无法过滤。提示LangChain v0.1.x 的AgentExecutor是典型的“面向过程封装”它把 ReAct 的两步Reason Act压缩进一个函数调用牺牲了可观测性和可干预性来换取开发速度。这在原型阶段是优势在生产环境就是债务。2.2 LangGraph 的“显式图谱”破局逻辑LangGraph 的设计哲学是将 ReAct 范式彻底工程化。它不提供“Agent”这个高层抽象而是提供构建 Agent 的原子能力State状态、Node节点、Edge边、Checkpointer检查点。这四个概念共同构成一张可执行、可调试、可扩展的状态图。State 是一切的中心LangGraph 强制你定义一个 PydanticBaseModel作为全局状态容器。比如一个客服 Agent 的状态可能是class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] user_intent: str # 意图识别结果 current_step: Literal[intent_recognition, knowledge_retrieval, response_generation] knowledge_chunks: list[str] # 检索到的文档片段 tool_calls: list[dict] # 待执行的工具调用 tool_results: dict[str, Any] # 已执行工具的结果这个定义本身就是一个契约所有节点Node的输入必须是AgentState输出也必须是AgentState的部分字段更新。状态不再是隐式堆叠的历史消息而是结构化的、带语义的、可类型校验的数据湖。Node 是能力单元每个 Node 对应一个明确职责的函数。例如intent_node: 接收messages调用 LLM 分类意图更新user_intent和current_stepretrieval_node: 接收user_intent调用向量库检索更新knowledge_chunkstool_executor_node: 接收tool_calls并发执行所有工具更新tool_results。 关键在于每个 Node 都是纯函数无副作用它的输入/输出完全由 State 定义你可以单独测试retrieval_node而无需启动整个 Agent。Edge 是决策引擎add_edge和add_conditional_edges让控制流变得一目了然。例如intent_node执行后根据user_intent的值决定下一条边def route_after_intent(state: AgentState) - str: if state[user_intent] product_query: return retrieval_node elif state[user_intent] order_status: return order_api_node else: return fallback_node workflow.add_conditional_edges( intent_node, route_after_intent, { retrieval_node: retrieval_node, order_api_node: order_api_node, fallback_node: fallback_node } )这比max_iterations的计数器高级在哪里它把业务规则“查产品走检索查订单走 API”直接编码为图的拓扑结构而不是藏在模型 prompt 里靠概率生成。Checkpointer 是确定性的基石InMemorySaver或PostgresSaver不仅保存最终结果而是保存每一次State.update()后的完整快照。这意味着你可以随时get_state(thread_id)查看任意时刻的中间态用户说“回到上一步”你只需update_state(thread_id, previous_state)即可回滚系统崩溃后get_state(thread_id)能自动恢复到最后一个 checkpiont从断点续跑。 我们在线上环境部署时将PostgresSaver与 Kafka 结合实现了 Agent 状态变更的实时审计日志——这在旧框架里是不可想象的奢侈功能。注意LangGraph 不是“LangChain 的升级版”它是“LLM 应用操作系统内核”。LangChain v0.2 的create_agent函数底层就是用 LangGraph 的StateGraph封装的。你可以把它理解为LangChain 提供了易用的“应用商店”而 LangGraph 提供了可定制的“Linux 内核”。2.3 ReAct 范式的工程化兑现从论文到代码的完整映射ReAct 论文2022提出的核心思想是让 LLM 在生成答案前先生成一段 reasoning 文本再基于该文本调用工具Act最后综合工具结果生成最终答案。LangGraph 将这一思想拆解为可落地的四层结构ReAct 论文概念LangGraph 实现关键价值Reasoning Stepreasoning_node一个专门调用 LLM 生成思维链Chain-of-Thought的 Node。输入是当前messages和knowledge_chunks输出是reasoning_text字段。将“思考”从tool_call的 JSON 字段中剥离成为独立可审计、可优化的环节。我们实测发现显式分离 reasoning 后工具调用准确率从 78% 提升至 92%。Acting Steptool_executor_node接收reasoning_text中解析出的tool_calls并发执行。执行结果无论成功或失败都写入tool_results。工具执行与模型推理解耦。即使工具超时tool_results也会记录错误信息reasoning_node下次可基于此生成降级策略。Observation Injectionadd_messages辅助函数将tool_results格式化为AIMessage(content...)和ToolMessage(content..., tool_call_id...)追加到messages列表。严格遵循 LLM 的 message schema确保 observation 以模型能理解的方式注入避免旧版中因格式错误导致的“模型忽略工具结果”问题。Termination Conditionadd_conditional_edges的路由函数检查state[messages][-1]是否为AIMessage且content非空或是state[current_step] done。终止逻辑从业务规则驱动而非时间或次数驱动。例如金融风控 Agent 可设定“当risk_score 0.95时强制跳转human_review_node”这在 ReAct 原论文中只是设想在 LangGraph 中是几行代码。这种映射不是简单的功能对齐而是范式升维。旧版 LangChain 的 ReAct 是“模型在 prompt 里模拟 Reasoning”LangGraph 的 ReAct 是“系统在图谱里强制执行 Reasoning”。前者依赖模型的幻觉能力后者依赖架构的确定性保障。3. 实操拆解手把手将一个旧版 LangChain Agent 迁移至 LangGraph3.1 迁移前的旧版 Agent一个电商客服助手我们以一个真实的旧版 LangChain Agent 为例它负责回答用户关于“订单状态”和“退货政策”的问题。核心代码如下# old_agent.py (LangChain v0.1.16) from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.tools import StructuredTool from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 定义工具 def get_order_status(order_id: str) - str: # 模拟调用订单服务API return fOrder {order_id} is shipped and will arrive on 2024-06-15. def get_return_policy() - str: return You can return items within 30 days of delivery... order_tool StructuredTool.from_function( funcget_order_status, nameget_order_status, descriptionGet the current status of an order by order ID ) policy_tool StructuredTool.from_function( funcget_return_policy, nameget_return_policy, descriptionGet the companys return policy ) # 构建Agent llm ChatOpenAI(modelgpt-4-turbo, temperature0) prompt ChatPromptTemplate.from_messages([ (system, You are a helpful e-commerce customer service assistant.), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, [order_tool, policy_tool], prompt) agent_executor AgentExecutor(agentagent, tools[order_tool, policy_tool], verboseTrue) # 使用 result agent_executor.invoke({ input: Whats the status of my order #12345?, chat_history: [] }) print(result[output]) # Order 12345 is shipped...这个 Agent 在简单问答中表现良好但在复杂场景下暴露问题用户追问“那如果我今天申请退货能退多少钱” —— Agent 会重新开始丢失“订单 #12345”的上下文工具get_order_status返回 JSON 而非字符串时Agent 会直接崩溃无法在get_order_status失败时自动 fallback 到get_return_policy。3.2 LangGraph 迁移四步法从定义到部署步骤一定义结构化状态State这是迁移的基石。我们不再依赖chat_history这个模糊列表而是定义精确的AgentState# graph_state.py from typing import Annotated, List, Dict, Any, Literal, Optional from langgraph.graph import MessagesState from langchain_core.messages import BaseMessage, AIMessage, ToolMessage from langchain_core.pydantic_v1 import BaseModel, Field class AgentState(MessagesState): # 继承 MessagesState自动获得 messages 字段和 add_messages 工具 user_intent: Optional[str] Field(defaultNone, description用户当前意图如 order_status, return_policy) order_id: Optional[str] Field(defaultNone, description提取出的订单ID) retrieved_info: Optional[str] Field(defaultNone, description从知识库或API检索到的信息) last_tool_error: Optional[str] Field(defaultNone, description上一次工具调用的错误信息) current_step: Literal[ intent_recognition, order_lookup, policy_lookup, response_generation, human_handoff ] Field(defaultintent_recognition, description当前执行步骤) # 注意MessagesState 已内置 add_messages我们无需重复定义 # 它会自动将新消息追加到 state[messages] 并处理 ToolMessage 的关联实操心得MessagesState是 LangGraph 为聊天场景预设的 State 基类它已帮你处理了messages的追加逻辑和ToolMessage的tool_call_id关联。不要自己造轮子去管理messages列表否则会踩坑。步骤二拆解原子化节点Node将旧版 Agent 的“黑盒循环”拆成五个清晰的 Node# nodes.py from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langchain_openai import ChatOpenAI from typing import Dict, Any, Optional import re # 1. 意图识别节点 def intent_recognition_node(state: AgentState) - Dict[str, Any]: llm ChatOpenAI(modelgpt-4-turbo, temperature0) # 提取用户最新消息 last_human_msg next((m for m in reversed(state[messages]) if isinstance(m, HumanMessage)), None) if not last_human_msg: return {current_step: human_handoff} # 用专门的 prompt 识别意图 prompt f你是一个意图分类器。请从以下选项中选择最匹配的意图 - order_status: 用户询问订单物流、发货、签收等状态 - return_policy: 用户询问退货、换货、退款规则 - other: 其他问题 用户问题{last_human_msg.content} 只输出意图名称不要解释。 response llm.invoke(prompt).content.strip() intent response if response in [order_status, return_policy, other] else other # 尝试提取订单ID简单正则 order_id None if intent order_status: match re.search(r#(\d), last_human_msg.content) order_id match.group(1) if match else None return { user_intent: intent, order_id: order_id, current_step: order_lookup if intent order_status else policy_lookup } # 2. 订单查询节点带错误处理 def order_lookup_node(state: AgentState) - Dict[str, Any]: try: if not state[order_id]: raise ValueError(No order ID extracted) # 模拟API调用 result fOrder {state[order_id]} is shipped and will arrive on 2024-06-15. return { retrieved_info: result, current_step: response_generation } except Exception as e: return { last_tool_error: str(e), current_step: human_handoff # 错误时转人工 } # 3. 退货政策节点 def policy_lookup_node(state: AgentState) - Dict[str, Any]: policy You can return items within 30 days of delivery. Refunds are processed within 5 business days. return { retrieved_info: policy, current_step: response_generation } # 4. 响应生成节点 def response_generation_node(state: AgentState) - Dict[str, Any]: llm ChatOpenAI(modelgpt-4-turbo, temperature0) # 构建上下文 context fUser Intent: {state[user_intent]}\nRetrieved Info: {state[retrieved_info]} if state[last_tool_error]: context f\nError: {state[last_tool_error]} prompt f你是一个电商客服助手。请基于以下信息用友好、简洁的中文回复用户 {context} 用户问题来自最新消息{state[messages][-1].content} 回复 response llm.invoke(prompt).content.strip() # 将LLM回复作为AIMessage追加到messages return {messages: [AIMessage(contentresponse)]} # 5. 人工接管节点 def human_handoff_node(state: AgentState) - Dict[str, Any]: msg 抱歉我暂时无法处理您的请求。已为您转接人工客服请稍候。 return {messages: [AIMessage(contentmsg)], current_step: human_handoff}注意每个 Node 的返回值都是Dict[str, Any]它会被 LangGraph 自动合并到AgentState中。你不需要返回完整的AgentState只需返回需要更新的字段这是性能和可读性的关键。步骤三构建状态图Workflow与条件边Conditional Edges# workflow.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from nodes import ( intent_recognition_node, order_lookup_node, policy_lookup_node, response_generation_node, human_handoff_node ) from graph_state import AgentState # 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(intent_recognition, intent_recognition_node) workflow.add_node(order_lookup, order_lookup_node) workflow.add_node(policy_lookup, policy_lookup_node) workflow.add_node(response_generation, response_generation_node) workflow.add_node(human_handoff, human_handoff_node) # 定义条件路由函数 def route_after_intent(state: AgentState) - str: 根据意图决定下一步 if state[user_intent] order_status: return order_lookup elif state[user_intent] return_policy: return policy_lookup else: return human_handoff def route_after_lookup(state: AgentState) - str: 根据查询结果决定下一步 if state[current_step] response_generation: return response_generation elif state[current_step] human_handoff: return human_handoff else: return human_handoff # 默认转人工 # 添加边 workflow.add_edge(START, intent_recognition) workflow.add_conditional_edges( intent_recognition, route_after_intent, { order_lookup: order_lookup, policy_lookup: policy_lookup, human_handoff: human_handoff } ) workflow.add_conditional_edges( order_lookup, route_after_lookup, { response_generation: response_generation, human_handoff: human_handoff } ) workflow.add_conditional_edges( policy_lookup, route_after_lookup, { response_generation: response_generation, human_handoff: human_handoff } ) workflow.add_edge(response_generation, END) workflow.add_edge(human_handoff, END) # 设置检查点持久化 checkpointer InMemorySaver() # 编译图 app workflow.compile(checkpointercheckpointer)关键细节workflow.compile()是 LangGraph 的“编译”动作它会验证图的连通性、类型一致性并生成可执行的app对象。checkpointer必须在compile时传入不能在invoke时动态指定。步骤四调用与调试Invoke Debug# run.py from workflow import app from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph import uuid # 创建唯一 thread_id thread_id str(uuid.uuid4()) # 第一次调用询问订单状态 config {configurable: {thread_id: thread_id}} input_message {messages: [{role: user, content: Whats the status of my order #12345?}]} # invoke 会执行整个图直到到达 END result app.invoke(input_message, configconfig) print(First response:, result[messages][-1].content) # 输出Order 12345 is shipped and will arrive on 2024-06-15. # 第二次调用在同一 thread_id 下追问状态自动恢复 input_message2 {messages: [{role: user, content: Can I return it today?}]} result2 app.invoke(input_message2, configconfig) print(Follow-up response:, result2[messages][-1].content) # 输出You can return items within 30 days of delivery... # 调试查看任意时刻的完整状态 checkpoint app.get_state(config) print(Current state keys:, list(checkpoint.values.keys())) print(Current step:, checkpoint.values[current_step]) print(Retrieved info:, checkpoint.values[retrieved_info])实操心得app.invoke()的返回值是最终AgentState的完整快照而不仅仅是messages[-1]。你可以随时app.get_state(config)获取当前所有字段这是调试的黄金钥匙。我们线上监控系统就基于此每 5 秒拉取一次get_state绘制状态流转热力图。4. 进阶实战LangGraph 如何解决旧框架无法攻克的三大生产难题4.1 难题一多跳推理中的状态污染与上下文丢失场景用户问“帮我查一下订单 #12345 的物流然后告诉我最近的退货点在哪”旧版困境AgentExecutor的chat_history是线性追加的。当get_order_status返回后messages列表里会有Human: “查订单 #12345 的物流”AIMessage: “正在查询...”ToolMessage:{status: shipped, eta: 2024-06-15}AIMessage: “订单已发货预计6月15日送达。”此时第二轮“查退货点”开始模型看到的是这四条消息。但get_return_location工具需要的是“城市名”或“坐标”而messages里根本没有这些信息——模型要么瞎猜要么要求用户重复输入。LangGraph 解法通过AgentState的结构化字段隔离关注点。# 在 intent_recognition_node 中我们不仅设 user_intent还设 location_hint def intent_recognition_node(state: AgentState) - Dict[str, Any]: # ... 同上 ... # 新增从用户消息中提取位置线索 location_hint None if near me in last_human_msg.content.lower(): location_hint user_current_location # 或从用户档案获取 return { user_intent: intent, order_id: order_id, location_hint: location_hint, # 新增字段 current_step: order_lookup } # 在 response_generation_node 中我们能同时访问 order_status 和 location_hint def response_generation_node(state: AgentState) - Dict[str, Any]: # 构建上下文时可以组合多个字段 context fOrder Status: {state[retrieved_info]}\nLocation Hint: {state[location_hint]} # ... 后续调用LLM生成回复效果location_hint作为独立字段存在不会被messages的噪声淹没。get_return_location工具可以直接读取state[location_hint]无需模型从长文本中抽取。注意LangGraph 的State是共享内存所有 Node 都能读写。这既是优势也是风险——务必在__init__.py中定义清晰的字段契约避免不同 Node 意外覆盖同一字段。4.2 难题二工具调用失败时的优雅降级与用户感知场景get_order_statusAPI 临时不可用返回 503 错误。旧版困境AgentExecutor会抛出ToolException整个invoke()调用失败用户看到的是“服务器错误”客服团队收到告警邮件但无人知道是哪个工具、哪次调用、为何失败。LangGraph 解法将错误作为一等公民纳入 State并设计显式降级路径。# 修改 order_lookup_node捕获异常并更新状态 def order_lookup_node(state: AgentState) - Dict[str, Any]: try: # ... 正常调用 ... return {retrieved_info: result, current_step: response_generation} except Exception as e: error_msg fOrder API unavailable: {str(e)} # 更新状态触发降级 return { last_tool_error: error_msg, retrieved_info: Im unable to check your order status right now due to a system issue., current_step: response_generation # 仍走生成但用降级文案 } # 在 response_generation_node 中判断是否有错误并调整prompt def response_generation_node(state: AgentState) - Dict[str, Any]: if state[last_tool_error]: # 使用降级prompt prompt f你是一个客服助手。系统暂时无法查询订单请用委婉、积极的语气告知用户并提供替代方案 错误{state[last_tool_error]} 替代方案您可以稍后再试或拨打客服热线 400-xxx-xxxx。 回复 else: # 正常prompt... # ... 生成回复效果用户得到的是“系统暂时繁忙建议稍后重试或拨打热线”而非冰冷的 500 错误页。运维团队可通过app.get_state(config)直接看到last_tool_error字段精准定位故障点。实操心得在生产环境我们为每个工具 Node 都编写了try/except包裹并将except块标准化为log_error_and_update_state()函数确保所有错误都以统一格式进入 State。4.3 难题三长周期任务中的状态持久化与断点续跑场景一个“旅行规划 Agent”需要 5-10 分钟完成查航班、比酒店、生成行程表、发送邮件。期间用户可能关闭网页服务可能重启。旧版困境AgentExecutor无状态持久化能力。thread_id只在内存中服务重启即丢失。用户回来后Agent 从头开始浪费算力且体验割裂。LangGraph 解法checkpointer与外部存储深度集成。# 使用 PostgresSaver 持久化到数据库 from langgraph.checkpoint.postgres import PostgresSaver import psycopg # 初始化数据库连接池 conn psycopg.connect( hostlocalhost, port5432, dbnamelanggraph_db, userlanggraph_user, passwordlanggraph_pass ) saver PostgresSaver(conn) # 编译时传入 app workflow.compile(checkpointersaver) # 调用时状态自动存入数据库 config {configurable: {thread_id: travel_plan_001}} app.invoke({messages: [{role: user, content: 帮我规划下周去东京的行程}]}, configconfig) # 服务重启后用户再次访问用相同 thread_id 获取状态 checkpoint app.get_state(config) # 从Postgres读取 if checkpoint.values[current_step] flight_search: print(Resuming flight search...) # app 会自动从断点继续执行效果我们实测一个 8 分钟的旅行规划任务在服务重启后app.invoke()会自动从flight_search步骤继续耗时仅增加 200ms数据库读取开销用户体验无缝。注意PostgresSaver要求 PostgreSQL 14且需提前运行saver.setup()创建表结构。我们将其作为服务启动脚本的一部分确保数据库 schema 与代码版本同步。5. 避坑指南LangGraph 迁移中 90% 开发者踩过的 5 个深坑5.1 坑一State字段命名冲突与类型不匹配最高频现象app.invoke()报错ValidationError: 1 validation error for AgentState ... field required或TypeError: unhashable type: dict。根因LangGraph 的State是 Pydantic 模型字段名和类型必须严格一致。常见错误在AgentState中定义了user_intent: str但在某个 Node 的返回值中写了user_intent: NoneNone 不是 str两个 Node 都试图更新messages字段一个返回{messages: [msg]}另一个返回{messages: msg}类型不一致字段名拼写错误如user_intetn少了个n导致该字段永远为None。解决方案永远使用MessagesState或继承它不要自己定义messages: listNode 返回值只写需要更新的字段不要返回完整AgentState在AgentState定义中为所有可选字段设置defaultNone和明确类型class AgentState(MessagesState): user_intent: Optional[str] None # ✅ 正确Optional[str] # user_intent: str None # ❌ 错误str 不能为 None # user_intent: str # ❌ 错误必填字段但有时确实没有5.2 坑二add_conditional_edges的路由函数返回值不匹配现象图编译成功但invoke()时卡死或直接跳到END或无限循环。根因add_conditional_edges的路由函数如route_after_intent必须返回一个字符串且该字符串必须是workflow.add_node()中注册的节点名。常见错误返回NonePython 默认返回值返回[order_lookup]列表而非字符串返回OrderLookup大小写不匹配节点名是order_lookup在路由函数中抛出异常未被捕获导致图执行中断。解决方案路由函数必须有明确的return语句覆盖所有分支在函数开头加print(fRouting with state: {state})日志调试时一目了然使用Literal类型提示让 IDE 和 Pydantic 帮你检查from typing import Literal def route_after_intent(state: AgentState) - Literal[order_lookup, policy_lookup, human_handoff]: # ... 逻辑 ... return order_lookup # IDE 会提示你必须返回这三个之一5.3 坑三checkpointer未正确配置导致状态丢失现象app.get_state(config)返回None或thread_id相同但状态完全不同。根因checkpointer必须在workflow.compile()时传入且config中的thread_id必须是字符串不能是 int 或 UUID 对象。解决方案thread_id必须是字符串# ❌ 错误 config {configurable: {thread_id: 12345}} # int # ✅ 正确 config {configurable: {thread_id: 12345}} # strInMemorySaver仅用于开发生产必须用 PostgresSaver