企业官网建设流程全解析

引言：能调工具的 Chatbot，离"可用"还有多远

过去一年，几乎每个团队都写过类似这样的代码：把工具 schema 丢给 LLM，拿回tool_calls，for循环里逐个执行，把结果拼回消息列表，再调一次模型……一个while True套着if，两三百行，能跑 demo，一上生产就原形毕露：

• 状态不可恢复：进程重启、用户刷新页面，整段对话上下文蒸发；

• 流程不可控：模型偶尔陷入"调工具—报错—再调同一个工具"的死循环，只能靠超时兜底；
• 流式不可用：用户盯着空白屏幕等 20 秒，因为你在等最后一个 tool 执行完才一次性返回；
• 不可观测：出问题时只能print一大坨 message 列表，根本不知道卡在哪个环节。

LangGraph 的价值不在于"又一个 Agent 框架"，而在于它把这坨while循环显式建模成带检查点的有状态图：节点是可观测的执行单元，边是显式的控制流，状态由 checkpointer 持久化。本文从状态机设计讲起，一路做到可上线的流式输出服务，并给出踩过的坑。

先交代一下本文的路线图，方便你按需跳读：第一节讲清楚 LangGraph 的四个核心抽象以及状态设计原则；第二节拆解 tool calling 在图上究竟是怎么跑的、有哪些必须守住的不变式；第三节是重头戏，讲流式输出——这是绝大多数 demo 转生产时最先翻车的地方；第四节讲检查点与多轮会话的持久化；第五节是踩坑清单与优化建议；第六节用 FAQ 收尾。全文的代码都基于langgraph >= 0.2，示例围绕一个电商客服场景展开（查订单、发起退款），因为这个场景天然带"读工具 + 写工具 + 高风险操作需审批"三类复杂度，很适合演示。

一、把 Agent 看成状态机：LangGraph 的核心抽象

LangGraph 的心智模型只有四个词：

ReAct 循环在图上就是一个二元环：agent节点产出AIMessage（可能带tool_calls），条件边判断"有工具调用就去tools，否则结束"；tools节点执行完再指回agent。环的每一圈都会被 checkpointer 打一个快照，thread_id就是这场对话的主键。

这里值得多解释一句"为什么是图，而不是链"。LangChain 时代的Chain是 DAG（有向无环图），只能一路向前；而 Agent 的本质是带反馈的循环——模型看完工具结果还要再推理一次，直到它认为信息足够。循环一旦出现，你就必须回答三个问题：循环什么时候终止（终止条件）、循环到一半断了怎么办（持久化）、循环卡住了怎么救（熔断与人工介入）。LangGraph 把这三个问题分别交给条件边、checkpointer、以及你自己写的节点逻辑，边界清晰，责任到人。这正是它比"手写 while 循环"值钱的地方：不是省代码，而是把隐式控制流显式化。

状态设计的三条原则

1. 状态必须可序列化。别把数据库连接、httpx.Client塞进 State，checkpointer 序列化时会直接炸。外部资源走依赖注入或全局单例。

这条原则在实践中比听起来更严苛。MemorySaver只是把状态存在内存里，看起来什么都能塞；一旦你换成SqliteSaver或PostgresSaver（生产必然要换），序列化就会走JsonPlusSerializer，datetime、Decimal、UUID、自定义 dataclass 都会出问题——要么提前转成字符串/数字，要么注册对应的序列化器。我见过最常见的翻车是把Pydantic模型对象直接放进 State，本地测试全绿，上了 Postgres 才报Type is not msgpack serializable。稳妥的做法是：State 里只放基本类型、dict、list 和 LangChain 的 Message 对象，任何复杂对象都在进入 State 前"拍平"。

2. 用 reducer 声明"合并语义"，而不是"覆盖语义"。这是最容易翻车的地方。默认情况下节点返回的字段会覆盖旧值，只有用Annotated[..., add_messages]标注的字段才会做追加 + 按 ID 去重。写错 reducer，表现就是"多轮对话突然失忆"。

展开讲讲 reducer 的工作机制：每个节点返回的是一个partial state（只包含它想更新的字段），LangGraph 在合并这个 partial state 到全局 state 时，对每个字段查找它是否声明了 reducer。声明了add_messages的字段，会用"旧列表 + 新消息"的方式合并，并且依据message.id去重——这意味着如果你手动构造AIMessage时没给id，去重就退化成了简单追加，同一个消息可能被重复写入。没声明 reducer 的字段（比如上面的turn: int），则是直接覆盖，所以return {"turn": state["turn"] + 1}这种写法是必须的——你不能指望框架帮你累加。

除了add_messages，LangGraph 还提供了operator.add（列表拼接、数字相加）、以及自定义 reducer。自定义 reducer 的签名是(old_value, new_value) -> merged_value，你可以用它实现"取最大值"、“合并 dict”、"保留最近 N 条"等语义。比如下面这个"只保留最近 20 条消息"的裁剪 reducer，就是控制上下文长度的低成本手段：

fromtypingimportAnnotated,TypedDictfromlangchain_core.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesdefkeep_last_n(n:int):def_reduce(old:list,new:list)->list:merged=add_messages(old,new)# 先按标准语义合并去重returnmerged[-n:]# 再截断，注意别切断 tool_call 配对return_reduceclassTrimmedState(TypedDict):messages:Annotated[list[AnyMessage],keep_last_n(20)]

注意：按条数截断是有风险的，因为它可能把一条AIMessage(tool_calls=...)和它对应的ToolMessage拆散，从而触发下一节要讲的那个"400 不变式"错误。更安全的做法是按"完整的工具调用轮次"裁剪，或者用摘要压缩（见第四节）。

3. 业务状态与对话状态分离。messages是给模型看的，user_id、order_snapshot、retry_count是给你自己看的。混在一起会让 prompt 越来越脏，也让状态裁剪变得不可能。

这一条的收益会在三个地方体现出来：其一，可观测性——你能直接从 checkpointer 里读出retry_count判断是不是模型在钻牛角尖；其二，可测试性——你可以脱离 LLM 单独对业务字段做断言；其三，可演进性——messages的结构受模型厂商约束，而业务字段是你自己的 schema，改起来互不牵连。

fromtypingimportAnnotated,TypedDict,Literalfromlangchain_core.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesclassAgentState(TypedDict):# add_messages: 追加 + 按 message.id 去重，这是多轮记忆的基石messages:Annotated[list[AnyMessage],add_messages]user_id:strturn:int# 已执行的 agent 轮次，用于熔断intent:str# 业务侧打标，不参与 prompt 拼接

补充一个工程细节：TypedDict适合轻量场景，但它没有运行时校验。如果你希望状态在写入时就报错（比如turn必须是int），可以改用pydantic.BaseModel作为 State 的 schema，LangGraph 同样支持。代价是每次状态合并都要过一次 Pydantic 校验，性能略低，但在调试期非常值得。

二、核心原理：Tool Calling 在图上到底怎么跑

一次完整的工具调用，本质是三条消息的严格配对：

1. AIMessage(tool_calls=[{id: "call_abc", name: "query_order", args: {...}}])
2. 你的工具函数被执行
3. ToolMessage(content="...", tool_call_id="call_abc")

OpenAI / Anthropic 的接口都强制要求：每个tool_call.id必须在下一条 assistant 消息之前，有一条同 id 的ToolMessage回应。少一条、多一条、id 对不上，直接 400。ToolNode帮你做了这件事，但只要你手写过一次并行工具或人工中断，就会明白这个不变式有多脆。

这个不变式之所以重要，是因为它同时约束了三件事：顺序、数量、对应关系。顺序上，ToolMessage必须紧跟在带tool_calls的AIMessage之后，中间不能插别的消息；数量上，模型一次返回几个tool_calls，你就必须回几个ToolMessage（并行调用时尤其容易漏）；对应关系上，tool_call_id必须一字不差。手动实现时，最常见的错误是"工具抛异常了就不返回消息"——你以为只是少了一条，实际上整段对话从此无法再送给模型。正确做法是即使工具失败，也要返回一条ToolMessage，把错误信息写进 content，让模型自己决定重试还是放弃。

tools_condition就是那个条件边：检查state["messages"][-1].tool_calls是否非空，非空走"tools"，否则走END。

代码示例 1：一个带熔断的完整图

fromlangchain_core.messagesimportSystemMessagefromlangchain_core.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.prebuiltimportToolNode,tools_conditionfromlanggraph.checkpoint.memoryimportMemorySaver@tooldefquery_order(order_id:str)->str:"""根据订单号查询物流状态。order_id 形如 'SO20240517'。"""# 真实场景：查 DB / 调内部 HTTP 接口returnf"订单{order_id}已发货，当前在杭州转运中心。"@tooldefrefund(order_id:str,reason:str)->str:"""对指定订单发起退款。高风险操作，需人工确认。"""returnf"订单{order_id}的退款申请已提交，原因：{reason}。"tools=[query_order,refund]llm=ChatOpenAI(model="gpt-4o-mini",temperature=0).bind_tools(tools)SYSTEM=SystemMessage(content="你是一个电商客服助手，必要时调用工具查询订单或发起退款。")defagent_node(state:AgentState)->dict:# 熔断：超过 6 轮直接兜底，避免模型陷入死循环ifstate["turn"]>=6:return{"messages":[AIMessage(content="这个问题我暂时处理不了，已转人工。")]}resp=llm.invoke([SYSTEM,*state["messages"]])return{"messages":[resp],"turn":state["turn"]+1}defroute_after_agent(state:AgentState)->Literal["tools","__end__"]:last=state["messages"][-1]ifgetattr(last,"tool_calls",None):return"tools"returnEND builder=StateGraph(AgentState)builder.add_node("agent",agent_node)builder.add_node("tools",ToolNode(tools))builder.add_edge(START,"agent")builder.add_conditional_edges("agent",route_after_agent,{"tools":"tools",END:END})builder.add_edge("tools","agent")graph=builder.compile(checkpointer=MemorySaver())

这段代码有几处值得单独说明：

其一，熔断放在agent_node内部而不是条件边里。因为turn只有 agent 节点在累加，把它和判断放在同一个函数里，语义最清晰。如果你把熔断做成一条条件边（turn >= 6 -> END），模型仍会在第 6 轮调用一次 LLM，白白烧钱。另外，熔断后返回的AIMessage必须是"干净"的（不带tool_calls），否则条件边会误判为要去tools。

其二，route_after_agent里用了getattr(last, "tool_calls", None)而不是last.tool_calls。因为HumanMessage、ToolMessage上没有tool_calls属性，直接访问会抛AttributeError。虽然按图的结构，agent 节点之后必然是AIMessage，但防御性写法能让你在改图（比如加一个从 START 直接到 tools 的边）时不至于踩雷。实际上 LangGraph 的tools_condition内部也是这么做的，你完全可以删掉自己写的路由函数、直接用tools_condition——这里保留自定义版本，是为了演示"路由逻辑可以插业务判断"，比如某些工具调用必须走人工审批分支。

其三，ToolNode(tools)默认会捕获工具异常并转成ToolMessage（content里是错误字符串）。这个默认行为在多数场景下是好的——保证不变式不被破坏——但它也有副作用：模型可能看不到"这是异常"，继续拿错误字符串做推理。如果希望显式区分，可以设置ToolNode(tools, handle_tool_errors=True)或传入自定义的异常处理函数，把错误包装成结构化 JSON 塞进 content。

其四，工具函数的 docstring 就是工具描述。@tool装饰器会把函数名、类型注解、docstring 一起转成 JSON Schema 发给模型。docstring 写得含糊，模型就会乱调参数——比如query_order里那句"order_id 形如 ‘SO20240517’"，就是在教模型格式，能显著降低无效调用率。这是一个零成本但收益极高的优化点。

一个容易忽略的细节是并行工具调用。当模型一次返回多个tool_calls（比如同时查两个订单），ToolNode会并发执行它们，然后把多个ToolMessage按原顺序写回。这里有两个坑：一是工具函数内部要保证线程安全（ToolNode用的是线程池）；二是如果某个工具很慢，整轮都会等它，流式输出时用户会看到"卡住"——这正是下一节要解决的问题。

三、流式输出实战：别让用户盯着空白屏幕

到这一步，图能跑通了，但你把它包成 HTTP 接口时，会发现体验极差：用户发一句话，前端转圈十几秒，然后"啪"地弹出一整段回复。原因很简单——graph.invoke()是同步阻塞的，它要等整个 ReAct 循环（可能包含两三次 LLM 调用和若干次工具执行）全部结束才返回。

要做出"打字机"效果，必须理解 LangGraph 的流式是分层的，不同stream_mode对应不同粒度的数据：

实际做产品时，通常是messages+updates组合：messages负责正文的逐字输出，updates负责在工具执行阶段给用户一个"正在查询订单…"的状态提示，避免中间那几秒的空白。

3.1 拿到 token 流

asyncdefstream_chat(graph,message:str,thread_id:str):config={"configurable":{"thread_id":thread_id}}inputs={"messages":[HumanMessage(content=message)]}asyncformode,chunkingraph.astream(inputs,config=config,stream_mode=["messages","updates"]):ifmode=="messages":token,meta=chunk# meta 里有 langgraph_node / langgraph_step，可用来区分节点ifmeta.get("langgraph_node")=="agent"andtoken.content:yield{"type":"token","data":token.content}elifmode=="updates":fornode_name,updateinchunk.items():ifnode_name=="tools":yield{"type":"status","data":"正在执行工具…"}

几个关键点：

• stream_mode传列表时，astream产出的是(mode, chunk)二元组，你必须先按mode分流，否则会把 token 和状态更新混在一起。
• messages模式产出的chunk是(message_chunk, metadata)，metadata["langgraph_node"]告诉你这个 token 来自哪个节点。如果你的图里有多个会调 LLM 的节点（比如一个"意图识别"节点 + 一个"主回复"节点），不加这个判断就会把两段回复搅在一起。
• 工具调用时，模型会先流出一段tool_calls的增量片段（AIMessageChunk的tool_call_chunks），此时content往往是空的。所以if token.content这个判断不能省，否则会往前端推一堆空字符串。

3.2 用 custom 模式推送工具进度

updates模式只会在节点结束时产出一次，粒度太粗。如果你想让用户看到"正在查询订单 SO20240517…"，可以在工具函数内部用writer主动推送：

fromlanggraph.configimportget_stream_writer@tooldefquery_order(order_id:str)->str:"""根据订单号查询物流状态。order_id 形如 'SO20240517'。"""writer=get_stream_writer()writer({"type":"progress","data":f"正在查询订单{order_id}…"})# 真实场景：查 DB / 调内部 HTTP 接口returnf"订单{order_id}已发货，当前在杭州转运中心。"

对应的消费端加上stream_mode="custom"（或把它放进列表），就能收到这条消息。注意get_stream_writer()只在流式上下文中有效，在invoke()里调用会拿到一个空 writer，所以工具函数要对此做兼容，不要依赖它做关键逻辑。

3.3 封装成 SSE 接口

fromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponseimportjson app=FastAPI()@app.post("/chat")asyncdefchat(payload:dict):asyncdefevent_gen():asyncforeventinstream_chat(graph,payload["message"],payload["thread_id"]):yieldf"data:{json.dumps(event,ensure_ascii=False)}\n\n"yield"data: [DONE]\n\n"returnStreamingResponse(event_gen(),media_type="text/event-stream")

三个工程注意事项：

1. SSE 需要禁用缓冲。Nginx 默认会缓冲响应，导致 token 攒成一大块才吐出来。要在 location 里加proxy_buffering off;，并把X-Accel-Buffering: no写进响应头。
2. 必须处理客户端断连。用户关掉页面时，astream应该被取消，否则后台会继续跑完整个循环、白烧 token。FastAPI 里可以用request.is_disconnected()检测，或依赖StreamingResponse在连接断开时抛CancelledError，配合try/finally做清理。
3. thread_id由前端生成并持久化。不要用后端 session，因为刷新页面后前端需要拿同一个thread_id恢复对话。通常用 UUID，存在 localStorage 里。

四、检查点与多轮会话：让对话"记得住、查得到、能回滚"

MemorySaver只能用于本地调试——进程一重启全丢。生产环境至少要换SqliteSaver（单机）或PostgresSaver（多实例），它们把每一步的 State 序列化后写库，thread_id是分区键，checkpoint_id是版本号。

换 checkpointer 只需要改一行：

fromlanggraph.checkpoint.postgres.aioimportAsyncPostgresSaverasyncwithAsyncPostgresSaver.from_conn_string(DB_URL)ascheckpointer:awaitcheckpointer.setup()# 首次运行建表graph=builder.compile(checkpointer=checkpointer)

这带来三个能力：

其一，断点恢复。用户发消息后服务重启，只要thread_id不变，下次invoke会自动从最后一次检查点继续，messages完好无损。这就是开头说的"状态可恢复"。

其二，状态历史与时间旅行。graph.get_state_history(config)能拿到这个 thread 的所有快照。你可以实现"重新生成上一条回复"——找到倒数第二个checkpoint_id，用graph.update_state(config, ..., checkpoint_id=...)回滚，再重跑。这在客服场景里非常实用，因为用户经常会说"你刚才答得不对，重来"。

其三，人工介入（human-in-the-loop）。配合interrupt()，可以在refund这类高风险工具执行前暂停，等人工审批后再用Command(resume=...)继续。这也是为什么前面强调"业务状态与对话状态分离"——审批人需要看到的是结构化的order_snapshot，而不是一坨 messages。

还有一个绕不开的话题是上下文长度。多轮对话越滚越长，迟早撑爆窗口。常见策略有三种：滑动窗口（保留最近 N 轮，注意别拆散 tool 配对）、摘要压缩（用一个 LLM 把旧消息压成一段 summary 塞进 SystemMessage）、以及"把工具结果替换成引用"（ToolMessage里只存order_id，需要时再查）。生产上通常是三者组合，按 token 数动态触发。

五、踩坑清单与优化建议

以下每一条都是真实踩过的，按"症状 → 原因 → 解法"整理：

1. 症状：多轮对话突然失忆。原因：messages字段忘了加add_messages，节点返回时把整个列表覆盖了。解法：检查 State 定义，或者打印graph.get_state(config).values["messages"]的长度来确认。
2. 症状：调用模型报 400，提示 tool_call 与 tool_message 不匹配。原因：工具抛异常时直接raise，没有返回ToolMessage；或者手动构造消息时漏了tool_call_id。解法：用ToolNode，或在自定义节点里try/except后构造ToolMessage(content=str(e), tool_call_id=call["id"], status="error")。
3. 症状：换 Postgres checkpointer 后启动报序列化错误。原因：State 里塞了datetime、Decimal或 Pydantic 对象。解法：入库前转成字符串/数字；或给 checkpointer 配置自定义 serializer。
4. 症状：模型反复调同一个工具，停不下来。原因：工具返回的错误信息对模型没有指导性（比如只返回"error"）。解法：错误信息里写清楚"参数 X 无效，应为 Y 格式"；同时加turn熔断做兜底。
5. 症状：流式输出时前端收到一堆空 token。原因：没过滤AIMessageChunk中content为空、只有tool_call_chunks的片段。解法：if chunk.content再推。
6. 症状：两个请求同时打到同一个thread_id，状态错乱。原因：checkpointer 是乐观并发，同一 thread 的并发写会互相覆盖。解法：前端保证同一会话串行发送；后端用分布式锁按thread_id加锁。
7. 优化：给工具加超时。工具内部调外部 HTTP 一定要设timeout，否则一个慢接口会拖垮整轮。可以在工具里try/except TimeoutError后返回"查询超时，请稍后重试"。
8. 优化：记录 token 用量。在agent_node里从resp.usage_metadata读input_tokens/output_tokens，累加进业务状态（注意别放进messages），用于成本核算和限流。
9. 优化：给每个节点打日志。节点的输入输出都是可序列化的，把node_name、turn、tool_calls名字打出来，出问题时一眼能定位卡在哪。比print(messages)强一百倍。

六、FAQ

Q1：LangGraph 和直接用 OpenAI 的 function calling 有什么区别？
function calling 只解决"模型怎么表达要调工具"，不解决"调完之后怎么继续、状态存哪、怎么流式、怎么中断"。LangGraph 是在前者之上补了控制流和状态管理。你可以理解为：function calling 是协议，LangGraph 是运行时。

Q2：ToolNode能自定义吗？我需要在执行前后做鉴权。
可以。要么不用ToolNode、自己写一个节点函数（记得遵守 tool_call / ToolMessage 配对不变式），要么用ToolNode但把鉴权逻辑包在工具函数内部（推荐，因为工具函数天然知道user_id从哪来）。

Q3：流式输出时，工具调用的参数能提前展示给用户吗？
能。messages模式会流出tool_call_chunks，你可以把这些增量拼起来，在参数完整后展示"正在用 order_id=SO20240517 查询"。但要注意参数是分片到达的，必须做拼接，不能直接展示半截 JSON。

Q4：多个 Agent 节点共享一个 LLM，会不会串消息？
不会。每次invoke都是独立的，消息列表由 State 决定。真正会串的是"你忘了给不同节点传不同的 SystemMessage"——建议把 SystemMessage 作为节点常量，而不是放进 State。

Q5：checkpointer 会不会让数据库爆炸？
会，如果不做清理。每个 thread 的每一步都会存一份快照。生产上要么定期删老 thread，要么用 TTL 策略。另外可以对messages做摘要压缩，减少单条快照的体积。

**Q6：能不能不用 LangChain 的 Message，用自己定义的？

更多硬核网安与AI工具包，请扫码获取完整源码！
**
技术上可以，但add_messages、ToolNode、tools_condition都依赖 Message 的字段约定