先说我自己的经历。去年我把一个智能客服项目从单 Agent 升级成多 Agent 协作架构时团队里有人问一个 Agent 能干完的活拆给好几个 Agent除了增加出 bug 的概率还有什么意义我没急着反驳直接跑了一个对比测试。同样一条“帮我比较三款显卡的性价比并给出推荐理由”的复杂请求单 Agent 吭哧吭哧拆解、搜索、推理结果后半段上下文乱了把 A 卡的显存参数当成了 B 卡的而多 Agent 协作版本里检索 Agent 只负责找资料对比 Agent 只负责按指标计算评测 Agent 只负责写结论最后给出一份结构清晰的对比报告速度还快了近一倍。那次之后团队里反对的声音基本消失了。这篇内容我想把多 Agent 协作这件事讲透它到底解决什么问题架构上怎么设计代码层面如何落地并发参数怎么配才稳以及我在实际项目中踩过哪些坑。不管你是准备用 jiuwen swarm、CrewAI、LangGraph 这类开源框架还是想自己从零写一套轻量实现这篇都值得你花十分钟看完。1. 先搞清楚多 Agent 协作到底解决什么问题很多人在接触多 Agent 协作时第一反应是“花活”“炫技”。但真正推动大家从单 Agent 迁移到多 Agent 的往往不是好奇心而是单 Agent 模式在复杂任务面前确实顶不住了。1.1 单 Agent 模式卡在哪单 Agent 最核心的问题是上下文隔离和职责混杂。你用同一个模型实例同时承担检索、推理、写作、校验等工作它必须把每一步产生的中间结果都塞进同一个上下文窗口里不断累积。一旦任务链条变长早期信息会被后续内容冲淡模型就开始“忘事”。我做客服机器人时遇到过一个特别典型的场景用户咨询的是“我的订单 3 天没发货帮我查一下物流然后写一封投诉邮件”。单 Agent 要先调用订单查询工具再调用物流查询工具最后写邮件。问题在于查订单返回了一长串 JSON其中包含多个字段Agent 在执行后续步骤时很容易把“订单创建时间”误认为是“物流更新时间”原因就是有效信息被埋在了大量冗余文本里。另一个问题是工具调度混乱。当 Agent 需要调用七八个工具时它对“什么时候调哪个工具、工具返回后怎么处理”的决策质量会明显下降尤其是工具参数相似时经常出现调错接口、漏传参数的情况。你可以把单 Agent 想象成一个同时要接电话、写文档、订机票的助理每件事它都会一点但一旦同时涌进来就开始手忙脚乱。1.2 多 Agent 协作带来的核心价值多 Agent 协作本质上做的事情是“职责分解 上下文隔离 并行计算”。职责分解的好处最直观每个 Agent 只需要关注一件事它的系统提示词可以写得非常专注。比如“资料检索员”只需要知道“如何调用搜索工具、如何从结果中提取有效信息”不需要关心后续怎么写报告。这个分工让每个 Agent 的指令空间变小模型误操作的概率自然下降。上下文隔离是容易被低估的一点。每个 Agent 收到的是编排器分发的“与任务相关的上下文”而不是完整的历史会话。也就是说检索 Agent 拿到的只是“用户问题 搜索工具的返回结果”分析师拿到的只是“检索整理后的文档 分析指令”。这样每个 Agent 的上下文窗口都能被高效利用不会出现信息淹没。并行计算则是性能上的直接收益。用户给出一个包含 5 个子任务的需求如果这 5 个子任务之间没有依赖关系你可以同时启动 5 个 Agent而不是让一个 Agent 串行处理 5 遍。在后面谈并发配置时我会给出具体的数据同样 50 个任务并发数从 1 提高到 8总耗时能减少到原来的四分之一左右。提示多 Agent 协作不是银弹。任务本身如果逻辑上必须严格串行强行拆成多个 Agent 只会增加通信开销。判断标准很简单拆出来的子任务之间有没有明确的边界如果有并且边界之间可以靠结构化数据传递就适合多 Agent如果边界模糊靠“模糊上下文”衔接那就别拆。2. 多 Agent 协作的架构设计与角色编排架构设计是整个多 Agent 协作系统中最关键、也最容易出问题的一环。很多项目栽跟头不是模型能力不够而是角色编排一团糟。这一节我把常见的组织模式、角色划分粒度和通信机制讲清楚。2.1 三种常见的组织模式怎么选我在梳理多 Agent 开源项目时发现不管框架叫什么名字底层组织模式基本逃不出三种主管-下属模式、同事协商模式、流水线模式。主管-下属模式也称 Orchestrator-Worker是国内项目用得最多的方式。编排器先对用户请求做意图识别和任务分解再把子任务分发给各个 Worker最后收集结果并汇总。这种模式的好处是控制流清晰、容易排查问题适合任务结构相对稳定的场景比如“查资料—做分析—写报告”。同事协商模式类似于让多个 Agent 在同一个会话里自由发言彼此提问、反驳、补充模拟人类头脑风暴。它能激发出一些单 Agent 想不到的角度但缺点也很明显容易离题、token 消耗大、结果不可控。我自己的实践体会是这种模式最多控制在 3 个以内角色参与并且需要有一个“主持人 Agent”负责收敛结论否则聊到第五轮基本就偏了。流水线模式则是按顺序一个接一个处理前一个 Agent 的输出是后一个 Agent 的输入适合内容生产链路比如“大纲生成—章节扩写—通稿润色”。它的优点是实现简单、依赖关系明确缺点是整体延迟等于所有 Agent 延迟之和且前一个 Agent 出错会直接污染后续所有步骤。我整理了一个对比表格方便你快速选型模式适合场景优势劣势典型框架主管-下属任务可拆解、结构稳定控制流清晰、易排查编排器可能成为瓶颈LangGraph、jiuwen swarm同事协商头脑风暴、多角度分析信息覆盖面广易离题、token 消耗大AutoGen、ChatDev流水线内容生产、ETL 类任务实现简单、依赖清晰延迟叠加、误差传播自研居多2.2 角色粒度别把角色拆得太碎也别什么都塞一个角色角色划分的粒度是设计中最考验经验的部分。拆得太细每个 Agent 都要消耗一套独立的提示词和上下文协调成本急剧上升拆得太粗本质上又回到了单 Agent 的老路。我自己的经验是三条原则每个角色必须有一项“不能被别人替代”的专属能力每个角色的系统提示词控制在 500 字以内角色数量在单次协作中尽量控制在 5 个以内。角色命名也有讲究。我给团队定的规范是“动词短语 领域名”比如“资料检索员”“数据分析师”“报告撰写人”。避免使用“QA Agent”“Helper Agent”这种意义模糊的名字因为编排器在生成任务调度时角色名本身是一种语义信号命名越清晰模型对任务的理解越准确。另外角色描述不要写成“你是一个负责数据分析的 AI 助手”这种空话而要写清楚“你的输入是什么、你要做什么、你的输出格式是什么”。我通常会固定成这样的模板角色数据分析师 职责根据给定的原始数据计算出销售额增长率、客单价和复购率 输入结构化数据通常为 JSON 数组 输出一段包含具体数值和结论的 Markdown 分析 约束如果数据量少于 5 条明确提示“数据不足”不要强行下结论2.3 通信机制共享黑板还是消息队列Agent 之间怎么交换数据直接决定了系统的复杂度天花板。最原始、也最常用的方式是“函数调用返回”也就是子 Agent 把结果以返回值形式交还给编排器由编排器统一分发。这种方式实现简单、可观测性强适合中小规模的协作系统。如果你想做更灵活的协作可以考虑共享黑板模式。所有 Agent 往一个公共区域写结果其他 Agent 按需读取。这个模式适合任务之间有信息依赖、但依赖关系不固定场景。但也别盲目用多个 Agent 同时写同一个字段时会出现状态竞争共享黑板里的数据会越来越多最终变成一锅粥。消息队列是面向大规模并发的方案Agent 之间通过消息传递解耦你可以随时增加消费者来提升吞吐。代价是引入额外的基础设施调试时看不到全局状态问题定位难度直线上升。我自己的建议是团队项目初期老老实实用“编排器加函数返回”的模型把状态流画清楚。等到并发量上去了再考虑引入消息队列也不迟。很多开源项目包括我参考过的 jiuwen swarm 协同架构核心也是这个思路——编排器持有全局调度逻辑Agent 保持相对无状态通信靠结构化数据。3. 从零搭建一个多 Agent 协作系统以 jiuwen swarm 风格为例理论讲完上实操。这一节我参考 jiuwen swarm 这类 Swarm 风格项目的协同架构设计思想带大家从零写一套轻量级多 Agent 协作系统。不依赖重框架核心代码 200 行内能跑通。代码使用 Python模型接口我以 OpenAI SDK 风格为例但换成 Qwen、DeepSeek 或本地模型也通用。3.1 环境准备与项目结构基础环境需要 Python 3.10 以上安装openai和pydantic两个包就够了其余按需引入。my_swarm/ ├── agents.py # Agent 定义 ├── orchestrator.py # 编排器 ├── tools.py # Agent 可用工具 ├── config.py # 并发与模型配置 └── main.py # 入口项目的核心思路是Agent只是一个描述不包含复杂逻辑Orchestrator负责调度所有 Agent 并传递上下文。这样一个 Agent 可以被多个任务复用也可以在编排器里灵活调整协作关系。3.2 定义 Agent 角色与工具我用一个Agent类来表示角色里面只存名字、系统提示词和可用的工具函数列表。from dataclasses import dataclass, field from typing import Callable, Optional dataclass class Agent: name: str system_prompt: str tools: dict[str, Callable] field(default_factorydict) model: str qwen-plus temperature: float 0.3 def execute(self, task: str, context: str) - str: # 简化实现把上下文和任务拼进提示词交给 LLM messages [ {role: system, content: self.system_prompt}, {role: user, content: f任务{task}\n\n可参考上下文{context}} ] # 这里默认你已经配置好模型客户端可以换成任何 SDK response call_llm(messages, modelself.model, temperatureself.temperature) return response这里有一个很关键的设计我不在Agent内部维护长期记忆所有上下文由编排器每次显式传入。这避免了“Agent 越干越糊涂”的问题也让每个 Agent 的实际运行逻辑完全可控。工具函数的定义也很简单比如让“资料检索员”能调用一个网页搜索函数def search_web(query: str) - str: # 假设这里调用了搜索 API返回结果文本 return f关于{query}的搜索摘要包含来源链接和关键段落 collector Agent( name资料检索员, system_prompt你负责根据用户问题搜索并整理相关资料。输出为不带主观判断的要点列表。, tools{search_web: search_web} )3.3 编排器Orchestrator的核心逻辑编排器承担三件事任务分解、依赖调度、结果聚合。最简单的编排器可以用固定流程实现不引入额外的规划模型。class Orchestrator: def __init__(self, agents: dict[str, Agent]): self.agents agents def run(self, user_query: str) - str: # 第一步检索 Agent 搜集资料 search_task f搜集与以下问题相关的资料{user_query} raw_materials self.agents[资料检索员].execute(search_task, user_query) # 第二步分析 Agent 对资料进行整理和分析 analysis_task 根据已有资料提炼关键论点和数据给出结构化分析。 analysis self.agents[数据分析师].execute(analysis_task, raw_materials) # 第三步报告 Agent 输出最终书面结论 report_task 基于分析结果写一份条理清晰、有结论的书面回答。 final_report self.agents[报告撰写人].execute(report_task, analysis) return final_report这种写法的好处是流程是显式的、可读的。你一眼就能看到每一步谁在执行、输入是什么、输出给了谁。相比让模型自己决定调用哪个 Agent 和以什么顺序调用固定流程在大多数业务场景下更稳。如果你需要更灵活的编排可以让编排器在第一步先调用一个“规划 LLM”来把任务拆成 JSON 格式的子任务。但要注意多增加一次 LLM 调用就会多增加一次延迟和出错概率。我在生产环境里的做法是能用规则拆的任务就不让模型拆只有模型发现“这是一个全新的、没见过的任务模式”时才走动态规划路由。3.4 任务分发与结果汇总当子任务之间存在并行关系时固定顺序执行就浪费了资源。我一般会在编排器里维护一个简单任务状态机import concurrent.futures class Task: def __init__(self, name, agent_name, prompt, depends_onNone): self.name name self.agent_name agent_name self.prompt prompt self.depends_on depends_on or [] self.result None def dispatch(self, user_query): tasks self.plan(user_query) with concurrent.futures.ThreadPoolExecutor(max_workersconfig.MAX_WORKERS) as pool: future_map {} for task in tasks: future pool.submit(self._execute_single, task) future_map[future] task.name for future in concurrent.futures.as_completed(future_map): task_name future_map[future] try: future.result() except Exception as e: # 记录失败决定是重试还是降级 logger.error(f任务 {task_name} 失败: {e}) # 所有任务完成后由汇总 Agent 或直接拼接结果 return self._aggregate(tasks)结果汇总阶段我习惯单独加一个“总结 Agent”而不是让编排器用规则硬拼。因为子 Agent 的结构化输出往往是碎片化的直接拼接会显得生硬总结 Agent 的作用是把这些碎片整理成用户能直接阅读的完整回答。注意结果汇总时切忌把全部子 Agent 的原始输出原封不动塞进总结 Agent 的上下文要先做长度裁剪和关键信息抽取。否则总结 Agent 的上下文会被大量噪音占据反而影响输出质量。4. agent 多并发配置压测与调优实操多 Agent 协作项目上线后大家最关心的往往是“并发高了会不会崩”。这一节专门讲 agent 多并发配置从并发模型选型到参数设置再给一份我实测过的调优记录。4.1 并发模型线程、异步还是多进程选并发模型前先判断你的任务属于什么类型。LLM 调用是典型的 IO 密集型任务因为大部分时间都花在等待网络返回上CPU 几乎不参与。这种情况下线程池和异步 IO 都能很好地提升吞吐。线程池的好处是写起来简单配合ThreadPoolExecutor很容易入门异步 IO 的性能上限更高但会强迫你把所有代码写成 async 风格调试成本略高。如果你的场景涉及本地部署的开源模型推理那么任务属于 CPU/GPU 密集型这时应该考虑多进程部署多个推理实例或者干脆用模型服务自带的并发能力而不是在自己的业务进程里开一堆线程去抢同一个显存。我自己在绝大多数业务场景下的选择是线程池 ThreadPoolExecutor。原因很简单业务代码里还有大量同步的数据库调用、文件读取、日志写入硬要全部改成异步反而容易引入新的问题。4.2 关键参数并发数、超时、重试、限流并发配置不是只调一个“并发数”就完事。真正决定稳定性的是一组参数缺一个都可能出问题。第一个是最大并发数max_workers。这个值的设置需要同时考虑上游 API 的速率限制和你自己的机器资源。我的经验公式是先从 1 开始每次翻倍压测直到出现 429 限流错误或超时报错然后把并发数回调到报错值的 50% 到 70%。第二个是请求超时request_timeout。LLM 接口在复杂任务下响应时间波动很大设置太短会导致大量原本会成功的请求被误杀。我一般设为 60 秒长文本生成任务会放到 120 秒。如果某个 Agent 的任务特别重我会单独给它更长超时而不是全局统一。第三个是重试策略。重试要配指数退避不要抢着重试。第一次失败后等 1 秒第二次等 2 秒第三次等 4 秒最多重试 3 次。如果你的上游 API 明确返回了限流错误退避时间还要拉得更长。第四个是限流。即使你对某一个 API 设置了 max_workers 为 8但如果这个数字超过了上游分配的配额照样会被熔断。必要时候要自己实现一个令牌桶限流器把每秒请求数控制在上游限制以内。4.3 一个可复用的并发配置样例下面是我在实际项目中使用过的一份配置你可以直接抄过去改改。# config.yaml agent_concurrency: max_workers: 8 # 同时运行的 Agent 数 request_timeout: 60 # 单次 LLM 请求超时秒 max_retries: 3 # 失败重试次数 retry_backoff: [1, 2, 4] # 退避间隔秒 rate_limit_rps: 5 # 每秒最多请求数 max_context_tokens: 4096 # 单个 Agent 最大上下文 token result_validation: true # 是否校验子任务输出结构配合这段 Python 伪代码限制实际并发的核心逻辑大概长这样import threading import time class RateLimiter: def __init__(self, rps: int): self.rps rps self.lock threading.Lock() self.window [] def acquire(self): now time.time() with self.lock: self.window [t for t in self.window if now - t 1.0] if len(self.window) self.rps: self.window.append(now) return time.sleep(0.1) self.acquire() # 简化实现实际应使用条件变量 def execute_with_retry(agent_task, timeout, backoff): for attempt, wait in enumerate(backoff): try: return agent_task.run(timeouttimeout) except TimeoutError: time.sleep(wait) except APIError as e: if e.is_rate_limit(): time.sleep(wait * 2) else: raise raise RuntimeError(任务重试后仍然失败)4.4 实测数据与调优心得我拿一批真实客服工单做过压测一共 50 个任务每个任务里包含“检索资料、提取关键信息、生成回复”三个子任务平均每个子任务调一次 LLM。并发 1总耗时约 15 分钟每个任务平均 18 秒。并发 4总耗时约 4 分半吞吐提升 3.3 倍。并发 8总耗时约 2 分 40 秒吞吐提升 5.6 倍。并发 16总耗时降至 2 分 10 秒但出现 429 限流错误部分任务需要重试实际稳定性下降。最终我把生产环境的并发数定在 8配合 1、2、4 秒的三次退避重试整体表现最稳。这个结果也印证了并发调参的一个基本原则不要盲目追求并发数高只要吞吐增长明显减缓、报错率开始上升就应该往回撤。提示如果项目同时对接多个模型服务商建议在配置里把并发数拆成“全局并发”和“模型级并发”两层。全局并发控制整体并行度模型级并发控制单一模型调用的压力两层配合才能避免某个模型服务被压垮。5. 多 Agent 协作中的常见问题与排查技巧实录多 Agent 协作系统的坑不少是单 Agent 时代根本遇不到的。这里把我在项目上踩过、也帮别人排查过的几类高频问题整理出来每一类都会附上排查思路和解决办法。5.1 Agent 之间互相踩脚上下文与状态冲突现象两个子 Agent 明明各自独立工作结果 A 的状态把 B 的状态覆盖了或者 A 引用 B 的结果时拿到的已经是 C 修改过的版本。原因通常是共享了同一个可变容器。比如你把所有 Agent 的结果都写进同一个全局字典某个任务的编号重复或者字段名冲突就会覆盖。解决思路约定每个任务的键是“任务 ID 加 Agent 名”而不是裸字段名只有一个写入方也就是编排器负责写入共享状态子 Agent 的结果先返回给编排器再由编排器决定是否需要落盘。如果项目里确实需要多个 Agent 同时写共享黑板给每条记录加版本号并在读取时做冲突检测。5.2 上下文越滚越长记忆管理的失控现象上线一段时间后开始频繁出现“Token 超限”错误或者 Agent 输出质量明显下降答非所问。原因多 Agent 协作中编排器为了让子 Agent 有足够信息把历史会话全部传给每个 Agent导致每个 Agent 的上下文窗口迅速堆满。这是我见过最多的问题。解决办法也最简单传输上下文之前做裁剪。我通常只把三类信息传给子 Agent——用户原始请求、当前子任务需要的数据、其他 Agent 产出的结构化摘要。绝不在子任务上下文里放完整聊天记录。更进一步可以定期把过去 N 轮对话压缩成摘要再传给下一轮 Agent。5.3 单个 Agent 故障拖垮整个流程降级策略现象数据分析 Agent 偶发超时由于编排器是同步等待整个任务直接被拖住用户在等了 90 秒后收到一个失败提示。这个问题的本质是缺少故障隔离。我们需要给每个 Agent 调用包一层独立的异常处理子任务失败时先写失败日志再决定是否重试、跳过或换一个轻量模型重新执行。我习惯把“重试脚手架”做成装饰器统一给所有 Agent 执行函数加超时、重试和降级逻辑而不是在每个编排函数里散落一套 try except。降级的另一层含义是结果降级如果最终只有部分子任务成功编排器也应该能拼出一个带有“部分结果”标识的回答而不是直接报错。用户在大多数时候宁愿看到“已获取资料但对比分析暂不可用”也不愿意被一个通用错误打断。5.4 常见问题速查表我把高频问题整理成一张速查表方便你遇到问题时直接对号入座症状可能原因定位方法解决建议两个 Agent 的结论矛盾各自的评分指标不一致检查 prompt 中的评价标准统一指标交给裁判 Agent 做最终裁决子任务结果缺失异常未被捕获导致结果未写入查看执行日志检查是否有 RuntimeError给每个 Agent 增加结果校验与失败日志请求频繁超时或 429并发数超过上游限制查看 API 返回头和错误码降低并发数增加退避重试Token 消耗暴涨上下文历史被完整透传统计单次任务的 token 消耗使用上下文裁剪/摘要限制传参长度Agent 输出 JSON 无法解析模型没有严格遵守格式打印原始输出定位解析报错位置用 JSON Schema 约束或增加格式解析兜底函数任务整体响应变慢某个子任务无谓地等待锁用时间戳统计每步耗时为每个子任务设置独立超时避免全局阻塞我在排查这些问题时有个习惯每个子任务在执行前和执行后都会打印一条结构化日志包含任务名、耗时、输入摘要、输出长度和返回码。不要嫌日志多多 Agent 协作系统的排查难度随着 Agent 数量指数级上升没有日志出了问题你就只能对着黑盒猜。6. 多 Agent 协作的项目化落地建议最后这部分不谈代码谈谈怎么把多 Agent 协作真正用进项目里。技术能跑通是一回事能在业务上稳定运行又是另一回事。我自己的经验是分三步走先用固定流程的编排器把整个链路打通跑一个最小可用版本不要上来就搞复杂路由然后逐步引入并行和动态规划每次只改一个变量比如先加并发稳定了再加动态任务拆解最后再做完整的可观测性建设让每个 Agent 的执行过程都在日志里有迹可循。6.1 不要一开始就追求“全自动编排”市面上的多 Agent 框架很喜欢强调“全自动”模型自己决定怎么拆任务、调哪些 Agent。这个概念听起来很美但实际落地时会发现模型的规划能力即使足够好也不一定稳定同一个任务今天拆成三步明天可能拆成四步。我在生产环境里的做法是“骨架固定边界灵活”。骨架固定指大流程是写死的比如检索、分析、生成这三个阶段永远存在边界灵活指每个阶段内部根据用户问题子 Agent 可以有所选择。这样既有一定的灵活性又不会失控。6.2 评估收益时算总账多 Agent 协作确实能提升输出质量和并发能力但也要看到成本更多的 LLM 调用意味着更高的 API 费用更多的中间上下文意味着更长的响应时间更多的组件意味着更高的维护复杂度。所以我建议项目决策前先把账算清楚。比如你的业务场景是简单问答单 Agent 一次调用就能解决引入 5 个 Agent 只会让事情更慢更贵。而如果你的场景是“写行业分析报告”需要检索多篇资料、交叉分析、结构化输出那么多 Agent 协作带来的质量提升绝对值回票价。注意多 Agent 协作的收益要定量验证不要凭感觉。上线前后用同一批测试集跑分别记录输出达标率、平均耗时、单次任务成本用数据说话。6.3 记得为后续扩展留好接口最后小提醒无论你现在是 3 个 Agent 还是 5 个 Agent都建议把 Agent 注册表、任务定义、工具函数这三个维度解耦。Agent 注册表维护“有哪些角色”任务定义维护“每个任务由哪个角色执行”工具函数维护“Agent 能调用什么能力”。当你想加一个新 Agent 或者换一个工具时只需要改对应的注册表或配置而不需要动编排器的主体逻辑。我见过太多项目在 Agent 数量超过 10 个之后代码里到处是硬编码角色名改一处崩三处原因就是没有提前做解耦。提前半小时做的设计往往能省下后面好几天重构的时间。