多智能体协作这件事我在过去一年里反复折腾过不少框架。从最早的纯 Prompt 编排到后来用状态机硬写流程再到接触 LangGraph 之后才算真正找到一条能落地的路子。今天要聊的这套“子图模式”是我在做 AI 客服系统时踩了无数坑之后沉淀下来的方案。它解决的核心问题很具体当一个客服系统需要同时处理售前咨询、订单查询、售后投诉、技术排查这几类完全不同的任务时你不可能把所有逻辑塞进一个 Agent 里那样提示词会膨胀到无法维护工具调用也会互相干扰。子图模式的价值就在于把每个业务域拆成一个独立的子图每个子图有自己的状态、自己的工具集、自己的决策逻辑最后由一个主图来负责路由和汇总。这套方案适合谁看如果你已经写过基础的 LangGraph 单图流程知道 StateGraph 怎么定义、节点怎么连边但一到多业务线就感觉状态管理混乱、节点爆炸那这篇内容就是为你准备的。如果你还没接触过 LangGraph建议先补一下基础概念否则后面讲子图的状态隔离和消息传递时会有点吃力。整篇内容我会按照实际项目推进的顺序来讲先讲清楚为什么选子图而不是其他方案再拆解状态设计这个最容易翻车的地方然后逐个模块搭建子图最后讲主图怎么编排和实际跑起来之后遇到的那些坑。1. 为什么客服系统非要用子图模式不可1.1 单图硬扛多业务线的真实后果我最初的做法很直接一个 StateGraph 走天下。所有工具函数注册在同一个 ToolNode 里所有业务逻辑用条件边来分流。刚开始只有售前咨询和订单查询两个场景时跑得还挺顺。但加到第四个业务域的时候问题集中爆发了。第一个问题是提示词冲突。售前咨询需要 Agent 热情、主动推荐、引导下单售后投诉需要 Agent 克制、共情、先安抚再解决。这两种语气放在同一个系统提示里模型会表现得精神分裂该热情的时候端着该克制的时候又过度推销。第二个问题是工具集污染。订单查询工具需要订单号参数技术排查工具需要设备型号和错误码当所有工具都暴露给同一个 Agent 时模型经常选错工具把订单号传给技术排查接口直接报参数错误。第三个问题是状态字段爆炸。每个业务域都有自己的中间状态售前有推荐商品列表订单有查询结果缓存售后有工单号技术有排查步骤记录。这些字段全塞进一个 State 里类型定义越来越长调试时根本看不清哪个字段是哪个环节写入的。实测下来单图方案在业务域超过三个之后维护成本呈指数上升。每次改一个业务域的逻辑都要回归测试其他三个域有没有被影响。1.2 子图模式到底解决了什么子图模式的本质是“分而治之”。每个业务域是一个独立的 StateGraph拥有自己独立的 State 定义、独立的节点集合、独立的工具绑定。主图只负责一件事根据用户输入判断该走哪个子图然后把子图的输出汇总成最终回复。这样做的好处很实在。状态隔离之后售前子图的 State 里只有推荐相关字段售后子图的 State 里只有工单相关字段互不干扰。工具隔离之后每个子图只绑定自己需要的工具模型选错工具的概率大幅下降。提示词隔离之后每个子图可以用完全不同的系统提示语气和策略各自独立调优。还有一个容易被忽略的好处子图可以独立测试。我可以单独跑售前子图用一组测试用例验证它的推荐逻辑不需要启动整个客服系统。这在迭代阶段节省的时间非常可观。1.3 子图与多 Agent 的区别在哪这里要澄清一个概念。很多人把子图模式和多 Agent 协作混为一谈其实两者有本质区别。多 Agent 协作通常指多个独立的 Agent 实例通过消息传递来协商完成任务每个 Agent 有自己的决策循环。而子图模式是在同一个图执行引擎内把不同的业务逻辑封装成可复用的子图单元由主图来调度。区别体现在三个地方。第一子图共享同一个执行上下文状态传递是显式的、类型安全的多 Agent 之间的消息传递往往是自然语言容易丢失结构化信息。第二子图的执行是确定性的图遍历你可以精确控制每一步走哪个节点多 Agent 的协商过程带有随机性调试困难。第三子图模式更容易做权限和边界控制因为每个子图的输入输出都是明确定义的。对于客服系统这种需要强流程控制的场景子图模式比多 Agent 协商更合适。当然如果你要做的是开放式创意协作多 Agent 可能更灵活。2. 状态设计子图模式最容易翻车的地方2.1 主图 State 和子图 State 的字段划分原则状态设计是子图模式的地基地基没打好后面全是坑。我总结了一条核心原则主图 State 只放跨子图共享的字段子图 State 只放本业务域内部使用的字段。具体来说主图 State 需要包含这些字段用户原始输入、对话历史、路由决策结果、当前活跃子图标识、最终回复。这些字段是所有子图都可能读写的或者主图路由逻辑必须访问的。子图 State 则包含各自业务域特有的字段。比如订单子图需要订单查询结果、订单状态、物流信息售后子图需要工单号、投诉类型、处理进度技术子图需要设备信息、错误码、排查步骤。这里有个关键细节子图 State 必须继承主图 State 的字段定义。LangGraph 的子图机制要求子图的 State schema 是主图 State 的超集或者至少兼容。我通常的做法是定义一个 BaseState 包含共享字段然后每个子图的 State 继承 BaseState 并追加自己的字段。from typing import TypedDict, Annotated, Literal from langgraph.graph import MessagesState class BaseState(MessagesState): user_input: str route_decision: str final_response: str class OrderState(BaseState): order_id: str order_status: str logistics_info: str class AfterSalesState(BaseState): ticket_id: str complaint_type: str progress: str这样定义之后子图在执行时既能访问共享字段又能操作自己的专属字段类型检查也能通过。2.2 消息历史在子图间的传递陷阱消息历史的处理是另一个高频翻车点。LangGraph 的 MessagesState 用 add_messages 注解来管理消息列表但在子图场景下消息的追加和读取需要特别注意。我遇到的问题是主图把用户输入传给子图时如果直接把完整的对话历史塞进子图的 messages子图会看到其他业务域的历史消息导致上下文污染。比如用户在售前聊了半天商品推荐然后转到售后投诉售后子图如果看到之前的推荐记录可能会在回复里莫名其妙地提到商品。解决方案是在子图入口做消息裁剪。只保留最近若干轮对话和当前用户输入把其他业务域的历史消息过滤掉。具体实现可以在子图的第一个节点里做预处理def prepare_subgraph_input(state: BaseState): recent_messages state[messages][-4:] return {messages: recent_messages}这个裁剪窗口的大小需要根据业务调整。售前咨询可能需要更多历史来理解用户偏好售后投诉可能只需要当前这一条。我一般设成 4 到 6 条实测下来够用且不会引入太多噪声。2.3 子图输出如何干净地回写主图子图执行完之后输出怎么回写到主图 State这里也有讲究。最直接的做法是把子图的整个 State 返回给主图但这样会把子图的专属字段也带回来污染主图 State。正确的做法是在子图末尾加一个汇总节点只提取需要回写的字段。比如订单子图执行完查询后只把 order_status 和 logistics_info 拼成一段自然语言回复写入 final_response 字段其他中间字段不往外传。def summarize_order_result(state: OrderState): summary f订单{state[order_id]}当前状态{state[order_status]}物流信息{state[logistics_info]} return {final_response: summary}主图拿到 final_response 之后直接返回给用户不需要关心子图内部经历了什么。这种“黑盒”式的接口设计让主图和子图的耦合度降到最低后续替换子图实现时主图完全不用改。3. 逐个搭建业务子图从售前到技术排查3.1 售前咨询子图意图识别与商品推荐链路售前咨询子图的核心任务是理解用户想买什么然后给出推荐。这个子图的节点链路我设计成四步意图澄清、需求提取、商品检索、推荐生成。意图澄清节点负责判断用户的问题是否足够明确。如果用户只说“我想买个耳机”这个信息太模糊需要追问使用场景、预算范围、有线还是无线。这个节点用一个小模型做二分类即可成本低响应快。需求提取节点从对话中抽取结构化需求比如品类、预算区间、关键功能点。这里我用的是带 structured output 的模型调用直接输出 JSON 格式的需求对象。商品检索节点根据需求对象查询商品库。这里有个经验不要把商品库查询做成纯向量检索客服场景下用户需求往往是硬性条件过滤预算不超过 500、必须支持降噪纯向量检索会返回不符合硬条件的结果。我的做法是先做结构化过滤再在过滤结果里做语义排序。推荐生成节点把检索结果组织成自然语言推荐。这里要注意推荐数量控制在 3 个以内超过 3 个用户会选择困难。每个推荐要包含商品名、核心卖点、价格以及一句“为什么适合你”的个性化说明。def build_presales_subgraph(): graph StateGraph(PresalesState) graph.add_node(clarify, clarify_intent) graph.add_node(extract, extract_requirements) graph.add_node(search, search_products) graph.add_node(recommend, generate_recommendation) graph.add_edge(clarify, extract) graph.add_edge(extract, search) graph.add_edge(search, recommend) graph.set_entry_point(clarify) return graph.compile()3.2 订单查询子图参数校验与多轮补全订单查询子图看起来简单实际上参数校验这块很容易出问题。用户可能直接说“查一下我的订单”但没给订单号。这时候不能直接报错要做多轮补全。我的设计是三个节点参数校验、参数补全、订单查询。参数校验节点检查 order_id 是否存在且格式合法。如果缺失进入参数补全节点向用户追问订单号。如果格式不对提示用户重新输入。这里有个细节订单号格式校验要用正则做前置过滤但不要过于严格。我见过有的实现要求订单号必须是 18 位纯数字结果用户输入带字母的订单号直接被拒。实际上订单号格式应该从订单系统动态获取规则或者用宽松匹配加后端二次校验。订单查询节点调用后端接口获取订单详情。这里必须做超时和异常处理。后端接口挂了的时候不能让整个客服系统卡死要返回“查询服务暂时不可用请稍后再试”这样的兜底回复。def validate_order_params(state: OrderState): order_id state.get(order_id, ) if not order_id: return {next_action: ask_for_id} if not re.match(r^[A-Za-z0-9]{8,24}$, order_id): return {next_action: invalid_format} return {next_action: query}3.3 售后投诉子图情绪识别与工单创建售后投诉子图是最需要小心处理的。用户带着情绪来如果回复不当很容易升级成舆情事件。这个子图的节点设计我加了情绪识别环节。情绪识别节点判断用户当前的情绪状态平静、不满、愤怒。平静状态走标准处理流程不满状态先安抚再处理愤怒状态直接转人工并创建高优先级工单。工单创建节点把投诉内容、用户信息、情绪等级打包成工单写入工单系统。这里要注意工单描述要结构化不能直接把用户原话扔进去。我的做法是提取投诉类型、涉及订单、具体问题描述、用户诉求四个字段。安抚回复节点根据情绪等级生成不同语气的回复。愤怒状态下回复要简短、诚恳、明确给出解决时间承诺不要长篇大论解释原因那样只会火上浇油。def detect_emotion(state: AfterSalesState): emotion emotion_classifier(state[messages][-1].content) if emotion angry: return {emotion_level: high, next_action: escalate} elif emotion dissatisfied: return {emotion_level: medium, next_action: soothe} return {emotion_level: low, next_action: standard}3.4 技术排查子图分步引导与知识库检索技术排查子图的特点是交互轮次多需要一步步引导用户排查问题。这个子图我用的是状态机式的设计每个排查步骤是一个节点根据用户反馈决定下一步走哪个节点。排查步骤的设计要遵循“从简到繁”的原则。先让用户检查最简单的可能性比如设备是否重启过、网络是否正常再逐步深入到复杂排查。每步只让用户做一个操作不要一次给三个步骤用户会漏做。知识库检索节点在标准排查步骤走完之后触发用用户描述的错误现象去检索知识库返回匹配的解决方案。这里检索的 query 构造很关键要把设备型号、错误码、操作场景都拼进去提高检索命中率。def build_tech_support_subgraph(): graph StateGraph(TechSupportState) graph.add_node(identify_device, identify_device) graph.add_node(basic_check, basic_check) graph.add_node(advanced_check, advanced_check) graph.add_node(kb_search, search_knowledge_base) graph.add_conditional_edges(basic_check, route_after_basic) graph.add_conditional_edges(advanced_check, route_after_advanced) return graph.compile()4. 主图编排路由决策与子图调度4.1 路由节点的分类逻辑怎么设计主图的路由节点决定了整个系统的分流准确性。我的做法是用一个轻量级分类模型把用户输入映射到四个业务域之一外加一个“闲聊/其他”兜底类别。分类模型的提示词要精心设计。不能只给类别名称要给每个类别配上典型示例和边界说明。比如“订单查询”和“售后投诉”容易混淆用户说“我买的东西还没到”到底算查询还是投诉我的界定是询问进度算查询表达不满算投诉。这个边界要在提示词里写清楚。路由节点还要处理多意图的情况。用户可能一句话里既有查询又有投诉“我的订单三天了还没发货你们怎么回事”这时候应该优先路由到售后投诉子图因为情绪处理优先级更高。我的做法是在分类时加一个优先级规则投诉类意图优先于查询类意图。def route_decision(state: BaseState) - Literal[presales, order, aftersales, tech, fallback]: user_msg state[messages][-1].content category classifier(user_msg) priority_map {aftersales: 1, tech: 2, order: 3, presales: 4, fallback: 5} return category4.2 子图作为节点的注册方式LangGraph 支持把编译好的子图直接作为主图的节点添加。这个机制用起来很顺手但有几个细节要注意。子图注册时主图会把子图的 State schema 合并进来。如果子图 State 和主图 State 有同名字段但类型不同会在编译时报错。所以前面强调的 BaseState 继承机制在这里就体现出价值了所有子图共享同一套基础字段定义不会出现类型冲突。子图节点的返回值只会包含子图 State 中定义的字段。主图拿到返回值后只会更新主图 State 中存在的字段子图专属字段会被忽略。这个行为是符合预期的但初次使用时容易困惑以为子图输出丢了。from langgraph.graph import StateGraph, END def build_main_graph(): main_graph StateGraph(BaseState) main_graph.add_node(router, route_decision) main_graph.add_node(presales, presales_subgraph) main_graph.add_node(order, order_subgraph) main_graph.add_node(aftersales, aftersales_subgraph) main_graph.add_node(tech, tech_subgraph) main_graph.add_node(fallback, fallback_handler) main_graph.set_entry_point(router) main_graph.add_conditional_edges(router, lambda x: x[route_decision], { presales: presales, order: order, aftersales: aftersales, tech: tech, fallback: fallback }) main_graph.add_edge(presales, END) main_graph.add_edge(order, END) main_graph.add_edge(aftersales, END) main_graph.add_edge(tech, END) main_graph.add_edge(fallback, END) return main_graph.compile()4.3 多轮对话中路由的重新触发客服场景下用户经常在一个会话里切换业务域。先问售前然后查订单最后投诉物流。这就要求主图支持多轮路由而不是一次路由定终身。我的做法是在主图里加一个循环机制。每个子图执行完之后不直接 END而是回到路由节点重新判断。但这里要加一个终止条件否则会无限循环。终止条件有两个一是子图返回了 final_response 且用户没有新的输入二是循环次数超过阈值。实现上可以用一个 turn_count 字段来计数每次路由加一超过 5 次强制结束并返回当前汇总结果。这个阈值根据业务调整一般客服场景 5 轮足够覆盖绝大多数情况。def should_continue(state: BaseState): if state.get(turn_count, 0) 5: return end if state.get(final_response) and not state.get(needs_followup): return end return route5. 实测中暴露的问题与调优记录5.1 子图间状态污染的真实案例上线第一周就遇到一个诡异问题用户查完订单之后转到售前咨询售前推荐的商品里居然包含了订单里的商品。排查了半天发现是订单子图的 order_status 字段被写进了主图 State售前子图读取 messages 时通过某种路径访问到了这个字段。根因是子图 State 继承 BaseState 时如果 BaseState 里定义了可选字段子图写入后主图会保留。解决方案是在主图 State 里严格区分“共享字段”和“子图专属字段”子图专属字段不要定义在 BaseState 里而是定义在各子图自己的 State 扩展里。这样主图 State 里根本不存在这些字段自然不会污染。这个坑让我重新审视了状态设计原则BaseState 只放真正需要跨子图共享的字段宁少勿多。每多一个字段就多一份污染风险。5.2 路由误判的边界情况处理路由误判主要集中在两个边界售前和订单的边界、订单和售后的边界。售前和订单的边界案例“这个商品有货吗”这句话既可以理解为售前咨询询问商品信息也可以理解为订单相关库存查询。我的处理是把这类模糊问题默认路由到售前因为售前子图可以处理库存查询而且售前子图的回复更友好。订单和售后的边界案例“我的订单怎么还没到”这句话如果用户语气平静路由到订单查询如果带有明显不满路由到售后。这里情绪识别的准确率直接影响路由效果。我实测下来纯文本情绪识别的准确率大概在 75% 左右所以加了一个兜底策略路由到订单子图后如果订单子图检测到用户情绪升级主动转交给售后子图。def order_subgraph_with_escalation(state: OrderState): result order_subgraph.invoke(state) if detect_escalation(result[messages]): return aftersales_subgraph.invoke(result) return result5.3 子图执行超时的降级方案子图执行超时是生产环境必须处理的问题。订单查询依赖后端接口技术排查依赖知识库检索这些外部依赖都可能超时。我的降级方案分三层。第一层是子图内部超时每个外部调用设置 3 秒超时超时后返回缓存结果或兜底话术。第二层是子图整体超时主图给每个子图设置 10 秒执行上限超时后中断子图返回“正在处理中请稍后”的提示。第三层是全局超时整个对话轮次超过 30 秒强制结束并转人工。这三层超时机制配合使用之后系统的可用性从 95% 提升到了 99.5% 以上。关键是每层超时都要有明确的用户提示不能让用户干等。5.4 提示词隔离带来的维护便利最后说一个正面收益。子图模式的提示词隔离在实际维护中带来的便利超出预期。以前单图方案时改一句系统提示要回归测试所有业务域。现在每个子图的提示词独立存放改售前提示词只影响售前子图其他子图完全不受影响。我们甚至可以让不同的人负责不同子图的提示词调优互不干扰。提示词文件我按子图分目录存放每个子图目录下有 system_prompt.txt、few_shot_examples.json、tool_descriptions.json 三个文件。这样版本管理清晰回滚也方便。prompts/ presales/ system_prompt.txt few_shot_examples.json tool_descriptions.json order/ system_prompt.txt few_shot_examples.json aftersales/ system_prompt.txt few_shot_examples.json tech/ system_prompt.txt few_shot_examples.json这套结构跑了大半年子图从 4 个扩展到 7 个新增业务域只需要复制目录结构、改提示词、注册子图主图几乎不用动。这种扩展性是我最终选择子图模式而不是其他方案的核心原因。如果你也在做多业务线的客服系统我的建议是不要一开始就追求大而全的单图方案先把业务域拆清楚每个域用子图独立实现主图只做路由和汇总。前期多花一点时间设计状态边界后期能省下大量调试和维护成本。子图之间的状态隔离和消息传递是最容易出问题的地方建议在开发阶段就写好单元测试每个子图单独跑通再接入主图。