多Agent编排实战：LangGraph核心概念与生产环境避坑指南

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：

举个我实际做过的例子：一个合同审查系统，需要提取条款、比对标准模板、识别风险点、生成修改建议、输出审查报告。这五个环节每个都需要不同的 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。生产项目里一定要把版本号锁死，比如langgraph==0.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 编排图的完整实现

下面是一个我实际项目里用过的编排图，任务是"分析一份文档并生成结构化报告"。涉及四个 Agent：Reader（读取解析）、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(checkpointer=memory)

跑起来就是：

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 的输入上下文设一个上限，超了就按重要性排序截断。别指望模型自己"忽略无关信息"，它做不到。

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% 的常见问题：

4.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_steps=20): 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%，性价比很高。