多智能体协作这件事我从去年就开始折腾了。最开始用单 Agent 硬扛所有任务结果 prompt 越写越长工具越挂越多最后模型自己都绕晕了——用户问一句“我要退款”它既想查订单又想调物流还想直接触发退款流程三件事搅在一起输出质量断崖式下跌。后来我把这套东西拆成多个专职 Agent用 LangGraph 的子图模式做编排才真正把 AI 客服系统跑顺。这篇就聊聊我怎么用 StateGraph 把“接待—查询—决策—执行”这条链路拆开、拼起来以及中间踩过的那些坑。如果你正在做 AI 客服、任务型对话或者任何需要多个角色分工的 Agent 系统这套子图协作的思路可以直接抄。哪怕你刚接触 LangGraph只要会写 Python 函数、理解“状态”这个概念跟着走一遍就能搭出可运行的骨架。1. 为什么客服场景必须上多智能体子图1.1 单 Agent 的三个死穴先说清楚为什么不能一个 Agent 干到底。我最早那版就是一个大 ReAct Agent挂了查订单、查物流、退款、转人工四个工具。上线第一天就出问题意图混淆用户说“我上周买的东西怎么还没到”模型有时候调查订单有时候调查物流甚至两个都调然后自己编一个“预计明天到”的答案。上下文污染退款流程需要确认订单号、金额、退款原因这些信息在长对话里被前面的闲聊稀释模型经常漏问关键字段。无法局部重试查物流失败时整个 Agent 要重跑一遍前面查订单的结果也丢了token 哗哗烧。这三个问题的本质是一个 Agent 同时承担了路由、查询、决策、执行四种职责而每种职责对 prompt、工具集、上下文的要求完全不同。1.2 子图模式到底解决了什么LangGraph 的子图Subgraph本质是把一个编译好的 StateGraph 当作另一个 StateGraph 的节点。听起来简单但它带来的好处很实在维度单 Agent多智能体子图职责边界模糊靠 prompt 约束每个子图一个职责代码级隔离状态管理全局共享易污染子图有独立 state按需读写父状态重试粒度整体重跑只重跑失败子图可测试性难要 mock 整个链路每个子图可单独跑单测扩展性加工具就改 prompt加子图就是加节点我实测下来拆成子图后退款流程的字段完整率从 62% 提到了 94%因为退款子图有自己独立的 state schema强制校验必填字段缺了就回到追问节点不会漏。1.3 客服系统的角色拆分思路我最终拆成四个子图对应客服的四个阶段接待子图Reception识别意图、提取实体、判断是否需要转人工。查询子图Query根据意图调对应工具查订单、查物流、查售后政策。决策子图Decision根据查询结果判断能不能自助解决能就生成方案不能就升级。执行子图Execution真正触发退款、改地址、发补发单等写操作。为什么这么拆因为读操作和写操作的风险等级完全不同。查询可以随便重试执行必须幂等且可审计。把它们放在不同子图我可以在执行子图里加人工确认节点而查询子图完全自动化。2. 核心状态设计父子图之间怎么传数据2.1 父图 State 的字段规划父图是整个客服会话的主状态我用 TypedDict 定义关键字段如下from typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages class CustomerServiceState(TypedDict): messages: Annotated[list, add_messages] # 完整对话历史 user_id: str # 用户标识 intent: Literal[query_order, query_logistics, refund, human, unknown] entities: dict # 提取的实体如 order_id query_result: dict # 查询子图回填 decision: Literal[self_solve, escalate, need_more_info] execution_result: dict # 执行子图回填 need_human: bool这里有个关键点messages 用 add_messages 注解这是 LangGraph 内置的 reducer保证多节点写入时消息是追加而不是覆盖。我一开始没加结果子图返回的消息把父图历史冲掉了排查了半天。2.2 子图 State 的独立与共享子图可以有自己的 state schema也可以复用父图的。我的做法是接待子图独立 state只读 messages输出 intent 和 entities。查询子图独立 state输入 intent entities输出 query_result。决策子图复用父图 state因为要综合看 intent、entities、query_result。执行子图独立 state输入 decision entities输出 execution_result。为什么查询和执行用独立 state因为这两个子图内部有循环比如查询失败重试、执行前确认独立 state 让循环条件更清晰不会误触发父图的其他节点。2.3 状态映射的两种写法子图接入父图时状态怎么对接有两种方式方式一共享 key直接透传。如果子图 state 的字段名和父图一致LangGraph 会自动映射。比如父图有messages子图也有messages就直接传。方式二显式包装节点。我更喜欢这种可控性强def call_reception_subgraph(state: CustomerServiceState): sub_input {messages: state[messages], user_id: state[user_id]} sub_output reception_graph.invoke(sub_input) return { intent: sub_output[intent], entities: sub_output[entities], }这样父图只拿到它需要的字段子图内部的中间变量不会污染父状态。踩过的坑是如果直接add_node(reception, reception_graph)子图所有输出字段都会往父图塞字段名冲突时行为很诡异。3. 四个子图的实操实现3.1 接待子图意图识别与实体抽取接待子图的核心是一个结构化输出节点。我用with_structured_output让模型直接吐 JSONfrom pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class ReceptionOutput(BaseModel): intent: Literal[query_order, query_logistics, refund, human, unknown] order_id: str | None Field(defaultNone, description订单号如 ORD12345) reason: str | None Field(defaultNone, description退款原因等) llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(ReceptionOutput) def reception_node(state): result structured_llm.invoke(state[messages]) return {intent: result.intent, entities: result.model_dump()}注意事项temperature 一定要设 0意图分类这种任务不需要创造性。另外 order_id 的 description 要写清楚格式否则模型会把“上周买的”当成订单号。接待子图还要加一个转人工判断如果 intent 是 human或者连续两轮 unknown直接置need_humanTrue父图路由到结束节点。3.2 查询子图工具调用与结果归一化查询子图内部是一个 ReAct 循环挂三个工具get_order、get_logistics、get_refund_policy。关键设计是结果归一化——不同工具返回的字段不一样我统一转成一个 dictdef normalize_query_result(raw, query_type): if query_type query_order: return {order_status: raw[status], amount: raw[total], items: raw[items]} elif query_type query_logistics: return {logistics_status: raw[state], eta: raw[estimated_arrival]} ...为什么要归一化因为决策子图要基于统一结构做判断如果每个工具返回格式不同决策逻辑会变成一堆 if-else。归一化后决策子图只看order_status、logistics_status这些标准字段。查询子图的重试策略工具调用失败时最多重试 2 次每次间隔用指数退避。超过 2 次就返回query_result{error: ...}让决策子图走降级路径。3.3 决策子图规则与模型混合判断决策子图我没有完全交给模型而是规则优先、模型兜底。因为退款能不能自助有明确的业务规则def decision_node(state): qr state[query_result] if qr.get(error): return {decision: escalate} if state[intent] refund: if qr[order_status] delivered and qr[days_since_delivery] 7: return {decision: self_solve} else: return {decision: escalate} if state[intent] query_logistics: return {decision: self_solve} # 规则覆盖不到的交给模型 return {decision: llm_decide(state)}这样做的理由是业务规则是确定的模型判断有随机性。退款这种涉及钱的场景规则能覆盖 80% 的情况剩下 20% 的模糊场景再让模型判断既保证准确性又控制成本。3.4 执行子图幂等与人工确认执行子图是最危险的部分因为它会真正改数据。我加了两道保险第一道幂等键。每次执行前生成一个idempotency_key f{user_id}_{intent}_{order_id}执行接口收到相同 key 直接返回上次结果防止用户连点两次退款。第二道人工确认节点。对于退款金额超过阈值的执行子图内部先走到await_confirmation节点把确认请求写进 state父图路由到等待用户回复。用户确认后才继续执行。def execute_refund(state): key f{state[user_id]}_refund_{state[entities][order_id]} if state[entities].get(amount, 0) 500: return {execution_result: {status: pending_confirmation}, need_human: False} result refund_api(order_id..., idempotency_keykey) return {execution_result: result}4. 父图编排路由与条件边4.1 主图的节点与边父图结构很清晰from langgraph.graph import StateGraph, END builder StateGraph(CustomerServiceState) builder.add_node(reception, call_reception_subgraph) builder.add_node(query, call_query_subgraph) builder.add_node(decision, decision_node) builder.add_node(execution, call_execution_subgraph) builder.add_node(human_handoff, human_handoff_node) builder.set_entry_point(reception) builder.add_conditional_edges(reception, route_after_reception, { query: query, human: human_handoff, end: END, }) builder.add_edge(query, decision) builder.add_conditional_edges(decision, route_after_decision, { execution: execution, human: human_handoff, end: END, }) builder.add_edge(execution, END)4.2 条件路由的判断逻辑route_after_reception的逻辑def route_after_reception(state): if state[need_human]: return human if state[intent] unknown: return end # 追问一轮等用户补充 return queryroute_after_decision的逻辑def route_after_decision(state): if state[decision] self_solve: return execution if state[intent] refund else end if state[decision] escalate: return human return end这里有个细节查询类意图查订单、查物流决策为 self_solve 后直接 end因为查询结果已经在 query_result 里生成回复的节点可以放在 query 子图内部也可以单独加一个 respond 节点。我选择在 query 子图内部生成自然语言回复减少父图节点数。4.3 循环与终止条件父图本身没有循环但子图内部有。LangGraph 的终止靠recursion_limit默认 25。我实测客服场景设 15 就够超过说明有死循环。设置方式app builder.compile() result app.invoke(input_state, config{recursion_limit: 15})踩坑记录有一次查询子图的重试逻辑写错了失败后没有正确返回 error导致子图内部无限循环直接把 recursion_limit 打满报错。后来我在子图的重试节点加了计数器超过 2 次强制走 error 分支。5. 常见问题与排查实录5.1 状态字段丢失或覆盖现象子图返回后父图的 messages 只剩最后一条。原因子图 state 的 messages 没有用add_messages注解或者子图返回时直接返回了新的 list 而不是增量。解决确保父子图的 messages 字段都用Annotated[list, add_messages]子图返回时只返回新增消息。5.2 子图调用后父图路由不生效现象接待子图返回 intent 后条件边没有按预期走。原因包装节点返回的 dict 字段名和父图 state 不一致。比如返回了{intent_result: ...}但父图字段是intent。解决包装节点的返回值 key 必须和父图 state 的字段名严格一致。我现在的习惯是写一个assert set(return_dict.keys()) set(ParentState.__annotations__.keys())做校验。5.3 工具调用参数错误现象查询子图调 get_order 时传了order_idNone。原因接待子图没抽到 order_id但查询子图没做空值检查。解决查询子图入口加守卫if not state[entities].get(order_id): return {query_result: {error: missing_order_id}}让决策子图走追问路径。5.4 常见问题速查表问题排查方向快速修复子图输出丢失检查 state 字段名和 reducer统一字段名加 add_messages路由不跳转检查条件边映射 key打印 state 确认字段值无限循环检查子图重试计数加重试上限超限走 error意图识别不准检查 prompt 和 temperaturetemperature0补充 few-shot执行重复扣款检查幂等键用 user_idintentorder_id 做 key转人工不及时检查 need_human 触发条件连续 unknown 或负面情绪词触发5.5 独家避坑技巧技巧一给每个子图加 trace 标签。LangGraph 支持在 config 里传tags我在调用每个子图时打上标签配合 LangSmith 一眼就能看出哪个子图耗时最长、token 最多。技巧二子图单测用假 state。不要每次都跑完整链路直接构造一个 dict 传给子图invoke几秒钟就能验证逻辑。我每个子图都有 5-8 个单测用例覆盖正常、缺字段、工具报错三种情况。技巧三决策子图的规则要可配置。我把退款天数阈值、金额阈值抽到配置文件运营改规则不用改代码。上线三个月改了四次阈值全靠这个设计。技巧四执行子图先 dry-run。所有写操作接口都支持dry_runTrue参数先在测试环境跑一遍确认参数正确再切生产。这个习惯帮我避免了一次批量误退款。6. 性能与成本优化6.1 模型分级使用不是所有节点都需要 GPT-4o。我的配置接待子图gpt-4o-mini意图分类够用。查询子图gpt-4o-mini工具调用对模型要求不高。决策子图规则优先模型兜底用 gpt-4o。执行子图不用模型纯代码。这样整体成本比全用 gpt-4o 降了约 70%而准确率只降了不到 2 个百分点。6.2 子图缓存查询子图的结果可以缓存。同一个 order_id 在 5 分钟内重复查询直接返回缓存。我用 Redis 做了一层key 是query:{intent}:{order_id}TTL 300 秒。实测客服场景重复查询率约 18%缓存命中后响应从 1.2 秒降到 80 毫秒。6.3 并发与超时父图调用子图是同步的如果查询子图要调多个工具可以在子图内部用asyncio.gather并发。但要注意LangGraph 的节点默认是同步执行要用异步节点需要async def并且用ainvoke。超时设置每个子图调用包一层asyncio.wait_for超时 10 秒。超时后返回 error让决策子图走降级。这个在线上救过我好几次某个工具接口挂了不会拖垮整个会话。7. 后续可以怎么扩展这套骨架跑通后扩展方向很多。我目前在做的是多轮记忆把用户历史会话存进向量库接待子图先检索相似历史辅助意图判断。另一个方向是子图热插拔把子图注册成配置不同业务线挂不同的查询子图父图不用改。还有一个我觉得很有价值的点执行子图的操作审计。每次写操作都记一条审计日志包含 idempotency_key、操作前后状态、操作人用户还是人工。出了问题能快速回溯也满足合规要求。如果你也在做多智能体客服建议先从两个子图开始——接待和查询跑通后再加决策和执行。一上来就拆四个调试成本会很高。我第一版就是四个一起上光状态映射就调了两天。