多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

多Agent系统成本与干扰治理:从LangChain迁移到LangGraph的实战指南

多Agent系统成本与干扰治理:从LangChain迁移到LangGraph的实战指南 多 Agent 系统这两年从论文里的概念一路卷到生产环境我身边不少团队都在做同一件事把原来那个一个 Agent 包打天下的 LangChain 脚本拆成几个各司其职的 Agent 协作。拆完之后一部分人跑通了效果确实比单 Agent 好另一部分人拆完发现成本翻了三倍延迟涨了一倍Agent 之间还互相打架最后又灰溜溜地合回去了。我自己完整走过一遍这条路从 LangChain 的AgentExecutor迁到 LangGraph 的StateGraph中间踩的坑足够写一篇长文。这篇就把账算清楚多 Agent 到底贵在哪、乱在哪LangGraph 用什么机制把这两件事管住以及迁移过程中那些文档里不会写的细节。1. 先搞清楚多 Agent 到底在解决什么问题1.1 单 Agent 的能力天花板不是模型不够强很多人以为单 Agent 做不好任务是因为模型能力不够换个更强的模型就行了。实际跑下来你会发现瓶颈往往不在模型本身而在上下文被污染和工具选择空间过大这两件事上。一个典型的单 Agent 场景你给它挂了 15 个工具包括查数据库、调 API、读文件、发邮件、算指标。用户问一个简单问题模型要在 15 个工具里选。工具描述加起来可能就占了 2000 个 token再加上系统提示词、历史对话、中间结果上下文很快就膨胀到几万 token。上下文越长模型注意力越分散选错工具、重复调用、参数填错的概率就越高。我实测过一个客服场景单 Agent 挂 12 个工具时工具选择准确率大概在 78% 左右把工具按领域拆成三个子 Agent每个只挂 4 个工具整体准确率能到 91%。模型没换提示词没大改纯粹是缩小了每次决策的搜索空间。1.2 多 Agent 的本质是职责隔离不是数量堆砌这里有个特别容易走偏的点多 Agent 不是Agent 越多越好。我见过有人把一个任务拆成七个 Agent结果每个 Agent 都在等上一个的输出串行链路拉得极长一次请求跑了 40 秒。正确的拆法应该按职责边界来而不是按步骤数量来。判断标准很简单如果两个环节用的是完全不同的工具集、完全不同的提示词风格、或者需要完全不同的上下文那它们就该拆开如果只是流程上的先后顺序那用一个 Agent 多轮调用就够了。举个具体的例子。一个分析销售数据并生成报告的任务合理的拆法是数据查询 Agent只负责把自然语言转成 SQL 并执行工具就是数据库连接分析 Agent拿到数据后做统计和洞察工具是计算函数撰写 Agent把分析结果组织成报告工具是模板渲染这三个 Agent 的工具集完全不重叠提示词风格也完全不同一个偏结构化、一个偏推理、一个偏表达拆开是合理的。但如果你把撰写再拆成写标题和写正文两个 Agent那就是过度拆分纯属给自己找麻烦。1.3 拆开之后两笔账立刻浮出水面一旦你决定拆两笔账马上要算第一笔是 token 账。每个 Agent 都有自己的系统提示词都要带上必要的上下文。三个 Agent 意味着三份系统提示词、三次上下文组装。如果 Agent 之间还要传递中间结果那这些结果会在多个 Agent 的上下文里重复出现。我统计过一个三 Agent 的流程同样的任务多 Agent 版本的 token 消耗是单 Agent 的 2.3 到 3.1 倍取决于中间结果的大小。第二笔是干扰账。Agent A 的输出要传给 Agent B但 A 的输出里往往包含 B 不需要的信息。比如数据查询 Agent 返回了完整的 SQL 和原始数据分析 Agent 其实只需要聚合后的数字但原始数据全塞进它的上下文了。这些冗余信息就是干扰会让 B 的判断变差。LangGraph 的价值恰恰就是针对这两笔账设计的。它用StateGraph把状态流转显式化让你能精确控制每个 Agent 看到什么、传什么而不是像 LangChain 的链式调用那样中间结果一股脑往下传。2. LangChain 的链式结构为什么管不住多 Agent2.1 AgentExecutor 的黑盒特性LangChain 的AgentExecutor设计初衷是给一个 Agent 一个任务让它自己循环直到完成。它的内部是一个 while 循环模型思考、选工具、执行工具、把结果塞回上下文、再思考直到模型说我完成了。这个结构对单 Agent 很友好但对多 Agent 就很别扭。因为AgentExecutor把思考-行动-观察这个循环封装成了一个黑盒你很难在循环中间插入另一个 Agent也很难精确控制哪部分历史传给下一个 Agent。我最早的做法是用AgentExecutor套AgentExecutor外层 Agent 把内层 Agent 当成一个工具来调。这个方案能跑但问题很明显内层 Agent 的完整对话历史包括它调了哪些工具、看到了什么中间结果会作为工具返回值整个塞进外层 Agent 的上下文。外层 Agent 的 token 消耗直接爆炸。2.2 中间结果无处安放的尴尬LangChain 里传递数据主要靠两种方式一是通过output_keys把结果传给下一步二是通过Memory组件。这两种方式在多 Agent 场景下都不够用。output_keys的问题是它只能传最终输出传不了中间状态。比如你想让分析 Agent 知道数据查询 Agent 执行了哪条 SQL这个信息在AgentExecutor的输出里可能被压缩掉了你拿不到。Memory的问题是它是全局的所有 Agent 共享同一份记忆。这就导致 Agent A 的中间思考过程会污染 Agent B 的上下文。我遇到过最离谱的情况撰写 Agent 在生成报告时把数据查询 Agent 的调试信息比如正在连接数据库...也写进了报告里因为那些信息都在共享 Memory 里。2.3 一个真实的翻车案例说个具体的。我之前做一个合同审查的多 Agent 系统三个 Agent条款提取、风险识别、建议生成。用 LangChain 的链式结构串起来跑测试集的时候发现风险识别 Agent 经常幻觉出合同里没有的条款。排查了半天最后定位到问题条款提取 Agent 的输出里除了提取出的条款还包含了它的思考过程比如这段文字看起来像是免责条款但表述模糊我倾向于归类为责任限制条款。风险识别 Agent 看到这段思考过程就把它当成了合同的实际内容然后基于这个实际内容去识别风险。这就是典型的干扰问题。条款提取 Agent 的不确定判断被下游 Agent 当成了事实。在 LangChain 的链式结构里你很难干净地把思考过程和最终结论分开传递因为它们混在同一个输出字符串里。3. LangGraph 的 StateGraph 是怎么把账算明白的3.1 状态是一等公民而不是副产品LangGraph 最核心的改变是把**状态State**提升成了一等公民。在 LangChain 里状态是隐式的藏在各个组件的输入输出里在 LangGraph 里你要显式定义一个 State 结构所有节点都读写这个 State。这个改变看起来只是 API 层面的实际影响很大。因为一旦状态显式化了你就能精确控制每个节点能读到 State 的哪些字段每个节点往 State 里写什么字段之间怎么合并比如多个节点都往同一个列表里追加内容我用 TypedDict 定义 State 的典型写法from typing import TypedDict, Annotated from operator import add class ContractState(TypedDict): raw_text: str clauses: Annotated[list, add] risks: Annotated[list, add] suggestions: Annotated[list, add] current_step: str这里Annotated[list, add]是关键。它告诉 LangGraph当多个节点都往clauses里写东西时用add函数合并也就是列表拼接而不是覆盖。这个机制叫reducer是多 Agent 协作的基础。3.2 节点只做一件事边界清晰LangGraph 里每个节点就是一个函数输入是 State输出是要更新的字段。节点不需要知道全局流程只需要关心我读什么、我写什么。还是合同审查的例子条款提取节点大概长这样def extract_clauses(state: ContractState) - dict: raw state[raw_text] # 只把 raw_text 喂给模型不传其他字段 result llm.invoke( f从以下合同文本中提取所有条款只返回条款内容不要返回你的分析过程\n{raw} ) clauses parse_clauses(result.content) return {clauses: clauses, current_step: extracted}注意这里的关键设计节点的返回值只包含它要更新的字段。它不返回整个 State只返回clauses和current_step。这样风险识别节点读 State 时拿到的clauses是干净的条款列表不包含提取节点的任何思考过程。这就是 LangGraph 解决干扰问题的核心机制通过 State 的字段隔离让每个 Agent 只看到它该看的东西。3.3 条件边让流程可控而不是让模型自由发挥LangChain 的AgentExecutor里下一步做什么是模型决定的。LangGraph 里你可以用**条件边conditional edge**把控制权拿回来。比如风险识别完之后如果发现高风险条款要走人工复核分支如果都是低风险直接进建议生成。这个逻辑用条件边表达def route_after_risk(state: ContractState) - str: high_risks [r for r in state[risks] if r[level] high] if high_risks: return human_review return generate_suggestions graph.add_conditional_edges( identify_risks, route_after_risk, {human_review: human_review, generate_suggestions: generate_suggestions} )这个设计的好处是流程的确定性部分用代码控制不确定性部分才交给模型。模型只负责它擅长的理解文本、判断风险流程编排这种确定性的事交给图结构。这比让模型自己决定下一步该干嘛要可靠得多也省 token——不用在提示词里反复描述流程规则。4. token 账本多 Agent 的成本到底花在哪4.1 拆解一次多 Agent 调用的 token 构成要管住成本先得知道钱花在哪。一次三 Agent 的调用token 消耗大致分四块消耗项说明占比实测系统提示词每个 Agent 的角色定义、工具说明、输出格式要求25%-35%上下文传递上游 Agent 的输出传给下游30%-45%工具调用开销工具描述、参数、返回结果15%-25%模型推理模型实际的思考输出10%-20%最容易被忽视的是上下文传递这一块。很多人优化 token 只盯着提示词把系统提示词压缩了又压缩结果上下文传递这块漏得像个筛子。4.2 用 State 字段裁剪上下文LangGraph 的 State 机制给了你裁剪上下文的天然抓手。核心原则是下游节点只读它真正需要的字段。我做过一个对比实验。同一个合同审查任务两种 State 设计设计 A粗放型State 里放一个history列表每个节点把自己的完整输出追加进去下游节点读整个history。设计 B精细型State 里按语义分字段clauses、risks、suggestions各管各的节点只读自己需要的字段。结果设计 B 的 token 消耗比设计 A 低了 42%。原因很简单设计 A 里建议生成节点读到了条款提取节点的原始输出包含大量格式标记和冗余描述而设计 B 里它只读到结构化的clauses列表。4.3 中间结果的压缩策略有些中间结果确实需要传递但可以压缩。几个我常用的策略策略一结构化替代自然语言。让上游 Agent 输出 JSON 而不是自然语言段落。JSON 的信息密度高同样的信息量 token 能省一半以上。比如条款提取不要输出我找到了以下条款第一甲方应当在...第二乙方应当...而是输出[{id: 1, content: 甲方应当...}, ...]。策略二摘要替代全文。如果下游只需要知道上游做了什么不需要知道具体内容就让上游输出一句话摘要。比如数据查询 Agent 不需要把完整结果集传给分析 Agent只需要传查询返回了 1523 行包含 date、amount、region 三列。策略三引用替代内联。大块数据不要直接塞进 State而是存到外部比如文件或数据库State 里只放引用 ID。下游需要时再按需读取。这个策略在处理长文档时特别有用。提示压缩中间结果时要注意保留决策依据。如果下游 Agent 需要基于上游的判断做决策那上游的判断理由不能省否则下游会重新推理一遍反而更费 token。4.4 一个可复用的 token 预算表我在项目里会维护一张 token 预算表给每个节点设定上限超了就报警。大致长这样节点输入预算输出预算说明条款提取30001500输入含合同全文输出结构化条款风险识别20001000输入只含条款不含原文建议生成25002000输入含条款风险输出建议有了这张表每次改动提示词或 State 结构都能快速判断有没有超预算。我见过太多团队改着改着 token 就翻倍了就是因为没有预算意识。5. 干扰问题Agent 之间怎么才能不互相添乱5.1 干扰的三种典型形态多 Agent 系统里的干扰不是玄学它有具体的表现形式。我总结了三类形态一上下文污染。上游 Agent 的思考过程、调试信息、格式标记混进了下游的输入。前面合同审查的例子就是这种。形态二指令冲突。不同 Agent 的系统提示词里有矛盾的指令。比如 Agent A 被要求尽可能详细地输出Agent B 被要求只关注关键信息A 的详细输出到了 B 那里就成了噪音。形态三状态竞争。多个 Agent 同时往同一个 State 字段写数据顺序不确定导致结果不可复现。这个在并行节点场景下特别常见。5.2 用输出契约把接口钉死解决上下文污染最有效的办法是给每个 Agent 定义输出契约明确它必须输出什么格式、什么字段多余的一律不要。我在实际项目里会用 Pydantic 模型来定义契约from pydantic import BaseModel, Field class ClauseOutput(BaseModel): clauses: list[str] Field(description提取出的条款列表每条为纯文本) confidence: float Field(description提取置信度0到1之间) # 注意不包含 reasoning 字段思考过程不对外暴露然后在节点里强制模型按这个 schema 输出。LangChain 的with_structured_output可以做到这点structured_llm llm.with_structured_output(ClauseOutput) result structured_llm.invoke(prompt) return {clauses: result.clauses}这样下游拿到的永远是干净的clauses列表模型的思考过程被挡在契约之外。这个做法看起来多了一层约束实际省下的调试时间远超这点开销。5.3 提示词的一致性检查指令冲突这个问题靠人工 review 提示词很难发现因为提示词分散在各个节点里。我的做法是建一个提示词清单把所有 Agent 的系统提示词集中管理定期做一致性检查。检查的重点是这几类词输出长度相关的详细、简洁、完整、精简语气相关的专业、友好、客观格式相关的JSON、markdown、纯文本如果 Agent A 要求详细输出而 Agent B 要求简洁输入那中间就需要一个转换层不能直接对接。5.4 并行节点的状态合并陷阱LangGraph 支持并行执行节点这在多 Agent 场景下很有用比如同时做风险识别和合规检查。但并行会带来状态合并的问题。假设两个并行节点都往risks字段写数据。如果你用的是默认的覆盖语义后写的会覆盖先写的数据就丢了。必须用 reducerclass State(TypedDict): risks: Annotated[list, add] # 用 add 合并不覆盖但用add也有坑如果两个节点写了重复的数据合并后会有重复项。我一般会在合并后加一个去重节点或者让每个节点写入时带上来源标记方便后续去重。注意并行节点的执行顺序是不确定的所以任何依赖顺序的逻辑都不能放在并行分支里。我踩过一次坑两个并行节点都依赖同一个上游字段但其中一个节点会修改这个字段结果另一个节点读到的值时有时无。后来把修改操作单独抽成一个串行节点才解决。6. 从 LangChain 迁到 LangGraph 的实操路径6.1 迁移不是重写是逐步替换很多人一听迁移就头大觉得要把整个系统推倒重来。其实不用。我的做法是保留 LangChain 的组件只替换编排层。具体来说LangChain 里的这些东西可以原样保留LLM 客户端ChatOpenAI之类工具定义tool装饰的函数提示词模板ChatPromptTemplate输出解析器需要替换的只有编排部分把AgentExecutor换成StateGraph把链式调用换成节点边。这样迁移的风险可控因为大部分代码没动出问题容易定位。6.2 一个最小可运行的迁移示例先看迁移前的 LangChain 版本from langchain.agents import AgentExecutor, create_openai_tools_agent agent create_openai_tools_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools) result executor.invoke({input: user_query})迁移到 LangGraph 后from langgraph.graph import StateGraph, END class State(TypedDict): input: str output: str def agent_node(state: State) - dict: result executor.invoke({input: state[input]}) return {output: result[output]} graph StateGraph(State) graph.add_node(agent, agent_node) graph.set_entry_point(agent) graph.add_edge(agent, END) app graph.compile() result app.invoke({input: user_query})这个最小版本看起来比原来还麻烦但它是个起点。一旦你有了这个骨架往里加节点、加条件边、加状态字段就都是增量操作了。6.3 迁移过程中的三个高频坑坑一State 字段的可变性。LangGraph 的 State 默认是不可变的节点返回的字典会被合并到 State 里生成新状态。如果你在节点里直接修改了 State 里的可变对象比如往列表里 append可能会引发难以追踪的 bug。我的习惯是节点里永远返回新对象不原地修改。坑二递归深度限制。LangGraph 默认有递归深度限制防止无限循环默认值是 25。如果你的图有循环结构比如 Agent 反复调用工具直到满足条件可能会撞到这个限制。可以通过graph.compile()时传参调整但更好的做法是检查你的循环退出条件是不是写得太宽松了。坑三流式输出的处理。LangChain 的AgentExecutor有现成的astream_eventsLangGraph 的流式输出机制不太一样。如果你需要把中间过程实时推给前端得用app.astream()并处理每个节点的事件。这块我踩了不少坑建议单独花时间研究。6.4 迁移后的验证清单迁完之后别急着上线按这个清单过一遍单节点功能是否和原来一致逐个节点对比输出State 字段是否都有明确的 reducer 定义条件边的分支是否都覆盖到了包括异常分支token 消耗是否在预算内对比迁移前后的账单并行节点的状态合并是否正确构造并发场景测试循环结构是否有明确的退出条件避免无限循环我一般会写一组回归测试用固定的输入跑迁移前后的系统对比输出和 token 消耗。差异超过阈值就人工 review。7. 多 Agent 编排的几个反直觉经验7.1 不是所有任务都值得拆多 Agent这是我最想强调的一点。多 Agent 有明确的成本只有当收益大于成本时才值得拆。判断标准我总结成三条工具集是否真的不重叠如果两个 Agent 用的工具高度重合拆开没意义上下文是否真的需要隔离如果下游本来就需要上游的全部信息隔离反而增加传递成本任务是否有明确的阶段划分如果任务本身是模糊的、需要反复横跳的硬拆成阶段反而降低效果我见过太多为了架构好看而拆多 Agent 的项目最后都因为成本和复杂度回退了。7.2 状态设计比节点设计更重要新手做 LangGraph 容易把精力放在怎么设计节点上其实State 的设计才是决定成败的关键。节点是函数改起来容易State 是契约改起来牵一发动全身。我的经验是先把 State 的字段设计清楚想明白每个字段谁写谁读、怎么合并、生命周期多长再动手写节点。State 设计对了节点就是水到渠成的事。7.3 给每个 Agent 留一个我不确定的出口多 Agent 系统里最危险的情况是某个 Agent 在信息不足时强行给出结论然后这个错误结论被下游 Agent 当成事实继续加工。错误会沿着链路放大。解决办法是给每个 Agent 设计一个不确定的输出通道。比如风险识别 Agent 如果拿不准就输出{level: uncertain, reason: ...}而不是硬猜一个high或low。然后在条件边里处理这个分支要么转人工要么触发补充信息查询。这个设计看起来增加了复杂度但它把错误传播变成了错误拦截在多 Agent 系统里价值极高。7.4 监控要盯住Agent 之间的接口单 Agent 系统的监控主要看输入输出和延迟。多 Agent 系统还要额外盯住 Agent 之间的接口每个节点的输入是否符合预期、输出是否符合契约、有没有异常大的中间结果。我一般会在每个节点的入口和出口打点记录 State 的关键字段大小。一旦某个字段异常膨胀比如某个列表突然从 10 项变成 1000 项就能立刻发现。8. 关于 token 和干扰最后再聊几句实在的多 Agent 系统的成本优化本质上是在信息传递的完整性和经济性之间找平衡。传得太少下游 Agent 信息不足会瞎猜传得太多token 爆炸还引入干扰。这个平衡点没有通用答案只能靠实测调。我的做法是先按最小必要原则设计 State跑起来看效果效果不够再逐步加字段每加一个字段就测一次 token 和准确率。这样能找到一个相对优的点而不是一上来就传一大堆。干扰问题的根治办法说到底就是契约化。每个 Agent 的输入输出都用 schema 钉死思考过程不对外暴露中间结果结构化。做到这几点干扰问题能解决八成以上。剩下的两成靠监控和回归测试兜底。LangGraph 这套东西刚上手会觉得比 LangChain 啰嗦要定义 State、要写节点函数、要连边。但等你真正管过一个多 Agent 系统的成本和稳定性之后会发现这些啰嗦恰恰是把账算明白的前提。LangChain 让你快速跑起来LangGraph 让你跑得久、跑得省。这两个阶段的需求不一样工具自然也不一样。
返回列表