ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

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

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

如果要用一句话概括最近这轮架构折腾的核心,那就是:我把一条跑得挺稳的线性链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。节点是图的骨架,状态是图的血肉。所有节点共享同一个状态对象,每个节点执行完,返回的是一个dict,LangGraph会自动把返回值合并进状态里。

我们的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(checkpointer=MemorySaver())

这段代码里有一个管理者模式的雏形:指挥官是唯一的决策者,专家只负责执行,执行完回传结果,循环直到完成。这种"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( checkpointer=MemorySaver(), 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 从线性链迁移的最小改造步骤

如果你决定迁移,我的建议是走最小路径,别一步到位推翻重写:

  1. 先把线性链的每一步抽象成独立的"节点函数",输入输出显式定义好,这步不需要LangGraph,纯重构。
  2. 引入LangGraph,用固定边按原顺序把节点串起来,此时运行效果和原来一致。
  3. 把最痛的决策点改造成条件边,比如意图路由。先只改一处,跑几天观察。
  4. 再加入循环,让关键子任务能回传指挥节点。这里建议加上recursion_limit和日志。
  5. 最后根据线上表现,决定要不要加interrupt人工确认、子图拆分、外部checkpointer。

整个过程可以分两周做,风险可控,每一步都能回归验证。

最后聊点个人体会。我实际跑下来最大的感受是:LangGraph真正解决的,不是"让开发者少写几行代码",而是给了你一个把Agent运行过程显式化的框架。线性链时代,Agent的决策路径藏在代码逻辑里,出了问题只能靠猜;图编排时代,每一步都能看到、能暂停、能回放,这带来的确定感远远超过框架本身的功能列表。

如果让我给一个最核心的落地建议,那就是:无论你最后选什么框架,先花最多的时间把状态Schema设计清楚。图编排就像一座城市,节点是地标,边是道路,但真正决定城市能不能运转的,是每天在路上流动的人和货——对应到系统里,就是状态里传的那些数据。这个搞定了,后面一切都是水到渠成的事。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进