1. 多 Agent 编排到底在解决什么问题1.1 从单 Agent 到多 Agent 的必然演进先说结论单 Agent 能做的事天花板比大多数人想象的低得多。我去年帮一个团队做智能客服的 POC一开始就是一个 Agent 挂几个工具——查订单、查物流、发邮件。demo 阶段跑得挺漂亮一上真实流量就崩了。问题不是模型不行而是一个 Agent 同时要理解意图、规划步骤、调用工具、校验结果、生成回复上下文里塞了七八种职责稍微复杂一点的请求就开始胡言乱语工具调用参数错得离谱。这就是单 Agent 的根本瓶颈职责耦合导致上下文污染。你可以把它想象成一个公司只雇了一个人既当前台又当销售又当财务还当技术客服短期能撑业务一复杂必然出错。多 Agent 编排的核心思路就是分而治之把一个大任务拆成若干子任务每个子任务交给一个专职 Agent再由一个编排层Orchestrator负责调度、传递状态、汇总结果。这跟微服务架构的思路是一模一样的——单体应用拆成服务每个服务只干一件事通过明确的接口通信。1.2 编排层到底编排什么很多人对编排这个词有误解以为就是按顺序调用几个 Agent。实际上编排层要处理的事情远比想象中复杂我把它拆成四个核心职责任务分解把用户的自然语言请求拆成可执行的子任务图。比如帮我分析这份财报并生成 PPT要拆成读取文件→提取关键指标→计算同比环比→生成图表→组织文案→输出 PPT。路由决策决定每个子任务交给哪个 Agent。是走检索 Agent 还是走计算 Agent是走代码执行还是走知识库查询。状态管理在 Agent 之间传递上下文。这里有个大坑——不能把上一个 Agent 的全部输出无脑塞给下一个否则上下文会爆炸必须做摘要和结构化提取。结果聚合与校验多个 Agent 的输出可能冲突编排层要负责仲裁、重试、兜底。提示编排层本身也是一个Agent只不过它的工具是其他 Agent。这个视角转换很关键想通了之后整个架构就顺了。1.3 什么场景真的需要多 Agent不是所有项目都值得上多 Agent。我见过太多团队为了技术先进硬上多 Agent结果复杂度翻了三倍效果还不如一个调好的单 Agent。判断标准很简单满足下面任意两条才考虑多 Agent判断维度单 Agent 够用建议上多 Agent任务步骤数3 步以内5 步以上且步骤间有依赖工具数量5 个以内10 个以上且分属不同领域上下文长度稳定在 8K 以内经常超过 32K职责类型单一如只做问答混合检索计算生成校验错误容忍度低错了重来高需要交叉验证举个我实际做过的例子一个合同审查系统需要提取条款、比对标准模板、识别风险点、生成修改建议、输出审查报告。这五个环节每个都需要不同的 prompt 策略和工具集硬塞进一个 Agent 里prompt 写到 3000 字还是顾此失彼。拆成五个专职 Agent 之后每个 prompt 控制在 500 字以内准确率直接从 62% 提到 89%。2. 主流编排框架选型与 LangGraph 核心概念2.1 框架选型的几个真实考量现在市面上做多 Agent 编排的框架不少我实际用过并且踩过坑的主要有这么几类第一类是 LangGraph 这种图结构编排。核心抽象是节点边状态你把每个 Agent 当成图里的一个节点用边定义流转逻辑状态在整个图里共享。它的优势是控制流极其明确支持条件分支、循环、并行而且有 checkpoint 机制可以做断点续跑。缺点是学习曲线陡得先理解状态机和图的概念。第二类是 CrewAI、AutoGen 这种角色扮演式编排。你定义几个角色Researcher、Writer、Reviewer框架自动帮你协调它们对话。上手快但控制粒度粗一旦流程复杂起来Agent 之间的对话会失控token 消耗也吓人。第三类是自己撸。用 Python 原生写一个调度循环Agent 就是函数状态就是一个 dict。灵活度最高但所有轮子都得自己造重试、超时、并发、日志全要手写。我的建议是流程明确、需要精细控制的选 LangGraph快速验证想法选 CrewAI生产环境且团队有工程能力LangGraph 是当前最稳的选择。下面重点讲 LangGraph因为它的抽象最接近生产需求。2.2 LangGraph 的三个核心概念LangGraph 看着复杂其实就三个概念理解了就能上手State状态一个贯穿整个图的共享数据结构通常是个 TypedDict。每个节点读取它、修改它。关键是要定义好 reducer——比如多个节点都想往一个列表里 append 消息你得告诉 LangGraph 用operator.add来合并否则后写的会覆盖先写的。Node节点就是一个函数输入是 State输出是要更新的 State 字段。一个节点可以是一个 Agent、一次工具调用、一段纯 Python 逻辑。Edge边定义节点之间的流转。普通边是固定的 A→B条件边是add_conditional_edges根据当前 State 决定下一步走哪。条件边是多 Agent 编排的灵魂路由决策全靠它。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[list, operator.add] next_agent: str task_result: str def researcher_node(state: AgentState): # 检索 Agent 的逻辑 return {messages: [(assistant, 检索完成)], task_result: ...} def router_node(state: AgentState): # 根据任务结果决定下一步 if 需要计算 in state[task_result]: return {next_agent: calculator} return {next_agent: writer} graph StateGraph(AgentState) graph.add_node(researcher, researcher_node) graph.add_node(router, router_node) graph.add_edge(researcher, router) graph.add_conditional_edges( router, lambda s: s[next_agent], {calculator: calc_node, writer: writer_node} )这段代码就是最小可用的多 Agent 编排骨架。注意Annotated[list, operator.add]这个写法这是 LangGraph 里最容易踩的坑不加 reducer 的话多个节点写 messages 会互相覆盖。2.3 状态设计决定架构成败我踩过最大的坑就是状态设计得太随意。一开始我把整个对话历史都塞进 State结果跑几轮之后上下文爆了token 费用飙升模型还开始失忆——因为关键信息被淹没在噪音里。后来我总结出一套状态分层设计全局状态只放任务 ID、用户 ID、最终结果这类必须贯穿全程的字段。阶段状态每个 Agent 有自己的局部状态用完就清理不往全局塞。消息通道Agent 之间的通信走结构化的消息对象而不是裸字符串。比如{from: researcher, type: finding, content: ..., confidence: 0.9}。这样设计之后上下文长度稳定控制在 4K 以内成本降了 60%准确率反而升了。3. 从零搭建一个多 Agent 编排系统3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.10 以上LangGraph 对类型注解依赖比较重低版本会有各种奇怪的报错。# 创建虚拟环境强烈建议别在全局环境里装 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装核心依赖 pip install langgraph langchain langchain-openai pip install python-dotenv # 管理 API key如果你还没装 Python去官网下载安装包安装时务必勾选 Add Python to PATH否则后面命令行调不起来。装完用python --version验证一下。numpy 这类科学计算库如果后面要用直接pip install numpy就行别去折腾源码编译。注意LangGraph 的版本迭代很快API 偶尔会有 breaking change。生产项目里一定要把版本号锁死比如langgraph0.2.x别用latest。3.2 定义 Agent 的标准化接口多 Agent 系统最容易乱的地方就是每个 Agent 的输入输出格式不统一。我见过一个项目检索 Agent 返回字符串计算 Agent 返回 dict生成 Agent 又返回 list编排层光做格式转换就写了一堆胶水代码。我的做法是定义一个基类所有 Agent 都继承它from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): task: str context: dict {} history: list [] class AgentOutput(BaseModel): result: str confidence: float metadata: dict {} needs_next: str | None None class BaseAgent(ABC): name: str abstractmethod def run(self, inp: AgentInput) - AgentOutput: pass这样编排层只需要处理AgentInput和AgentOutput两种类型所有 Agent 对编排层来说都是黑盒替换、新增、删除都不影响其他部分。这就是面向接口编程在多 Agent 场景下的价值。3.3 编排图的完整实现下面是一个我实际项目里用过的编排图任务是分析一份文档并生成结构化报告。涉及四个 AgentReader读取解析、Analyzer分析提取、Calculator数值计算、Writer生成报告。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class ReportState(TypedDict): doc_path: str raw_content: str analysis: dict calc_result: dict final_report: str error: str | None retry_count: int def reader_node(state: ReportState): try: content read_document(state[doc_path]) return {raw_content: content, error: None} except Exception as e: return {error: str(e), retry_count: state.get(retry_count, 0) 1} def analyzer_node(state: ReportState): analysis analyze(state[raw_content]) return {analysis: analysis} def calculator_node(state: ReportState): result calculate(state[analysis]) return {calc_result: result} def writer_node(state: ReportState): report generate_report(state[analysis], state[calc_result]) return {final_report: report} def should_retry(state: ReportState): if state.get(error) and state.get(retry_count, 0) 3: return retry if state.get(error): return fail return continue builder StateGraph(ReportState) builder.add_node(reader, reader_node) builder.add_node(analyzer, analyzer_node) builder.add_node(calculator, calculator_node) builder.add_node(writer, writer_node) builder.set_entry_point(reader) builder.add_conditional_edges( reader, should_retry, {retry: reader, continue: analyzer, fail: END} ) builder.add_edge(analyzer, calculator) builder.add_edge(calculator, writer) builder.add_edge(writer, END) memory MemorySaver() app builder.compile(checkpointermemory)跑起来就是config {configurable: {thread_id: task-001}} result app.invoke({doc_path: ./report.pdf, retry_count: 0}, config) print(result[final_report])这里有几个设计决策值得说清楚第一为什么 reader 要单独成节点而不是塞进 analyzer因为文档读取是最容易失败的一步文件不存在、格式不支持、编码错误单独成节点才能针对性地做重试。这就是失败隔离原则。第二为什么用条件边做重试而不是 try-except 循环因为 LangGraph 的 checkpoint 机制会在每个节点执行后保存状态用条件边重试的话即使进程崩了重启后能从上次的 checkpoint 继续不用从头跑。这在长流程任务里能省大量时间和 token。第三MemorySaver 只是开发用的。生产环境要换成SqliteSaver或PostgresSaver否则进程一重启状态就没了。3.4 并行编排与结果聚合有些子任务之间没有依赖可以并行跑。比如分析文档时提取关键指标和识别风险点互不干扰串行跑就是浪费时间。LangGraph 支持并行节点但并行分支的写入必须用 reducer 合并否则会报错。我一般这样处理class ParallelState(TypedDict): content: str metrics: Annotated[list, operator.add] risks: Annotated[list, operator.add] summary: str def extract_metrics(state): return {metrics: [extract_metrics_logic(state[content])]} def detect_risks(state): return {risks: [detect_risks_logic(state[content])]} def merge_node(state): summary summarize(state[metrics], state[risks]) return {summary: summary} builder.add_node(extract_metrics, extract_metrics) builder.add_node(detect_risks, detect_risks) builder.add_node(merge, merge_node) # 从同一个节点分叉出去就是并行 builder.add_edge(start, extract_metrics) builder.add_edge(start, detect_risks) builder.add_edge(extract_metrics, merge) builder.add_edge(detect_risks, merge)并行跑下来整体耗时从 12 秒降到 7 秒效果立竿见影。但要注意并行节点之间不能有共享的可变状态否则会有竞态问题。4. 生产环境的关键问题与排查实录4.1 上下文爆炸与 token 成本控制多 Agent 系统最大的成本黑洞就是 token。我统计过一个项目单次任务平均消耗 45K token其中 70% 是 Agent 之间传递的冗余上下文。控制手段我总结了三条按效果排序第一条Agent 间通信必须结构化。别传自然语言传 JSON。自然语言里全是好的我已经完成了检索找到了以下内容……这种废话结构化消息能砍掉一半 token。第二条做上下文摘要。上一个 Agent 的输出如果超过 500 字先让一个小模型比如 gpt-4o-mini压缩成 200 字以内的要点再传给下一个。这一步能再省 30%。第三条设置硬性截断。每个 Agent 的输入上下文设一个上限超了就按重要性排序截断。别指望模型自己忽略无关信息它做不到。控制手段实施难度token 节省副作用结构化通信中40-50%需要定义 schema上下文摘要低25-35%可能丢细节硬性截断低15-25%可能截掉关键信息缓存重复调用中10-20%需要设计缓存键4.2 Agent 死循环与超时处理多 Agent 系统最恶心的 bug 就是死循环。A 让 B 做事B 觉得信息不够让 A 补充A 补充完 B 还是觉得不够……两个 Agent 能来回扯皮几十轮token 烧光任务还没完成。我的处理方案是三重保险全局步数上限整个图最多执行 N 步我一般设 20超了直接终止并返回当前最优结果。单 Agent 调用次数上限同一个 Agent 在一次任务里最多被调用 3 次超了就走兜底逻辑。超时熔断单次任务总耗时超过 60 秒强制结束。def check_limits(state): if state.get(step_count, 0) 20: return force_end if state.get(elapsed, 0) 60: return force_end return continue提示兜底逻辑一定要设计好。宁可返回一个信息不完整但可用的结果也不要让用户干等或者拿到一个报错。4.3 常见问题速查表下面这张表是我和团队踩坑踩出来的基本覆盖了 90% 的常见问题现象可能原因排查方向解决方案状态字段被覆盖没定义 reducer检查 TypedDict 注解加Annotated[type, reducer]图跑不起来入口点没设检查 set_entry_point显式设置入口条件边不生效路由函数返回值不匹配打印路由函数输出确保返回值在映射表里Agent 反复调用路由逻辑有环画图看流转路径加步数上限token 超限上下文没清理统计每步 token加摘要和截断结果不稳定温度参数太高检查 LLM 配置降到 0.1 以下checkpoint 失效用了 MemorySaver检查 saver 类型换持久化 saver4.4 一个真实的排查案例有次线上任务成功率突然从 92% 掉到 67%日志里全是analyzer 节点超时。我一开始以为是模型服务不稳定查了半天 API 延迟正常。后来把每个节点的输入输出打出来对比发现reader 节点返回的内容长度从平均 3K 涨到了 15K——因为上游数据源改了格式把整个 PDF 的原始文本都塞进来了。analyzer 处理 15K 的输入自然超时。解决方案是在 reader 和 analyzer 之间加了一个预处理节点做文本清洗和分段把输入压回 3K 以内。改完之后成功率恢复到 94%。这个案例的教训是多 Agent 系统里节点之间的数据契约比节点本身更重要。任何一个上游节点的输出格式变化都可能引发下游雪崩。所以我在每个节点入口都加了输入校验长度、格式、必填字段全检查一遍不合法直接走错误分支别让它污染下游。5. 多 Agent 编排的进阶实践5.1 Agent 记忆的分层设计多 Agent 系统里记忆是个绕不开的话题。我的经验是分三层短期记忆当前任务内的上下文存在 State 里任务结束就清。中期记忆跨任务的会话历史存数据库按用户 ID 索引定期摘要压缩。长期记忆沉淀下来的知识和经验存向量库用检索的方式按需调用。关键点是别把三层混在一起。我见过把全部历史都塞进 State 的做法跑十几个任务之后上下文直接爆炸。正确的做法是State 里只放当前任务相关的需要历史的时候主动去查中期记忆需要知识的时候去检索长期记忆。5.2 Agent 安全与权限隔离多 Agent 系统一旦接入真实工具文件系统、数据库、外部 API安全问题就必须重视。我的做法是最小权限原则每个 Agent 只给它完成本职工作的最小工具集。检索 Agent 只能读不能写计算 Agent 只能跑沙箱里的代码不能碰文件系统写报告 Agent 只能生成文本不能调用外部 API。另外所有 Agent 的工具调用都要过一层审计记录谁在什么时候调了什么工具、传了什么参数、返回了什么。出问题的时候能快速定位。注意如果 Agent 能执行代码一定要跑在沙箱里限制 CPU、内存、执行时间禁止网络访问。这是底线别省这一步。5.3 从 LangGraph 到自研编排的取舍LangGraph 很好用但不是银弹。当你的流程复杂到一定程度或者有特殊的性能要求时可能需要考虑自研编排层。我判断的标准是如果 LangGraph 的抽象能覆盖你 80% 的需求就用它如果超过 30% 的逻辑是在跟框架对抗就该考虑自研了。自研的核心其实就是一个调度循环加一个状态机没那么神秘class Orchestrator: def __init__(self, agents: dict, router): self.agents agents self.router router def run(self, task, max_steps20): state {task: task, history: [], step: 0} while state[step] max_steps: next_agent self.router(state) if next_agent END: break agent self.agents[next_agent] output agent.run(state) state[history].append(output) state[step] 1 return state就这么几十行核心逻辑全在里面。自研的好处是完全可控坏处是所有基础设施checkpoint、并发、可观测性都得自己搭。所以我的建议是先用 LangGraph 快速验证验证通过后再评估要不要自研别一上来就造轮子。5.4 可观测性建设多 Agent 系统不上可观测性等于闭着眼睛开车。我一般会埋这几类数据每个节点的执行耗时找出性能瓶颈。每个节点的 token 消耗找出成本黑洞。Agent 之间的流转路径可视化整个任务的执行轨迹。失败率和失败原因分布定位系统性问题。这些数据用 LangSmith 或者自己接 OpenTelemetry 都能采集。关键是别等到出问题才想起来加日志一开始就埋好后面排查问题能省一半时间。6. 我踩过的坑和几条实在建议做多 Agent 编排这一年多踩的坑比写的代码还多。挑几条最有价值的分享出来希望能帮你少走弯路。第一条别为了多 Agent 而多 Agent。我见过太多项目明明一个 Agent 加几个工具就能搞定非要拆成五个 Agent结果复杂度翻倍效果还更差。先跑通单 Agent遇到明确的瓶颈再拆这是最务实的路径。第二条状态设计要克制。State 里每多一个字段就多一份维护成本和上下文开销。我现在的原则是能推导出来的字段不放能放局部的不放全局能结构化的不存原文。第三条错误处理比正常流程更重要。多 Agent 系统里任何一个节点都可能失败而且失败会沿着图传播。所以每个节点都要有明确的失败语义编排层要有统一的兜底策略。宁可返回降级结果也不要让整个任务崩掉。第四条测试要覆盖流转路径。单元测试测单个 Agent 不够一定要测整个图的流转。我一般会构造几类典型输入正常流程、需要重试的、需要并行的、会触发兜底的确保每条路径都跑通。第五条成本要实时监控。多 Agent 系统的 token 消耗是单 Agent 的好几倍不监控的话月底账单能吓死人。我一般会设一个日消耗上限超了就告警别等烧完了才发现。最后分享一个我最近在用的技巧给每个 Agent 加一个自信度字段。Agent 输出结果的时候让它自己评估一下对这个结果的把握有多大0-1。编排层根据自信度决定要不要交叉验证——自信度低于 0.7 的就再派一个 Agent 独立跑一遍两个结果对比。这个机制让我的系统准确率又提了 8 个百分点成本只增加了 15%性价比很高。