1. 项目概述与整体设计思路1.1 为什么客服系统需要多智能体子图模式做 AI 应用开发这一年多我最大的感受是单智能体看起来什么都能干一旦落到真实业务场景Prompt 堆到 300 行工具编排链绕成一团最后还是应付不了跨部门、跨流程的复杂请求。比如用户问“我的手机流量套餐可以改吗会不会影响话费如果不能改帮我查一下附近营业厅”——这条消息背后至少涉及三个能力套餐业务规则查询、订单/账户信息查询、位置服务。你当然可以写一个超级 Agent把所有工具装进去用一个 System Prompt 把业务规则、API 逻辑、对话策略全塞进去。结果就是这个 Agent 的 Prompt 越来越长工具选择越来越慢稍微来一个不在主路径上的问题它就开始乱调工具。我在生产环境里见过太多次这样的翻车现场用户问一句“我上个月账单怎么多了 20 块”模型在十几个工具里来回试错最后调了一个查天气的 API答非所问。多智能体方案的思路是分而治之把客服业务按职责拆成几个独立 Agent每个 Agent 只负责一个子领域彼此之间通过图结构协同。这不只是“把代码分开写”而是从运行机制上让每个 Agent 的状态空间变窄决策更稳也更容易定位问题。子图模式在这套思路上更进一步不仅职责分开连图的执行流程都是分层的——父图负责调度子图负责执行某一类完整子任务子图内部还可以继续嵌套。1.2 子图模式相比其他多智能体方案的取舍多智能体常见的协作方式有几种手工编排一个循环里按顺序调多个 Agent、ReAct 式的动态工具路由、以及 LangGraph 的图结构协作。手工编排写起来简单但流程一旦带分支、带循环、带状态同步代码就变成意大利面。动态工具路由靠模型自己选 Agent灵活是灵活但不可控模型选错 Agent 的代价很高尤其客服场景里一旦走错流程用户的耐心就没了。LangGraph 子图模式的核心价值在于把“可控”和“灵活”的平衡点放到图结构上。路由逻辑可以交给模型LLM Router但路由之后每个分支的执行是确定性的子图子图内部也可以有自己的模型决策点。翻译成大白话就是关键岔路口让大模型做选择题但每条路内部怎么走由你画好图不会跑偏。这也是我实际项目里最看中的一点——客户不会为 90% 的准确率买单但会为 95% 以上的可控性鼓掌。再从开发维护的角度说子图模式把复杂的会话流程拆成了几个可以独立测试的模块。我在项目里给每个子图都配了单独的测试用例比如直接给order_agent输入一条“查一下订单 SO-1024”验证返回结构是否符合预期。这种隔离测试能力在大模型应用的工程化里面非常宝贵因为模型的输出天然带随机性如果不把链路拆细出了问题你真的不知道是哪一环在捣乱。2. 核心细节解析与实操要点2.1 先理解 Parent State 与 Child State 的边界用 LangGraph 写子图第一步不是写代码而是理清状态State的边界。子图可以共享父图的 State也可以定义自己的 State关键在子图的input_schema和output_schema怎么设置。我在实际项目里常用的方式是父图只管三个字段——user_query、current_step、final_reply而每个子图内部维护自己的临时状态比如intent、order_info、policy_answer子图结束时只把需要对外暴露的结果写回父图。这种设计的理由很容易理解如果你让所有 Agent 共享一个大 State字段会越来越多内存占用增大还好说关键是某个子图改了共享字段别的子图再读的时候可能拿到脏数据排查起来非常痛苦。所以我的经验是子图之间的通信越“接口化”越好——子图收到的输入是父图传进来的元组或字典子图吐出的结果也是明确的结构内部细节完全隔离。具体到代码层面定义状态时我强烈建议用TypedDict而不是普通的dict。TypedDict能让你在类型检查阶段就发现字段名拼写错误而且配合 IDE 的自动补全写节点函数的时候心里踏实很多。下面是一个实际项目里用过的状态定义示例from typing import TypedDict, Optional class ParentState(TypedDict): user_query: str route: Optional[str] history: list[str] final_reply: str class OrderState(TypedDict): user_query: str order_info: Optional[str] reply: str这里的history字段值得单独说一下。客服场景天然是多轮对话用户经常说“那个订单现在怎么样了”“那个”指代上一轮的订单信息。如果每个子图都只盯着当前一句孤立的话就答非所问了。所以我在父图 State 里加一个history: list[str]每次对话结束后把user_query assistant_reply追加进去Router 和子图都能读取这段历史来做上下文理解。这是子图模式里“状态隔离”和“共享上下文”两者之间最常踩的平衡点。2.2 子图的两种挂载方式节点挂载与函数调用LangGraph 里挂子图有两种常见方式。第一种是直接把StateGraph编译后的app作为一个普通节点挂到父图上builder.add_node(order_agent, order_subgraph_app)这样父图跑到这个节点时会执行整个子图的完整生命周期。第二种是在父图的节点函数里调用子图的ainvoke走函数调用的路子def call_order_agent(state: ParentState): result await order_app.ainvoke({user_query: state[user_query]}) return {final_reply: result[reply]}两者效果类似但区别在于节点挂载方式下子图的内部节点和边在父图的调试面板里是展开的而且子图可以拿到父图的RunnableConfig比如recursion_limit、callbacks函数调用方式则更像“在一个节点里程序化调用”调试时是一个黑盒。实际项目里我更推荐节点挂载尤其你后面要接 LangSmith 或者 Langfuse 做链路追踪的时候子图展开与否直接决定你能不能看清一次完整回答的中间过程。另外节点挂载方式下子图内部若发生异常父图可以通过 try 节点捕获而函数调用方式的异常处理会更绕你可能得在节点函数里自己用 try-except 包一层再把异常信息写入 State。提示如果你刚开始学建议先用函数调用方式写逻辑更容易理解等你要上线生产环境了再改成节点挂载调试和排障会轻松很多。2.3 Router 节点设计LLM 决策 确定性兜底AI 客服系统里 Router 节点至关重要。父图收到用户问题后Router 决定走哪个子图订单查询、业务咨询、还是人工客服转接。我见过不少团队在 Router 节点里直接让模型输出一个 JSON比如{next: order_agent}但实际跑起来经常出现路由 key 和节点名不匹配的问题。我的做法是三层校验。第一用with_structured_output让模型输出固定 schema不要裸 JSON。第二在 Router 节点函数里做 key 映射把模型输出映射到实际节点名同时准备一个fallback节点当置信度低于阈值比如 0.5时直接走兜底路径。第三在 System Prompt 里给足约束样例明确告诉模型“当识别到用户在问套餐变更时必须返回policy_agent而不是order_agent”。这里有一个很值得说的细节Router 不是非黑即白。实际场景里用户可能一个问题涉及两个 Agent。比如“我的订单 SO-1024 能不能退退的话怎么收费”这既涉及订单状态查询又涉及退款政策。如果只走一个 Agent信息一定不完整。我的解决方案是让 Router 输出一个primary_intent和一个secondary_intent父图先把主意图给对应子图跑如果主意图返回的结果里带了一个needs_followup标志再触发第二个子图做补充。这样做可以把复杂问题从串行等待变成一个可控的接力流程体感上回复会更快。Router 的模型选择也有讲究。我用gpt-4o-mini这类轻量模型做路由因为路由决策本身不需要很强的推理能力但需要的是稳定和快。复杂的意图分类丢给轻量模型实测延迟能压到 300 到 500 毫秒以内而如果让大模型来做可能多花一倍的时间对客服场景来说不可接受。3. 实操过程与核心环节实现3.1 先搭一个可运行的最小框架三个子图的 AI 客服下面我手写一个可以跑通的最小案例业务背景用户进线询问套餐、查订单、或者索要人工客服。定义三个子图order_agent查订单、policy_agent业务政策咨询、human_agent转人工。父图只做路由和聚合。第一步定义共享状态和子图状态。这一步决定了整个系统的数据流。我把父图 State 控制在三个字段加一个历史列表子图各自维护私有字段。你不要嫌这一步啰嗦状态定义的好坏直接决定后面掉不掉头发。我见过太多人上来就写节点写到一半发现状态对不上全盘推倒重来。from typing import TypedDict, Optional from langgraph.graph import StateGraph, START, END class ParentState(TypedDict): user_query: str route: Optional[str] final_reply: str class OrderState(TypedDict): user_query: str order_info: Optional[str] reply: str class PolicyState(TypedDict): user_query: str policy_answer: Optional[str] reply: str class HumanState(TypedDict): user_query: str reply: str第二步构建三个子图。每个子图内部可以只有一个节点但为了示范我让order_agent内部带一个条件分支模拟“命中订单号”和“未命中订单号”两种情况。def order_match(state: OrderState): # 实际项目里这里接数据库/工具调用 if 订单号 in state[user_query]: return {order_info: 订单号 SO-1024当前已发货, reply: 查到您的订单 SO-1024 已在运输途中。} return {order_info: None, reply: 抱歉我没有找到您的订单信息建议提供订单号。} order_builder StateGraph(OrderState) order_builder.add_node(match, order_match) order_builder.add_edge(START, match) order_builder.add_edge(match, END) order_app order_builder.compile()policy_agent和human_agent的结构完全一样只是节点函数内部逻辑不同。注意我这里没有给order_agent加条件分支到 human因为你真正写的时候肯定希望在子图内部自己决定“要不要转人工”而不是把决定权交给父图。这个设计留给你扩展下面会讲更完整的版本。第三步构建父图。父图的 Router 节点我用一个简单规则函数代替实际项目里接 LLM 即可。def router_node(state: ParentState): if 订单 in state[user_query]: return {route: order_agent} if 转人工 in state[user_query] or 套餐 in state[user_query]: return {route: policy_agent} return {route: human_agent} def call_order_agent(state: ParentState): result order_app.invoke({user_query: state[user_query]}) return {final_reply: result[reply]} builder StateGraph(ParentState) builder.add_node(router, router_node) builder.add_node(order_agent, call_order_agent) builder.add_node(policy_agent, call_policy_agent) builder.add_node(human_agent, call_human_agent) builder.add_edge(START, router) builder.add_conditional_edges(router, lambda s: s[route], {order_agent: order_agent, policy_agent: policy_agent, human_agent: human_agent}) builder.add_edge(order_agent, END) builder.add_edge(policy_agent, END) builder.add_edge(human_agent, END) app builder.compile()这里有一个需要解释清楚的技术点我在call_order_agent里用的是order_app.invoke而不是把order_app直接作为节点挂载。两种方式都能跑但函数调用方式需要手动传递输入状态并从返回结果里挑出需要的字段再返回给父图。节点挂载方式则更自动化你只需要告诉 LangGraph 这个节点是哪个子图 app状态传递和回调追踪都会更友好。注意lambda s: s[route]这种写法在条件边里很常见但有个隐藏问题——如果s[route]是 None或者包含不在映射表里的值LangGraph 会直接抛异常。稳妥做法是加一个 fallback 映射比如在条件边函数里最后return human_agent确保任何意外值都有人接住。3.2 让子图真正“协作”引入共享记忆与跨 Agent 上下文上面的最小框架每个子图都是独立的拿到什么问题就处理什么问题。真实客服场景里用户经常说“那个订单现在怎么样了”这里的“那个”指代上文的订单如果每个子图都只盯着当前一句话就答非所问了。解决办法有两个层级。第一层级用messages列表把对话历史传进父图 State。我在前面已经给ParentState加了history字段Router 节点和子图节点都能读取它。子图内部如果需要提取实体我通常在order_match节点里先用 LLM 做一个extract_order_no的动作把历史里出现的订单号抽出来再拼到当前查询里。这个做法实现简单效果立竿见影。第二层级上记忆持久化。LangGraph 提供了MemorySaver和SqliteSaver分别做内存态和持久态的记忆。我在客服系统里用SqliteSaver这样服务重启后用户的历史上下文还在。注意一点MemorySaver实测只适合 demo 或单机短会话多实例部署时会话状态不同步一定要换持久化的SqliteSaver或PostgresSaver否则生产环境分分钟翻车。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(:memory:) as saver: app builder.compile(checkpointersaver)关于记忆还有一层很多人忽略的问题记忆不只对子图内部有用Router 节点同样需要。用户先说“帮我查一下我的订单”Router 路由到order_agent子图回复了订单信息。紧接着用户说“如果这个订单能退的话退费规则是什么”这时候 Router 需要知道“这个订单”指代的是刚才那个订单号才能正确路由到policy_agent同时在传参数的时候带上订单号。这个跨 Agent 的上下文传递靠的就是history里的完整会话记录。我在 Router 的 prompt 里明确加了一句话“根据对话历史判断指代对象如果用户说的是‘那个订单’请从历史中提取订单号并附加在意图中。”实测下来指代消解的准确率能到九成以上。3.3 人工客服转接多智能体与真实世界的握手AI 客服系统跑得再好也逃不过“转人工”这三个字。子图模式把转人工做成一个子图human_agent而不是简单的END好处是能承接后面的“排队状态更新”“工单创建”等动作。我的设计里human_agent子图内部有两个节点create_ticket和notify_backend。create_ticket调用工单系统的 API把用户问题、AI 的初步判断、用户联系方式一并写入工单notify_backend是模拟推送一条 Webhook 给客服工作台。这两个节点之间是串行边确保工单创建成功后再通知顺序不能反。下面给出完整的human_agent子图实现def create_ticket(state: HumanState): # 实际项目里这里调用工单系统 API ticket_id TK-8888 return {ticket_id: ticket_id} def notify_backend(state: HumanState): # 模拟推送 Webhook return {notified: True, reply: 已为您转接人工客服工单号 TK-8888请耐心等待。} human_builder StateGraph(HumanState) human_builder.add_node(create_ticket, create_ticket) human_builder.add_node(notify_backend, notify_backend) human_builder.add_edge(START, create_ticket) human_builder.add_edge(create_ticket, notify_backend) human_builder.add_edge(notify_backend, END) human_app human_builder.compile()这时候如果子图内部节点需要维护多个临时字段比如ticket_id、notified、reply就需要把这些字段都定义在HumanState里。我在上面的HumanState里只写了user_query和reply你实际用的时候记得补全字段。子图内部节点共用子图 State这是子图和父图的一个关键区别——子图内部通信自由但跨图通信必须通过输入输出 schema 显式约定。这里有一个最常见的坑子图返回值不写全。如果human_agent只返回reply没把ticket_id写进父图 State后面的记录统计就没法做。所以我在每个子图的出口都会统一封装一个finish节点把需要对外暴露的字段一次性 merge 进父图 State。这个习惯是从几次线上事故里学来的——有一次客服系统上线后运营发现工单记录全部缺失查了半天发现是子图返回的ticket_id字段名和父图预期的ticket_id不一致数据流静默丢失。4. 常见问题与排查技巧实录4.1 子图返回的 State 没有被父图接收怎么回事这是子图模式最容易踩的坑。如果你在父图节点里用order_app.invoke(...)invoke 返回的是子图的整个 State 字典你拿到后要自己挑选需要对外暴露的字段 return 给父图。很多新手直接return {final_reply: result}结果发现final_reply变成了整个字典前端渲染时直接报错或者说出一串 JSON。我的自查清单是第一步打印子图的result看结构第二步确认父图节点返回的字段和ParentState中声明的字段类型一致第三步在父图里加一个log_node在 END 前打印final_reply的前 50 个字符验证聚合是否正确。三步走完绝大多数 State 传递问题都能定位。如果用的是“把子图 app 直接 add_node”的挂载方式父图和子图之间的字段映射规则不同LangGraph 默认会做局部更新但只针对同名字段。也就是说子图 State 和父图 State 的同名字段会被自动合并不同名字段忽略。想要精准控制就在子图编译时显式指定input_schema和output_schema。class OrderInput(TypedDict): user_query: str class OrderOutput(TypedDict): order_info: Optional[str] reply: str order_app order_builder.compile( input_schemaOrderInput, output_schemaOrderOutput )这样写的好处是子图对外只暴露OrderOutput里声明的字段内部的临时字段比如matched、confidence完全不会污染父图 State。这在多智能体协作里是很强的封装性保障。4.2 递归深度限制 recursion_limit 踩坑LangGraph 默认的recursion_limit我印象中是 25。子图嵌套子图、Router 再进子图、子图内部还有循环很容易踩到GraphRecursionError: The maximum recursion depth was exceeded。这个报错不是说你代码死循环了而是图执行的步数超过了限制。排查思路先看是哪个环节步数暴涨。常见原因有三个一是子图内部的循环边没有设置出口条件比如我在policy_agent里让模型回答不满意就重试结果没设最大重试次数二是 Router 在两个子图之间反复横跳A 没处理完跳 BB 又说需要 A三是单个子图本身节点就多父图每走一步都算一次 recursion嵌套三层后步数很容易上万。我的处理方式是给每个循环边都加一个max_retries计数器计数器用 State 字段维护给 Router 节点加一个route_attempts字段第二次进入同一路由时强制走 fallback最后在app.invoke(..., config{recursion_limit: 100})按需调大限制但调大是最后手段优先检查循环逻辑。def my_node(state): if state.get(retries, 0) 3: return {route: fallback} return {retries: state.get(retries, 0) 1} builder.add_conditional_edges(my_node, my_node, {continue: my_node, fallback: fallback, done: END})4.3 模型路由误判当 Router 把“退订”识别成“投诉”AI 客服里最让人心累的就是语义识别偏差。用户说“我要退订套餐”本来是policy_agent的业务Router 识别成“投诉”进了human_agent用户体验非常差。我的解决方法是给 Router 加一个“意图确认”动作当模型输出路由结果的置信度低于阈值时先让 Router 反问一句“您是想办理套餐变更业务吗”而不是直接进任何子图。具体实现上我维护了一个route_confidence: float字段。Router 节点返回route和route_confidence父图在条件路由前先判断置信度低于 0.6 时走clarify_node该节点用模板生成确认话术并把awaiting_confirmationTrue写进 State下一轮用户确认后再把历史里“确认意图”的信息加入上下文重新走 Router。这个双轮确认机制上线后误路由比例下降了一半以上代价是大约 8% 的简单请求多了一轮交互整体可接受。下面是一个 Router 节点的真实代码结构简化版from pydantic import BaseModel from langchain_openai import ChatOpenAI class RouteDecision(BaseModel): next_node: str confidence: float router_prompt 你是客服系统的路由引擎。根据用户的提问决定将该请求分配给哪个子 Agent。 可选路由order_agent订单查询、policy_agent业务政策、human_agent人工客服。 注意不要将套餐退订归为投诉退订属于 policy_agent除非用户明确表达强烈不满情绪才走 human_agent。 def router_node(state: ParentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(RouteDecision) decision structured_llm.invoke( f用户提问{state[user_query]}对话历史{state[history]} ) if decision.confidence 0.6: return {route: clarify, route_confidence: decision.confidence} return {route: decision.next_node, route_confidence: decision.confidence}4.4 子图内部节点报错会导致整个链路失败吗会。这是很多人初期没料到的。当你在父图里挂载一个子图 app 作为节点时子图内部任何节点抛异常默认都会向上传播把整次ainvoke调用打挂。客服场景里绝不允许一次报错让用户收到“服务器错误”所以必须做异常兜底。我的做法是在父图层面加一个全局 fallback 节点。具体来说给父图加一条从 START 开始的并行分支不现实更实用的做法是利用 LangGraph 的try机制——在父图编译时可以用with上下文捕获或者在节点函数层面对子图调用包一层 try-except。我实际生产代码里是下面这么写的async def safe_call_subgraph(subgraph, input_state): try: result await subgraph.ainvoke(input_state) return result except Exception as e: return {reply: 很抱歉系统暂时开小差了已为您转接人工客服。, error: str(e)}这样即使order_agent内部调用的数据库 API 超时了用户拿到的还是一句有温度的兜底话术而不是光秃秃的异常堆栈。后续再接上日志上报把error字段打到追踪平台方便定位。5. 工具选型与生产落地建议5.1 模型选型为什么我优先用带 function calling 的模型子图模式里Router 节点是模型决策点子图内部也可能需要模型调用工具。为了这套架构跑得稳我建议优先选择带 function calling / structured output 能力的模型而不是只靠 Prompt 让模型输出 JSON。原因很简单结构化输出的模型可以直接绑定预期 schema模型层保证输出合法靠 Prompt 硬凹 JSON三个小时调一个斜杠转义的日子我是真不想再过了。我的习惯是Router 节点用gpt-4o-mini这类轻量模型子图内的实体提取、政策问答用略强的模型最终话术润色可以再走一次模型。分层用模型成本差异不止一点但效果上完全够用。另外所有模型调用都建议配temperature0客服场景对一致性要求极高宁可答案稳一点也不要灵光一现的自由发挥。工具调用还有一个需要重点把关的环节工具的输入参数校验。大模型填参数经常丢三落四比如查订单的工具要求传order_id模型可能只传了一个user_query整个字符串进去。我的办法是在子图节点函数里加一层数据清洗凡是工具调用前都先跑一个 validator 函数参数缺失就自动补默认值或返回错误信息让模型重新生成。这个干净的参数校验层能让整个客服系统的工具调用成功率从八十几提到九十五个点。5.2 可观测性LangSmith/Langfuse 与子图展开子图模式真正的威力只有在可观测时才能发挥出来。我建议从第一天就接入追踪。LangSmith 对 LangGraph 的集成最顺能自动把每个节点、每条边、每次模型调用都记录下来Langfuse 则适合团队已经有自建可观测平台的情况可以跟 OpenTelemetry 打通。在追踪平台上重点看三件事Router 到了哪个子图、子图内部每个节点的耗时、模型的 token 消耗。有一次我们线上反馈“AI 客服响应慢”我打开追踪一看问题不在 Router也不在工具调用而是policy_agent内部有个政策问答节点模型在一个 3000 token 的上下文里反复做无关检索耗时占整个链路的 70%。这种问题没有子图展开的追踪根本定位不到。LangSmith 的 trace 面板里子图节点会默认展开成树状结构。一开始你可能觉得节点太多很乱但其实这是最大的优势——你能清晰地看到每一次调用是从哪个父节点发出的、是在哪个子图内完成的。如果用了函数调用方式挂载子图trace 里只有一个黑盒节点排查效率至少打对折。这也是我前面反复强调用节点挂载的原因之一。5.3 部署形态与稳定性FastAPI 异步态生产环境部署 LangGraph 应用我最常用的形态是 FastAPI 服务启动时编译一次图请求进来后await app.ainvoke(...)拉通全流程。LangGraph 官方也提供了langgraph serve快速起一个 HTTP 服务但自定义程度较低一旦需要鉴权、限流、外部依赖注入还是自己写 FastAPI 更顺手。稳定性上有两个容易被忽略的点。一是线程安全LangGraph 的app.invoke在并发下是否安全取决于内部状态存储建议一律用ainvoke异步写法并且把状态存储切到线程安全的 saver。二是超时控制客服场景不希望用户等太久我会在ainvoke外层包asyncio.wait_for比如 30 秒超时超时后返回“正在为您转接人工客服”的兜底话术。这个兜底逻辑其实也能作为一个fallback_agent子图挂到父图上超时分支直接路由过去架构上更统一。import asyncio from fastapi import FastAPI from pydantic import BaseModel app FastAPI() graph_app compile_graph() # 启动时编译一次 class ChatRequest(BaseModel): user_query: str session_id: str app.post(/chat) async def chat(req: ChatRequest): try: result await asyncio.wait_for( graph_app.ainvoke( {user_query: req.user_query, history: []}, config{configurable: {thread_id: req.session_id}} ), timeout30 ) return {reply: result[final_reply]} except asyncio.TimeoutError: return {reply: 正在为您转接人工客服请稍候。}这里configurable.thread_id的用法值得展开说LangGraph 的 checkpointer 会按thread_id维护独立的会话状态。这意味着你不用自己在 State 里维护history而是把多轮对话的累积交给 LangGraph 的持久化层处理。每次请求带同一个session_id图就能自动恢复上下文。我实际项目里把这两套方案都验证过——早期是自己拼history后来切换到 checkpointer 之后代码量明显减少而且会话恢复的可靠性更高。当然子图内部的私有状态比如ticket_id也会随 checkpointer 一起持久化所以你必须在子图结束时清理不需要保留的临时字段避免 State 无限膨胀。6. 实战体会与进阶扩展如果只看一个心得我想说子图模式本质上不是 LangGraph 的专属技巧而是一种“控制复杂度”的工程思维。把一个大而全的 Agent 分解成若干个高内聚、低耦合的子图让每个子图的边界清楚、出口干净整体的可靠性就会指数级上升。这套思维在客服系统之外同样适用于电商售后、企业知识库问答、工单分类、内容审核等场景。你甚至可以把policy_agent从一个客服项目里抽出来直接复用到另一个销售助手项目里因为它的输入输出是标准化的user_query到reply父图只要把它当作插件挂上就行。再分享一个我正在玩的扩展思路让子图自己注册“能力标签”。比如order_agent在初始化时对外声明自己“能查订单能查物流”policy_agent声明“能查套餐能算退费”。父图的 Router 不再是硬编码的几个分支而是变成一个动态的服务发现机制——根据用户意图到子图能力列表里做匹配。这样新增一个业务子图完全不需要改动父图的逻辑只要在新的 StateGraph 编译后注册进去即可。这个方向再往下走就是多智能体系统的微服务化每个子图独立部署、独立扩容通过 gRPC 或者 HTTP 与父图通信LangGraph 的add_node传的不再是本地的app而是一个远程调用的封装节点。最后提醒一句无论架构怎么演进别忘了给每个子图配上独立的测试用例。我的习惯是给每个子图写一份“输入样例-期望输出”的表格比如测试输入期望路由期望回复要点我要退订套餐policy_agent说明退订规则查订单 SO-1024order_agent返回订单状态转人工human_agent生成工单号这些用例既是回归测试的基准也是你和业务方对齐需求时的沟通语言。多智能体系统最大的风险不是代码写不出来而是没人说得清每个 Agent 的职责边界。表格写清楚代码自然就顺了。 LangGraph多智能体子图模式：构建可控AI客服系统实践
1. 项目概述与整体设计思路1.1 为什么客服系统需要多智能体子图模式做 AI 应用开发这一年多我最大的感受是单智能体看起来什么都能干一旦落到真实业务场景Prompt 堆到 300 行工具编排链绕成一团最后还是应付不了跨部门、跨流程的复杂请求。比如用户问“我的手机流量套餐可以改吗会不会影响话费如果不能改帮我查一下附近营业厅”——这条消息背后至少涉及三个能力套餐业务规则查询、订单/账户信息查询、位置服务。你当然可以写一个超级 Agent把所有工具装进去用一个 System Prompt 把业务规则、API 逻辑、对话策略全塞进去。结果就是这个 Agent 的 Prompt 越来越长工具选择越来越慢稍微来一个不在主路径上的问题它就开始乱调工具。我在生产环境里见过太多次这样的翻车现场用户问一句“我上个月账单怎么多了 20 块”模型在十几个工具里来回试错最后调了一个查天气的 API答非所问。多智能体方案的思路是分而治之把客服业务按职责拆成几个独立 Agent每个 Agent 只负责一个子领域彼此之间通过图结构协同。这不只是“把代码分开写”而是从运行机制上让每个 Agent 的状态空间变窄决策更稳也更容易定位问题。子图模式在这套思路上更进一步不仅职责分开连图的执行流程都是分层的——父图负责调度子图负责执行某一类完整子任务子图内部还可以继续嵌套。1.2 子图模式相比其他多智能体方案的取舍多智能体常见的协作方式有几种手工编排一个循环里按顺序调多个 Agent、ReAct 式的动态工具路由、以及 LangGraph 的图结构协作。手工编排写起来简单但流程一旦带分支、带循环、带状态同步代码就变成意大利面。动态工具路由靠模型自己选 Agent灵活是灵活但不可控模型选错 Agent 的代价很高尤其客服场景里一旦走错流程用户的耐心就没了。LangGraph 子图模式的核心价值在于把“可控”和“灵活”的平衡点放到图结构上。路由逻辑可以交给模型LLM Router但路由之后每个分支的执行是确定性的子图子图内部也可以有自己的模型决策点。翻译成大白话就是关键岔路口让大模型做选择题但每条路内部怎么走由你画好图不会跑偏。这也是我实际项目里最看中的一点——客户不会为 90% 的准确率买单但会为 95% 以上的可控性鼓掌。再从开发维护的角度说子图模式把复杂的会话流程拆成了几个可以独立测试的模块。我在项目里给每个子图都配了单独的测试用例比如直接给order_agent输入一条“查一下订单 SO-1024”验证返回结构是否符合预期。这种隔离测试能力在大模型应用的工程化里面非常宝贵因为模型的输出天然带随机性如果不把链路拆细出了问题你真的不知道是哪一环在捣乱。2. 核心细节解析与实操要点2.1 先理解 Parent State 与 Child State 的边界用 LangGraph 写子图第一步不是写代码而是理清状态State的边界。子图可以共享父图的 State也可以定义自己的 State关键在子图的input_schema和output_schema怎么设置。我在实际项目里常用的方式是父图只管三个字段——user_query、current_step、final_reply而每个子图内部维护自己的临时状态比如intent、order_info、policy_answer子图结束时只把需要对外暴露的结果写回父图。这种设计的理由很容易理解如果你让所有 Agent 共享一个大 State字段会越来越多内存占用增大还好说关键是某个子图改了共享字段别的子图再读的时候可能拿到脏数据排查起来非常痛苦。所以我的经验是子图之间的通信越“接口化”越好——子图收到的输入是父图传进来的元组或字典子图吐出的结果也是明确的结构内部细节完全隔离。具体到代码层面定义状态时我强烈建议用TypedDict而不是普通的dict。TypedDict能让你在类型检查阶段就发现字段名拼写错误而且配合 IDE 的自动补全写节点函数的时候心里踏实很多。下面是一个实际项目里用过的状态定义示例from typing import TypedDict, Optional class ParentState(TypedDict): user_query: str route: Optional[str] history: list[str] final_reply: str class OrderState(TypedDict): user_query: str order_info: Optional[str] reply: str这里的history字段值得单独说一下。客服场景天然是多轮对话用户经常说“那个订单现在怎么样了”“那个”指代上一轮的订单信息。如果每个子图都只盯着当前一句孤立的话就答非所问了。所以我在父图 State 里加一个history: list[str]每次对话结束后把user_query assistant_reply追加进去Router 和子图都能读取这段历史来做上下文理解。这是子图模式里“状态隔离”和“共享上下文”两者之间最常踩的平衡点。2.2 子图的两种挂载方式节点挂载与函数调用LangGraph 里挂子图有两种常见方式。第一种是直接把StateGraph编译后的app作为一个普通节点挂到父图上builder.add_node(order_agent, order_subgraph_app)这样父图跑到这个节点时会执行整个子图的完整生命周期。第二种是在父图的节点函数里调用子图的ainvoke走函数调用的路子def call_order_agent(state: ParentState): result await order_app.ainvoke({user_query: state[user_query]}) return {final_reply: result[reply]}两者效果类似但区别在于节点挂载方式下子图的内部节点和边在父图的调试面板里是展开的而且子图可以拿到父图的RunnableConfig比如recursion_limit、callbacks函数调用方式则更像“在一个节点里程序化调用”调试时是一个黑盒。实际项目里我更推荐节点挂载尤其你后面要接 LangSmith 或者 Langfuse 做链路追踪的时候子图展开与否直接决定你能不能看清一次完整回答的中间过程。另外节点挂载方式下子图内部若发生异常父图可以通过 try 节点捕获而函数调用方式的异常处理会更绕你可能得在节点函数里自己用 try-except 包一层再把异常信息写入 State。提示如果你刚开始学建议先用函数调用方式写逻辑更容易理解等你要上线生产环境了再改成节点挂载调试和排障会轻松很多。2.3 Router 节点设计LLM 决策 确定性兜底AI 客服系统里 Router 节点至关重要。父图收到用户问题后Router 决定走哪个子图订单查询、业务咨询、还是人工客服转接。我见过不少团队在 Router 节点里直接让模型输出一个 JSON比如{next: order_agent}但实际跑起来经常出现路由 key 和节点名不匹配的问题。我的做法是三层校验。第一用with_structured_output让模型输出固定 schema不要裸 JSON。第二在 Router 节点函数里做 key 映射把模型输出映射到实际节点名同时准备一个fallback节点当置信度低于阈值比如 0.5时直接走兜底路径。第三在 System Prompt 里给足约束样例明确告诉模型“当识别到用户在问套餐变更时必须返回policy_agent而不是order_agent”。这里有一个很值得说的细节Router 不是非黑即白。实际场景里用户可能一个问题涉及两个 Agent。比如“我的订单 SO-1024 能不能退退的话怎么收费”这既涉及订单状态查询又涉及退款政策。如果只走一个 Agent信息一定不完整。我的解决方案是让 Router 输出一个primary_intent和一个secondary_intent父图先把主意图给对应子图跑如果主意图返回的结果里带了一个needs_followup标志再触发第二个子图做补充。这样做可以把复杂问题从串行等待变成一个可控的接力流程体感上回复会更快。Router 的模型选择也有讲究。我用gpt-4o-mini这类轻量模型做路由因为路由决策本身不需要很强的推理能力但需要的是稳定和快。复杂的意图分类丢给轻量模型实测延迟能压到 300 到 500 毫秒以内而如果让大模型来做可能多花一倍的时间对客服场景来说不可接受。3. 实操过程与核心环节实现3.1 先搭一个可运行的最小框架三个子图的 AI 客服下面我手写一个可以跑通的最小案例业务背景用户进线询问套餐、查订单、或者索要人工客服。定义三个子图order_agent查订单、policy_agent业务政策咨询、human_agent转人工。父图只做路由和聚合。第一步定义共享状态和子图状态。这一步决定了整个系统的数据流。我把父图 State 控制在三个字段加一个历史列表子图各自维护私有字段。你不要嫌这一步啰嗦状态定义的好坏直接决定后面掉不掉头发。我见过太多人上来就写节点写到一半发现状态对不上全盘推倒重来。from typing import TypedDict, Optional from langgraph.graph import StateGraph, START, END class ParentState(TypedDict): user_query: str route: Optional[str] final_reply: str class OrderState(TypedDict): user_query: str order_info: Optional[str] reply: str class PolicyState(TypedDict): user_query: str policy_answer: Optional[str] reply: str class HumanState(TypedDict): user_query: str reply: str第二步构建三个子图。每个子图内部可以只有一个节点但为了示范我让order_agent内部带一个条件分支模拟“命中订单号”和“未命中订单号”两种情况。def order_match(state: OrderState): # 实际项目里这里接数据库/工具调用 if 订单号 in state[user_query]: return {order_info: 订单号 SO-1024当前已发货, reply: 查到您的订单 SO-1024 已在运输途中。} return {order_info: None, reply: 抱歉我没有找到您的订单信息建议提供订单号。} order_builder StateGraph(OrderState) order_builder.add_node(match, order_match) order_builder.add_edge(START, match) order_builder.add_edge(match, END) order_app order_builder.compile()policy_agent和human_agent的结构完全一样只是节点函数内部逻辑不同。注意我这里没有给order_agent加条件分支到 human因为你真正写的时候肯定希望在子图内部自己决定“要不要转人工”而不是把决定权交给父图。这个设计留给你扩展下面会讲更完整的版本。第三步构建父图。父图的 Router 节点我用一个简单规则函数代替实际项目里接 LLM 即可。def router_node(state: ParentState): if 订单 in state[user_query]: return {route: order_agent} if 转人工 in state[user_query] or 套餐 in state[user_query]: return {route: policy_agent} return {route: human_agent} def call_order_agent(state: ParentState): result order_app.invoke({user_query: state[user_query]}) return {final_reply: result[reply]} builder StateGraph(ParentState) builder.add_node(router, router_node) builder.add_node(order_agent, call_order_agent) builder.add_node(policy_agent, call_policy_agent) builder.add_node(human_agent, call_human_agent) builder.add_edge(START, router) builder.add_conditional_edges(router, lambda s: s[route], {order_agent: order_agent, policy_agent: policy_agent, human_agent: human_agent}) builder.add_edge(order_agent, END) builder.add_edge(policy_agent, END) builder.add_edge(human_agent, END) app builder.compile()这里有一个需要解释清楚的技术点我在call_order_agent里用的是order_app.invoke而不是把order_app直接作为节点挂载。两种方式都能跑但函数调用方式需要手动传递输入状态并从返回结果里挑出需要的字段再返回给父图。节点挂载方式则更自动化你只需要告诉 LangGraph 这个节点是哪个子图 app状态传递和回调追踪都会更友好。注意lambda s: s[route]这种写法在条件边里很常见但有个隐藏问题——如果s[route]是 None或者包含不在映射表里的值LangGraph 会直接抛异常。稳妥做法是加一个 fallback 映射比如在条件边函数里最后return human_agent确保任何意外值都有人接住。3.2 让子图真正“协作”引入共享记忆与跨 Agent 上下文上面的最小框架每个子图都是独立的拿到什么问题就处理什么问题。真实客服场景里用户经常说“那个订单现在怎么样了”这里的“那个”指代上文的订单如果每个子图都只盯着当前一句话就答非所问了。解决办法有两个层级。第一层级用messages列表把对话历史传进父图 State。我在前面已经给ParentState加了history字段Router 节点和子图节点都能读取它。子图内部如果需要提取实体我通常在order_match节点里先用 LLM 做一个extract_order_no的动作把历史里出现的订单号抽出来再拼到当前查询里。这个做法实现简单效果立竿见影。第二层级上记忆持久化。LangGraph 提供了MemorySaver和SqliteSaver分别做内存态和持久态的记忆。我在客服系统里用SqliteSaver这样服务重启后用户的历史上下文还在。注意一点MemorySaver实测只适合 demo 或单机短会话多实例部署时会话状态不同步一定要换持久化的SqliteSaver或PostgresSaver否则生产环境分分钟翻车。from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(:memory:) as saver: app builder.compile(checkpointersaver)关于记忆还有一层很多人忽略的问题记忆不只对子图内部有用Router 节点同样需要。用户先说“帮我查一下我的订单”Router 路由到order_agent子图回复了订单信息。紧接着用户说“如果这个订单能退的话退费规则是什么”这时候 Router 需要知道“这个订单”指代的是刚才那个订单号才能正确路由到policy_agent同时在传参数的时候带上订单号。这个跨 Agent 的上下文传递靠的就是history里的完整会话记录。我在 Router 的 prompt 里明确加了一句话“根据对话历史判断指代对象如果用户说的是‘那个订单’请从历史中提取订单号并附加在意图中。”实测下来指代消解的准确率能到九成以上。3.3 人工客服转接多智能体与真实世界的握手AI 客服系统跑得再好也逃不过“转人工”这三个字。子图模式把转人工做成一个子图human_agent而不是简单的END好处是能承接后面的“排队状态更新”“工单创建”等动作。我的设计里human_agent子图内部有两个节点create_ticket和notify_backend。create_ticket调用工单系统的 API把用户问题、AI 的初步判断、用户联系方式一并写入工单notify_backend是模拟推送一条 Webhook 给客服工作台。这两个节点之间是串行边确保工单创建成功后再通知顺序不能反。下面给出完整的human_agent子图实现def create_ticket(state: HumanState): # 实际项目里这里调用工单系统 API ticket_id TK-8888 return {ticket_id: ticket_id} def notify_backend(state: HumanState): # 模拟推送 Webhook return {notified: True, reply: 已为您转接人工客服工单号 TK-8888请耐心等待。} human_builder StateGraph(HumanState) human_builder.add_node(create_ticket, create_ticket) human_builder.add_node(notify_backend, notify_backend) human_builder.add_edge(START, create_ticket) human_builder.add_edge(create_ticket, notify_backend) human_builder.add_edge(notify_backend, END) human_app human_builder.compile()这时候如果子图内部节点需要维护多个临时字段比如ticket_id、notified、reply就需要把这些字段都定义在HumanState里。我在上面的HumanState里只写了user_query和reply你实际用的时候记得补全字段。子图内部节点共用子图 State这是子图和父图的一个关键区别——子图内部通信自由但跨图通信必须通过输入输出 schema 显式约定。这里有一个最常见的坑子图返回值不写全。如果human_agent只返回reply没把ticket_id写进父图 State后面的记录统计就没法做。所以我在每个子图的出口都会统一封装一个finish节点把需要对外暴露的字段一次性 merge 进父图 State。这个习惯是从几次线上事故里学来的——有一次客服系统上线后运营发现工单记录全部缺失查了半天发现是子图返回的ticket_id字段名和父图预期的ticket_id不一致数据流静默丢失。4. 常见问题与排查技巧实录4.1 子图返回的 State 没有被父图接收怎么回事这是子图模式最容易踩的坑。如果你在父图节点里用order_app.invoke(...)invoke 返回的是子图的整个 State 字典你拿到后要自己挑选需要对外暴露的字段 return 给父图。很多新手直接return {final_reply: result}结果发现final_reply变成了整个字典前端渲染时直接报错或者说出一串 JSON。我的自查清单是第一步打印子图的result看结构第二步确认父图节点返回的字段和ParentState中声明的字段类型一致第三步在父图里加一个log_node在 END 前打印final_reply的前 50 个字符验证聚合是否正确。三步走完绝大多数 State 传递问题都能定位。如果用的是“把子图 app 直接 add_node”的挂载方式父图和子图之间的字段映射规则不同LangGraph 默认会做局部更新但只针对同名字段。也就是说子图 State 和父图 State 的同名字段会被自动合并不同名字段忽略。想要精准控制就在子图编译时显式指定input_schema和output_schema。class OrderInput(TypedDict): user_query: str class OrderOutput(TypedDict): order_info: Optional[str] reply: str order_app order_builder.compile( input_schemaOrderInput, output_schemaOrderOutput )这样写的好处是子图对外只暴露OrderOutput里声明的字段内部的临时字段比如matched、confidence完全不会污染父图 State。这在多智能体协作里是很强的封装性保障。4.2 递归深度限制 recursion_limit 踩坑LangGraph 默认的recursion_limit我印象中是 25。子图嵌套子图、Router 再进子图、子图内部还有循环很容易踩到GraphRecursionError: The maximum recursion depth was exceeded。这个报错不是说你代码死循环了而是图执行的步数超过了限制。排查思路先看是哪个环节步数暴涨。常见原因有三个一是子图内部的循环边没有设置出口条件比如我在policy_agent里让模型回答不满意就重试结果没设最大重试次数二是 Router 在两个子图之间反复横跳A 没处理完跳 BB 又说需要 A三是单个子图本身节点就多父图每走一步都算一次 recursion嵌套三层后步数很容易上万。我的处理方式是给每个循环边都加一个max_retries计数器计数器用 State 字段维护给 Router 节点加一个route_attempts字段第二次进入同一路由时强制走 fallback最后在app.invoke(..., config{recursion_limit: 100})按需调大限制但调大是最后手段优先检查循环逻辑。def my_node(state): if state.get(retries, 0) 3: return {route: fallback} return {retries: state.get(retries, 0) 1} builder.add_conditional_edges(my_node, my_node, {continue: my_node, fallback: fallback, done: END})4.3 模型路由误判当 Router 把“退订”识别成“投诉”AI 客服里最让人心累的就是语义识别偏差。用户说“我要退订套餐”本来是policy_agent的业务Router 识别成“投诉”进了human_agent用户体验非常差。我的解决方法是给 Router 加一个“意图确认”动作当模型输出路由结果的置信度低于阈值时先让 Router 反问一句“您是想办理套餐变更业务吗”而不是直接进任何子图。具体实现上我维护了一个route_confidence: float字段。Router 节点返回route和route_confidence父图在条件路由前先判断置信度低于 0.6 时走clarify_node该节点用模板生成确认话术并把awaiting_confirmationTrue写进 State下一轮用户确认后再把历史里“确认意图”的信息加入上下文重新走 Router。这个双轮确认机制上线后误路由比例下降了一半以上代价是大约 8% 的简单请求多了一轮交互整体可接受。下面是一个 Router 节点的真实代码结构简化版from pydantic import BaseModel from langchain_openai import ChatOpenAI class RouteDecision(BaseModel): next_node: str confidence: float router_prompt 你是客服系统的路由引擎。根据用户的提问决定将该请求分配给哪个子 Agent。 可选路由order_agent订单查询、policy_agent业务政策、human_agent人工客服。 注意不要将套餐退订归为投诉退订属于 policy_agent除非用户明确表达强烈不满情绪才走 human_agent。 def router_node(state: ParentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) structured_llm llm.with_structured_output(RouteDecision) decision structured_llm.invoke( f用户提问{state[user_query]}对话历史{state[history]} ) if decision.confidence 0.6: return {route: clarify, route_confidence: decision.confidence} return {route: decision.next_node, route_confidence: decision.confidence}4.4 子图内部节点报错会导致整个链路失败吗会。这是很多人初期没料到的。当你在父图里挂载一个子图 app 作为节点时子图内部任何节点抛异常默认都会向上传播把整次ainvoke调用打挂。客服场景里绝不允许一次报错让用户收到“服务器错误”所以必须做异常兜底。我的做法是在父图层面加一个全局 fallback 节点。具体来说给父图加一条从 START 开始的并行分支不现实更实用的做法是利用 LangGraph 的try机制——在父图编译时可以用with上下文捕获或者在节点函数层面对子图调用包一层 try-except。我实际生产代码里是下面这么写的async def safe_call_subgraph(subgraph, input_state): try: result await subgraph.ainvoke(input_state) return result except Exception as e: return {reply: 很抱歉系统暂时开小差了已为您转接人工客服。, error: str(e)}这样即使order_agent内部调用的数据库 API 超时了用户拿到的还是一句有温度的兜底话术而不是光秃秃的异常堆栈。后续再接上日志上报把error字段打到追踪平台方便定位。5. 工具选型与生产落地建议5.1 模型选型为什么我优先用带 function calling 的模型子图模式里Router 节点是模型决策点子图内部也可能需要模型调用工具。为了这套架构跑得稳我建议优先选择带 function calling / structured output 能力的模型而不是只靠 Prompt 让模型输出 JSON。原因很简单结构化输出的模型可以直接绑定预期 schema模型层保证输出合法靠 Prompt 硬凹 JSON三个小时调一个斜杠转义的日子我是真不想再过了。我的习惯是Router 节点用gpt-4o-mini这类轻量模型子图内的实体提取、政策问答用略强的模型最终话术润色可以再走一次模型。分层用模型成本差异不止一点但效果上完全够用。另外所有模型调用都建议配temperature0客服场景对一致性要求极高宁可答案稳一点也不要灵光一现的自由发挥。工具调用还有一个需要重点把关的环节工具的输入参数校验。大模型填参数经常丢三落四比如查订单的工具要求传order_id模型可能只传了一个user_query整个字符串进去。我的办法是在子图节点函数里加一层数据清洗凡是工具调用前都先跑一个 validator 函数参数缺失就自动补默认值或返回错误信息让模型重新生成。这个干净的参数校验层能让整个客服系统的工具调用成功率从八十几提到九十五个点。5.2 可观测性LangSmith/Langfuse 与子图展开子图模式真正的威力只有在可观测时才能发挥出来。我建议从第一天就接入追踪。LangSmith 对 LangGraph 的集成最顺能自动把每个节点、每条边、每次模型调用都记录下来Langfuse 则适合团队已经有自建可观测平台的情况可以跟 OpenTelemetry 打通。在追踪平台上重点看三件事Router 到了哪个子图、子图内部每个节点的耗时、模型的 token 消耗。有一次我们线上反馈“AI 客服响应慢”我打开追踪一看问题不在 Router也不在工具调用而是policy_agent内部有个政策问答节点模型在一个 3000 token 的上下文里反复做无关检索耗时占整个链路的 70%。这种问题没有子图展开的追踪根本定位不到。LangSmith 的 trace 面板里子图节点会默认展开成树状结构。一开始你可能觉得节点太多很乱但其实这是最大的优势——你能清晰地看到每一次调用是从哪个父节点发出的、是在哪个子图内完成的。如果用了函数调用方式挂载子图trace 里只有一个黑盒节点排查效率至少打对折。这也是我前面反复强调用节点挂载的原因之一。5.3 部署形态与稳定性FastAPI 异步态生产环境部署 LangGraph 应用我最常用的形态是 FastAPI 服务启动时编译一次图请求进来后await app.ainvoke(...)拉通全流程。LangGraph 官方也提供了langgraph serve快速起一个 HTTP 服务但自定义程度较低一旦需要鉴权、限流、外部依赖注入还是自己写 FastAPI 更顺手。稳定性上有两个容易被忽略的点。一是线程安全LangGraph 的app.invoke在并发下是否安全取决于内部状态存储建议一律用ainvoke异步写法并且把状态存储切到线程安全的 saver。二是超时控制客服场景不希望用户等太久我会在ainvoke外层包asyncio.wait_for比如 30 秒超时超时后返回“正在为您转接人工客服”的兜底话术。这个兜底逻辑其实也能作为一个fallback_agent子图挂到父图上超时分支直接路由过去架构上更统一。import asyncio from fastapi import FastAPI from pydantic import BaseModel app FastAPI() graph_app compile_graph() # 启动时编译一次 class ChatRequest(BaseModel): user_query: str session_id: str app.post(/chat) async def chat(req: ChatRequest): try: result await asyncio.wait_for( graph_app.ainvoke( {user_query: req.user_query, history: []}, config{configurable: {thread_id: req.session_id}} ), timeout30 ) return {reply: result[final_reply]} except asyncio.TimeoutError: return {reply: 正在为您转接人工客服请稍候。}这里configurable.thread_id的用法值得展开说LangGraph 的 checkpointer 会按thread_id维护独立的会话状态。这意味着你不用自己在 State 里维护history而是把多轮对话的累积交给 LangGraph 的持久化层处理。每次请求带同一个session_id图就能自动恢复上下文。我实际项目里把这两套方案都验证过——早期是自己拼history后来切换到 checkpointer 之后代码量明显减少而且会话恢复的可靠性更高。当然子图内部的私有状态比如ticket_id也会随 checkpointer 一起持久化所以你必须在子图结束时清理不需要保留的临时字段避免 State 无限膨胀。6. 实战体会与进阶扩展如果只看一个心得我想说子图模式本质上不是 LangGraph 的专属技巧而是一种“控制复杂度”的工程思维。把一个大而全的 Agent 分解成若干个高内聚、低耦合的子图让每个子图的边界清楚、出口干净整体的可靠性就会指数级上升。这套思维在客服系统之外同样适用于电商售后、企业知识库问答、工单分类、内容审核等场景。你甚至可以把policy_agent从一个客服项目里抽出来直接复用到另一个销售助手项目里因为它的输入输出是标准化的user_query到reply父图只要把它当作插件挂上就行。再分享一个我正在玩的扩展思路让子图自己注册“能力标签”。比如order_agent在初始化时对外声明自己“能查订单能查物流”policy_agent声明“能查套餐能算退费”。父图的 Router 不再是硬编码的几个分支而是变成一个动态的服务发现机制——根据用户意图到子图能力列表里做匹配。这样新增一个业务子图完全不需要改动父图的逻辑只要在新的 StateGraph 编译后注册进去即可。这个方向再往下走就是多智能体系统的微服务化每个子图独立部署、独立扩容通过 gRPC 或者 HTTP 与父图通信LangGraph 的add_node传的不再是本地的app而是一个远程调用的封装节点。最后提醒一句无论架构怎么演进别忘了给每个子图配上独立的测试用例。我的习惯是给每个子图写一份“输入样例-期望输出”的表格比如测试输入期望路由期望回复要点我要退订套餐policy_agent说明退订规则查订单 SO-1024order_agent返回订单状态转人工human_agent生成工单号这些用例既是回归测试的基准也是你和业务方对齐需求时的沟通语言。多智能体系统最大的风险不是代码写不出来而是没人说得清每个 Agent 的职责边界。表格写清楚代码自然就顺了。