企业官网建设流程全解析

1. 项目概述：从“LangChain 旧时代”到“LangGraph 新底座”的真实迁移图谱

“旧版 LangChain 已死”——这句话不是标题党，而是我在过去三个月里亲手拆解、重写、压测、上线六个生产级 LLM 应用后，写在团队周会白板上的第一行字。它背后没有情绪宣泄，只有一连串无法绕开的技术事实：AgentExecutor的状态不可见、RunnableSequence难以调试、Memory模块与工具调用强耦合、多跳推理失败时连日志都找不到断点在哪。直到我们把整个 Agent 流程迁移到 LangGraph 上，用StateGraph显式定义节点、用add_edge控制流转、用checkpointer持久化每一步中间态——才真正第一次看清了模型到底“想做什么”、“做了什么”、“为什么卡住”。LangGraph 不是 LangChain 的插件，它是整套 LLM 应用开发范式的重铸：把隐式的“循环调用工具”变成显式的“有向状态图”，把黑盒的“agent.invoke()”变成可追踪、可中断、可分支、可回滚的确定性工作流。这正是 ReAct（Reasoning + 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_method（generate或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_step；
  • retrieval_node: 接收user_intent，调用向量库检索，更新knowledge_chunks；
  • tool_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 将这一思想拆解为可落地的四层结构：

这种映射不是简单的功能对齐，而是范式升维。旧版 LangChain 的 ReAct 是“模型在 prompt 里模拟 Reasoning”，LangGraph 的 ReAct 是“系统在图谱里强制执行 Reasoning”。前者依赖模型的幻觉能力，后者依赖架构的确定性保障。

3. 实操拆解：手把手将一个旧版 LangChain Agent 迁移至 LangGraph

3.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 f"Order {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( func=get_order_status, name="get_order_status", description="Get the current status of an order by order ID" ) policy_tool = StructuredTool.from_function( func=get_return_policy, name="get_return_policy", description="Get the company's return policy" ) # 构建Agent llm = ChatOpenAI(model="gpt-4-turbo", temperature=0) 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(agent=agent, tools=[order_tool, policy_tool], verbose=True) # 使用 result = agent_executor.invoke({ "input": "What's 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(default=None, description="用户当前意图，如 'order_status', 'return_policy'") order_id: Optional[str] = Field(default=None, description="提取出的订单ID") retrieved_info: Optional[str] = Field(default=None, description="从知识库或API检索到的信息") last_tool_error: Optional[str] = Field(default=None, description="上一次工具调用的错误信息") current_step: Literal[ "intent_recognition", "order_lookup", "policy_lookup", "response_generation", "human_handoff" ] = Field(default="intent_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(model="gpt-4-turbo", temperature=0) # 提取用户最新消息 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 = f"Order {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(model="gpt-4-turbo", temperature=0) # 构建上下文 context = f"User 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(content=response)]} # 5. 人工接管节点 def human_handoff_node(state: AgentState) -> Dict[str, Any]: msg = "抱歉，我暂时无法处理您的请求。已为您转接人工客服，请稍候。" return {"messages": [AIMessage(content=msg)], "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(checkpointer=checkpointer)

关键细节：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": "What's the status of my order #12345?"}]} # invoke 会执行整个图，直到到达 END result = app.invoke(input_message, config=config) 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, config=config) 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列表里会有：

1. Human: “查订单 #12345 的物流”
2. AIMessage: “正在查询...”
3. ToolMessage:{"status": "shipped", "eta": "2024-06-15"}
4. 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 = f"Order 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 = f"Order API unavailable: {str(e)}" # 更新状态，触发降级 return { "last_tool_error": error_msg, "retrieved_info": "I'm 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( host="localhost", port="5432", dbname="langgraph_db", user="langgraph_user", password="langgraph_pass" ) saver = PostgresSaver(conn) # 编译时传入 app = workflow.compile(checkpointer=saver) # 调用时，状态自动存入数据库 config = {"configurable": {"thread_id": "travel_plan_001"}} app.invoke({"messages": [{"role": "user", "content": "帮我规划下周去东京的行程"}]}, config=config) # 服务重启后，用户再次访问，用相同 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": None（None 不是 str）；
• 两个 Node 都试图更新messages字段，一个返回{"messages": [msg]}，另一个返回{"messages": msg}（类型不一致）；
• 字段名拼写错误，如user_intetn（少了个n），导致该字段永远为None。

解决方案：

1. 永远使用MessagesState或继承它，不要自己定义messages: list；
2. Node 返回值只写需要更新的字段，不要返回完整AgentState；
3. 在AgentState定义中，为所有可选字段设置default=None和明确类型：

   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()中注册的节点名。常见错误：

• 返回None（Python 默认返回值）；
• 返回["order_lookup"]（列表，而非字符串）；
• 返回"OrderLookup"（大小写不匹配，节点名是"order_lookup"）；
• 在路由函数中抛出异常，未被捕获，导致图执行中断。

解决方案：

1. 路由函数必须有明确的return语句，覆盖所有分支；
2. 在函数开头加print(f"Routing with state: {state}")日志，调试时一目了然；
3. 使用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 对象）。

解决方案：

1. thread_id必须是字符串：

   # ❌ 错误 config = {"configurable": {"thread_id": 12345}} # int # ✅ 正确 config = {"configurable": {"thread_id": "12345"}} # str

2. InMemorySaver仅用于开发，生产必须用 `PostgresSaver