做客服方向的大模型应用大多数团队都会卡在同一个地方单个智能体确实能跑通 demo但一旦同时挂上订单查询、退换货、商品咨询、转人工这些能力Prompt 和工具列表就迅速膨胀模型开始分不清该调用哪个工具业务规则也越改越乱。我在这个项目里换了一种组织方式用 LangGraph 的子图模式把 AI 客服系统拆成“一个主图 多个能力子图”主图只负责意图路由和状态分发每个客服能力封装成独立的子图内部再挂自己的工具和流程。这套方案最终形成了一个可落地的多智能体协作框架用户消息进来后主图先判断意图再按需调用订单查询、售后办理、商品咨询或人工转接子图每个子图内部又是独立的 ReAct 智能体。项目全部基于 LangGraph 实现模型走 OpenAI 兼容协议所以换任意国产大模型都能直接跑最后用 FastAPI 包一层 HTTP 接口就能上线。如果你正在搭智能客服、工单系统或售前导购这类多流程系统或者已经试过 LangGraph 但总觉得链路乱、状态管理糊这篇的建模思路、主图与子图挂载方式、路由节点写法和排查记录可以直接拿来抄作业。1. 为什么用子图模式做 AI 客服系统1.1 单个智能体在客服场景里为什么撑不住客服系统的第一特征是业务边界不清晰。用户说一句“我那个耳机没收到想退款”这句话同时涉及物流查询和售后办理两个意图任何一个单独的智能体面对这种信息都会纠结先调哪个工具。我一开始图省事把所有工具挂到一个 agent 里Prompt 写了三千字还是压不住实测十次里有两三次模型会选错工具比如用户问“能开电子发票吗”它反而去调了库存查询接口。这类问题在单条 Prompt 里很难彻底解决。客服需求的另一个特点是规则密集退换货有时效限制、运费险有赔付范围、某些品类不支持无理由退换这些规则如果全塞进同一个 Prompt或者是堆进同一个工具描述改一次业务策略就要做一轮全量回归。而且单智能体链路意味着每一次修改的爆炸半径是整个客服系统上线风险极大。把能力按业务域拆开之后规则就散落到各个子图里谁的业务谁维护模型的工具选择空间也小得多准确率自然上来。1.2 三种多智能体协作模式我为什么选子图LangGraph 项目里做多智能体目前主流有三条路supervisor 模式、handoff 模式和子图模式。我做这个客服系统前把三种都试过一遍放在真实场景里对比差异非常明显。协作模式组织结构优点缺点典型适用场景supervisor 模式一个主管智能体统一调度多个 worker编排灵活能处理意图交织的动态任务主管上下文压力大链路不透明主管决策错了难排查需求变化频繁、需要动态规划的任务handoff 模式智能体之间直接转移会话控制权会话交接自然模拟真实坐席转接控制流不透明容易互相“踢皮球”有明确会话转交语义的系统子图模式主图按意图分发到独立能力子图边界清晰、状态隔离、子图可独立复用跨子图动态协同能力偏弱意图边界明确的业务系统如客服选子图模式的核心原因是客服系统的意图分布相对稳定无非就是订单、物流、售后、商品咨询和转人工这几类用户不会要求智能体现场编排出一个新流程。子图模式把每条业务线变成独立编译的图主图只关心“这次请求该进哪个子图”子图之间互相不感知业务团队完全可以各自维护自己那部分。更直白一点这就是把 AI 客服做成了电话总机加专席坐席总机按按键分线每线坐席只处理自己那一摊事。1.3 项目架构与功能边界整个系统的顶层结构非常清晰一共就三层。最外层是 FastAPI 服务负责接收 HTTP 请求、维护会话、返回结果中间层是主图承担意图路由和状态分发最里层是四个子图分别对订单查询、退换货办理、商品咨询、人工转接四个能力域负责。子图之下再挂工具就是订单中心接口、售后工单接口、商品知识库这类业务系统。FastAPI HTTP 服务 └─ 主图 main_graph ├─ intent_router 意图路由 ├─ order_agent 订单查询子图 ├─ after_sale_agent 退换货子图 ├─ consult_agent 商品咨询子图 └─ human_agent 人工转接 工具层订单中心 API、售后工单系统、商品知识库这个分层最大的意义在于主图不感知任何业务细节它不需要知道订单查询具体调了什么 API也不需要理解退换货的七日限制它只维护一个路由表把带“订单”字眼的请求丢给订单子图把带“退换”字眼的请求丢给售后子图。业务变复杂时比如售后流程里加一个“运费险理赔”环节我只需要修改售后子图内部的节点主图和其他子图一行代码都不用动。这就是子图模式给工程维护带来的直接好处。2. 先建模再写代码状态与工具准备2.1 LangGraph 的最小运行逻辑五分钟说清很多朋友第一次看 LangGraph 的代码会有点懵其实它的运行逻辑就三句话定义状态类型声明节点再用边把节点串起来。节点是一个普通函数接收当前状态字典返回一个字典表示要更新哪些字段边定义执行顺序条件边则会根据状态里的值决定下一跳走向。整个图编译之后调用 invoke 就是在状态图上从起点走到终点。子图模式之所以成立是因为 LangGraph 里的图本身可以用来当一个节点。子图内部可以继续有自己的节点、边和工具调用编译之后扔进主图的一个节点位置对外部来说它就是一个带输入输出的黑盒。这种“图里有图”的嵌套结构天然适合把复杂业务分层建模。我这里要强调一个容易踩坑的点LangGraph 的状态更新不是自动累积的除了少数用加法合并器处理的字段返回的键值对会直接覆盖对应字段。所以设计状态时要想清楚哪些字段需要累积、哪些字段只需要存最近一次结果。2.2 任务状态 KbState 定义状态是贯穿整个系统的主线多智能体协作最忌讳的就是每个子图自己搞一套状态定义主图传不下去、子图回不来。我的做法是全系统共用一个 KbState字段尽量精简用哪些取哪些。下面这个定义是整个项目的核心数据结构from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage, HumanMessage import os class KbState(TypedDict): messages: Annotated[list, add_messages] user_id: str order_id: str intent: str need_human: bool model ChatOpenAI( modelos.getenv(LLM_MODEL, qwen-plus), api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), temperature0, )这里最关键的是 messages 字段用了Annotated[list, add_messages]。这个声明表示每当一个节点往父状态返回 messages 时LangGraph 不是覆盖整个消息列表而是把新消息追加到已有消息后面。你可以把它理解成一张不断加长的聊天记录纸每个节点做的事情只是在纸的末尾追加内容。如果没有这个合并器第二轮对话时子图返回的新消息会把第一轮的全部历史消息覆盖掉客服系统立刻失忆。其余字段按业务需要定义即可比如 need_human 用来标记是否要转人工order_id 存当前关联的订单号节点可以直接读写这些字段而不依赖大模型。2.3 工具层让智能体真正能干活的基础模型自身不产生业务数据所有订单状态、售后政策、商品信息都来自工具层。这一步我用 LangChain 的 tool 装饰器定义函数函数体内部在真实项目里就是调用订单中心 API 或读取数据库这里为了可演示先返回模拟结果。有一点非常重要tool 下面的 docstring 会被作为工具描述原样交给模型模型靠这个描述来决定何时调用工具所以描述字段必须写清楚“这个工具是干什么的”“应该传什么参数”。from langchain_core.tools import tool tool def get_order_status(order_id: str) - str: 根据订单号查询物流和订单状态。 return f订单 {order_id} 已发货快递单号 SF1234567预计 3 天内送达。 tool def check_return_eligibility(order_id: str) - str: 检查订单是否支持退换货。 return 该订单支持 7 天无理由退换货。 tool def create_after_sale_ticket(order_id: str, reason: str) - str: 创建售后工单传入订单号和售后原因。 return f售后工单 AF-{order_id} 已创建售后同事会在 24 小时内联系您。 tool def search_products(keywords: str) - str: 从商品知识库中检索商品信息。 return 这款降噪耳机支持 7 天无理由退换售价 399 元支持蓝牙 5.3。工具设计上有一条我实测得到的经验工具粒度要按业务子图切分而不是把所有工具堆到一个模型面前。订单查询子图只看得到订单和物流工具售后子图只看得到退换货和工单工具这样模型选错工具的概率会大幅下降。原因很简单工具数量一多模型在 function calling 阶段的 token 就长、注意力就分散误调用的几率随工具数量上升。3. 子图构建实战三个坐席级智能体3.1 订单查询子图基于 create_react_agent 的轻量实现每个子图本质上是一个独立的智能体。订单查询子图是其中最典型也最简单的一种直接用 LangGraph 的 create_react_agent 就能构建。这个函数内置了 ReAct 循环模型思考要不要调工具、调用工具、看结果、再回答整个过程在主图里只体现为一个节点。系统因此有了“多智能体”的效果每个子图都是独立决策的大脑带着自己的工具和行动准则工作。from langgraph.prebuilt import create_react_agent order_subgraph create_react_agent( model, tools[get_order_status], )子图构建出来之后不能直接在用的时候把大段历史一股脑塞给它。我这里做了一层包装函数把主图里最近的几轮消息截取出来再传给子图子图返回的答案再塞回主图的消息流。为什么要截取而不直接传全量因为每个子图都是独立的 ReAct agent每次触发都会重新处理输入消息如果每次把这个会话几个小时的历史全部灌进去token 成本会迅速失控而且历史里其他子图的业务内容对这个子图是纯噪音。def run_order_agent(state: KbState) - dict: recent_messages state[messages][-6:] result order_subgraph.invoke({messages: recent_messages}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}我这里截取最近 6 条消息是处理“用户说‘那个耳机呢’但前文提到过耳机”这类指代问题。如果只传最新一条用户消息子图没有任何上下文根本听不懂“那个”指什么传最近一屏对话一方面保留了指代信息另一方面不至于把半小时前的维修工单记录也带进来干扰它。3.2 退换货子图把“资格校验提交工单”封装成流程退换货子图是整张图里业务最重的一块。它挂在两个工具上一个是 check_return_eligibility 负责校验订单是否还能退换一个是 create_after_sale_ticket 负责真正提交售后工单。这两个工具的组合让子图内部形成了一条隐性的业务链先校验再提交模型负责判断顺序。after_sale_subgraph create_react_agent( model, tools[check_return_eligibility, create_after_sale_ticket], ) def run_after_sale_agent(state: KbState) - dict: if state.get(need_human): return {messages: [AIMessage(content正在为您转接人工客服请稍候。)]} result after_sale_subgraph.invoke({messages: state[messages][-6:]}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}真实业务里退换货流程比这个复杂得多可能涉及优惠券回滚、库存退回校验、审批流。更严格的工程化做法是在子图内部再用手写 StateGraph 做显式流程一个节点做资格校验返回 allowed 和 reason 两个字段条件边根据字段走提交节点或拒绝节点。这样流程完全可控不依赖模型的临场编程。我在项目里的建议是涉及资金、优惠、库存流转的关键路径尽量用显式 StateGraph 固定住步骤只有类似“解释退换政策”这种开放对话才交给 ReAct 的模型自由发挥。3.3 商品咨询与人工转接子图不是所有子图都需要大模型商品咨询子图本质是一个轻量 RAG 入口。演示代码里我用 search_products 模拟知识库检索真实环境换成向量库检索把命中片段塞给模型生成回答就行。这类子图的 ReAct 循环不需要太复杂一个检索工具加一个拼接 Prompt 的节点已经足够因为商品咨询的回答大多来自知识库原文模型主要做润色和口语化表达不需要自己编造商品参数。consult_subgraph create_react_agent( model, tools[search_products], ) def run_consult_agent(state: KbState) - dict: result consult_subgraph.invoke({messages: state[messages][-6:]}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}人工转接子图就更特殊了它完全不调用大模型。用户表达转人工意图后人工转接子图要做的只是把 need_human 置为 True然后给一个固定回复。很多项目把转人工也塞给模型生成既浪费 token又容易在用户情绪激动时生成不合适的回复。我的实现是把它做成一个普通节点函数不经过任何 LLM 调用def run_human_agent(state: KbState) - dict: return { messages: [AIMessage(content已为您转接人工客服请描述您的问题。)], need_human: True, }这里也体现了一个设计原则能用代码解决的问题不要浪费模型。子图本身只是一种组织方式节点里可以是任意逻辑可以是模型调用也可以是普通函数。4. 主图编排多个子图协作的真正难点4.1 子图挂载的两种方式和我的选择LangGraph 官方支持两种把子图挂到主图的方式。一种是直接把编译好的子图对象传给 add_node这是最省事的做法另一种是把子图的调用包进一个普通节点函数再把这个函数 add_node也就是前文展示的做法。我在这里明确推荐第二种。直接挂载很快但有几个隐藏约束要求子图状态字段必须是主图状态字段的子集否则运行时会因为状态键对不上而报错而且子图在执行过程中产生的中间状态容易污染主图排查问题的时候很难分清楚哪个字段是哪一层写入的。包装函数则相当于给子图加了一层显式的输入输出边界我可以控制传给子图的历史长度可以拦截子图的返回结果做二次处理还可以在节点这一层打日志记录每个子图的耗时和 token 消耗。对于正式上线的客服系统这层拦截的价值非常大。main_graph StateGraph(KbState) main_graph.add_node(intent_router, intent_router_node) main_graph.add_node(order_agent, run_order_agent) main_graph.add_node(after_sale_agent, run_after_sale_agent) main_graph.add_node(consult_agent, run_consult_agent) main_graph.add_node(human_agent, run_human_agent) main_graph.add_edge(START, intent_router) main_graph.add_conditional_edges( intent_router, route_by_intent, { order_agent: order_agent, after_sale_agent: after_sale_agent, consult_agent: consult_agent, human_agent: human_agent, }, ) for node_name in [order_agent, after_sale_agent, consult_agent, human_agent]: main_graph.add_edge(node_name, END)这里有个重要的工程决策四个子图节点都直接连到 END。也就是说在一次 invoke 里主图只选择一条子图路径执行执行完就结束不会出现子图之间来回跳转。多轮对话的持续协作不靠一次 invoke 内的循环而是靠下一次用户请求再次进入主图、通过 intent_router 重新路由到对应子图。这样做的好处是每轮请求的链路都是确定的出了问题可以非常快速定位是哪一条路径。4.2 意图路由节点的设计先规则后模型意图路由是多智能体协作的第一道闸门它决定会话进入哪个子图。这里我用了一套基于关键词的规则路由写起来直接跑起来零成本而且完全可测试def intent_router_node(state: KbState) - dict: last_message state[messages][-1] text last_message.content.lower() if any(word in text for word in [人工, 转人工, 投诉, 经理]): return {intent: human_agent} if any(word in text for word in [订单, 物流, 快递, 发货, 签收]): return {intent: order_agent} if any(word in text for word in [退, 换货, 退款, 售后, 维修]): return {intent: after_sale_agent} return {intent: consult_agent} def route_by_intent(state: KbState) - str: return state[intent]为什么路由不直接用大模型因为我实测发现客服意图分类这种低频、高确定性的任务用规则能覆盖九成以上场景而大模型每次路由都会消耗 token 和延迟并且偶尔还会把“我遇到骗子了想退款”这种消息错分到投诉子图。规则匹配的缺点是遇到语义变化会失灵比如“我买的东西在路上了吗”虽然能命中“路上”但和 logistics 关系不大所以我在规则后面留了一条升级路径规则命中不了时可以再挂一个轻量分类模型或小 LLM 做兜底。上线初期用纯规则收集一批真实用户语料后再决定要不要加模型兜底这个节奏比较稳。4.3 会话记忆如何让多个子图共享上下文多智能体系统最容易翻车的部分不是路由而是记忆。用户第一轮说“帮我查一下 20241128 的订单”第二轮说“它什么时候到”这里的“它”必须能指回到 20241128。LangGraph 处理这个问题的标准做法是 Checkpointer 加 thread_idCheckpointer 负责保存每一轮执行完成后的完整状态thread_id 是会话的唯一标识同一会话的多次请求会基于保存的状态继续执行。from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() compiled_main_graph main_graph.compile(checkpointermemory) config {configurable: {thread_id: session-10001}}这里要克服一个常见误区Checkpointer 保存的是主图状态里的 messages但每个子图在包装函数里只接收最近 6 条消息所以会话记忆的“长期性”体现在主图而子图只做局部推理。这个分层很有好处子图不需要知道完整的历史它只要知道足够回答当前问题的上下文。生产环境里 InMemorySaver 只适合单机演示正式服务建议把 Checkpointer 换成 Postgres 或 Redis 版本这样即使进程重启用户之前的对话上下文也不会丢。5. 编译、调试和对外服务5.1 编译主图并跑通一次完整调用子图全部构建完、主图节点和边都接好之后编译主图然后就可以调用了。编译这个步骤很像吹一口气把节点、边、状态定义绑定成一个可执行的图结构Common errors 基本都集中在这一层暴露。跑通一次完整调用代码如下compiled_main_graph main_graph.compile(checkpointermemory) result compiled_main_graph.invoke( {messages: [HumanMessage(content帮我查一下订单 20241128 到哪了)]}, config{configurable: {thread_id: session-10001}}, ) print(result[messages][-1].content)第一次请求会走 查询订单去向 的消息进入 intent_router命中“订单”关键词后走到 order_agent 子图子图里的 ReAct 循环调用 get_order_status 工具最终返回物流信息。接着同一会话再问一句“它什么时候能送到”注意这次的输入里没有订单号但因为 InMemorySaver 保存了上一次的完整消息主图状态里仍然存着 20241128 这个订单号子图看到最近六条消息里包含上下文就能正确回答。这种跨轮指代能力是客服系统区别于普通单轮问答的核心体验。5.2 上线前先看图画出来的结构不会骗人LangGraph 自带图可视化能力编译后的图可以直接导出结构图我每次改完图结构都会导出一张 PNG 贴到会议室白板上。这一步对排查链路特别管用尤其是节点多、边复杂的情况人眼看代码容易漏看图的连接关系一眼就能发现某个节点没被任何一条路径引用或者某个条件边映射到了不存在的节点名。# 导出结构图检查节点与边的连接关系 graph_image compiled_main_graph.get_graph().draw_mermaid_png() with open(kb_graph.png, wb) as f: f.write(graph_image)我的调试习惯是分两段走。第一段用 mock 函数替代真实工具调用和大模型调用只验证路由逻辑和状态流转是否符合预期第二段才接上真实模型逐步打开日志看每条链路的完整轨迹。如果直接一步到位跑全链路一旦出问题你很难分清是路由错、工具错、模型错还是状态更新错。线上的话建议把 LangSmith 或同类 tracing 工具接上多智能体系统每次请求会涉及两三层嵌套调用没有追踪日志排查一个“用户转人工失败”的 case 可能要翻半小时代码。5.3 用 FastAPI 包装成 HTTP 接口AI 客服系统最终要暴露给前端或呼叫中心调用我用 FastAPI 做了一层薄薄的 HTTP 接口。这里不需要把 LangGraph 的图对象暴露出去后端只接收 session_id、用户ID和消息文本内部用 thread_id 绑定会话返回最后一条 AI 消息和是否需要转人工的标志。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAI 客服系统) class ChatRequest(BaseModel): session_id: str user_id: str message: str class ChatResponse(BaseModel): answer: str need_human: bool False app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): config {configurable: {thread_id: req.session_id}} result compiled_main_graph.invoke( {messages: [HumanMessage(contentreq.message)], user_id: req.user_id}, configconfig, ) return ChatResponse( answerresult[messages][-1].content, need_humanresult.get(need_human, False), )启动服务后直接调 uvicorn main:app --port 8000前端 POST 一个 JSON 就能拿到结果。如果需要打字机式的流式输出可以把 invoke 换成 astream接口改成 SSE 或 WebSocket不过流式会增加不少复杂度第一版建议先同步返回把链路跑稳再说。6. 常见问题速查表与工程优化建议6.1 高频问题排查速查表我把这个项目里踩过的高频问题整理成一张表这些问题在 LangGraph 多智能体项目里基本都会碰到直接对着查能省很多时间。问题现象可能原因处理方式报 Recursion limit reached子图内部 ReAct 循环没触发退出条件或主图错误地把节点连回路由节点形成死循环检查子图工具是否返回足够明确的结束信号检查主图条件边是否闭环报 Invalid key 错误节点函数返回了 KbState 里没有的字段要么在状态定义里补上该字段要么精简节点返回值只返回已定义字段多轮对话失忆回答对不上前文没配 Checkpointer或 thread_id 每次请求都在变compile(checkpointer...)invoke 时用同一 thread_id模型频繁调错工具工具数量太多或工具描述写得太含糊按子图拆分工具列表把 tool docstring 写成具体、带边界说明的操作手册消息出现重复包装函数里同时手动 append 了一条消息主图又有 add_messages 合并器两边一叠加就重复节点返回的消息只通过 add_messages 合并不要手动修改 state[messages]API 名称找不到langgraph 不同版本之间的模块名有差异优先用当前稳定版本常见的差异点是 MemorySaver 与 InMemorySaver、graph.message.add_messages 与 graph.add_messages这里有个我反复强调的坑不要在包装节点里既手动修改state[messages]又通过返回字典用 add_messages 合并。手动 append 加自动合并等于消息写了两遍子图下次接收上下文时会看到重复内容模型的回答质量会肉眼可见地下降。正确的姿势是只做一件事要么不清空 messages 只靠合并器追加要么只在返回字典里给出要新增的消息。6.2 成本与 token 优化多智能体系统的 token 消耗比你想象中涨得快因为同一个用户请求会经过路由判断、子图 ReAct 循环、工具结果回填等多个环节。我的第一条优化实践是路由尽量用规则或小分类模型这一环节省掉的 token 非常可观。第二条是控制每次传给子图的历史窗口长度不要图省事把完整 messages 全灌进去历史每多一轮每个子图的 ReAct 循环就要重新处理一轮成本是叠乘的。第三条建议是设计工具调用时“能先拦截就先拦截”比如退换货场景里先让 check_return_eligibility 工具确认订单资质如果已经超期就直接生成拒绝文案不让模型再去走创建工单的流程。拦截逻辑放在工具里比放在 Prompt 里可靠得多。对更复杂的跨天会话还可以加一个“历史摘要压缩”节点把半小时前的对话先让模型总结成一段摘要后续请求只带摘要再加最近几轮原文这样既保住了上下文语义又把 token 控制在一个稳定区间。6.3 从 Demo 走到生产的扩展建议如果只是为了技术验证这套架构已经够用了一旦要真正上线我建议再补几块。第一块是中途转人工的能力。目前的架构里用户进入订单子图后如果中途说“算了你直接转人工”子图内部还在按 ReAct 循环执行外部很难打断。我在业务子图里额外挂了一个 handover_to_human 工具让模型自己识别转人工的信号并主动调用包装函数发现返回结果里带转人工标记后直接跳到人工转接逻辑整个链路就活了。第二块是离线评测集。多智能体系统上线后最大风险不是崩溃而是路由漂移比如大模型底座升级后意图识别行为发生变化。我建了一个意图路由评测集里面放几百条真实客服语料每次变更代码或者换模型就跑一遍回归准确率掉了就不上线。第三块是子图复用。售后子图不仅可以被客服主图调用也可以被工单系统、CRM 系统独立调用因为子图本身是独立编译的只要用不同的状态和上下文包一层就能复用到其他场景这也是子图模式相比单一大 Prompt 最大的长期价值。做这套系统我最大的体会是LangGraph 的子图模式真正解决的并不是性能问题而是组织问题。你把订单、售后、咨询、人工转接各自封装成独立子图之后业务边界、团队成员分工、代码仓库结构全部跟着清晰了起来真正的难点从写代码变成了画清楚业务边界。另外提醒一句无论多智能体系统看起来多简洁一定要保留完整的日志和追踪数据否则链路一长排查问题的成本会成倍增加。最后给你一个小技巧给节点和子图起业务化的中文名比如“售后办理子图”“人工转接节点”LangGraph 导出的可视化图里这些名字会直接显示出来贴到文档里比任何架构说明都直白。 LangGraph子图模式构建多智能体AI客服系统的实战方案
做客服方向的大模型应用大多数团队都会卡在同一个地方单个智能体确实能跑通 demo但一旦同时挂上订单查询、退换货、商品咨询、转人工这些能力Prompt 和工具列表就迅速膨胀模型开始分不清该调用哪个工具业务规则也越改越乱。我在这个项目里换了一种组织方式用 LangGraph 的子图模式把 AI 客服系统拆成“一个主图 多个能力子图”主图只负责意图路由和状态分发每个客服能力封装成独立的子图内部再挂自己的工具和流程。这套方案最终形成了一个可落地的多智能体协作框架用户消息进来后主图先判断意图再按需调用订单查询、售后办理、商品咨询或人工转接子图每个子图内部又是独立的 ReAct 智能体。项目全部基于 LangGraph 实现模型走 OpenAI 兼容协议所以换任意国产大模型都能直接跑最后用 FastAPI 包一层 HTTP 接口就能上线。如果你正在搭智能客服、工单系统或售前导购这类多流程系统或者已经试过 LangGraph 但总觉得链路乱、状态管理糊这篇的建模思路、主图与子图挂载方式、路由节点写法和排查记录可以直接拿来抄作业。1. 为什么用子图模式做 AI 客服系统1.1 单个智能体在客服场景里为什么撑不住客服系统的第一特征是业务边界不清晰。用户说一句“我那个耳机没收到想退款”这句话同时涉及物流查询和售后办理两个意图任何一个单独的智能体面对这种信息都会纠结先调哪个工具。我一开始图省事把所有工具挂到一个 agent 里Prompt 写了三千字还是压不住实测十次里有两三次模型会选错工具比如用户问“能开电子发票吗”它反而去调了库存查询接口。这类问题在单条 Prompt 里很难彻底解决。客服需求的另一个特点是规则密集退换货有时效限制、运费险有赔付范围、某些品类不支持无理由退换这些规则如果全塞进同一个 Prompt或者是堆进同一个工具描述改一次业务策略就要做一轮全量回归。而且单智能体链路意味着每一次修改的爆炸半径是整个客服系统上线风险极大。把能力按业务域拆开之后规则就散落到各个子图里谁的业务谁维护模型的工具选择空间也小得多准确率自然上来。1.2 三种多智能体协作模式我为什么选子图LangGraph 项目里做多智能体目前主流有三条路supervisor 模式、handoff 模式和子图模式。我做这个客服系统前把三种都试过一遍放在真实场景里对比差异非常明显。协作模式组织结构优点缺点典型适用场景supervisor 模式一个主管智能体统一调度多个 worker编排灵活能处理意图交织的动态任务主管上下文压力大链路不透明主管决策错了难排查需求变化频繁、需要动态规划的任务handoff 模式智能体之间直接转移会话控制权会话交接自然模拟真实坐席转接控制流不透明容易互相“踢皮球”有明确会话转交语义的系统子图模式主图按意图分发到独立能力子图边界清晰、状态隔离、子图可独立复用跨子图动态协同能力偏弱意图边界明确的业务系统如客服选子图模式的核心原因是客服系统的意图分布相对稳定无非就是订单、物流、售后、商品咨询和转人工这几类用户不会要求智能体现场编排出一个新流程。子图模式把每条业务线变成独立编译的图主图只关心“这次请求该进哪个子图”子图之间互相不感知业务团队完全可以各自维护自己那部分。更直白一点这就是把 AI 客服做成了电话总机加专席坐席总机按按键分线每线坐席只处理自己那一摊事。1.3 项目架构与功能边界整个系统的顶层结构非常清晰一共就三层。最外层是 FastAPI 服务负责接收 HTTP 请求、维护会话、返回结果中间层是主图承担意图路由和状态分发最里层是四个子图分别对订单查询、退换货办理、商品咨询、人工转接四个能力域负责。子图之下再挂工具就是订单中心接口、售后工单接口、商品知识库这类业务系统。FastAPI HTTP 服务 └─ 主图 main_graph ├─ intent_router 意图路由 ├─ order_agent 订单查询子图 ├─ after_sale_agent 退换货子图 ├─ consult_agent 商品咨询子图 └─ human_agent 人工转接 工具层订单中心 API、售后工单系统、商品知识库这个分层最大的意义在于主图不感知任何业务细节它不需要知道订单查询具体调了什么 API也不需要理解退换货的七日限制它只维护一个路由表把带“订单”字眼的请求丢给订单子图把带“退换”字眼的请求丢给售后子图。业务变复杂时比如售后流程里加一个“运费险理赔”环节我只需要修改售后子图内部的节点主图和其他子图一行代码都不用动。这就是子图模式给工程维护带来的直接好处。2. 先建模再写代码状态与工具准备2.1 LangGraph 的最小运行逻辑五分钟说清很多朋友第一次看 LangGraph 的代码会有点懵其实它的运行逻辑就三句话定义状态类型声明节点再用边把节点串起来。节点是一个普通函数接收当前状态字典返回一个字典表示要更新哪些字段边定义执行顺序条件边则会根据状态里的值决定下一跳走向。整个图编译之后调用 invoke 就是在状态图上从起点走到终点。子图模式之所以成立是因为 LangGraph 里的图本身可以用来当一个节点。子图内部可以继续有自己的节点、边和工具调用编译之后扔进主图的一个节点位置对外部来说它就是一个带输入输出的黑盒。这种“图里有图”的嵌套结构天然适合把复杂业务分层建模。我这里要强调一个容易踩坑的点LangGraph 的状态更新不是自动累积的除了少数用加法合并器处理的字段返回的键值对会直接覆盖对应字段。所以设计状态时要想清楚哪些字段需要累积、哪些字段只需要存最近一次结果。2.2 任务状态 KbState 定义状态是贯穿整个系统的主线多智能体协作最忌讳的就是每个子图自己搞一套状态定义主图传不下去、子图回不来。我的做法是全系统共用一个 KbState字段尽量精简用哪些取哪些。下面这个定义是整个项目的核心数据结构from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage, HumanMessage import os class KbState(TypedDict): messages: Annotated[list, add_messages] user_id: str order_id: str intent: str need_human: bool model ChatOpenAI( modelos.getenv(LLM_MODEL, qwen-plus), api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), temperature0, )这里最关键的是 messages 字段用了Annotated[list, add_messages]。这个声明表示每当一个节点往父状态返回 messages 时LangGraph 不是覆盖整个消息列表而是把新消息追加到已有消息后面。你可以把它理解成一张不断加长的聊天记录纸每个节点做的事情只是在纸的末尾追加内容。如果没有这个合并器第二轮对话时子图返回的新消息会把第一轮的全部历史消息覆盖掉客服系统立刻失忆。其余字段按业务需要定义即可比如 need_human 用来标记是否要转人工order_id 存当前关联的订单号节点可以直接读写这些字段而不依赖大模型。2.3 工具层让智能体真正能干活的基础模型自身不产生业务数据所有订单状态、售后政策、商品信息都来自工具层。这一步我用 LangChain 的 tool 装饰器定义函数函数体内部在真实项目里就是调用订单中心 API 或读取数据库这里为了可演示先返回模拟结果。有一点非常重要tool 下面的 docstring 会被作为工具描述原样交给模型模型靠这个描述来决定何时调用工具所以描述字段必须写清楚“这个工具是干什么的”“应该传什么参数”。from langchain_core.tools import tool tool def get_order_status(order_id: str) - str: 根据订单号查询物流和订单状态。 return f订单 {order_id} 已发货快递单号 SF1234567预计 3 天内送达。 tool def check_return_eligibility(order_id: str) - str: 检查订单是否支持退换货。 return 该订单支持 7 天无理由退换货。 tool def create_after_sale_ticket(order_id: str, reason: str) - str: 创建售后工单传入订单号和售后原因。 return f售后工单 AF-{order_id} 已创建售后同事会在 24 小时内联系您。 tool def search_products(keywords: str) - str: 从商品知识库中检索商品信息。 return 这款降噪耳机支持 7 天无理由退换售价 399 元支持蓝牙 5.3。工具设计上有一条我实测得到的经验工具粒度要按业务子图切分而不是把所有工具堆到一个模型面前。订单查询子图只看得到订单和物流工具售后子图只看得到退换货和工单工具这样模型选错工具的概率会大幅下降。原因很简单工具数量一多模型在 function calling 阶段的 token 就长、注意力就分散误调用的几率随工具数量上升。3. 子图构建实战三个坐席级智能体3.1 订单查询子图基于 create_react_agent 的轻量实现每个子图本质上是一个独立的智能体。订单查询子图是其中最典型也最简单的一种直接用 LangGraph 的 create_react_agent 就能构建。这个函数内置了 ReAct 循环模型思考要不要调工具、调用工具、看结果、再回答整个过程在主图里只体现为一个节点。系统因此有了“多智能体”的效果每个子图都是独立决策的大脑带着自己的工具和行动准则工作。from langgraph.prebuilt import create_react_agent order_subgraph create_react_agent( model, tools[get_order_status], )子图构建出来之后不能直接在用的时候把大段历史一股脑塞给它。我这里做了一层包装函数把主图里最近的几轮消息截取出来再传给子图子图返回的答案再塞回主图的消息流。为什么要截取而不直接传全量因为每个子图都是独立的 ReAct agent每次触发都会重新处理输入消息如果每次把这个会话几个小时的历史全部灌进去token 成本会迅速失控而且历史里其他子图的业务内容对这个子图是纯噪音。def run_order_agent(state: KbState) - dict: recent_messages state[messages][-6:] result order_subgraph.invoke({messages: recent_messages}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}我这里截取最近 6 条消息是处理“用户说‘那个耳机呢’但前文提到过耳机”这类指代问题。如果只传最新一条用户消息子图没有任何上下文根本听不懂“那个”指什么传最近一屏对话一方面保留了指代信息另一方面不至于把半小时前的维修工单记录也带进来干扰它。3.2 退换货子图把“资格校验提交工单”封装成流程退换货子图是整张图里业务最重的一块。它挂在两个工具上一个是 check_return_eligibility 负责校验订单是否还能退换一个是 create_after_sale_ticket 负责真正提交售后工单。这两个工具的组合让子图内部形成了一条隐性的业务链先校验再提交模型负责判断顺序。after_sale_subgraph create_react_agent( model, tools[check_return_eligibility, create_after_sale_ticket], ) def run_after_sale_agent(state: KbState) - dict: if state.get(need_human): return {messages: [AIMessage(content正在为您转接人工客服请稍候。)]} result after_sale_subgraph.invoke({messages: state[messages][-6:]}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}真实业务里退换货流程比这个复杂得多可能涉及优惠券回滚、库存退回校验、审批流。更严格的工程化做法是在子图内部再用手写 StateGraph 做显式流程一个节点做资格校验返回 allowed 和 reason 两个字段条件边根据字段走提交节点或拒绝节点。这样流程完全可控不依赖模型的临场编程。我在项目里的建议是涉及资金、优惠、库存流转的关键路径尽量用显式 StateGraph 固定住步骤只有类似“解释退换政策”这种开放对话才交给 ReAct 的模型自由发挥。3.3 商品咨询与人工转接子图不是所有子图都需要大模型商品咨询子图本质是一个轻量 RAG 入口。演示代码里我用 search_products 模拟知识库检索真实环境换成向量库检索把命中片段塞给模型生成回答就行。这类子图的 ReAct 循环不需要太复杂一个检索工具加一个拼接 Prompt 的节点已经足够因为商品咨询的回答大多来自知识库原文模型主要做润色和口语化表达不需要自己编造商品参数。consult_subgraph create_react_agent( model, tools[search_products], ) def run_consult_agent(state: KbState) - dict: result consult_subgraph.invoke({messages: state[messages][-6:]}) answer result[messages][-1].content return {messages: [AIMessage(contentanswer)]}人工转接子图就更特殊了它完全不调用大模型。用户表达转人工意图后人工转接子图要做的只是把 need_human 置为 True然后给一个固定回复。很多项目把转人工也塞给模型生成既浪费 token又容易在用户情绪激动时生成不合适的回复。我的实现是把它做成一个普通节点函数不经过任何 LLM 调用def run_human_agent(state: KbState) - dict: return { messages: [AIMessage(content已为您转接人工客服请描述您的问题。)], need_human: True, }这里也体现了一个设计原则能用代码解决的问题不要浪费模型。子图本身只是一种组织方式节点里可以是任意逻辑可以是模型调用也可以是普通函数。4. 主图编排多个子图协作的真正难点4.1 子图挂载的两种方式和我的选择LangGraph 官方支持两种把子图挂到主图的方式。一种是直接把编译好的子图对象传给 add_node这是最省事的做法另一种是把子图的调用包进一个普通节点函数再把这个函数 add_node也就是前文展示的做法。我在这里明确推荐第二种。直接挂载很快但有几个隐藏约束要求子图状态字段必须是主图状态字段的子集否则运行时会因为状态键对不上而报错而且子图在执行过程中产生的中间状态容易污染主图排查问题的时候很难分清楚哪个字段是哪一层写入的。包装函数则相当于给子图加了一层显式的输入输出边界我可以控制传给子图的历史长度可以拦截子图的返回结果做二次处理还可以在节点这一层打日志记录每个子图的耗时和 token 消耗。对于正式上线的客服系统这层拦截的价值非常大。main_graph StateGraph(KbState) main_graph.add_node(intent_router, intent_router_node) main_graph.add_node(order_agent, run_order_agent) main_graph.add_node(after_sale_agent, run_after_sale_agent) main_graph.add_node(consult_agent, run_consult_agent) main_graph.add_node(human_agent, run_human_agent) main_graph.add_edge(START, intent_router) main_graph.add_conditional_edges( intent_router, route_by_intent, { order_agent: order_agent, after_sale_agent: after_sale_agent, consult_agent: consult_agent, human_agent: human_agent, }, ) for node_name in [order_agent, after_sale_agent, consult_agent, human_agent]: main_graph.add_edge(node_name, END)这里有个重要的工程决策四个子图节点都直接连到 END。也就是说在一次 invoke 里主图只选择一条子图路径执行执行完就结束不会出现子图之间来回跳转。多轮对话的持续协作不靠一次 invoke 内的循环而是靠下一次用户请求再次进入主图、通过 intent_router 重新路由到对应子图。这样做的好处是每轮请求的链路都是确定的出了问题可以非常快速定位是哪一条路径。4.2 意图路由节点的设计先规则后模型意图路由是多智能体协作的第一道闸门它决定会话进入哪个子图。这里我用了一套基于关键词的规则路由写起来直接跑起来零成本而且完全可测试def intent_router_node(state: KbState) - dict: last_message state[messages][-1] text last_message.content.lower() if any(word in text for word in [人工, 转人工, 投诉, 经理]): return {intent: human_agent} if any(word in text for word in [订单, 物流, 快递, 发货, 签收]): return {intent: order_agent} if any(word in text for word in [退, 换货, 退款, 售后, 维修]): return {intent: after_sale_agent} return {intent: consult_agent} def route_by_intent(state: KbState) - str: return state[intent]为什么路由不直接用大模型因为我实测发现客服意图分类这种低频、高确定性的任务用规则能覆盖九成以上场景而大模型每次路由都会消耗 token 和延迟并且偶尔还会把“我遇到骗子了想退款”这种消息错分到投诉子图。规则匹配的缺点是遇到语义变化会失灵比如“我买的东西在路上了吗”虽然能命中“路上”但和 logistics 关系不大所以我在规则后面留了一条升级路径规则命中不了时可以再挂一个轻量分类模型或小 LLM 做兜底。上线初期用纯规则收集一批真实用户语料后再决定要不要加模型兜底这个节奏比较稳。4.3 会话记忆如何让多个子图共享上下文多智能体系统最容易翻车的部分不是路由而是记忆。用户第一轮说“帮我查一下 20241128 的订单”第二轮说“它什么时候到”这里的“它”必须能指回到 20241128。LangGraph 处理这个问题的标准做法是 Checkpointer 加 thread_idCheckpointer 负责保存每一轮执行完成后的完整状态thread_id 是会话的唯一标识同一会话的多次请求会基于保存的状态继续执行。from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() compiled_main_graph main_graph.compile(checkpointermemory) config {configurable: {thread_id: session-10001}}这里要克服一个常见误区Checkpointer 保存的是主图状态里的 messages但每个子图在包装函数里只接收最近 6 条消息所以会话记忆的“长期性”体现在主图而子图只做局部推理。这个分层很有好处子图不需要知道完整的历史它只要知道足够回答当前问题的上下文。生产环境里 InMemorySaver 只适合单机演示正式服务建议把 Checkpointer 换成 Postgres 或 Redis 版本这样即使进程重启用户之前的对话上下文也不会丢。5. 编译、调试和对外服务5.1 编译主图并跑通一次完整调用子图全部构建完、主图节点和边都接好之后编译主图然后就可以调用了。编译这个步骤很像吹一口气把节点、边、状态定义绑定成一个可执行的图结构Common errors 基本都集中在这一层暴露。跑通一次完整调用代码如下compiled_main_graph main_graph.compile(checkpointermemory) result compiled_main_graph.invoke( {messages: [HumanMessage(content帮我查一下订单 20241128 到哪了)]}, config{configurable: {thread_id: session-10001}}, ) print(result[messages][-1].content)第一次请求会走 查询订单去向 的消息进入 intent_router命中“订单”关键词后走到 order_agent 子图子图里的 ReAct 循环调用 get_order_status 工具最终返回物流信息。接着同一会话再问一句“它什么时候能送到”注意这次的输入里没有订单号但因为 InMemorySaver 保存了上一次的完整消息主图状态里仍然存着 20241128 这个订单号子图看到最近六条消息里包含上下文就能正确回答。这种跨轮指代能力是客服系统区别于普通单轮问答的核心体验。5.2 上线前先看图画出来的结构不会骗人LangGraph 自带图可视化能力编译后的图可以直接导出结构图我每次改完图结构都会导出一张 PNG 贴到会议室白板上。这一步对排查链路特别管用尤其是节点多、边复杂的情况人眼看代码容易漏看图的连接关系一眼就能发现某个节点没被任何一条路径引用或者某个条件边映射到了不存在的节点名。# 导出结构图检查节点与边的连接关系 graph_image compiled_main_graph.get_graph().draw_mermaid_png() with open(kb_graph.png, wb) as f: f.write(graph_image)我的调试习惯是分两段走。第一段用 mock 函数替代真实工具调用和大模型调用只验证路由逻辑和状态流转是否符合预期第二段才接上真实模型逐步打开日志看每条链路的完整轨迹。如果直接一步到位跑全链路一旦出问题你很难分清是路由错、工具错、模型错还是状态更新错。线上的话建议把 LangSmith 或同类 tracing 工具接上多智能体系统每次请求会涉及两三层嵌套调用没有追踪日志排查一个“用户转人工失败”的 case 可能要翻半小时代码。5.3 用 FastAPI 包装成 HTTP 接口AI 客服系统最终要暴露给前端或呼叫中心调用我用 FastAPI 做了一层薄薄的 HTTP 接口。这里不需要把 LangGraph 的图对象暴露出去后端只接收 session_id、用户ID和消息文本内部用 thread_id 绑定会话返回最后一条 AI 消息和是否需要转人工的标志。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAI 客服系统) class ChatRequest(BaseModel): session_id: str user_id: str message: str class ChatResponse(BaseModel): answer: str need_human: bool False app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): config {configurable: {thread_id: req.session_id}} result compiled_main_graph.invoke( {messages: [HumanMessage(contentreq.message)], user_id: req.user_id}, configconfig, ) return ChatResponse( answerresult[messages][-1].content, need_humanresult.get(need_human, False), )启动服务后直接调 uvicorn main:app --port 8000前端 POST 一个 JSON 就能拿到结果。如果需要打字机式的流式输出可以把 invoke 换成 astream接口改成 SSE 或 WebSocket不过流式会增加不少复杂度第一版建议先同步返回把链路跑稳再说。6. 常见问题速查表与工程优化建议6.1 高频问题排查速查表我把这个项目里踩过的高频问题整理成一张表这些问题在 LangGraph 多智能体项目里基本都会碰到直接对着查能省很多时间。问题现象可能原因处理方式报 Recursion limit reached子图内部 ReAct 循环没触发退出条件或主图错误地把节点连回路由节点形成死循环检查子图工具是否返回足够明确的结束信号检查主图条件边是否闭环报 Invalid key 错误节点函数返回了 KbState 里没有的字段要么在状态定义里补上该字段要么精简节点返回值只返回已定义字段多轮对话失忆回答对不上前文没配 Checkpointer或 thread_id 每次请求都在变compile(checkpointer...)invoke 时用同一 thread_id模型频繁调错工具工具数量太多或工具描述写得太含糊按子图拆分工具列表把 tool docstring 写成具体、带边界说明的操作手册消息出现重复包装函数里同时手动 append 了一条消息主图又有 add_messages 合并器两边一叠加就重复节点返回的消息只通过 add_messages 合并不要手动修改 state[messages]API 名称找不到langgraph 不同版本之间的模块名有差异优先用当前稳定版本常见的差异点是 MemorySaver 与 InMemorySaver、graph.message.add_messages 与 graph.add_messages这里有个我反复强调的坑不要在包装节点里既手动修改state[messages]又通过返回字典用 add_messages 合并。手动 append 加自动合并等于消息写了两遍子图下次接收上下文时会看到重复内容模型的回答质量会肉眼可见地下降。正确的姿势是只做一件事要么不清空 messages 只靠合并器追加要么只在返回字典里给出要新增的消息。6.2 成本与 token 优化多智能体系统的 token 消耗比你想象中涨得快因为同一个用户请求会经过路由判断、子图 ReAct 循环、工具结果回填等多个环节。我的第一条优化实践是路由尽量用规则或小分类模型这一环节省掉的 token 非常可观。第二条是控制每次传给子图的历史窗口长度不要图省事把完整 messages 全灌进去历史每多一轮每个子图的 ReAct 循环就要重新处理一轮成本是叠乘的。第三条建议是设计工具调用时“能先拦截就先拦截”比如退换货场景里先让 check_return_eligibility 工具确认订单资质如果已经超期就直接生成拒绝文案不让模型再去走创建工单的流程。拦截逻辑放在工具里比放在 Prompt 里可靠得多。对更复杂的跨天会话还可以加一个“历史摘要压缩”节点把半小时前的对话先让模型总结成一段摘要后续请求只带摘要再加最近几轮原文这样既保住了上下文语义又把 token 控制在一个稳定区间。6.3 从 Demo 走到生产的扩展建议如果只是为了技术验证这套架构已经够用了一旦要真正上线我建议再补几块。第一块是中途转人工的能力。目前的架构里用户进入订单子图后如果中途说“算了你直接转人工”子图内部还在按 ReAct 循环执行外部很难打断。我在业务子图里额外挂了一个 handover_to_human 工具让模型自己识别转人工的信号并主动调用包装函数发现返回结果里带转人工标记后直接跳到人工转接逻辑整个链路就活了。第二块是离线评测集。多智能体系统上线后最大风险不是崩溃而是路由漂移比如大模型底座升级后意图识别行为发生变化。我建了一个意图路由评测集里面放几百条真实客服语料每次变更代码或者换模型就跑一遍回归准确率掉了就不上线。第三块是子图复用。售后子图不仅可以被客服主图调用也可以被工单系统、CRM 系统独立调用因为子图本身是独立编译的只要用不同的状态和上下文包一层就能复用到其他场景这也是子图模式相比单一大 Prompt 最大的长期价值。做这套系统我最大的体会是LangGraph 的子图模式真正解决的并不是性能问题而是组织问题。你把订单、售后、咨询、人工转接各自封装成独立子图之后业务边界、团队成员分工、代码仓库结构全部跟着清晰了起来真正的难点从写代码变成了画清楚业务边界。另外提醒一句无论多智能体系统看起来多简洁一定要保留完整的日志和追踪数据否则链路一长排查问题的成本会成倍增加。最后给你一个小技巧给节点和子图起业务化的中文名比如“售后办理子图”“人工转接节点”LangGraph 导出的可视化图里这些名字会直接显示出来贴到文档里比任何架构说明都直白。