做AI Agent开发的朋友，最近应该没少听说“多Agent协作”这个概念。单Agent的边界很明显，复杂任务一拆多步就容易上下文混乱、工具调用失控、结果没法自检，于是很多人开始把目光投向“agency-agents”这种思路——把任务拆给一组各司其职的Agent去协同完成，像一个团队一样分工、流转、验收。这个仓库核心做的事，就是给出了一套多Agent协作系统的参考实现：定义角色、编排流程、管理任务状态、处理Agent之间的消息传递。我花了两周时间把它跑通，又在此基础上改了一版用于自己的自动化项目，今天把整个设计思路、关键实现细节和踩过的坑完整梳理一遍。无论你是刚接触Agent开发，还是已经在单Agent方案里遇到瓶颈，这篇文章都能帮你少走不少弯路。
1. 先搞清楚agency-agents到底在解决什么问题
1.1 单Agent不够用了，才需要“团队”
很多人第一次接触Agent开发时，最先实现的是一个“万能型”Agent：给它一个系统提示词，挂上几个工具，然后让大模型自己规划、调用工具、生成最终回答。这种模式在简单场景下很顺，比如“查天气”“算个表达式”，但一旦任务复杂度上去，问题就暴露了。
拿一个具体场景来说，假设你想让Agent完成“从一篇技术文档里提取关键信息，整理成一份结构化报告，并按团队风格写成邮件发给某个人”。单Agent接到这个任务后，要么把所有步骤一股脑塞进一次对话里，导致上下文窗口被大量中间过程占据；要么在工具调用之间反复横跳，容易出现规划了但没执行、执行了但忘了复核的问题。更麻烦的是，如果某个环节出错，你没有清晰的定位方式——是理解错了文档？是报告格式不对？还是邮件语气不符合要求？全都混在一起，排查成本非常高。
agency-agents走的完全是另一条路。它把“一个全能Agent”替换成“一组专职Agent”，每个Agent只负责一个环节，通过明确的输入输出协议衔接。就像公司里的项目组，有产品经理拆需求，有工程师写代码，有测试员做质检，各管一段，谁出问题就找谁，整体效率反而更高。
注意：多Agent并不是为了“显得高级”。当你的任务满足“多步骤、多角色视角、需要质检”这几个特征，多Agent架构才会真正体现出价值。反过来，简单任务硬拆多个Agent，只会徒增延迟和成本。
1.2 多Agent协作的几种常见形态
多Agent系统的组织方式，社区里已经有几种被验证过的模式，agency-agents的实现主要围绕其中两种展开，上手之前值得先分清。
第一种是流水线模式。任务按固定顺序流经多个Agent，每个Agent处理完自己的部分，把结果交给下一个。这种模式适合流程明确、顺序固定的任务，比如“生成摘要 -> 生成大纲 -> 生成全文 -> 校对润色”。优点是实现简单、执行链路清晰；缺点是中间任何一个环节失败，整条链路都要重跑，而且后续Agent无法回头修改前面的结果。
第二种是编排者模式，也是agency-agents的核心形态。一个Coordinator（编排者）Agent负责接收任务、拆解子任务，分发给不同的Worker Agent，再回收结果、进行汇总和验收。Worker之间不直接通信，所有消息都通过编排者转发，责任边界非常清楚。这种模式更适合“任务可以并行处理，但最终需要统一汇总”的场景，比如“分别调研三个不同主题，合成一份对比分析报告”。
实际使用中，这两种模式经常混用。agency-agents在架构上允许你在某个子任务内部再嵌套一组流水线Agent，这样既保证了顶层任务的灵活性，又能让局部步骤保持线性稳定。我在自己项目里就是把“调研”做成编排者模式，“报告排版”做成流水线模式，组合使用很顺手。
2. 整体架构设计与角色分工
2.1 角色怎么定义：谁做决策，谁干活，谁把关
agency-agents的项目定义里，核心角色一共四类，我实际跑下来觉得这个划分是经过考虑的，既不多到冗余，也不少到职责不清。
第一类是Planner（规划者），负责拆解用户任务。它不需要调用外部工具，核心产出是一份结构化的任务列表，每项包含任务描述、依赖关系、负责人角色和验收标准。这一步非常关键，因为拆解质量直接决定后续执行效率。如果规划者把任务拆得颗粒度过大，执行者拿到一个“巨无霸”子任务，依然会面临上下文超限；拆得过细，又会产生大量消息流转开销，成本上升。
第二类是Executor（执行者），负责实际干活。它接收Planner给出的子任务，按需调用各类工具——比如网页检索、代码执行、数据库查询——然后把结果以固定格式返回。执行者是整个系统里唯一真正接触外部资源的地方，其他角色都不直接操作工具。
第三类是Critic（质检者），负责检查执行者返回的结果是否符合要求。它不修改内容，只返回“通过”或者“失败+原因”，如果失败会附上修改建议。这个角色的价值在于让质量问题在环节内闭环，而不是等到最后汇总时才发现一堆问题。
第四类是Coordinator（协调者），作为整个系统的心脏。所有消息都经过它转发，它还负责维护任务状态机，追踪每个子任务从“待处理”到“执行中”到“已完成”的状态迁移。Coordinator本身不做具体业务，但它是整个系统的稳定器，也是排查问题时第一个应该看的地方。
2.2 任务流转的完整链路
理解多Agent架构，最有价值的不是看单个角色的定义，而是看一条消息从进入系统到最终产出，中间经历了什么。
我实操中的完整链路是这样的：用户输入先落到Coordinator，它会生成一个全局任务ID，并把原始诉求转发给Planner。Planner经过一轮思考，返回一个JSON格式的任务拆解方案，包含若干子任务，每个子任务都标注了类型（可并行、有依赖、需汇总）。Coordinator拿到后，按依赖关系建立有向无环图（DAG），然后开始调度：先执行没有依赖的节点，节点完成后通知依赖它的后续节点解锁，逐层推进。
每个子任务被分发到对应的Executor时，消息包里不只有任务描述，还包含“上下文摘要”和“验收标准”。上下文摘要是由Coordinator生成的，把与该子任务相关的信息浓缩成一小段文本，避免把整个对话历史都塞给每个Executor——这是控制token成本的关键手段。Executor返回结果后，Coordinator会先做一次格式校验，再交给Critic做内容质检。Critic返回“通过”后，该子任务才算真正完成。所有子任务完成后，Coordinator把各节点结果按Planner最初定义的模板做汇总，生成最终回复。
提示：这个流程里最容易被忽视的是“上下文摘要生成”。如果直接把上一个Agent的原始输出全部传给下一个Agent，多轮之后上下文体积会指数增长。我在实践里是让Coordinator在转发前专门调用一次“摘要函数”，把上一步结果压缩到300字以内，效果非常明显。
2.3 任务状态机与消息协议
状态机是整个多Agent系统里最需要严格设计的一部分。agency-agents里定义的子任务状态包括：pending（等待执行）、ready（依赖已满足，可被调度）、running（执行中）、completed（已完成并通过质检）、failed（执行失败）、blocked（依赖失败导致无法执行）。每个状态之间流转都有明确条件，不允许跳级，这样设计的好处是任何异常都能在状态层面快速定位。
消息协议方面，所有Agent之间的通信都统一使用JSON格式，必带字段是message_id、source_agent、target_agent、task_id、payload、timestamp。我在自己改造时额外加了一个schema_version字段，因为Agent返回的JSON结构难免随着需求调整而变化，有了版本号，解析逻辑可以按版本做兼容，避免某次升级后历史任务全部解析失败。
协议里还有一个细节值得单独说：Executor返回给Coordinator的消息中，payload部分分为“raw_output”和“summary”两层。raw_output是完整的执行结果，用于存档和需要全量信息的场景；summary是提炼后的摘要，用于后续Agent的上下文。这样拆分避免了“下游Agent不得不读完上游所有输出才能开工”的低效情况。
3. 实操实现：从一个最小可跑的版本说起
3.1 环境准备与依赖选择
说再多架构，不如看一份能跑起来的代码。先交代环境：我用的Python 3.10，依赖的核心库只有三个——一个用于调用大模型API的SDK（市面上主流的那几个都行，接口差异不大）、一个用于读写配置文件的标准库json/yaml、还有一个concurrent.futures用于并行调度子任务。没有引入重量级框架，原因是想保持实现透明，方便排查问题。
项目目录结构我建议这样组织：
agency-agents/ ├── agents/ │ ├── __init__.py │ ├── base.py # Agent基类，定义消息接收与发送 │ ├── planner.py # 规划者 │ ├── executor.py # 执行者 │ ├── critic.py # 质检者 │ └── coordinator.py # 协调者 ├── core/ │ ├── message.py # 消息协议定义 │ ├── state_machine.py # 子任务状态机 │ ├── dag.py # 任务依赖图 │ └── context.py # 上下文摘要生成 ├── config/ │ └── agent_config.yaml # 角色参数配置 └── main.py # 系统入口3.2 角色定义与系统提示词设计
每个Agent都是“系统提示词 + 模型参数 + 工具集”的组合体。系统提示词是灵魂，我写提示词的心得是：要定义“输入格式 -> 处理规则 -> 输出格式”三段式结构，而不是笼统地告诉模型“你是一个负责规划的人”。
以Planner为例，我的系统提示词关键内容是这样的：
PLANNER_PROMPT = """ 你是一个任务规划者。你的职责是将用户的复杂请求拆解为可执行子任务列表。 输入格式： - 用户请求：一段自然语言描述 处理规则： 1. 识别请求中的独立可执行部分 2. 判断子任务之间的依赖关系 3. 为每个子任务指定最合适执行者角色 4. 为每个子任务定义可验证的验收标准 输出格式（严格JSON）： { "tasks": [ { "id": "task_001", "description": "具体任务描述", "dependencies": [], "assignee": "executor", "acceptance_criteria": "通过XX指标判断任务是否完成" } ], "final_format": "最终汇总输出的模板要求" } 注意：只输出JSON，不输出任何解释性文字。 """这个提示词的设计要点有两个。一是强制“只输出JSON”，大幅降低解析失败率；二是让Planner为每个子任务定义验收标准，使得下游Critic质检时有明确依据，而不是靠模型临场发挥。
3.3 核心编排逻辑实现
Coordinator是整个系统的调度中枢，它的核心循环我简化成这段代码：
import json from concurrent.futures import ThreadPoolExecutor, as_completed from collections import deque class Coordinator: def __init__(self, agents_config): self.agents = {} self.task_status = {} # task_id -> state self.task_results = {} self.dependencies = {} # task_id -> list of dependent task_ids self.reverse_deps = {} # task_id -> list of task_ids it depends on def submit_task(self, user_request): planner = self.agents["planner"] plan = planner.run(user_request) tasks = json.loads(plan) for t in tasks["tasks"]: tid = t["id"] self.task_status[tid] = "pending" self.dependencies[tid] = t.get("dependencies", []) self.reverse_deps[tid] = [] for tid, deps in self.dependencies.items(): for dep in deps: self.reverse_deps.setdefault(dep, []).append(tid) self._schedule_ready_tasks() def _schedule_ready_tasks(self): ready_tasks = [ tid for tid, st in self.task_status.items() if st == "pending" and self._deps_met(tid) ] with ThreadPoolExecutor(max_workers=3) as executor: future_map = {} for tid in ready_tasks: self.task_status[tid] = "running" executor_agent = self.agents[self._assignee_for(tid)] future = executor.submit(executor_agent.run, self._build_task_payload(tid)) future_map[future] = tid for future in as_completed(future_map): tid = future_map[future] try: result = future.result() # 先做格式校验，再交给Critic质检 if not self._validate_format(result): self.task_status[tid] = "failed" continue critic_result = self.agents["critic"].run(tid, self.task_results.get(tid), self._criteria_for(tid)) if critic_result["pass"]: self.task_status[tid] = "completed" self.task_results[tid] = result else: self.task_status[tid] = "failed" self._try_unlock_dependents(tid) except Exception as e: self.task_status[tid] = "failed" self._handle_failure(tid, e)这段代码只是最小骨架，实际跑下来有几个点必须补强。一是“重试机制”，我给每个Executor包了一层重试装饰器，最多重试2次，而且第二次重试时会附带第一次的报错信息，让模型“知道自己上次错在哪里”。二是“失败传播”，如果某个子任务最终失败，所有依赖它的下游子任务都应该标记为blocked，而不是继续等待，否则会白白消耗后续Agent的调用次数。
3.4 消息协议与上下文摘要实现
我前面提到消息协议里包含raw_output和summary两层，这里的summary生成可以通过一个专用函数实现，也可以直接调用大模型做压缩。实测下来，用大模型生成摘要更稳定，但成本略高；如果追求性价比，可以先用规则截断（比如抽取首尾段、保留所有结构化字段），再加一段大模型润色。我项目里用的是后者，效果不错。
def build_summary(raw_output, max_len=300): # 规则截断：优先保留JSON结构化字段 if isinstance(raw_output, str): return raw_output[:max_len] if isinstance(raw_output, dict): important_keys = ["result", "conclusion", "key_points", "body"] parts = [] for k in important_keys: if k in raw_output: parts.append(str(raw_output[k])) joined = " ".join(parts) return joined[:max_len] return str(raw_output)[:max_len]3.5 配置驱动的角色参数管理
每个Agent除了系统提示词，还有一组运行参数：模型名称、temperature、max_tokens、重试次数、并发限制等。这些参数集中放在agent_config.yaml里统一管理，而不是散落在代码各处。
我在配置里最常调整的是temperature参数。Planner和Critic的temperature我调到0.2左右，要求输出稳定、可预期；Executor如果任务偏创造性（比如写作、头脑风暴），temperature可以调到0.7；如果偏工具调用和代码执行，最好也调到0.2，避免模型发挥过度调用错误工具。
agents: planner: model: "gpt-4o-mini" temperature: 0.2 max_tokens: 2000 executor: model: "gpt-4o" temperature: 0.3 max_tokens: 4000 tools: ["web_search", "code_executor", "db_query"] critic: model: "gpt-4o-mini" temperature: 0.2 max_tokens: 10004. 常见问题与排查技巧实录
4.1 Agent“各说各话”，协作结果一团糟
多Agent系统最常见的失败模式：每个Agent单独看都做得不错，但拼在一起就是不对。我调试过一个案例，Planner拆出三个调研子任务，三个Executor分别返回了三个主题的详细资料，结果汇总时发现三个报告风格不一致、结构不统一，甚至数据口径对不上。
根因在于Planner在拆解任务时没有给Executor定义统一的“输出模板”。Executor的自由度过高，每个人都按自己的习惯去组织回答。解决办法是让Planner生成的子任务描述里带上明确的输出模板，比如“请按以下小标题输出：1. 核心结论 2. 证据与数据 3. 局限性 4. 建议”，这样汇总阶段才能对齐。我把这个约束直接写进了Planner的提示词处理规则部分。
实操心得：多Agent系统里，“约束即生产力”。每个Agent的自由度都应该被明确限制，尤其是输出格式。凡是交给模型的自由，最后都会变成你排查问题的成本。
4.2 任务陷入死循环，或者反复重试不成功
我遇到过最典型的问题：Critic连续三次判定Executor结果不合格，场景是让Executor提炼一篇文章的摘要，Critic每次都说“信息不够完整，请补充XX部分”，Executor每次都重新生成一遍完整摘要，但Critic永远不满意。两边都在认真干活，整体却在空转。
排查后发现根因是Critic的验收标准太模糊。它只说了“信息不够完整”，却没有说明到底缺什么。后续给Critic的提示词加了一条强制要求：必须明确指出缺失的具体内容，并引用原文中的对应片段作为依据。这样Executor才能知道怎么改，Critic也更容易做出可验证的判断。另外一个有效做法是设置重试上限，达到上限后强制转人工，而不是无限循环。
4.3 上下文越滚越大，token成本飙升
多Agent系统的成本失控，通常不是单个模型贵，而是消息冗余导致token总量飞涨。每个Agent都有输入token，如果每个环节都把完整上游结果塞进去，两轮之后就可能破万。
解决措施就是前面提到的上下文摘要机制。另外要注意，所有Agent之间的消息都应该走摘要层，而不是原始输出。我自己在项目中统计过，加上摘要机制后，同一个多步任务的token消耗减少了大约40%，效果非常明显。
4.4 结构化输出解析失败
大模型输出JSON偶尔会带上额外的解释文字，比如在JSON前后加上“json”标记，或者直接输出“好的，这是结果：{...}”。解析时如果严格执行json.loads()，非常容易抛异常。
我写了解析函数做了三级容错：先剥离代码块标记；再尝试直接解析；失败后用正则截取最外层的{...}结构。这三步处理下来，解析成功率能到99%以上。
import json, re def parse_json_response(text): text = text.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text) try: return json.loads(text) except json.JSONDecodeError: match = re.search(r"\{.*\}", text, re.DOTALL) if match: return json.loads(match.group(0)) raise ValueError("Invalid JSON response")4.5 状态不同步导致任务丢失
还有一次比较棘手的bug：系统偶尔出现“任务明明执行完了，但最终结果里缺少某个子任务输出”的情况。排查后发现是并发调度时，多个子任务同时完成，回调函数里更新共享的task_results字典时产生了竞态条件，后写入的结果覆盖了先写入的。
解决办法是给整个状态更新区域加锁，或者改用线程安全的ConcurrentDict结构。另外还建议在每条子任务结果保存时带上message_id和时间戳，方便追踪“最后写入者是谁”。
结尾：一点个人体会
把agency-agents这套跑通，并改造出适合自己业务的版本，前后花了我两周多时间。最大的体会是：多Agent系统真正的难点不在单个Agent的提示词怎么写，而在整个系统的状态流转和消息协议设计。只要状态机清晰、消息结构统一、每个角色的输入输出被严格约束，即便大模型偶尔“发挥失常”，你也能在状态层面快速定位问题。另一个让我印象深刻的点是“约束导向设计”——多Agent场景下，给模型太多自由不是好事，精确到格式、期限、风格的约束，反而是让系统稳定输出的关键。如果你正在考虑做Agent团队，我的建议是先从3个角色起步：一个规划者、一个执行者、一个质检者，把这三个角色之间的消息链路打磨顺了，再逐步扩展角色数量和工具集，会稳得多。
