多Agent协作架构设计实践：从职责拆解到并发调优

先说我自己的经历。去年我把一个智能客服项目从单 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 出错会直接污染后续所有步骤。

我整理了一个对比表格，方便你快速选型：

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_factory=dict) 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, model=self.model, temperature=self.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_on=None): 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_workers=config.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(timeout=timeout) 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 协作系统的排查难度随着 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 个之后，代码里到处是硬编码角色名，改一处崩三处，原因就是没有提前做解耦。提前半小时做的设计，往往能省下后面好几天重构的时间。