去年我在做内部自动化工具的时候，第一次认真思考"agency-agents"该怎么落地。当时的需求并不复杂：给出一份杂乱的需求文档，让大模型把它拆成可执行的开发任务，再针对每个任务写出实现说明，最后整体自检一遍。听起来很常规，可单Agent模式跑了不到两周，问题就集中爆发了。后来我干脆把项目名改成"agency-agents"——一个像虚拟团队一样工作的多Agent协作层。这篇文章就把它的设计思路、核心代码和实测踩坑一次性讲清楚。

这个项目适合三类人：正在做AI应用开发、想尝试多Agent协作但不知道怎么组织的朋友；手里已经接了多个模型服务、想让不同角色各司其职的开发者；以及单Agent方案反复撞墙、想换一种协作方式的人。就算你暂时不写代码，光看架构思路，也会对"多Agent到底怎么协作"有一个比很多教程都更清楚的理解。

1. 为什么我会把多个Agent组织成"一家虚拟公司"

1.1 单Agent模式在复杂任务面前的四个信号

第一个信号是角色混乱。让模型"既当产品经理、又当架构师、还当测试"，它会在同一轮回答里一会儿改需求、一会儿写代码、一会儿给自己挑毛病，最后产出一锅乱炖。第二个信号是上下文稀释：任务越往后，最开始定义的目标和约束条件越容易被遗忘，我曾在第三轮之后反复看到它把"数据权限按团队隔离"这种硬需求漏掉。

第三个信号是错误难定位。一段错误的输出到底是在哪个环节被引入的，完全说不清，只能把整段历史翻出来人工核对。第四个信号是token浪费——每轮都要把全部历史对话重传一遍，成本随着轮次线性上涨，效果却逐步下降。

这四个信号背后其实是同一个原因：单个模型实例在同一个上下文里连续扮演多个角色，超过了它有效注意力能够覆盖的边界。打个比方，让一个人同时当项目经理、文案、财务和客服，忙起来必然顾此失彼；但如果你让四个人分工，每个人只需要记住自己那一亩三分地，整体反而更稳。我就是在这个对比里，决定把架构改成多角色协作的。

1.2 "agency"模式借鉴的是现实组织的分工逻辑

"agency"这个词的灵感，来自现实中的广告公司和咨询公司。它们很少让一个人从头干到尾：客户经理负责对接需求，策略团队出方案，创意团队做执行，最后还有一层质检。员工各管一摊、交接标准明确、有人对总结果负责。这套逻辑搬到Agent上，就是三件事：角色边界清楚、交接物标准化、调度与质检集中。

所以我把这个项目定位成一层"编排层"，不是一个大模型应用。它不关心你接的是哪个商用模型接口，也不替Agent写业务知识，它只负责几件小事：定义角色、派发任务、传递上下文、回收结果。项目因此很轻，核心代码没有动辄几千行，却把"多个Agent怎么协作"这个最常见的问题管住了。

在动手前我也想了很多：市面上已经有那么多Agent框架，自己再造一遍轮子有意义吗？答案是看你想解决的问题是什么。如果你只想快速搭一个能对话的Agent，现有框架够用；但如果你的痛点是"多个角色如何稳定交接、任务状态如何可审计、模型服务如何按角色分配"，那你就需要这样一层轻量编排逻辑。我的判断是，永远别让框架替你决定协作模式，你自己的业务才该决定协作模式。

2. agency-agents的核心架构：注册中心、任务调度与协作上下文

开始写代码之前，我先把整个系统的运作模式拆开——不是画漂亮的流程图，而是明确"谁手里有什么数据、谁对什么负责"。最终沉淀下来四个模块，下面这张表是我在项目文档里反复改了四五版才定下来的。

Registry解决"有哪些角色、各自能干什么"，Dispatcher解决"某件事该找谁"，Context Store解决"上个角色说了什么、下个角色该知道什么"，Worker解决"最终调用模型干活"。四者缺一不可。最容易被忽略的是Context Store——很多人以为多Agent就是多个模型互相传消息，忽略了上下文本身需要统一管理，结果协作起来到处丢信息。

2.1 Agent的能力声明：让调度器"看懂"每个角色

为了让Dispatcher不写死逻辑，每个Agent在上线前都要提交一份能力声明。我用的格式是YAML，既好读，又方便直接做成配置文件。

name: requirements_analyst role: 需求分析员 description: 将杂乱需求文档整理成结构化任务清单 capabilities: - requirement_analysis - task_decomposition accepts: type: raw_document max_size: 20000 emits: type: task_plan schema: task_plan_v1 model: provider: generic_commercial_llm temperature: 0.2 max_retries: 2

这份声明里几个字段很关键。capabilities是给调度器做匹配用的；accepts和emits定义了交接物的类型和格式，这是"标准化交接"的基础；model允许不同角色用不同模型服务，比如需求分析用便宜快速的模型就够，质量检查用能力更强的模型效果更好；max_retries是执行单元的重试上限。新增一个Agent不需要改调度器代码，只需要在注册中心登记一份这样的声明，这是整个项目可扩展性的来源。

为什么用YAML而不是在代码里写死？因为能力声明本质上是一份"接口文档"。把接口从代码里抽出来，意味着你可以用配置驱动的方式动态加入新角色。我在后来接入一个数据清洗Agent时，就是靠加一份声明文件完成的，没有动一行调度代码。这种"配置优先"的思路，会省掉后期大量的改造成本。

2.2 Job对象与状态机：让任务"干净"地交到下一个人手里

每次任务进来，Dispatcher会生成一个Job对象。Job不只是消息，它携带完整状态。

{ "job_id": "job_7f3a", "trace_id": "trace_8e21", "type": "task_plan", "state": "pending", "payload": {}, "result": null, "attempts": 0, "assigned_to": null }

state字段是我反复强调绝对不要省略的部分。它让每个任务在任意时刻都是可审计的：pending表示等待派发，running表示有Agent在处理，done表示完成，failed表示执行失败，needs_review表示结果需要人工介入。显式状态机带来的好处是，你可以随时回答"这个任务现在到底怎么样了"，排错时一眼就能看清停在哪一步。

很多多Agent项目做得乱，就是因为任务在Agent之间传来传去，却没有任何状态记录，出了问题只能靠猜。我甚至要求所有环节都只能通过Job状态来判断下一步动作，禁止任何Agent"看到对方返回了就顺手改别人状态"这种旁路操作。状态只属于Job本身，这算是这个项目里最有用的一条纪律。

2.3 任务路由与协作上下文：Dispatcher怎么决定"下一个环节"

Dispatcher在收到Agent完成事件后，会读取Context Store里该Job的最新产物类型，再根据一张简单的路由表决定下一步。

ROUTE_TABLE = { ("requirements_analyst", "done"): "solution_designer", ("solution_designer", "done"): "quality_checker", ("quality_checker", "done"): "final", ("quality_checker", "needs_review"): "human_gate", }

路由表的key是"上一个Agent加它的结束状态"，value是"下一个处理者"。这种写法直观且容易改：你想在中间加一道人工确认，就加一行路由；你想让质量不合格的方案退回设计师而不是走向终稿，就改一个映射。注意这里没有写任何业务逻辑，只有"谁完成后交给谁"。业务判断发生在Agent内部，流程控制发生在路由表里，各管一段，改起来互不干扰。

3. 关键模块的代码实现与设计取舍

架构想清楚之后，写代码反而很快。核心类就两个：Agent基类和Dispatcher调度器，外加一个上下文存储模块。我把关键实现和取舍逻辑放在这一节。

3.1 Agent基类：封装调用、重试与错误隔离

先看基类代码。我用Python实现，核心思想是"把调用大模型的通用流程沉淀在基类里，子类只负责写提示词和结果解析"。

import json from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, config, context_store, llm_client): self.name = config["name"] self.config = config self.context_store = context_store def handle(self, job): state = "running" attempts = job.get("attempts", 0) try: payload = self.context_store.read(job["job_id"], "in") prompt = self.build_prompt(payload, job) result_raw = self.llm_client.chat( messages=prompt, temperature=self.config["model"]["temperature"], max_tokens=1024, ) result = self.parse_result(result_raw) self.context_store.write(job["job_id"], "out", result) state = "done" except Exception as e: attempts += 1 if attempts >= self.config.get("max_retries", 2): state = "failed" self.context_store.write(job["job_id"], "error", str(e)) else: state = "pending" finally: job["state"] = state job["attempts"] = attempts return job @abstractmethod def build_prompt(self, payload, job): ... @abstractmethod def parse_result(self, result_raw): ...

基类里做了三件事：读上下文、调模型、回写结果。比较重要的设计决策，是把错误处理和状态流转都放在基类里。这样每个Agent子类只需要专注写自己的提示词和解析函数，重试、异常、状态更新这类横切逻辑不会散落在各个Agent中。错误隔离也体现在这里：一个Agent抛异常不会连累别的任务，任务只会变成failed，或者回到pending等待下一次重试。

这里有一个容易踩的细节：重试逻辑放进基类之后，要小心"无限重试"问题。我在早期版本没有设置上限，结果网络故障时一个Job能被重试几十次，下游收到一堆重复结果。所以max_retries字段从一开始就写进能力声明，把重试次数当成一种资源，按需配置。

3.2 调度器：拉模型与推模型之间的折中

调度器我试过两种思路，都有明显缺点。

推模型是Dispatcher直接调用Agent的handle方法，代码简单直接，但问题也很明显：一个Agent处理慢会把整个调度线程卡住；Agent内部的重试会让调度器被迫空等；测试时想单独模拟某个Agent的行为，必须把整条链都跑起来。

拉模型是Agent处理完就主动向Dispatcher申请下一个任务，好处是天然有背压——Agent忙就不会积压新任务，系统不容易被打爆；坏处是调度逻辑被拆散到各个Agent里，协作链路不好观察，出了问题更难定位。

我最终采用的是折中的队列模型：Dispatcher负责匹配任务，并把Job投递到对应Agent的队列中；Worker空闲时从自己的队列里拉取。调度策略集中在Dispatcher里，执行节奏又掌握在每个Worker手里，两头兼顾。投递动作是原子的，同一个Job只会进入一条队列，不会出现两个Worker同时抢任务的情况。

import queue class Dispatcher: def __init__(self, agents, route_table): self.agents = agents self.route_table = route_table self.queues = {name: queue.Queue() for name in agents} def submit(self, job): target = self.agents.get(job.get("type")) if target: self.queues[target].put(job) def on_agent_done(self, job): key = (job.get("assigned_to"), job.get("state")) next_agent = self.route_table.get(key) if next_agent: job["assigned_to"] = next_agent self.queues[next_agent].put(job) else: job["state"] = job.get("state", "done")

实际应用时，我在on_agent_done里还加了路由日志：每个Job每经过一次路由，就追加一行"从X到Y，状态Z"的记录。调试的时候把这一串日志拉出来，基本就是整个协作链路的骨架，执行过程一目了然。

3.3 上下文存储的三个设计原则

多Agent系统里最容易被低估的是上下文管理。踩了几次数据错乱的坑之后，我定下三个原则。

第一个原则是命名空间隔离。每个Job有自己的独立命名空间，不存在"全局共享变量"；Agent只能读写自己这个Job的数据，碰不到别人的。第二个原则是结构化交接。Agent之间传递的是结构化字段，比如{"task_plan": {...}}，不是一大段散文式对话历史；散文没法被下一环节稳定解析，字段可以。第三个原则是交接前压缩。Agent在回写结果时会生成一份摘要，下个Agent优先读摘要而不是原文。

class ContextStore: def __init__(self): self._data = {} def write(self, job_id, key, value): self._data.setdefault(job_id, {})[key] = value def read(self, job_id, key): return self._data.get(job_id, {}).get(key) def snapshot(self, job_id): return dict(self._data.get(job_id, {}))

实现本身不复杂，难的是坚持这几个原则。我见过很多多Agent实现，上下文就是一个全局变量，所有Agent往里塞对话历史，最后模型被一坨互相矛盾的信息搞晕，质量崩盘。宁可少传、传精华，也不要多传、传垃圾——这句话我写在项目文档第一行。

4. 跑通第一个多Agent协作Demo的完整过程

讲完设计理念，用一个实际Demo说明整个链路。目标：让三个Agent协作完成"把一段需求描述变成一份产品方案，并完成自查"。

4.1 最小配置：三个角色加一份任务清单

我先定义了三个角色：requirements_analyst负责读需求原文、输出结构化任务清单；solution_designer根据任务清单输出产品方案；quality_checker负责检查方案，输出问题清单和修订建议。三个角色用YAML声明注册，再启动Dispatcher。整个初始化流程就是"读取配置、注册Agent、开启队列、等待任务提交"四步。这一步我建议不要贪多，三个Agent已经足够跑通机制，角色越多越难排查。

这次Demo提交的需求是一段真实场景描述："做一个面向团队成员的周报汇总工具，支持自动收集每个人提交的周报，并按项目维度和时间维度生成汇总视图。要求操作简单，数据权限按团队隔离。"

4.2 从提交任务到拿到成品的完整链路

任务提交之后，Dispatcher根据type字段把它路由给requirements_analyst。整个过程一共五步：

1. requirements_analyst收到raw_document，输出结构化任务清单。
2. Dispatcher看到结果是task_plan，将其作为输入路由给solution_designer。
3. solution_designer生成产品方案，包含信息架构、核心页面、数据权限说明。
4. Dispatcher把方案交给quality_checker做自检。
5. quality_checker输出问题清单；若问题严重则退回设计师修订一次，否则输出终稿。

我实测跑完这条链路，产物质量比单Agent直接生成高出不少。尤其"数据权限按团队隔离"这条需求，三个角色接力之后被完整保留，而单Agent方案在后期经常把这类约束忘掉。这其实不是巧合：每个角色只负责一个环节，约束作为结构化字段被显式携带，自然更不容易丢。

4.3 调试多Agent系统的三板斧

多Agent系统调试起来和单Agent完全不同。我摸索出三个办法。

第一个是给每个任务分配trace_id，全链路日志都带上它，排查时一条命令就能拉出完整执行轨迹。第二个是上下文快照：每完成一个环节，把Context Store里该Job的命名空间整体导出，人工检查中间产物是否合理，比如看看任务清单有没有漏掉关键需求。第三个是重放：把某一份快照喂给单个Agent，单独验证它的处理逻辑，避免每次都要跑完整链路才能复现问题。

这三个办法配合Job状态机，基本解决多Agent系统里九成的问题："任务走到哪了""这个Agent拿到了什么输入""它输出了什么""哪个环节出的错"。建议你在自己的项目里尽早把这些观测能力做进去，别等到出问题再补，那是真的来不及。

5. 实测中踩过的坑与优化经验

这个项目跑了几周之后，我从"能跑"进入"跑得稳"的阶段。这一阶段踩的坑最有价值，下面三个问题值得展开讲。

5.1 链路加深导致的质量滑坡与控制手段

最开始我把协作链设计得很长：需求分析、代码设计、测试用例、文档生成，四五个环节接力。结果发现环节越多，最终结果反而越差。原因不复杂：每一跳的模型都在"转述"上一环节的内容，转述过程中信息必然损耗；环节越多损耗越大，最后只剩下一些表面内容，真正关键的细节反而丢了。

我用的控制手段是"交接物压缩"和"关键约束显式携带"。每个Agent回写结果时，除了结构化产物，还必须把上游明确的关键约束写进constraints字段，原样传给下一环节。这样一来，中间步骤的信息损耗再大，最核心的约束也不会丢。同时我把链路从五跳压成三跳，能合并的环节直接合并，牺牲一点角色的单一职责，换回整体质量的大幅提升。

5.2 上下文膨胀与"遗忘"之间的平衡

第二个坑正好和第一个相反。有时候为了让信息不丢，我让Context Store把上游所有内容都传给下一个Agent，结果上下文飞速膨胀。一个Job处理到第三环节时，传给模型的token量已经是初始的几倍，响应变慢，模型对关键信息的注意力反而被稀释。信息不丢和信息可用，完全是两回事。

我的解决方案是分级记忆：Agent的输入分成必读字段和参考字段。必读字段是结构化关键信息，直接进Prompt；参考字段可能很大，只给摘要，Agent需要时再按需读取。这也逼着我在设计交接物时更克制——真正需要传给下一环的，通常就那么几个字段，其余都是噪音。配置好必读和参考分类之后，上下文体积基本稳定在初始的一倍多一点，质量也没再因为"信息太多"而下滑。

5.3 死循环、重复执行与并发风暴

多Agent系统跑起来之后，还会出现三类"失控"问题，我每个都遇到过至少一次。

死循环：某Agent发现结果不合格，把Job退回上一环节，上一环节重新处理后结果还是不合格，又退回来，循环往复。我加了最大回流次数限制，默认1次，超过就把Job标记成needs_review，转人工处理。退回机制一定要画上限，否则模型偶尔的"坚持己见"就会让整条链卡住。

重复执行：同一个Job因为网络超时被重试，结果下游收到两份几乎一样的产物。解决方式是给Job加幂等键，下游在写Context Store之前，先检查同样的job_id加产物类型是否已经存在。宁可丢弃一个可疑请求，也不要让重复产物污染下游。

并发风暴：多个Worker同时向同一个模型服务发起高并发请求，触发限流，反而拖慢所有任务。我在每个Worker外加了信号量限制同时进行的任务数，后来又加了一个简单的熔断器，连续失败超过阈值就暂停新任务。别看这两个机制简单，它们能保证系统在模型服务不稳定时，不会自己把自己拖死。

6. 项目后续还能怎么扩展

把基础链路做稳之后，我在几个方向上做了扩展，效果都不错，列出来供你参考。

6.1 人工介入节点：让系统在该停的时候停下

性价比最高的扩展是加人工介入节点。我在Dispatcher里加了一类gate节点，任务流到某个环节时先挂起，等人工确认再继续。比如quality_checker给出needs_review时，系统会带着问题清单请求人工裁决，而不是闷头自己改。多Agent系统最怕的不是出错，而是"信心满满地错"。人工闸门就是在关键分歧点上叫停，加了这个之后，长链路任务的最终质量明显提升。

6.2 成本追踪与角色-模型解耦

第二个值得做的扩展是成本追踪。每个Agent调用模型时记录token数和耗时，汇总到每个Job的成本账单里。有了账单，你能直观发现哪个环节最贵、哪次跑链路是无谓浪费。再往后，我给不同角色配置了不同的模型服务：需求分析用便宜快速的模型就够，质量检查用能力更强的模型效果更好。这其实是能力声明里model字段带来的自由度——角色和模型解耦，按需分配性价比。

6.3 给新手的最后几条建议

如果让我给打算动手的人几句实在话：别一上来铺开十个Agent，先用两三个角色把一条最关键的链路跑通，把状态机、上下文交接、质量检查这三件事做扎实，再谈扩展。我在实际使用中发现，多数场景下三到五个角色已经能覆盖大部分需求；多Agent的价值从来不是数量，而是清晰的协作结构和可控的交接质量。对于一个Agent就能完成的任务，不要硬拆；只有当单一角色确实顾不过来时，分工才有意义。