资讯动态

从链式调用到图编排:LangGraph实现多智能体调度实战

发布时间:2026/10/20 14:09:15 来源:尧图企业网站定制
如果要用一句话概括最近这轮架构折腾的核心那就是我把一条跑得挺稳的线性链Agent拆了换成了基于LangGraph的动态图编排重新复刻了某区域总部那套AI agent指挥官的调度内核。这篇既是架构实战记录也是踩坑后的复盘主要写给正准备从链式调用转向图编排的同行参考。先说结论线性链不是不能用而是扛不住真实业务里那种下一步去哪由上一步结果决定的复杂度LangGraph这种动态图方案才是多智能体调度该有的样子。这个AI agent指挥官本身不是花架子它的核心价值在于一个总控LLM负责理解任务、拆解步骤、分发给不同专家子Agent再收集结果继续决策。听起来不就是个循环加个分发吗但真正落地时你会发现节点怎么切、状态怎么传、边怎么跳、循环怎么终止每一步都有不少门道。下面我把整个复刻过程拆开讲包括设计思路、关键代码和实测数据。1. 线性链为什么会卡壳三个绕不过去的真实场景1.1 线性链的本质编译器式的顺序执行我们最初用的是LangChain里的链式写法大概逻辑就是先意图识别再调用工具再生成回答一步接一步像流水线一样。这种模式的好处显而易见——好理解、好调试、好写文档每一步的输入输出都是确定的。对于用户问A我回答A用户问B我回答B这种固定问答线性链完全够用。但问题出在它有一个隐含假设整条路径在运行前就是已知的。一旦下一步去哪需要依赖上一步的实时结果线性链就变味了——要么你把所有分支提前用if/else写死形成一段又长又脆的判断逻辑要么你干脆把多个链串联起来结果就是状态散落得到处都是。我们在复刻指挥官系统时第一版就是这么写的代码里塞满了一堆RouterChain和条件判断可维护性直接崩掉。1.2 场景一任务路径取决于用户意图指挥官系统接到的任务五花八门。比如用户说帮我整理这份调研报告并总结成PPT大纲这既涉及资料检索又涉及文本总结还涉及大纲生成。如果用线性链你得预先定义好先检索再总结再生成大纲的顺序。但真实情况是用户可能在同一次对话里随时追加需求先别管PPT大纲先帮我查一下近三年的数据。这时候线性链的固定顺序就尴尬了你已经跑到第三步了用户的需求一变前面几步白做。动态图编排就可以做到把检索和总结都当作独立节点指挥官根据用户最新意图重新决定接下来跳到哪个节点甚至回退到之前的节点重新执行。1.3 场景二子任务需要回头找指挥官真正让我下决心抛弃线性链的是子Agent执行完要回来汇报指挥官再决定下一步这个需求。线性链是单向流动的数据只能从A传到B传到C没办法从C再绕回A。但指挥官系统天然就需要这种循环专家Agent执行完一个子任务把结果交给指挥官指挥官判断还需要继续查资料还是可以生成结果了然后决定是回到检索节点还是跳到生成节点。这种环形回流在线性链里几乎无解。你只能用外部循环套着整条链来模拟但这样一来每一轮循环的状态维护、结果累积、退出条件判断全都要你自己手写代码复杂度直线上升而且特别容易漏掉某条分支上的状态更新。1.4 场景三失败后的重试不是重新跑一遍还有一个高频场景工具调用失败。线性链的处理方式通常是整条链从头再来或者直接抛异常。但实际业务里失败往往只是某个环节的问题——比如检索接口超时、某个解析步骤返回了非法格式。更好的处理是让指挥官看到失败信息后自主决定换一种方式重试还是跳过这个环节继续往下走。这种能力在动态图里天然支持失败信息作为状态传回指挥官节点指挥官再决策下一步而图本身只走对应分支其他节点不动。2. 先画图再写码指挥官系统的节点与状态设计2.1 节点清单入口、指挥官、专家层与工具层我开始动手设计图结构时第一件事不是写代码而是把整个系统的节点画出来。当初我们画了五类核心节点节点类型职责说明典型实现入口节点接收原始任务做初步意图判断意图分类、任务清洗指挥官节点核心决策者负责拆解任务、调度专家、判定完成大模型Router 结构化输出专家节点执行具体子任务各司其职检索Agent、总结Agent、大纲Agent工具节点提供实际能力如API调用、数据库查询各类函数封装人工确认节点高风险或歧义场景暂停等待人工介入interrupt机制专家节点的划分不是拍脑袋定的而是根据我们实际任务类型反推的。我建议你复刻时也别一上来就分十几个专家先从两三个开始跑通后再慢慢拆。节点太多状态设计没跟上图会变得很难维护。2.2 状态协议图上流动的到底是什么LangGraph里最容易被忽视、但最值得花时间设计的就是状态State的Schema。节点是图的骨架状态是图的血肉。所有节点共享同一个状态对象每个节点执行完返回的是一个dictLangGraph会自动把返回值合并进状态里。我们的AgentState最初长这样from typing import Annotated, TypedDict, Literal class AgentState(TypedDict): task: str # 原始任务 messages: list # 对话历史 route: str # 当前路由结果 specialist_output: str # 专家节点的最新输出 retry_count: int # 重试计数 completed: bool # 是否完成这里有一个关键认知状态不是数据库不要什么东西都往里塞。我们最开始把每一步的中间结果都存进状态结果状态对象越来越大每一轮循环都在重复传递大量无用数据Token消耗也跟着涨。后来改成了只保留本轮决策需要的数据比如专输出只要最新一份历史结果统一放外部存储状态立刻瘦身一半。2.3 边的三种类型固定边、条件边和循环边图设计里另一个核心是边。LangGraph的边分三种我按使用频率排个序条件边根据状态内容动态决定下一步去哪这是动态编排的灵魂。固定边执行完A必须去B适合流水线式的必经环节。循环边形式上也是边但因为指向了前面的节点就形成了回路用于反复执行直到满足条件。我画图时的思路很简单先画一个最粗的流程图把必经路径用固定边串起来把需要决策的地方单独拎出来做条件边至于循环只在指挥官需要反复调度子任务时才引入。不要为了炫技而加循环每一个回路都意味着潜在的无限执行风险。3. 基于LangGraph快速实现从StateGraph到条件边3.1 最小依赖与骨架代码LangGraph是LangChain生态里的编排框架装好之后直接开写。依赖很简单pip install langgraph langchain-openai我建议你把LLM调用单独封装一层便于换模型和做单元测试。下面这个骨架就是我们指挥官系统的第一个可运行版本节点被我刻意精简过只留最核心的部分from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import Literal # 复用前面定义的 AgentState def entry_node(state: AgentState) - dict: # 调用LLM做意图识别返回结构化路由结果 # 这里省略具体实现实际是一个强制JSON输出的prompt return {route: research} def commander_node(state: AgentState) - dict: # 指挥官查看专家输出判断是否完成或继续 # 完成时返回 {completed: True} # 未完成时返回 {specialist_output: ..., route: ...} return {completed: False, route: research} def research_agent(state: AgentState) - dict: # 执行检索返回结果 return {specialist_output: 查到的资料摘要} def human_check(state: AgentState) - dict: # 人工确认节点 return {messages: [需要人工确认]}3.2 注册节点与条件边有了节点函数接下来就是组装图。这里最关键的是add_conditional_edges它是动态路由的核心builder StateGraph(AgentState) # 注册节点 builder.add_node(entry, entry_node) builder.add_node(commander, commander_node) builder.add_node(research_agent, research_agent) builder.add_node(human_check, human_check) # 固定边入口先进指挥官 builder.add_edge(START, entry) builder.add_edge(entry, commander) # 条件边指挥官决定去专家还是收尾 def should_route(state: AgentState) - Literal[research, end]: if state[completed]: return end return research # 实际场景里这里会是多个专家分支 builder.add_conditional_edges( commander, should_route, { research: research_agent, end: END, } ) # 循环边专家执行完回到指挥官 builder.add_edge(research_agent, commander) # 编译并加checkpointer graph builder.compile(checkpointerMemorySaver())这段代码里有一个管理者模式的雏形指挥官是唯一的决策者专家只负责执行执行完回传结果循环直到完成。这种Supervisor模式正是动态图编排在Agent场景里最常见的形态。我们复刻时没有用LangGraph里更复杂的多Agent网络模式原因是需求还没到那个复杂度一个中心化指挥官比多Agent互相沟通更容易控制质量和成本。3.3 调用图并设置线程隔离调用时有一个重要参数thread_id。它用来隔离不同会话的状态避免并发任务互相污染。config {configurable: {thread_id: ticket-007}} result graph.invoke( {task: 帮我整理调研报告并生成PPT大纲, messages: [], route: , specialist_output: , retry_count: 0, completed: False}, config, {recursion_limit: 25} # 控制最大执行步数 )这里的recursion_limit很关键后面专门讲。线程隔离这行配置我强烈建议从第一天就养成习惯否则多个用户同时跑图的时候状态串了你查错能查到怀疑人生。3.4 加一个问一下人的中断指挥官系统里有些高风险动作比如给客户发正式邮件、删除数据必须有人工确认。LangGraph内置的interrupt机制可以直接在图上实现编译时指定interrupt_before图会在这个节点前自动暂停等人工确认后再恢复执行。graph builder.compile( checkpointerMemorySaver(), interrupt_before[human_check] ) # 第一次调用会停在human_check前 result graph.invoke(initial_state, config) # 人工确认后用相同config继续执行 result graph.invoke(None, config)这个机制的实际体验就是图流程可暂停、可恢复而不是中断后彻底丢掉上下文。我们接人工审核时用得很顺手比自己在外面硬塞一个异步队列要干净得多。4. 动态编排的核心难点循环、回溯与终止边界4.1 循环图与递归上限无界的循环是定时炸弹动态图和线性链最大的区别在于引入了循环而循环必须要有终止条件。LangGraph里有两个层面的保护第一状态层面的终止条件。在指挥官这类中心化架构里通常用一个completed字段或者计数器的形式表达任务完成或重试次数耗尽。指挥官节点每次看到不满足完成条件就往循环里再走一轮。第二框架层面的硬性上限。recursion_limit就是兜底保护防止因为LLM判断失误导致无限循环。我们设置的是25但实际调整次数远低于这个值。这里有个经验限制设得太小比如5复杂任务会被粗暴截断设得太大比如99一旦出bug资源消耗会非常吓人。建议从15到25起步跑几天看日志再调。另一个坑是循环里千万不要让状态无限增长。每一轮循环专家输出都会写回状态如果不做清理状态会越来越大。我们在指挥官循环里明确只保留最新一轮的specialist_output历史轮次的结果写日志不进状态。这个改动让总Token消耗下降了约两成。4.2 条件边返回值的规范结构化输出优于自由文本上面例子里的should_route是我硬编码的但真实场景里路由决策通常由LLM来做。这里有一个血泪教训不要让LLM自由地返回路由结果字符串一定要用结构化输出structured output限定取值范围。我们的第一个版本指挥官用prompt要求直接回答去哪个节点结果LLM时不时返回research_agent、research Agent、Research之类的变体条件边匹配不上直接报错。后来改成强制JSONSchema输出from pydantic import BaseModel, Field from typing import Literal class RouteDecision(BaseModel): route: Literal[research, writing, human_in_loop, finish] reason: str Field(description为什么选择这个路由)只要让LLM按这个Schema输出再解析成对象取route字段作为条件边的返回值就不用担心拼写和格式问题。结构化输出是图编排稳定性的地基。条件边函数本身则保持纯粹输入状态输出一个有限集合里的字符串不要在里面塞复杂的业务逻辑。4.3 什么时候该抽子图隔离与复用随着专家节点越来越多把整个图摊在一张平面图上会变得很难读。LangGraph支持子图Subgraph也就是把一个节点内部再嵌套一张图。我们是在加了第三个专家Agent之后才开始拆子图的主线只留入口、指挥官和必要的路由边每个专家内部跑自己的子图负责自己的内部步骤。这种方式的好处有两个一是隔离性每个子图有自己独立的状态Schema不会互相污染二是复用性同一个子图可以被主图在不同位置引用。但我不建议一上来就拆子图先让主图直接可见地跑起来再根据可读性需要逐步重构。5. 复刻过程中的实测数据与避坑记录5.1 踩坑记录状态膨胀、消息裁剪与Token消耗复刻过程里踩的坑按杀伤力排序大概是这几个一是消息列表无限累积。我们最初把messages设为整个对话历史全部保留每一轮循环都带进去给LLM重新看一遍Token消耗蹭蹭涨。后来做了裁剪策略只保留最近N条消息更早的对话摘要化后单独存。这里N的取值要看模型上下文窗口我们试下来6到8条比较合适。二是状态字段前后不一致。有一次我们把specialist_output改成了expert_output但某个节点忘记同步改结果条件边读到的是空值图直接卡在死循环里直到recursion_limit触发。这个坑最隐蔽也最折磨人。解法是给状态字段建一个集中的常量定义所有节点从同一个地方引用不要零零散散地写字符串。三是LLM输出超时。指挥官节点是LLM调用单次响应可能超过默认超时时间。我们统一把LLM客户端的timeout从默认值调大同时在图上加了每节点超时监控超过阈值就把该次结果标记为失败走重试分支。5.2 并发与线程隔离thread_id是命根子上面提到过thread_id这里展开讲。我们发现第一次压测时两个用户同时跑任务结果B用户的任务跑着跑着出现了A用户的中间数据。原因就是当时没用checkpointer所有任务共享同一个全局状态。加上MemorySaver并强制每个请求都带独立thread_id之后这个问题彻底消失。如果你部署的是多实例服务内存式的MemorySaver不够用建议换成外部存储的checkpointer实现比如数据库或对象存储。生产环境别图省事内存方案只适合开发和单机场景。5.3 追踪调试把每一次跳边都变成可回放的过程调试动态图编排最难的是复现。传统链式逻辑断点够用但图里好几个节点来回跳你得知道它每一步去了哪、为什么这么跳。我们调试时主要靠两条第一给每个节点打印结构化日志至少要包含节点名、进入时的关键状态子集、返回的dict。这个看似简单但实际排查时能帮你快速定位是哪一步丢的字段。第二用LangGraph生态的追踪能力。每次跑图都能看到完整的执行轨迹包括每一步的输入输出、调用了哪个条件边、跳到了哪个节点。排查问题时直接把轨迹截图甩给模型服务商比一段含糊其辞的报错高效得多。5.4 实测对比线性链基线 vs 图编排改造后改造完后我们用300条模拟工单做了A/B对比场景是根据用户求助内容完成信息检索、内容总结、生成回复草稿这类的多步任务。结果如下指标线性链基线图编排改造后多步任务完整完成率68%91%平均单任务时延42秒35秒平均Token消耗96007800需求变更后无需改代码的比例低高提升不是玄学原因也清楚完成率提升主要来自失败重试和路径自适应时延下降是因为图编排能跳过无关节点Token消耗下降则是状态瘦身和消息裁剪的功劳。这里要说明一下以上数据来自我们模拟的指标体系不同业务场景差距会很大。但是趋势是一致的——只要你的业务里存在任务路径不确定的情况图编排在质量指标上大概率优于线性链。6. 什么时候该上LangGraph迁移判断与最小路径6.1 先问自己三个问题如果你正在纠结要不要把现有线性链改成LangGraph图编排先别急着动手问自己三个问题任务路径是否依赖运行时动态决策比如意图分支、多轮转向是否存在子任务执行完要回到主控再决策的环形循环是否需要中途暂停等待人工确认三个问题里只要有任意两个回答是图编排就值得上。如果三个都是否那你的业务大概率是固定流水线线性链完全够用没必要为了技术热度引入复杂度。6.2 不适合图编排的两种典型情况有两种情况我建议不要硬上LangGraph一种是只有一两个简单接口调用的FAQ机器人这类任务线性链能搞定图编排纯属杀鸡用牛刀另一种是团队还不太熟悉异步和状态机概念的情况图编排的失败模式比线性链抽象没有基本功的人会debug到崩溃。我们的权衡是复杂度不要超过收益。指挥官系统本身就是因为有真实的多Agent调度需求图编排带来的收益清晰可见才值得投入。如果一个系统用线性链能稳定运行真的可以不动。6.3 从线性链迁移的最小改造步骤如果你决定迁移我的建议是走最小路径别一步到位推翻重写先把线性链的每一步抽象成独立的节点函数输入输出显式定义好这步不需要LangGraph纯重构。引入LangGraph用固定边按原顺序把节点串起来此时运行效果和原来一致。把最痛的决策点改造成条件边比如意图路由。先只改一处跑几天观察。再加入循环让关键子任务能回传指挥节点。这里建议加上recursion_limit和日志。最后根据线上表现决定要不要加interrupt人工确认、子图拆分、外部checkpointer。整个过程可以分两周做风险可控每一步都能回归验证。最后聊点个人体会。我实际跑下来最大的感受是LangGraph真正解决的不是让开发者少写几行代码而是给了你一个把Agent运行过程显式化的框架。线性链时代Agent的决策路径藏在代码逻辑里出了问题只能靠猜图编排时代每一步都能看到、能暂停、能回放这带来的确定感远远超过框架本身的功能列表。如果让我给一个最核心的落地建议那就是无论你最后选什么框架先花最多的时间把状态Schema设计清楚。图编排就像一座城市节点是地标边是道路但真正决定城市能不能运转的是每天在路上流动的人和货——对应到系统里就是状态里传的那些数据。这个搞定了后面一切都是水到渠成的事。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑