前阵子接了个自动化调研的活儿，需求看起来很简单：让AI定期帮我产出竞品动态周报。我一开始也是老思路，单Agent加一段超长Prompt，把搜索、分析、总结全塞进去。跑了两周就发现不对劲，上下文越堆越长，输出质量忽上忽下，有时候连数据来源都开始编。后来我把这套东西彻底推倒，重新按"agency-agents"的思路组织——说白了，就是不再养一个全能选手，而是养一支各司其职的AI小团队，有做研究的、有做分析的、有做质检的，再由一个调度角色统一分派任务。这篇文章就聊聊我完整搭建这套多智能体协作体系的过程，从架构设计到核心代码，再到实测数据和一个又一个的坑。适合那些已经跑通过单Agent、正打算往多Agent编排方向走的开发者，也适合想搞清楚Agent协作到底怎么落地的技术决策者。
1. 为什么我会自建"agency-agents"而不是直接堆Prompt
先说结论：单Agent不是不能用，但凡是需要"多步骤、多角色、可复现"的任务，硬塞给一个Agent就是在给自己埋雷。
1.1 单Agent模式的三个致命瓶颈
我那段时间天天跟一个超长Prompt较劲，网上的、文档里的、别人分享的各种结构都试过。表面看是提示词工程问题，实际是架构问题。具体来说有三个瓶颈特别明显：
第一是上下文膨胀。任务链条一长，前面搜到的资料、中间的分析草稿、后面的总结措辞，全堆在同一个上下文窗口里。Token占用指数级上涨不说，模型越往后越容易"忘记"开头的要求，经常出现前面说A方案、后面又按B方案总结的情况。
第二是职责混乱。既让Agent做事实检索，又让它做观点提炼，还让它做格式排版——这等于让一个人同时当记者、编辑和美术设计。每切换一次角色，模型就要做一次"思维转换"，转换不彻底就会串味。我实测下来，串味最典型的症状是：分析部分里忽然冒出一段"以下是搜索结果"，或者总结部分开始重复搜索关键词。
第三是返工成本。单Agent模式下如果中间某一步质量不行，你没法精准定位是哪一环出了问题，只能整体重跑。而整体重跑又意味着从头搜资料、从头生成，费时费力费钱，最后还不一定比上次好。
1.2 一个后厨团队的类比
想明白这个事之后我换了个思路。你看一个餐厅后厨，不可能让一个厨师既配菜又掌勺又洗碗——配菜师只负责切配，炒锅师傅只负责调味，洗碗工只负责清洁，各干各的，最后通过传菜口协作。放到AI这边一模一样：检索Agent只负责搜集和整理原始素材，分析Agent只负责在素材基础上提炼观点，编审Agent只负责核查前后矛盾和事实错误。每个Agent的Prompt都短而专，上下文干净，角色不会串，出了问题也能精准定位是哪一环。
这就是"agency-agents"的核心思想：把一群各有所长的Agent组织成一个"代理机构"，有管理岗、有执行岗、有质检岗，通过明确的通信协议协作。
1.3 这种模式适合什么场景
不是所有任务都必须上多Agent。我梳理了一下，适合"agency-agents"的任务有几个特征：子任务之间有明确的边界、不同子任务需要不同的技能或知识侧重、任务结果需要多次检查迭代。比如周期性的行业报告、数据处理流水线、需要调用多个外部工具的复杂业务流程。
反之，如果是单轮问答、简单的文本改写、一次性小任务，直接单Agent反而更快更省。多Agent不是银弹，它解决的是"复杂任务的可控性"，代价是更高的系统复杂度和推理成本。我后面会详细算这笔账。
2. 体系架构拆解：角色注册、消息总线与任务编排是怎么串起来的
想清楚了"要组建AI团队"，下一步就是搭骨架。我把这套体系拆成了五个核心组件，每个组件只干一件事，彼此之间通过标准化的消息格式通信。
组件职责对照表：
| 组件 | 职责 | 类比 |
|---|---|---|
| RoleRegistry | Agent的注册、查找和健康检查 | 公司花名册 |
| MessageBus | 任务消息的分发与回传 | 公司内部邮箱系统 |
| TaskScheduler | 任务分解、依赖管理和调度执行 | 项目经理 |
| SharedContext | 跨Agent共享中间产物与状态 | 共享文档库 |
| MemoryStore | 长期记忆与历史结果沉淀 | 档案室 |
这五个组件的分工逻辑我展开说一下。RoleRegistry是基础，所有Agent启动时都要在这里登记，说明自己是谁、擅长什么、接收什么格式的消息。TaskScheduler是大脑，接到用户请求后先做任务分解，拆成若干子任务，再根据依赖关系决定哪些能并行、哪些必须串行。MessageBus不承载业务逻辑，只负责把消息从A传到B，这样Agent之间不会互相直接调用，降低耦合。SharedContext用来传递中间产物，比如检索Agent找了30条资料，分析Agent需要读这30条资料，就通过SharedContext拿，不用重新搜索一遍。MemoryStore则是超长期记忆，比如上周的周报结论、上个月的错误教训，下次任务可以直接引用。
2.1 数据流长什么样
我按一次完整的任务走一遍数据流，这样更容易理解各个组件是怎么配合的：
用户请求先进入入口Agent。入口Agent不干活，只做意图识别，判断任务类型后交给TaskScheduler。TaskScheduler把任务拆成"调研-分析-编审-成稿"四个子任务，按依赖关系排好顺序。调研类子任务交给专门负责检索的Agent，该Agent从MessageBus拿到任务，干完活把结果写进SharedContext，再通过MessageBus回传一个"完成"信号。TaskScheduler收到信号后，把分析子任务投递给分析Agent。依此类推，最后一个Agent产出最终结果，回传给用户。
2.2 为什么消息不能直接点对点传
我刚开始偷懒，让TaskScheduler直接拿到每个Agent的对象引用，然后一个个调用。跑通是能跑通，但很快发现三个问题：第一，新增一个Agent要改调度器代码，违背开闭原则；第二，Agent之间如果产生间接依赖，代码里查都查不出来；第三，想加一个"任务日志审计"的功能，发现无从下手，因为调用散落在各处。
改成MessageBus之后这些问题迎刃而解。所有通信走统一通道，调度器只需要知道"发什么消息给谁"，不需要知道"谁在什么时候具体怎么处理"。日志审计也简单，在MessageBus挂一个全局监听器，所有消息记录一目了然。这就是为什么我宁可多写几百行基础设施代码，也不省这个事。
3. 核心代码路径：从Agent基类到Director调度器的落地实现
这套体系我用Python实现，核心依赖只有ASGI和Pydantic，没上重量级框架——后面会解释为什么。先看最底层的BaseAgent。
3.1 Agent基类设计
from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Optional @dataclass class AgentRequest: task_id: str role: str payload: dict context_id: Optional[str] = None max_retries: int = 2 @dataclass class AgentResult: task_id: str role: str success: bool output: dict error: Optional[str] = None iterations: int = 1 cost: float = 0.0 class BaseAgent(ABC): """所有Agent的基类：声明角色、定义能力边界、实现统一入口""" role: str = "general" description: str = "" def __init__(self, model_name: str = "qwen-plus", temperature: float = 0.3): self.model_name = model_name self.temperature = temperature self.system_prompt = self.build_system_prompt() @abstractmethod def build_system_prompt(self) -> str: """每个Agent必须定义自己的系统提示词，这是角色边界的核心""" pass @abstractmethod def process(self, context: dict) -> dict: """核心业务逻辑：接收上下文，产出结果""" pass def run(self, request: AgentRequest, context_store) -> AgentResult: """统一执行入口，处理重试、超时、成本统计""" for attempt in range(request.max_retries + 1): try: context = context_store.get(request.context_id) if request.context_id else {} context.update(request.payload) output = self.process(context) result = AgentResult( task_id=request.task_id, role=self.role, success=True, output=output, iterations=attempt + 1 ) return result except Exception as e: if attempt >= request.max_retries: return AgentResult( task_id=request.task_id, role=self.role, success=False, error=str(e), iterations=attempt + 1 )这个基类有几个设计点值得说。role字段是Agent的身份标识，调度器靠它路由任务；build_system_prompt是强制要求每个Agent自己定义角色Prompt，不允许在实例化的时候临时传一个，这样能保证角色边界在设计期就被"焊死"；run方法内部统一处理了重试和成本统计，业务子类只需要实现process，不用关心这些横切逻辑。
3.2 Director调度器：任务分解与路由
Director是整个体系里最关键的组件，它的职责是：把用户请求翻译成子任务序列，再按Agent能力路由出去。
class DirectorAgent: def __init__(self, registry: dict[str, BaseAgent], model): self.registry = registry self.model = model def decompose(self, user_request: str) -> list[dict]: """将用户请求拆解为子任务序列，返回[{"role": "...", "prompt": "..."}]""" # 这里走一次LLM调用，让模型基于当前注册的Agent列表做任务规划 agent_roles = ", ".join(self.registry.keys()) planning_prompt = f""" 你是任务规划器。当前可用的Agent角色有：{agent_roles}。 请将这个用户请求拆解为不超过5个子任务，每个子任务指定一个最适合的角色。 用户请求：{user_request} 只输出JSON数组，格式： [{{"role": "researcher", "prompt": "具体指令"}}, ...] """ raw = self.model.chat(planning_prompt) return self._parse_json_safe(raw) def dispatch(self, request: AgentRequest, subtasks: list[dict], context_store): """按依赖顺序执行子任务，允许并行""" # 简化版：按顺序执行，实际可以按依赖拓扑并行 for st in subtasks: agent = self.registry[st["role"]] sub_request = AgentRequest( task_id=f"{request.task_id}:{st['role']}", role=st["role"], payload={"task": st["prompt"], "context_id": request.context_id}, context_id=request.context_id ) result = agent.run(sub_request, context_store) # 把每个子任务的输出写回共享上下文，供后续任务读取 context_store.append(request.context_id, st["role"], result) final_result = context_store.get(request.context_id) return final_resultDirector的核心是一个规划-执行循环。每次拿到用户请求，先让LLM基于"当前注册了哪些角色"来拆分任务，这一步很关键——任务分解不是硬编码规则，而是动态适配的。你新增了一个"数据可视化Agent"，下次用户请求里涉及图表需求时，规划器就会自动把"生成图表"拆成一个子任务，无需改代码。
3.3 注册机制与消息规范化
Agent注册我用了一个简单的装饰器实现，这样新增能力只需要写一个类加一行注册语句，Director那边不用动：
AGENT_REGISTRY = {} def register_agent(cls): AGENT_REGISTRY[cls.role] = cls() return cls @register_agent class ResearcherAgent(BaseAgent): role = "researcher" def build_system_prompt(self): return "你是一名研究员，负责检索并整理事实性资料，只输出原文摘录和来源链接，不做主观判断。" def process(self, context): # 调搜索API，把结果去重、提取摘要 query = context["task"] results = search_web(query, top_k=10) return {"materials": dedupe(results)}每个Agent通过@register_agent把自己登记进AGENT_REGISTRY，Director启动时加载这个字典。这套做法看起来不起眼，但它保证了两件事：Agent列表是运行时可见的，规划器能实时感知可用能力；新增Agent不影响任何既有代码路径。
消息格式我全部用Pydantic的dataclass约束，字段不可随意增删。这么做看起来啰嗦，实际好处是：每个Agent收到的都是结构一致的任务包，不会出现"A传了b字段、B期待c字段"的对接错位；后续加消息追踪、审计、重放，全都基于这套格式做，不用返工。
4. 一次真实任务的完整旅程：从任务拆解到报告产出
架构讲得再多，不如看一次实际执行过程。我带了一个具体的业务场景走完整流程。
4.1 场景设定：竞品动态周报
某团队需要每周产出一份竞品动态周报，内容包括：主要竞品的产品发布动态、价格或策略调整、行业趋势信号、以及对自家产品的潜在影响评估。之前这个活靠人工，每人每周要花半天。我的目标是用"agency-agents"把这套流程跑起来。
4.2 四个Agent的职责定义
我注册了四个Agent，角色分别是：researcher（研究员）、analyst（分析师）、reviewer（编审）、writer（撰稿人）。
- researcher：对接搜索API，搜集一周内目标竞品的公开动态，输出事实清单，附带原始链接和时间戳。
- analyst：接收研究员的事实清单，结合产品背景，提炼出"哪些动态值得关注"以及"可能的影响"，输出分析要点。
- reviewer：核查分析师输出的合理性，检查是否有夸大、前后矛盾、事实缺乏支撑，输出核查意见。
- writer：把分析要点和核查意见整合成一篇结构完整的周报，输出Markdown正文。
这四个角色的顺序是严格依赖的：researcher干完，analyst才能干；analyst干完，reviewer和writer才有输入。我把前三个当作串行链，writer放在最后。
4.3 实际执行数据
跑了三轮完整测试，其中一轮是模拟突发竞品发布会当天的数据采集。执行情况如下表：
| 环节 | 平均耗时 | 平均Token消耗 | 质量自评 |
|---|---|---|---|
| 任务规划（Director） | 3.5s | 约800 | 稳定 |
| researcher | 22s | 约6000 | 事实清单较全，偶有重复项 |
| analyst | 18s | 约4500 | 洞察质量高，偶尔出现跨领域脑补 |
| reviewer | 12s | 约3500 | 能抓住大部分问题，有少量误报 |
| writer | 15s | 约5000 | 成稿稳定，格式统一 |
整个链路跑下来约70秒。对比人工方式需要半天，效率提升接近两个数量级。Token消耗约2万左右，按通用模型的成本折算，单次生成成本在几毛钱级别。这个成本换一个团队半天的工时，性价比相当可观。
4.4 为什么坚持这个顺序而不是让一个Agent全干
我在跑测试的过程中试过"合并角色"的偷懒做法：让analyst同时兼任reviewer，也就是自己分析、自己检查。结果发现一个问题，分析Agent给自己挑错的时候，挑错意愿明显下降，很多明显的前后矛盾它自己看不出来。后来我专门查了一下，这跟模型的自我纠错能力限制有关，模型在自己生成的内容上做批判性审视，效果远不如让它去审视另一个角色的输出。
这也是"agency-agents"为什么强调角色分离的一个具体原因——不只是为了让Prompt短，更是为了利用"不同视角"带来的制衡效应。reviewer Agent的Prompt里有一句关键设定："你负责代表用户质疑分析师的观点，不要轻易认同。"这句话让它在大概率上保持挑剔姿态。
4.5 最终产出效果
成稿一篇周报大约800到1200字，结构固定为：概览、竞品动态要点、影响分析、下周关注。人工审阅后，初稿可用率在八成左右，剩余两成主要是有些动态抓取不全或者影响判断过于保守。相比纯单Agent生成，这套体系的优势在于：事实核查环节能把"编造数据"的概率压到很低，因为reviewer会要求每个结论给出支撑来源；成稿格式始终稳定，writer不像单Agent那样隔几篇就突然改变排版风格。
5. 跑通之后的排障实录：上下文污染、任务死锁与格式翻车
任何系统只要跑起来，问题都会冒出来。我集中说三个最典型的，附带完整排查链路。
5.1 上下文污染：Agent串味是怎么回事
现象：analyst的输出里出现researcher风格的句子，比如直接输出"以下是搜索到的原始URL列表"，但它根本没搜过网。
排查过程：我先看MessageBus日志，确认消息路由没有问题，analyst确实只收到了researcher的产出摘要。然后又检查了SharedContext，发现我一开始设计的是所有子任务共享同一个上下文对象，researcher把30条原始资料全塞了进去，analyst拿到的上下文里既有"整理好的事实清单"，也有"未清洗的搜索摘要"。我给analyst的Prompt说"基于研究资料进行分析"，模型一看上下文里有一大堆URL原文，就倾向于直接引用它们。根因找到了：Sharing Context时没做内容隔离，中间产物的形态不合格。
修复：每次子任务之间传递时，不直接传原始上下文对象，而是通过一个extract_for_role(context, role)函数，该函数按照目标角色的需求把上下文裁剪成"该角色需要的最小集合"。比如传给analyst的只保留researcher整理后的"fact_list"，丢弃所有原始搜索摘要和中间草稿。修完后串味现象彻底消失。
这个教训后来让我养成了一个习惯：SharedContext里必须显式声明"谁写的、给谁读的"，并加一条隔离规则——默认情况下，其他Agent只能读取指定字段，看不到完整内容。
5.2 任务死循环：两个Agent互相甩锅
现象：有一轮测试迟迟不返回结果，查日志发现Director一直在循环调度，任务卡了快10分钟。
排查：MessageBus日志显示，reviewer在核查时发现了analyst的一个疑点，于是它把"需要补充分析"的信令写回了SharedContext。我那个版本的调度器逻辑是"只要上下文里有未处理信令就继续投递"，于是analyst又收到"补充分析"的任务，补完之后reviewer又发现新的疑点……两人来回互相派活，形成死循环。
修复：我在Director里加了两道保险。第一道是每轮调度检查该子任务的最大执行次数上限，超了就强行终止并标记"需人工介入"；第二道是引入超时熔断机制，整个任务的执行时间超过预设阈值（我设的5分钟）就自动终止，避免无声无息地耗着。
这个坑本质上是我在设计任务依赖时没有考虑"回环"的情况。任务有向无环图（DAG）不是口头说说，要真正在做调度的时候校验依赖关系，哪怕用最简单的方式——每轮执行前检查"这个子任务是否已经被执行过N次"——也能挡住绝大部分死循环。
5.3 模型输出格式翻车：让Agent输出JSON它非要加Markdown
现象：Director的_parse_json_safe函数频繁报错，一看原始返回，模型偶尔会在JSON外面加一段解释性文字，比如"以下是JSON格式的结果："，导致解析失败。
排查：不是Prompt问题。我试过在提示词里强调"只输出JSON，不要其他内容"，正确率从70%涨到85%，但尾部仍有15%的概率翻车。再试了降低temperature从0.7降到0.3，正确率继续提升到92%，但仍然有漏网。这说明单靠提示词和参数控制解决不了这个随机性问题，必须靠代码兜底。
修复：_parse_json_safe里做三层兜底。第一层尝试直接json.loads；第二层用正则提取第一个[和最后一个]之间的内容再解析；第三层如果还是失败，就把这个子任务标记为"格式异常"，重新提交给同一个Agent并附带"上次你返回的内容格式不对，请重新输出严格JSON"的错误反馈。这三层下来正确率逼近100%。
这条经验让我对Agent系统有个新认识：不要把稳定性押注在模型的输出纪律上，而是要让系统能容忍模型偶尔的调皮，用解析兜底和自动重试去吸收这部分不确定性。
6. 事后复盘：这套体系到底值不值，我用下来的三点建议
跑通这套"agency-agents"，我最大的感受不是"AI真厉害"，而是"工程化真重要"。模型能力强只是一个前提，真正让系统稳定可依赖的，是角色边界、消息规范、重试机制和日志追踪这一整套工程手段。
6.1 什么时候不要用多Agent
如果只是单轮问答，或者任务链条很短且不需要交叉验证，直接单Agent就好，省事省钱还快。多Agent的价值只有在任务复杂到"单Agent无法可靠完成"时才体现。你可以用这个标准判断：如果任务链条超过三个步骤，且中间有需要不同技能的环节，且结果需要人工审核——这时候才值得上"agency-agents"。达不到这个复杂度，所有的编排代码都是负担。
6.2 三条实操建议
第一，先把流程写死再上调度器。我第一次做就想让Director完全动态规划任务，结果每次执行路径都不一样，排障全靠猜。后来老老实实先写好固定的执行链（比如固定的搜索、分析、核查、成稿四步骤），跑熟之后，再让Director做局部动态调整。稳定的主干加上有限的灵活性，远比全动态可靠。
第二，把每个Agent的能力边界钉死。这体现在Prompt设计上，也体现在输入输出格式上。我给每个Agent都规定死了输入类型和输出字段，researcher只能输出事实清单，analyst只能输出分析要点，谁也不能越界。边界一旦模糊，系统的不确定性就会指数上升——因为模型会自动"补全"你没说明的部分，而补全的方向你不可控。
第三，日志必须能看清谁在什么时候干了什么。我建议在MessageBus挂一个全局日志监听，记录每一次消息的发起角色、接收角色、耗时、成本、成功失败状态。这套日志平时不显眼，一旦要排查问题，它就是救命的。我排上下文污染和死循环的两次经历，全靠日志一步步反推，没有日志的话这两个问题几乎没法定位。
6.3 后续可以怎么玩
这套体系跑顺之后，我下一步打算做两件事：一是把MemoryStore用起来，让每周周报自动参考上个月的结论，实现跨任务的连续性；二是加上并行调度，目前子任务还是顺序执行，实际上researcher可以拆成多个并行搜索任务来压缩耗时。这个方向做下来，可能会让整条链路从70秒压到40秒以内，我到时候有新发现再回来分享。
