多智能体协作这件事我在过去一年里反复折腾过好几套方案。最早用纯 LangChain 的 AgentExecutor 串起来跑三个 Agent 互相调用就开始出现上下文污染、状态丢失、死循环后来换成手写状态机维护成本高得离谱加一个节点要改五处代码。直到 LangGraph 出来尤其是它的子图Subgraph模式才算真正把多智能体协作这件事从玩具级别拉到了能上生产的程度。这篇内容我拿一个真实场景来拆——AI 客服系统。为什么选客服因为它天然是多角色协作有负责意图识别的、有查订单的、有处理退换货的、有兜底转人工的每个角色的工具集、提示词、甚至模型都不一样。用一个大 Agent 硬塞所有工具提示词会长到模型开始精神分裂用多个独立 Agent 又没法共享状态。子图模式恰好卡在中间这个甜点位上。下面我会从架构设计、子图拆分逻辑、状态传递、工具调用、踩坑排查一路讲到实测效果代码都是能直接跑的。如果你正在做多智能体系统或者被 LangGraph 的StateGraph和CompiledGraph绕晕过这篇应该能帮你省下不少试错时间。1. 为什么 AI 客服是多智能体子图模式的天然试验场1.1 单 Agent 塞所有工具提示词会先崩先说个我踩过的真实坑。最开始做客服系统我图省事把订单查询、物流追踪、退换货申请、优惠券核销、投诉建议这五类工具全挂在一个 Agent 上系统提示词写了大概 1200 字。测试阶段还行一上真实流量就出问题用户问我上周买的鞋还没到能退吗模型会同时触发query_order和check_logistics然后拿着两个工具的返回结果开始编——它会把物流信息里的运输中理解成可以退直接调用apply_refund。这不是模型笨是工具语义边界模糊 上下文过长导致注意力稀释。当工具数量超过 5 个、提示词超过 800 字模型对什么时候该用哪个工具的判断准确率会明显下降。我做过一组对照测试同样 200 条客服对话工具数量提示词长度意图识别准确率工具误调用率3 个约 400 字94%3%5 个约 800 字87%11%8 个约 1300 字76%23%数据很直白工具越多误调用越严重。而客服场景的工具只会越来越多不可能靠精简工具解决。1.2 子图模式解决的是职责隔离而不是功能拆分很多人第一次接触 LangGraph 子图会以为它就是把大图拆小图方便管理。这个理解只对了一半。子图真正的价值在于每个子图是一个独立的、有自己状态和工具集的执行单元它对外只暴露一个入口和一个出口内部怎么折腾外面不用管。放到客服场景里这意味着意图路由子图只负责判断用户想干什么工具集极小就一个分类器准确率能拉到 95% 以上订单处理子图只处理订单相关工具是query_order、modify_order、cancel_order售后子图只处理退换货工具是check_refund_policy、apply_refund、query_logistics兜底子图处理无法识别或情绪激动的对话直接转人工每个子图的提示词都能控制在 300 字以内工具不超过 4 个。这就是职责隔离——不是把功能拆开而是把决策上下文拆开让每个决策点都足够干净。1.3 子图和多个独立 Agent的本质区别有人会问那我直接起四个独立的 Agent用代码 if-else 调度不就行了我试过问题出在状态共享上。独立 Agent 之间传状态你得手动序列化、手动拼 prompt、手动处理上一个 Agent 说了什么。用户说我要退那个蓝色的这个那个指代的是上一轮订单查询的结果独立 Agent 拿不到这个上下文只能重新问一遍体验直接崩。LangGraph 子图模式的核心优势是共享 State。父图和子图可以定义同一套 State schema子图执行完把结果写回 State父图下一个节点直接读。用户说退那个蓝色的售后子图能从 State 里拿到订单子图之前写入的订单列表直接定位到具体订单。这个能力是独立 Agent 拼凑方案给不了的。2. 客服系统的子图拆分按决策边界而不是业务模块2.1 拆分的判断标准一个子图只做一类决策拆分粒度是子图模式最容易做错的地方。我见过有人按业务模块拆订单一个图、物流一个图、售后一个图结果发现物流查询在订单和售后里都要用状态传来传去很乱。正确的拆分标准是决策边界一个子图内部只做一类决策且这类决策所需的上下文是自洽的。具体到客服路由子图决策用户意图是什么输入是用户原话 历史对话输出是意图标签订单子图决策订单状态是什么、要不要改输入是意图 用户身份输出是订单操作结果售后子图决策能不能退、退多少输入是订单信息 售后政策输出是处理方案兜底子图决策是否转人工输入是前面所有结果 情绪判断输出是转人工或安抚话术注意物流查询没有单独成图因为它不是一个决策而是一个数据获取动作被订单和售后子图各自作为工具调用即可。有决策才成图纯查询做成工具这条线划清楚架构就不会乱。2.2 父图只做路由不碰业务逻辑父图的职责要极度克制。我的做法是父图只有三个节点router调用路由子图、dispatch根据路由结果分发到对应子图、aggregate汇总结果生成回复。父图本身不写任何业务判断所有 if-else 都在dispatch的条件边里。这样做的好处是父图极其稳定。业务逻辑变化只影响子图父图几乎不用动。我上线三个月售后政策改了四次只改了售后子图父图一行没动。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, Literal import operator class CustomerServiceState(TypedDict): messages: Annotated[list, operator.add] user_id: str intent: str order_info: dict after_sale_result: dict final_response: str def build_parent_graph(): builder StateGraph(CustomerServiceState) builder.add_node(router, router_node) builder.add_node(order_agent, order_subgraph) builder.add_node(after_sale_agent, after_sale_subgraph) builder.add_node(fallback_agent, fallback_subgraph) builder.add_node(aggregate, aggregate_node) builder.set_entry_point(router) builder.add_conditional_edges( router, route_decision, { order: order_agent, after_sale: after_sale_agent, fallback: fallback_agent } ) builder.add_edge(order_agent, aggregate) builder.add_edge(after_sale_agent, aggregate) builder.add_edge(fallback_agent, aggregate) builder.add_edge(aggregate, END) return builder.compile()2.3 子图作为节点接入父图的两种方式LangGraph 里子图接入父图有两种写法我两种都用过各有适用场景。第一种把编译好的子图直接当节点加进去。builder.add_node(order_agent, compiled_order_subgraph)。这种方式最简洁子图的 State 和父图 State 字段名一致时LangGraph 会自动做映射。适合子图和父图 State 结构高度重合的情况。第二种包一层函数手动做 State 转换。子图有自己的 State schema父图调用时手动把需要的字段传进去执行完再把结果写回父图 State。这种方式灵活适合子图 State 和父图差异大的情况。def order_subgraph_wrapper(state: CustomerServiceState): sub_input { messages: state[messages], user_id: state[user_id], intent: state[intent] } sub_result compiled_order_subgraph.invoke(sub_input) return { order_info: sub_result[order_info], messages: sub_result[messages] }我实际项目里用的是第二种。因为订单子图内部需要维护当前查询到第几页是否已确认订单号这些临时状态父图不需要知道这些包一层做隔离更干净。提示子图 State 里如果有父图没有的字段用第一种方式会报 KeyError。这时候要么统一 schema要么用第二种包装方式。我建议新手直接用第二种虽然多写几行但调试时心智负担小很多。3. State 设计与跨子图数据传递的实操细节3.1 State 字段设计区分累积型和覆盖型State 设计是子图模式的地基设计错了后面全是坑。核心原则是区分累积型字段和覆盖型字段。累积型字段用Annotated[list, operator.add]比如messages每轮对话都往里追加不能覆盖。覆盖型字段直接写类型比如intent、order_info新值直接替换旧值。我见过最典型的错误是把order_info也写成累积型结果用户查了三次订单State 里堆了三个订单信息售后子图拿到之后不知道该用哪个。只有对话历史是累积的业务数据都是覆盖的这条记牢。class CustomerServiceState(TypedDict): # 累积型对话历史 messages: Annotated[list, operator.add] # 覆盖型业务数据 user_id: str intent: str order_info: dict after_sale_result: dict retry_count: int final_response: str3.2 子图之间怎么看到彼此的结果子图之间不直接通信全部通过父图 State 中转。订单子图把order_info写进 State售后子图从 State 读order_info。这个机制听起来简单但有个细节要注意子图读 State 时读到的是父图当前的最新值不是子图被创建时的快照。这意味着如果父图在dispatch之后、子图执行之前又改了 State子图会读到新值。这个特性在串行流程里没问题但在并行子图里要小心。LangGraph 支持并行执行多个子图如果两个子图同时写同一个字段后写的会覆盖先写的且顺序不确定。我的做法是并行子图只写各自独立的字段。比如订单子图只写order_info售后子图只写after_sale_result绝不交叉。需要汇总时在aggregate节点统一处理。3.3 用 reducer 处理并发写入冲突如果确实需要多个子图写同一个字段必须自定义 reducer。LangGraph 允许你给字段指定合并函数def merge_dict(existing: dict, new: dict) - dict: if existing is None: return new return {**existing, **new} class CustomerServiceState(TypedDict): messages: Annotated[list, operator.add] collected_data: Annotated[dict, merge_dict]merge_dict会把两个子图写入的字典合并而不是覆盖。这个技巧在处理多个子图各自收集了一部分用户信息时特别有用。但我要提醒一句能用独立字段解决就别用 reducerreducer 的合并逻辑一旦复杂起来调试难度是指数级上升的。3.4 状态持久化checkpointer 让对话可恢复客服系统有个硬需求用户中途离开回来要能接着聊。LangGraph 的 checkpointer 机制就是干这个的。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph build_parent_graph().compile(checkpointermemory) config {configurable: {thread_id: user_123_session_1}} result graph.invoke(input_state, config)thread_id是会话标识同一个thread_id的多次 invoke 会共享 State。用户下次进来带上同样的thread_id直接从上次的状态继续。生产环境别用MemorySaver它是内存存储重启就没了。我用的是SqliteSaver或者接 Redis 的自定义 checkpointer。这里有个坑checkpointer 存的是完整 State 快照如果 State 里有大对象比如整个订单列表存储会膨胀得很快。我的做法是 State 里只存订单 ID 和关键字段完整订单数据放数据库需要时再查。4. 工具调用在子图里的正确姿势4.1 工具绑定到子图而不是全局工具绑定位置很关键。我一开始把所有工具定义在全局然后每个子图自己去挑结果提示词里还是得列出所有工具让模型知道有哪些可用等于没隔离。正确做法是每个子图只绑定自己的工具。订单子图的 LLM 只 bind 订单相关工具售后子图只 bind 售后工具。这样模型在订单子图里根本看不到apply_refund从物理上杜绝了误调用。from langchain_openai import ChatOpenAI order_tools [query_order, modify_order, cancel_order] after_sale_tools [check_refund_policy, apply_refund, query_logistics] def build_order_subgraph(): llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(order_tools) builder StateGraph(OrderState) builder.add_node(agent, create_agent_node(llm_with_tools)) builder.add_node(tools, ToolNode(order_tools)) # ... 省略边定义 return builder.compile()4.2 工具返回结果的结构化处理工具返回的东西不能直接塞给模型尤其是查询类工具。query_order返回一个包含 20 个字段的订单对象全塞进 prompt 会浪费 token 还干扰判断。我的做法是工具内部就做结构化裁剪只返回模型决策需要的字段def query_order(order_id: str) - dict: raw db.get_order(order_id) return { order_id: raw[id], status: raw[status], amount: raw[amount], can_refund: raw[status] in [paid, shipped], refund_deadline: raw[refund_deadline] }注意can_refund这个字段它不是数据库原生字段是我根据业务规则算出来的。把业务判断前置到工具层而不是让模型去判断这是提升准确率的关键。模型看到can_refund: true就知道能退不用去理解已支付和已发货状态可退这种规则。4.3 工具调用失败的重试与降级工具调用失败是常态网络抖动、数据库超时、第三方接口限流都会发生。LangGraph 里处理工具失败有两种方式。方式一在工具内部 try-except返回错误信息让模型决策。适合可恢复的错误比如订单号格式不对模型可以引导用户重新提供。方式二用 LangGraph 的 retry 机制。builder.add_node(tools, ToolNode(tools), retryRetryPolicy(max_attempts3))。适合瞬时故障比如网络超时。我实际用的是组合策略网络类错误自动重试 3 次业务类错误返回结构化错误让模型处理。这里有个细节重试次数要写进 State避免无限重试。我在 State 里加了retry_count字段超过阈值直接走兜底子图转人工。from langgraph.pregel import RetryPolicy retry_policy RetryPolicy( max_attempts3, retry_onlambda e: isinstance(e, (TimeoutError, ConnectionError)) ) builder.add_node(tools, ToolNode(order_tools), retryretry_policy)4.4 工具调用的可观测性多智能体系统最难的不是写是调试。一个请求经过路由子图、订单子图、售后子图中间调了五次工具出问题时你根本不知道是哪一步错了。我的做法是每个子图的每个节点都打结构化日志记录输入 State 摘要、输出 State 摘要、耗时、工具调用记录。LangGraph 支持 callback可以挂 LangSmith 或者自建 callbackclass LoggingCallback(BaseCallbackHandler): def on_tool_start(self, serialized, input_str, **kwargs): logger.info(ftool_start: {serialized[name]}, input: {input_str}) def on_tool_end(self, output, **kwargs): logger.info(ftool_end: {output[:200]}) def on_chain_start(self, serialized, inputs, **kwargs): logger.info(fnode_start: {serialized.get(name)})这些日志在排查为什么售后子图没拿到订单信息这类问题时是救命的。我建议从第一天就加上别等出问题再补。5. 实测中暴露的五个典型问题与排查链路5.1 子图无限循环路由子图反复判定同一意图现象用户问我要退货系统卡住不返回日志显示路由子图被调用了十几次。排查过程先看路由子图的输出发现它每次都返回after_sale但父图没有正确分发。检查add_conditional_edges的映射发现route_decision函数返回的是after_sale但映射字典里写的是aftersale少了下划线。LangGraph 找不到匹配的边默认回到入口节点形成死循环。修复统一命名并且在route_decision里加默认分支def route_decision(state) - Literal[order, after_sale, fallback]: intent state.get(intent, unknown) if intent in [order_query, order_modify]: return order elif intent in [refund, return]: return after_sale return fallback # 兜底绝不返回未定义的值经验条件边的返回值必须有兜底分支且映射字典的 key 要和返回值严格一致。我后来养成了习惯把路由值定义成枚举从源头杜绝拼写错误。5.2 State 字段被覆盖售后子图读不到订单信息现象用户先查订单再说退这个售后子图报错说order_info为空。排查过程打印 State 发现order_info确实有值但售后子图读的时候是空的。检查发现订单子图返回时写的是{order_info: ...}但父图的order_subgraph_wrapper里写成了{order: ...}字段名不匹配LangGraph 静默忽略了未知字段。修复wrapper 返回的字段名必须和父图 State schema 完全一致。我后来加了个校验函数在 wrapper 返回前检查字段名def validate_state_update(update: dict, schema: type) - dict: valid_keys schema.__annotations__.keys() for k in update: if k not in valid_keys: raise ValueError(fUnknown state field: {k}) return update经验LangGraph 对未知 State 字段是静默忽略的不报错。这个设计很坑一定要自己加校验。5.3 工具调用参数错误模型传了不存在的订单号现象售后子图调用apply_refund时传了order_idunknown工具报错。排查过程看对话历史用户说退了吧模型没有从 State 里读order_info而是自己编了个订单号。原因是售后子图的提示词里没有明确告诉模型订单号从 State 的 order_info 字段取。修复在子图提示词里显式注入 State 数据def after_sale_agent_node(state): order_info state.get(order_info, {}) system_prompt f你是售后处理专员。 当前订单信息{json.dumps(order_info, ensure_asciiFalse)} 如果订单信息为空请先引导用户提供订单号不要自己编造。 # ...经验模型不会自动去 State 里找数据必须在提示词里显式喂给它。而且要给数据为空时怎么办的指令否则模型会幻觉。5.4 子图编译缓存导致的旧逻辑残留现象改了售后子图的提示词重启服务后行为没变。排查过程代码确实改了日志里打印的提示词却是旧的。原因是子图在模块加载时编译了一次缓存在全局变量里热重载没生效。修复把子图编译放到函数里每次构建父图时重新编译。或者用工厂模式def get_after_sale_subgraph(): # 每次调用重新编译确保拿到最新逻辑 return build_after_sale_subgraph()经验LangGraph 的compile()结果是有状态的别在模块顶层编译后全局复用尤其是在开发阶段。生产环境可以缓存但要有明确的失效机制。5.5 并发请求下 State 串号现象用户 A 的对话里出现了用户 B 的订单信息。排查过程这是最严重的问题。检查发现thread_id生成逻辑有 bug用了时间戳取模高并发下不同用户生成了相同的thread_id共享了 State。修复thread_id必须全局唯一用 UUIDimport uuid thread_id fuser_{user_id}_{uuid.uuid4().hex}经验checkpointer 的隔离完全依赖thread_id这个 ID 的生成逻辑要极其严谨。我后来直接用了user_id session_id的组合session_id 由前端生成 UUID 传入从源头保证唯一。6. 性能与成本子图模式到底值不值6.1 延迟对比多一次路由但省了长上下文子图模式比单 Agent 多了一次路由调用理论上延迟更高。但实测下来总延迟反而更低。原因是单 Agent 每次都要处理 1200 字的系统提示词 全部工具定义输入 token 在 2000 左右。子图模式下路由子图输入约 500 token业务子图输入约 800 token虽然调用了两次但每次的输入都短总 token 消耗反而下降。方案平均输入 token平均延迟意图准确率单 Agent 全工具21003.2s87%子图模式500 8002.6s95%延迟降低主要来自首 token 时间缩短输入短了模型开始生成更快。6.2 成本账路由调用是额外开销但误调用减少省更多路由子图用gpt-4o-mini就够了一次调用成本极低。真正省钱的地方在于误调用减少。单 Agent 方案下 11% 的误调用率意味着每 100 次对话有 11 次调用了错误的工具这些错误调用要么返回无用数据浪费 token要么触发错误操作需要人工介入。子图模式把误调用率压到 3% 以下省下的 token 和人工成本远超路由调用的开销。我算过一笔账日均 5000 次对话的场景下子图模式每月能省 30% 左右的 API 成本。6.3 什么情况下不该用子图子图不是银弹。如果你的场景满足以下任一条件单 Agent 可能更合适工具少于 3 个拆分收益不明显反而增加复杂度对话轮次极短比如单轮问答没有跨轮状态需求意图高度单一用户进来只干一件事不需要路由我做过一个内部工具就一个查数据功能硬套子图模式结果路由子图永远返回同一个意图纯属浪费。架构要匹配问题复杂度别为了用而用。7. 从 Demo 到生产还差哪几步7.1 子图的独立测试子图最大的好处之一是可以独立测试。每个子图编译后就是一个独立的Runnable可以单独 invokedef test_order_subgraph(): sub build_order_subgraph() result sub.invoke({ messages: [HumanMessage(content查一下订单 12345)], user_id: test_user, intent: order_query }) assert result[order_info][order_id] 12345我要求团队每个子图都要有独立的单元测试覆盖正常流程、工具失败、数据为空三种情况。父图的集成测试单独写。这样出问题时能快速定位是子图问题还是路由问题。7.2 灰度发布与子图版本管理子图模式让灰度发布变得容易。因为子图是独立编译的可以给不同用户路由到不同版本的子图def route_to_subgraph(state): if state[user_id] in beta_users: return order_agent_v2 return order_agent_v1我实际做的时候新版本子图先给 5% 流量观察一周准确率和用户满意度没问题再全量。这个能力在单 Agent 方案里很难实现因为改一处提示词影响全局。7.3 监控指标该看什么多智能体系统的监控和单 Agent 不一样要分层看路由层意图识别准确率、路由分布、兜底触发率子图层各子图调用次数、平均耗时、工具调用成功率工具层各工具调用次数、失败率、平均返回时间端到端对话完成率、转人工率、用户满意度我特别关注兜底触发率这个指标突然升高通常意味着路由子图出了问题或者出现了新的用户意图没被覆盖。还有子图间状态传递失败率这个指标能提前发现 State schema 不匹配的问题。7.4 提示词版本化子图的提示词要当代码管理进 Git有版本号。我见过太多团队提示词改来改去最后不知道线上跑的是哪版。我的做法是提示词单独放一个模块每个子图的提示词有明确的版本注释AFTER_SALE_PROMPT_V3 你是售后处理专员... # 变更记录 # v1 - 初始版本 # v2 - 增加退款政策说明 # v3 - 增加订单信息为空时的引导话术 配合灰度发布出问题能快速回滚到上一版提示词。8. 几个让我少走弯路的实操心得第一个心得关于子图粒度。我一开始拆得太细把查询订单和修改订单拆成两个子图结果发现它们共享 90% 的上下文拆开之后状态传递反而更复杂。后来合并成一个订单子图内部用条件边区分查询和修改清爽很多。子图粒度以是否需要独立的路由决策为准不需要独立路由的合并。第二个心得关于State 字段命名。我踩过字段名不一致导致静默失败的坑之后定了个规矩所有 State 字段用snake_case且子图写入的字段名必须和父图 schema 完全一致。为此我写了个装饰器在子图 wrapper 上自动校验字段名不匹配直接抛异常别让它静默通过。第三个心得关于兜底逻辑。多智能体系统一定要有兜底而且兜底要足够厚。我的兜底子图不只是转人工还会做三件事记录完整对话上下文供人工参考、给用户一个明确的预期正在为您转接预计等待 30 秒、尝试用通用话术安抚情绪。兜底做得好用户不会因为一次识别失败就流失。第四个心得关于日志的粒度。我现在的日志策略是每个子图入口打一条 INFO含 State 摘要每个工具调用打一条 INFO含参数和结果摘要每个异常打一条 ERROR含完整堆栈和 State 快照。日志量不小但排查问题时能省下大量时间。用结构化日志JSON 格式方便后续做分析和告警。最后一个心得也是最重要的别追求一次设计完美。我的客服系统架构改了五版从单 Agent 到双子图到四子图每次都是遇到具体问题才调整。子图模式的好处就是改一个子图不影响其他这让迭代成本很低。先跑起来遇到问题再拆比一开始就设计一个完美架构要务实得多。这套系统现在日均处理 5000 对话意图识别准确率稳定在 95% 以上转人工率从最初的 18% 降到了 7%。回头看子图模式最大的价值不是性能提升而是让复杂系统的每一部分都变得可理解、可测试、可独立迭代。当你的 Agent 开始不听话的时候先别急着换模型想想是不是该拆子图了。