企业官网建设流程全解析

多智能体协作这件事，我在实际项目里踩过的坑比看过的教程还多。大部分演示代码停留在“两个Agent互相聊天”的阶段，一旦要接入真实业务、处理多轮状态、还要保证每个环节可观测可回滚，代码结构就会迅速失控。LangGraph 的“子图模式”是我目前找到的、能把复杂度压下来的少数方案之一——它允许你把一个完整的多智能体流程封装成一张独立的图，再作为节点嵌入到更大的图里。这篇内容就围绕一个具体的 AI 客服系统，把子图模式从设计到落地的完整链路拆开讲清楚，包括状态怎么设计、子图怎么嵌套、工具调用怎么接、以及我在实测中遇到的几个反直觉问题。适合已经了解 LangGraph 基础概念、想往生产级多智能体架构推进的开发者参考。

1. 为什么客服系统是多智能体架构的典型试炼场

1.1 单Agent硬扛所有职责的必然崩溃

先说说为什么不能用一个 Agent 搞定客服。我最早的做法就是写一个“万能客服 Agent”，系统提示词里塞进退换货政策、订单查询流程、技术支持话术、投诉安抚模板，再挂上十几个工具。刚开始跑 demo 效果还行，一旦对话轮次超过五轮，问题就暴露了。

第一个问题是工具选择准确率断崖式下跌。当工具数量超过 8 个，模型在“用户说‘我上周买的东西还没到’”这种模糊表述下，经常在“订单查询”和“物流追踪”之间反复横跳，甚至调用“退款申请”这种明显不相关的工具。这不是模型能力问题，而是上下文里同时存在太多语义相近的工具描述，注意力被稀释了。

第二个问题是提示词内部逻辑冲突。退换货政策要求“先确认订单状态再判断是否符合条件”，而投诉安抚话术要求“先共情再处理问题”。这两套流程在同一个提示词里会互相干扰，模型有时候对着愤怒的用户直接开始念退货条款，体验极差。

第三个问题是状态管理混乱。单 Agent 的对话历史是一条线性增长的列表，但客服场景里其实存在多条并行的信息线：用户身份、订单上下文、当前诉求分类、已执行的动作。全部混在一条消息流里，后期想做“根据用户情绪动态调整回复策略”这种功能时，根本无从下手。

1.2 子图模式解决的核心痛点

LangGraph 的子图模式本质上是把一张编译好的 StateGraph 当作另一张图的节点来用。这个能力听起来简单，但它解决的是多智能体架构里最棘手的问题：职责隔离与状态边界。

我打个比方。单 Agent 就像一个小公司里只有一个员工，既要做销售又要做售后还要做技术，所有信息都堆在他脑子里。多智能体不加子图，就像把员工分开了，但他们共用一本混乱的公共账本，谁都能改，改完别人还不知道。而子图模式相当于给每个部门配了独立的账本，部门内部怎么记账自己决定，只在需要交接的时候把关键信息同步到公司总账上。

具体到客服系统，我用子图模式划分了三个独立的智能体团队：

• 意图识别子图：负责判断用户当前诉求属于哪一类（订单问题、退换货、技术故障、投诉建议），输出结构化的分类结果。
• 订单处理子图：内部包含订单查询 Agent、物流追踪 Agent、退换货判断 Agent，它们之间有自己的协作逻辑。
• 技术支持子图：内部包含知识库检索 Agent、故障诊断 Agent、解决方案生成 Agent。

每个子图有自己的内部状态结构、自己的工具集、自己的提示词体系。主图只负责路由和最终回复生成，不关心子图内部怎么运转。这种隔离带来的好处是：改订单处理逻辑时完全不会影响技术支持流程，测试也可以按子图独立进行。

1.3 子图模式与普通多Agent方案的关键差异

很多人会问：用 LangGraph 的SendAPI 做并行分支，或者用条件边做路由，不也能实现多智能体吗？为什么非要子图？

区别在于状态作用域。普通方案里所有 Agent 共享同一个 State 对象，每个 Agent 读写的是同一份字段。这在 Agent 数量少的时候没问题，但一旦超过三个，状态字段会膨胀到几十个，而且你无法从代码结构上看出“哪些字段是订单 Agent 该管的，哪些是技术 Agent 该管的”。

子图模式允许你定义独立的 State Schema。订单处理子图有自己的OrderState，里面可能有order_id、order_status、logistics_info这些字段；技术支持子图有自己的TechState，里面有issue_category、kb_results、solution_draft。主图的MainState只需要保留路由决策和最终输出相关的字段。子图在返回时，通过一个映射函数把内部状态转换成主图能理解的格式。

这个设计带来的直接好处是可测试性。我可以单独实例化订单处理子图，传入一个模拟的OrderState，验证它在各种订单状态下的行为，完全不需要启动整个客服系统。这在调试阶段节省的时间是巨大的。

2. 状态设计：子图模式最容易翻车的地方

2.1 主图状态该保留什么、该丢弃什么

状态设计是子图模式里最需要提前想清楚的事，我在这上面返工了三次。核心原则是：主图状态只保留路由和聚合所需的最小信息，所有领域细节下沉到子图。

我的MainState最终长这样：

from typing import TypedDict, Annotated, Literal from langgraph.graph import add_messages class MainState(TypedDict): messages: Annotated[list, add_messages] user_id: str intent: Literal["order", "tech", "complaint", "unknown"] | None subgraph_result: dict | None final_response: str | None

注意这里没有order_id、没有issue_type、没有kb_results。这些全部在子图内部处理。主图只需要知道“用户意图是什么”“子图返回了什么结果”“最终回复是什么”。

messages字段用了add_messages这个 reducer，这是 LangGraph 内置的消息合并器，保证多轮对话时消息按顺序追加而不是覆盖。这个细节很关键，我见过有人直接用list类型，结果子图返回时把主图的消息历史整个替换掉了。

intent字段用Literal限定取值范围，这样在条件边里做路由判断时，类型检查器能帮你发现拼写错误。我一开始用str，结果把"order"写成"oder"，路由直接走到默认分支，排查了半小时才发现。

2.2 子图内部状态的独立定义与字段收敛

订单处理子图的OrderState是这样的：

class OrderState(TypedDict): messages: Annotated[list, add_messages] user_id: str order_id: str | None order_status: str | None logistics_info: dict | None return_eligible: bool | None action_taken: str | None

这里有个设计决策值得展开：子图状态里也保留了messages。原因是子图内部的多个 Agent 之间需要传递对话上下文，比如订单查询 Agent 查到订单后，退换货判断 Agent 需要知道“用户具体想退哪个商品”。如果只传结构化字段，这些语义信息会丢失。

但子图的messages和主图的messages是隔离的。子图执行完毕后，只有action_taken和必要的摘要信息会被映射回主图。这样主图的对话历史不会被订单查询过程中的中间消息污染。

字段收敛方面，我遵循一个规则：每个字段都必须有明确的写入者和读取者。如果一个字段只有写入没有读取，说明它是冗余的；如果只有读取没有写入，说明设计有遗漏。用这个规则审查一遍，OrderState从最初的 12 个字段砍到了 7 个。

2.3 父子图状态映射的三种模式与选型

子图返回主图时，状态怎么映射？LangGraph 提供了几种方式，我在不同场景下用了不同的模式。

模式一：直接返回子图最终状态。适用于子图输出就是主图需要的全部信息。比如意图识别子图，它返回的intent字段直接就是主图要用的。

模式二：通过映射函数转换。适用于子图状态结构复杂、主图只需要其中一部分。订单处理子图返回后，我用一个map_order_result函数提取关键信息：

def map_order_result(order_state: OrderState) -> dict: return { "subgraph_result": { "action": order_state.get("action_taken"), "status": order_state.get("order_status"), "summary": order_state["messages"][-1].content if order_state["messages"] else "" } }

模式三：子图直接写入主图状态。这种方式耦合度最高，我只在子图需要更新主图messages时使用。具体做法是在子图节点里接收主图状态引用，但这样会让子图无法独立测试，所以我尽量不用。

选型建议：优先用模式二，它保持了子图的独立性，同时给了你控制映射逻辑的空间。模式一太死板，模式三太耦合。

3. 子图嵌套的工程实现细节

3.1 子图编译与节点注册的完整流程

把子图嵌入主图，核心就两步：编译子图、把编译后的子图作为节点加入主图。但这里面有几个容易忽略的细节。

先看子图的构建：

from langgraph.graph import StateGraph, END def build_order_subgraph(): builder = StateGraph(OrderState) builder.add_node("query_order", query_order_node) builder.add_node("check_logistics", check_logistics_node) builder.add_node("judge_return", judge_return_node) builder.add_node("execute_action", execute_action_node) builder.set_entry_point("query_order") builder.add_conditional_edges( "query_order", route_after_query, {"logistics": "check_logistics", "return": "judge_return", "done": END} ) builder.add_edge("check_logistics", "judge_return") builder.add_conditional_edges( "judge_return", route_after_judge, {"execute": "execute_action", "reject": END} ) builder.add_edge("execute_action", END) return builder.compile()

这里的关键点是set_entry_point和END的使用。子图必须有明确的入口和出口，否则嵌入主图后行为不可预测。我见过有人忘记设入口点，LangGraph 会报错但错误信息不太直观。

然后是主图注册：

def build_main_graph(): order_subgraph = build_order_subgraph() builder = StateGraph(MainState) builder.add_node("classify_intent", classify_intent_node) builder.add_node("order_team", order_subgraph) # 子图作为节点 builder.add_node("tech_team", build_tech_subgraph()) builder.add_node("generate_response", generate_response_node) builder.set_entry_point("classify_intent") builder.add_conditional_edges( "classify_intent", route_by_intent, {"order": "order_team", "tech": "tech_team", "complaint": "generate_response"} ) builder.add_edge("order_team", "generate_response") builder.add_edge("tech_team", "generate_response") builder.add_edge("generate_response", END) return builder.compile()

add_node("order_team", order_subgraph)这一行就是子图模式的核心。编译后的子图本身就是一个Runnable，可以直接当节点用。

3.2 子图输入输出与主图状态的桥接

这里有个必须理解清楚的机制：子图作为节点被调用时，它接收的是主图的完整状态，但只会读取自己 State Schema 里定义的字段。

什么意思？当主图执行到order_team节点时，LangGraph 会把当前的MainState传给子图。子图的OrderState里有messages和user_id，这两个字段会从MainState里自动提取。MainState里的intent、subgraph_result等字段，子图看不到，因为OrderState里没有定义。

子图执行完毕后，返回的OrderState会被合并回MainState。合并规则是：只合并两个 State 都有的字段。OrderState里的order_id、order_status这些字段，MainState里没有，所以会被丢弃。messages字段两边都有，会通过add_messagesreducer 合并。

这个机制意味着：如果你希望子图的某个输出能被主图使用，必须在主图 State 里定义同名字段，或者通过映射函数显式转换。我一开始没搞懂这个，子图算出了return_eligible，主图里死活拿不到，后来才发现是字段名不匹配。

3.3 子图内部工具调用的隔离与共享

工具调用是客服系统的核心。子图模式下，工具可以按子图隔离，也可以共享。

我的做法是：通用工具共享，领域工具隔离。比如“发送邮件通知”这个工具，订单子图和技术子图都可能用到，就放在一个公共模块里，两边都注册。但“查询订单数据库”这种工具只在订单子图里注册，“检索知识库”只在技术子图里注册。

这样做的好处是减少工具描述对模型的干扰。订单子图里的 Agent 看到的工具列表只有订单相关的五六个，选择准确率明显高于把十几个工具全塞给它。

工具注册的代码结构：

# 公共工具 common_tools = [send_notification, log_interaction] # 订单子图工具 order_tools = common_tools + [query_order_db, track_logistics, check_return_policy] # 技术子图工具 tech_tools = common_tools + [search_knowledge_base, diagnose_issue, generate_solution]

然后在各自的节点函数里，用llm.bind_tools(order_tools)绑定。注意bind_tools返回的是一个新的 LLM 实例，不会影响原来的。

有个坑要提醒：工具函数的 docstring 质量直接决定调用准确率。我一开始写的 docstring 很简略，比如"""查询订单"""，模型经常在“查询订单”和“查询物流”之间搞混。后来改成"""根据订单号查询订单的当前状态，包括支付状态、发货状态、预计送达时间。输入参数为订单号字符串。"""，准确率立刻上去了。docstring 要写清楚：这个工具做什么、什么时候用、输入是什么、输出是什么。

4. 意图识别子图的实现与路由策略

4.1 意图分类的提示词工程与结构化输出

意图识别子图是整个系统的入口，它的准确率直接影响后续所有环节。我的实现方案是用结构化输出强制模型返回固定格式。

from pydantic import BaseModel, Field from typing import Literal class IntentResult(BaseModel): intent: Literal["order", "tech", "complaint", "unknown"] = Field( description="用户意图分类：order=订单相关，tech=技术问题，complaint=投诉建议，unknown=无法判断" ) confidence: float = Field(description="分类置信度，0到1之间") reasoning: str = Field(description="分类理由，一句话说明")

然后用llm.with_structured_output(IntentResult)让模型直接输出 Pydantic 对象。这比让模型输出 JSON 字符串再解析要可靠得多，LangGraph 底层会处理格式验证和重试。

提示词方面，我用了 few-shot 的方式，给每个类别两个例子。关键是例子要覆盖边界情况。比如“我买的东西坏了”这句话，既可以归为订单问题（退换货），也可以归为技术问题（产品故障）。我在提示词里明确写了规则：涉及退换货、退款、物流的归为 order；涉及产品使用、功能异常、报错的归为 tech。这种边界规则不写清楚，模型会随机选。

4.2 条件边路由的写法与默认分支处理

意图识别完成后，主图需要根据intent字段路由到不同子图。条件边的写法：

def route_by_intent(state: MainState) -> str: intent = state.get("intent") if intent == "order": return "order" elif intent == "tech": return "tech" elif intent == "complaint": return "complaint" else: return "complaint" # 默认走投诉处理，由人工兜底 builder.add_conditional_edges( "classify_intent", route_by_intent, { "order": "order_team", "tech": "tech_team", "complaint": "generate_response" } )

这里有个设计决策：unknown意图走投诉处理分支。原因是投诉处理分支的提示词里包含了“如果无法确定用户诉求，引导用户提供更多信息”的逻辑，相当于一个兜底。如果走订单或技术分支，模型可能会强行处理一个它不理解的问题，产生幻觉。

条件边的映射字典必须覆盖路由函数所有可能的返回值，否则 LangGraph 会抛异常。我建议在路由函数里用Literal类型标注返回值，这样静态检查能帮你发现遗漏。

4.3 低置信度意图的降级与人工兜底

意图识别的confidence字段不是摆设。当置信度低于 0.6 时，我会走一个降级路径：不进入任何子图，直接生成一个澄清回复。

def route_by_intent(state: MainState) -> str: intent = state.get("intent") confidence = state.get("intent_confidence", 1.0) if confidence < 0.6: return "clarify" # ... 其余路由逻辑

澄清回复的提示词是：“用户的问题我暂时无法准确理解，请用更具体的描述告诉我您遇到了什么问题，比如‘我的订单还没发货’或‘产品无法开机’。” 这种引导能显著提高下一轮的识别准确率。

实测下来，加了置信度降级之后，误路由率从 12% 降到了 4% 左右。代价是多了一轮交互，但对于客服场景来说，问清楚比答错要好得多。

5. 订单处理子图的多Agent协作链路

5.1 订单查询Agent的工具绑定与参数提取

订单处理子图的第一个节点是订单查询。它的职责是从用户消息里提取订单号，然后调用工具查询订单状态。

参数提取用结构化输出：

class OrderQueryParams(BaseModel): order_id: str | None = Field(description="订单号，如果用户没有提供则为None") query_type: Literal["status", "logistics", "return"] = Field( description="查询类型：status=订单状态，logistics=物流信息，return=退换货咨询" )

如果order_id为 None，Agent 会生成一个回复向用户索要订单号，然后子图结束，等待下一轮对话。这里有个细节：子图结束时返回的状态里要保留order_id为 None 的信息，这样主图知道这次子图执行没有完成实际查询，需要在最终回复里体现“正在等待用户提供订单号”。

工具调用的实现：

def query_order_node(state: OrderState) -> dict: llm_with_tools = llm.bind_tools([query_order_db]) response = llm_with_tools.invoke(state["messages"]) if response.tool_calls: tool_call = response.tool_calls[0] order_id = tool_call["args"].get("order_id") result = query_order_db.invoke(tool_call) return { "messages": [response, result], "order_id": order_id, "order_status": result.get("status") } return {"messages": [response]}

注意这里返回的是dict，LangGraph 会自动合并到OrderState里。messages字段因为有add_messagesreducer，会追加而不是覆盖。

5.2 物流追踪与退换货判断的条件分支

订单查询完成后，根据query_type决定下一步。如果是logistics，走物流追踪节点；如果是return，走退换货判断节点。

def route_after_query(state: OrderState) -> str: last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: query_type = last_message.tool_calls[0]["args"].get("query_type") if query_type == "logistics": return "logistics" elif query_type == "return": return "return" return "done"

退换货判断节点会调用check_return_policy工具，传入订单状态和用户诉求，返回是否符合退换货条件。这个工具内部是一套规则引擎，不是 LLM 调用——能用确定性代码解决的问题，不要交给 LLM。规则引擎的输出是稳定的，LLM 每次调用可能有波动。

物流追踪节点类似，调用track_logistics工具获取物流信息，然后生成一个自然语言摘要。

5.3 子图内部循环与终止条件的控制

订单处理子图里有一个潜在的死循环风险：如果用户提供的订单号查不到，Agent 可能会反复尝试查询。我在子图里加了一个retry_count字段来控制。

class OrderState(TypedDict): # ... 其他字段 retry_count: int

在订单查询节点里，每次执行前检查retry_count，超过 2 次就直接返回“订单号查询失败，请确认后重试”，不再调用工具。

def query_order_node(state: OrderState) -> dict: if state.get("retry_count", 0) >= 2: return { "messages": [AIMessage(content="订单号查询失败，请确认订单号是否正确后重试。")], "action_taken": "query_failed" } # ... 正常查询逻辑 return {"retry_count": state.get("retry_count", 0) + 1, ...}

这个retry_count是子图内部状态，不会映射回主图。主图只看到最终的action_taken是query_failed，然后生成相应的用户回复。

6. 实测中暴露的三个反直觉问题

6.1 子图状态字段名冲突导致的静默覆盖

这个问题我排查了整整一个下午。现象是：订单子图明明设置了order_status，主图里读出来却是None。

原因是我在主图MainState里也定义了一个order_status字段，本意是想接收子图的结果。但 LangGraph 的合并逻辑是：子图返回的状态会覆盖主图同名字段。而子图执行过程中，order_status在某个分支里被设为了None（因为那个分支不需要订单状态），这个None覆盖了之前查询到的值。

解决方案是避免父子图字段名重叠。子图内部字段加前缀，比如order_status改成sub_order_status，或者干脆不在主图定义同名字段，通过映射函数显式转换。我最后选择了后者，主图里只有subgraph_result一个字段来承载子图输出。

6.2 工具调用结果在子图边界丢失的排查过程

另一个坑是：子图内部调用了工具，工具返回了结果，但子图结束后主图拿不到工具调用的详细信息。

排查发现，工具调用的结果是以ToolMessage的形式存在子图的messages里的。子图返回时，messages会通过add_messages合并到主图。但问题是，主图的generate_response节点在生成最终回复时，看到的是合并后的完整消息历史，里面包含了工具调用的原始 JSON 结果。这些结果对用户不友好，而且可能包含内部字段。

我的解决方案是在子图内部就把工具结果转换成自然语言摘要，存到action_taken字段里。主图只读action_taken，不直接读messages里的工具消息。同时，在映射函数里过滤掉ToolMessage，只保留AIMessage和HumanMessage。

def map_order_result(order_state: OrderState) -> dict: clean_messages = [ msg for msg in order_state["messages"] if not isinstance(msg, ToolMessage) ] return { "subgraph_result": { "action": order_state.get("action_taken"), "summary": clean_messages[-1].content if clean_messages else "" }, "messages": clean_messages[-1:] # 只保留最后一条AI消息 }

6.3 多轮对话中子图重复执行的性能损耗

客服场景是多轮对话，用户可能连续发好几条消息。我最初的实现是：每轮对话都重新执行整个主图，包括意图识别子图。

这导致了一个问题：用户第一轮说“我的订单没到”，意图识别为order，进入订单子图。第二轮用户补充“订单号是 12345”，意图识别又跑了一遍，可能识别成unknown（因为“订单号是12345”这句话本身没有明确的意图词），然后走了澄清分支，体验很割裂。

解决方案是在意图识别节点里加入对话历史感知。提示词里不仅包含当前用户消息，还包含最近三轮的对话摘要。这样模型能看到“上一轮用户说的是订单问题”，从而保持意图一致性。

def classify_intent_node(state: MainState) -> dict: recent_messages = state["messages"][-6:] # 最近三轮 prompt = f""" 根据以下对话历史判断用户当前意图。 对话历史： {format_messages(recent_messages)} 注意：如果用户当前消息是对上一轮问题的补充，应保持与上一轮相同的意图分类。 """ # ... 调用 LLM

这个改动让多轮场景下的意图一致性从 70% 提升到了 92%。代价是提示词变长了，每次调用的 token 消耗增加约 30%，但客服场景对延迟不敏感，这个 trade-off 是值得的。

7. 从子图模式延伸出的可观测性设计

7.1 每个子图节点的执行日志埋点

子图模式的一个隐藏优势是：每个子图天然就是一个可观测单元。我在每个子图的入口和出口都加了日志埋点，记录子图名称、输入状态摘要、输出状态摘要、执行耗时。

import time import logging logger = logging.getLogger("subgraph") def with_logging(subgraph_name: str, node_func): def wrapper(state): start = time.time() logger.info(f"[{subgraph_name}] 开始执行，输入字段：{list(state.keys())}") result = node_func(state) elapsed = time.time() - start logger.info(f"[{subgraph_name}] 执行完成，耗时 {elapsed:.2f}s，输出字段：{list(result.keys())}") return result return wrapper

这个装饰器可以套在子图内部的每个节点上，也可以套在整个子图节点上。我两种都用了：子图级别看整体耗时，节点级别看内部瓶颈。

实测发现，订单查询工具调用平均耗时 800ms，LLM 生成回复平均耗时 1.2s，意图识别平均耗时 600ms。这些数据帮助我定位了优化点：把订单查询工具改成异步的，整体响应时间从 3.5s 降到了 2.1s。

7.2 子图执行路径的可视化追踪

LangGraph 内置了get_graph().draw_mermaid()方法可以生成流程图，但我更推荐用 LangSmith 做执行追踪。它能记录每次运行的完整调用链，包括每个节点的输入输出、LLM 调用的提示词和回复、工具调用的参数和结果。

配置方式很简单，设置环境变量即可：

export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_key export LANGCHAIN_PROJECT=customer-service

然后在代码里不需要任何额外改动，LangGraph 会自动上报追踪数据。在 LangSmith 的界面上，你可以看到主图和子图的嵌套关系，点击子图节点可以展开看内部执行细节。这个对调试多智能体系统来说几乎是必备的。

7.3 基于子图粒度的A/B测试方案

子图模式让 A/B 测试变得非常自然。因为每个子图是独立的，你可以准备两个版本的订单处理子图，在运行时根据用户 ID 哈希决定用哪个版本。

def build_main_graph(ab_variant: str = "A"): if ab_variant == "A": order_subgraph = build_order_subgraph_v1() else: order_subgraph = build_order_subgraph_v2() # ... 构建主图

我实际用这个方案测试了两版退换货判断的提示词，A 版准确率 85%，B 版 91%，直接切换 B 版上线。整个过程不需要改动主图代码，也不需要重新部署整个服务。

8. 生产环境部署时的几个关键决策

8.1 子图编译时机与冷启动优化

子图的编译（builder.compile()）是有成本的，尤其是当子图内部节点多、边复杂的时候。我实测一个包含 6 个节点的子图，编译耗时约 200ms。

如果在每次请求时都重新编译，累积起来很可观。我的做法是在服务启动时预编译所有子图和主图，存为全局变量。FastAPI 的lifespan机制很适合做这件事：

from contextlib import asynccontextmanager from fastapi import FastAPI graphs = {} @asynccontextmanager async def lifespan(app: FastAPI): graphs["main"] = build_main_graph() graphs["order"] = build_order_subgraph() graphs["tech"] = build_tech_subgraph() yield graphs.clear() app = FastAPI(lifespan=lifespan)

这样每个请求进来时直接复用编译好的图，冷启动时间从 800ms 降到了 50ms 以内。

8.2 子图级别的超时与熔断策略

生产环境里，任何一个子图卡住都会拖垮整个请求。我给每个子图节点加了超时控制。

LangGraph 本身没有内置超时机制，但可以通过asyncio.wait_for包装：

import asyncio async def invoke_with_timeout(graph, state, timeout=10.0): try: return await asyncio.wait_for(graph.ainvoke(state), timeout=timeout) except asyncio.TimeoutError: return {"subgraph_result": {"action": "timeout", "summary": "处理超时，请稍后重试"}}

超时时间根据子图复杂度设置：意图识别 5s，订单处理 10s，技术支持 15s。超时后返回一个降级结果，而不是让整个请求挂起。

熔断方面，我用了一个简单的计数器：如果某个子图连续 5 次超时，接下来 60 秒内直接返回降级结果，不再尝试执行。这能防止某个下游服务故障时整个系统雪崩。

8.3 对话状态持久化与子图恢复

客服系统需要支持用户中断后继续对话。LangGraph 提供了 checkpointer 机制来做状态持久化。

from langgraph.checkpoint.sqlite import SqliteSaver checkpointer = SqliteSaver.from_conn_string("checkpoints.db") main_graph = build_main_graph().compile(checkpointer=checkpointer)

调用时传入config={"configurable": {"thread_id": user_session_id}}，LangGraph 会自动保存和恢复状态。

这里有个细节：子图的状态也会被持久化。如果用户在订单子图执行到一半时断开，下次连接时会从子图的断点继续。这个行为大部分时候是好的，但如果你更新了子图的逻辑，旧的状态可能和新逻辑不兼容。我的做法是给子图状态加一个version字段，版本不匹配时清空重来。

9. 写在最后的一些个人体会

子图模式不是银弹，它解决的是“多智能体职责隔离”这个特定问题。如果你的系统只有两个 Agent，用条件边直接路由可能更简单。但一旦 Agent 数量超过三个，或者你发现状态字段开始失控，子图模式的价值就体现出来了。

我在这个项目里最大的收获是：状态设计比 Agent 设计更重要。花在MainState和OrderState字段定义上的时间，远比花在提示词调优上的时间值得。字段定义清楚了，每个 Agent 的职责边界就清楚了，提示词反而变得简单。

另一个体会是：能用代码解决的问题不要交给 LLM。退换货判断、订单状态查询、物流信息提取，这些都有确定性答案，用工具和规则引擎处理，LLM 只负责理解用户意图和生成自然语言回复。这样系统的稳定性和可测试性都会好很多。

最后分享一个调试技巧：在开发阶段，给每个子图节点加一个debug模式，把输入输出状态打印到控制台。LangGraph 的执行流程有时候不太直观，尤其是条件边和子图嵌套的场景，打印出来看一遍比盯着代码想半天快得多。