多Agent协作框架agency-agents:如何让大模型团队有序工作

发布时间:2026/10/24 17:13:47
多Agent协作框架agency-agents:如何让大模型团队有序工作

1. 为什么我会做一个叫 agency-agents 的智能体协作框架

事情得从一次让我印象深刻的翻车现场说起。当时我在某个内部演示项目里用多个大语言模型 Agent 并行处理一批文档:有的负责摘要、有的负责分类、有的负责提取关键词。理想很丰满,结果却是一场混乱——三个 Agent 同时抢着改同一份 JSON 输出,一个覆盖另一个;负责摘要的 Agent 把分类结果也写了进去;下游流程根本不知道该信谁。我盯着日志看了半小时,唯一确认的事情是:把一堆通用 Agent 丢在一起,并不会自动产生协作,只会产生事故。

后来我一想,问题的本质不在模型能力,而在于我压根没有设计“协作结构”。现实世界里的公司能顺畅运转,靠的不是所有人聪明,而是岗位明确、流程清楚、文件交接有规范。于是就有了 agency-agents 这个实验项目——它不是又一个“多功能 Agent 全家桶”,而是一套把大模型任务编排成“虚拟小型团队”的运行框架。所谓 agency,就是那个接单并派单的“调度中枢”;所谓 agents,则是各司其职的“岗位员工”。

这套框架解决的是多 Agent 场景里最常见的三类痛点:任务分配靠猜、上下文互相污染、结果无人验收。如果你正在做 RAG 问答、批量文档处理、报表生成、网页信息抽取这类需要多个步骤协作的 AI 应用,而且已经厌倦了“写一堆 if-else 把不同模型输出拼在一起”的做法,那后面这些设计思路和踩坑记录应该能帮你少走不少弯路。

需要先说明一点:这篇文章讲的是我基于常见工程实践做的个人项目方案,不是某个标准化产品的官方文档。具体的技术选型你可以替换,但底层的协作思想和流程设计是通用可复用的。

2. 架构设计:把“一堆 Agent”变成“一家虚拟公司”

2.1 三个角色:agency、agent 与 tool

我第一版代码其实极其简单,就是一个 Python 类里面塞了 N 个函数,每个函数调用一次大模型接口。结果项目稍微一复杂就撑不住了。后来我反复重构,最终稳定下来的核心抽象只有三个:agency、agent、tool。

  • agency 是整个系统的大脑。它不直接处理业务细节,只负责接收任务、拆解任务、分派给合适的人、汇总结果。它自身不调用大模型做普通执行,而是通过一套规则和状态机来管理任务流转。
  • agent 是具体干活的执行单元。每个 agent 有明确的角色定义、职责范围、可用工具列表,以及自己的系统提示词。它不需要知道整个项目长什么样,只需要知道自己的岗位是什么。
  • tool 是 agent 能操作的外部能力,比如检索数据库、调用搜索引擎、读写文件、运行 SQL。tool 层把“大模型只能输出文字”这件事遮住了,让 agent 能真正对世界产生作用。

这个三件套的设计是参考了真实公司的运作方式。agency 是老板,agent 是员工,tool 是员工手上的工具。老板不需要掌握每个岗位的技能,员工不需要知道公司战略的全貌,工具需要申请才可以用。这样一层一层隔离下来,系统的复杂度就控制住了。

2.2 岗位分工的具体配置长什么样

我用 YAML 文件来定义每个 agent 的“岗位说明书”。这样做的直接好处是,调整一个角色的行为不需要改代码,改配置文件就行。下面是一个最小角色配置的例子:

roles: - name: content_writer description: 负责根据资料大纲撰写完整章节内容 system_prompt: | 你是一名技术领域的资深内容编辑。 你只负责写作,不负责查资料、不负责排版、不负责审核。 你的输入是资料卡片和章节大纲,你的输出只有一种格式:Markdown 正文。 如果资料不足,你需要在输出中显式标注[缺资料],不要自行编造。 tools: - read_file - write_file temperature: 0.7 max_retries: 2 - name: fact_checker description: 负责校验内容中的事实与数据引用是否准确 system_prompt: | 你是一名严谨的审核员。 你逐条检查输入内容里的数据、日期、引用来源。 你不对内容做润色和扩写,只输出一份问题清单。 每条问题必须包含:原文位置、问题描述、建议修改方式。 tools: - search_web - read_file temperature: 0.1 max_retries: 3

这里有两个细节值得注意。第一个是 temperature 的设置——写作类 agent 我用 0.7 保留一定表达灵活性,审核类 agent 我用 0.1 让它尽量稳定和保守。第二个是工具权限的限定——“审核员”有联网搜索工具而“写手”没有,从根上避免执行环节因为随手查资料而跑偏。

我给每个 agent 的 system_prompt 里都强制规定了输出格式和“不做什么”。在实际测试里,告诉模型“不要做什么”往往比只告诉“要做什么”更有效,因为大模型默认会倾向于多做一步。比如写手如果没被限制,会很自然地顺带去查资料、顺带做排版,然后整个流程就乱套了。

2.3 任务卡片:员工之间唯一的沟通语言

多个 agent 之间怎么传东西?我的答案很简单:不直接传。任何两个 agent 之间不互相聊天,所有信息都放在一张结构化的“任务卡片”(TaskCard)里,经过 agency 转交。

TaskCard 是这套系统的命根子,我用一个 Pydantic 模型来定义它的结构:

from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any from enum import Enum class TaskStatus(str, Enum): PENDING = "pending" IN_PROGRESS = "in_progress" NEED_REVIEW = "need_review" APPROVED = "approved" FAILED = "failed" class TaskCard(BaseModel): task_id: str = Field(description="全局唯一的任务标识") parent_id: Optional[str] = Field(default=None, description="父任务ID,用于任务拆解溯源") role: str = Field(description="当前负责的agent角色") status: TaskStatus = TaskStatus.PENDING input_payload: Dict[str, Any] = Field(default_factory=dict, description="该角色的输入数据") output_payload: Optional[Dict[str, Any]] = Field(default=None, description="该角色的产出") error_info: Optional[str] = Field(default=None, description="失败原因") created_at: str = Field(description="创建时间") updated_at: str = Field(description="最后更新时间")

你可以把 TaskCard 想象成工厂里的流转工单。每个工人不管上一个工序是谁做的,只看工单上写了什么。这样做有三个好处:第一是审计容易,出了问题直接查工单历史;第二是上下文干净,每个 agent 只处理自己该看的那一段数据;第三是流程可以暂停和恢复,因为每一张卡片的运行状态都清清楚楚。

3. 核心流程:从拆解任务到汇总结果

3.1 任务拆解的“三步法”

agency 接到一个初始任务后,不会把它直接丢给某一个 agent。它先做拆解。这一步在早期版本里我完全依赖大模型,后来发现不可靠——模型经常会拆出重复的子任务,或者把任务拆成十几个小得没法执行的操作。所以我改成“固定三步法”:

第一步,把原始任务做结构化描述。用一个独立的“规划 Prompt”让大模型输出任务的目标、范围、输入资源、输出产物、约束条件。这一步的输出是纯文本描述,不执行任何操作。

第二步,把结构化描述转成任务依赖图。我规定拆解出来的一级子任务只能有三种关系:串行(前一个完成才能开始下一个)、并行(互不依赖)、汇总(每个子任务的结果合并成最终产物)。我在提示词里要求模型输出的子任务必须同时带上依赖类型和前置条件。

第三步,将任务依赖图转成 TaskCard 集合。每个子任务生成一张 TaskCard,记录 role、parent_id 和输入输出约定。到这里,任务的执行就能交给调度器了。

一个实际的内容生产项目拆解结果大致长这样:

原任务:产出一篇某个开源工具的上手指南 1. 资料收集(并行执行) - 角色:researcher,工具:search_web、read_file - 产出:资料卡片集合 2. 大纲编写(串行,依赖步骤1) - 角色:outline_planner,工具:read_file - 产出:章节大纲和每章写作要点 3. 章节撰写(并行,依赖步骤2) - 角色:content_writer,工具:read_file、write_file - 产出:各章节Markdown初稿 4. 事实审核(串行,依赖步骤3) - 角色:fact_checker,工具:search_web、read_file - 产出:问题清单 5. 修订合成(串行,依赖步骤3和4) - 角色:content_writer,工具:read_file、write_file - 产出:最终完整Markdown文档

实际上,我测试过让大模型自己随便拆任务和按这个三步法拆任务的差距。前者经常会拆出“调研竞品”“整理知识体系”这种没法落地的模糊任务,后者则相对可控很多——因为结构化描述先行降低了模型自由发挥的空间。

3.2 状态机:把流程推进变成可靠的事

每张 TaskCard 都在六个状态之间流转:pending、in_progress、need_review、approved、failed。调度器本质上就是一个循环:扫描所有非终态的卡片,找到满足依赖条件的,交给对应的 agent 执行,执行完毕更新状态。

这里有个特别容易踩的坑:我在一开始用的是“按顺序执行列表”的简单循环,结果一旦某个任务失败,后面所有任务全部阻塞,而且失败任务还不会被自动重试。后来我把调度器重写成了基于状态机的版本,核心逻辑大约是下面这样:

def run_loop(self, cards: List[TaskCard]): while True: active = [c for c in cards if c.status not in (TaskStatus.APPROVED, TaskStatus.FAILED)] if not active: break for card in active: if not self.dependencies_met(card, cards): continue if card.status == TaskStatus.PENDING: self.dispatch(card) # 分派给角色并更新为 in_progress elif card.status == TaskStatus.IN_PROGRESS: self.collect_result(card) # 查询执行结果并更新状态 elif card.status == TaskStatus.NEED_REVIEW: self.run_review(card) # 调用审核角色,通过则置为 approved

加了状态机之后,系统的行为变得可预期了。即便某个子任务连续重试失败,也只是那一张卡片变成 failed,其他卡片不受影响。而且因为每张卡片的中间状态都落库,我可以在任何时间点把系统停下、查清楚当前卡在哪儿,再决定是人工介入还是调整配置重新跑。

3.3 上下文隔离与共享:只给该看的人看该看的事

多 Agent 系统里最贵的往往不是 API 调用次数,而是上下文长度。我见过很多人设计协作系统时,习惯性把前面所有 agent 的输出一股脑塞给后面的 agent——“你不是要审核吗?这个文档全文给你,你自己看”。这个做法在任务量小的时候无所谓,一旦文档变长,费用会以肉眼可见的速度飙升。

agency-agents 的做法是显式定义“输入数据清单”。每张 TaskCard 的 input_payload 里,由上游 agent 只写入下游真正需要的那部分。比如 fact_checker 审核的时候,并不需要拿整篇文档,它只需要拿到“涉及数据、引用的句子集合”就足够了。

具体实现上,我在每个 agent 完成工作后加了一个“输出裁剪”步骤,用一个小模型或者规则把输出中跟执行无关的思考过程、过程性描述去掉,只保留结构化的结果字段。这个裁剪在实践里大概能省掉 20% 到 40% 的 token 成本,算是一个性价比极高的优化点。

4. 实战中的坑:三个月跑下来,我踩过的五个典型问题

4.1 多个 Agent 同时写同一个文件:输出互相覆盖

第一次跑完整流程的时候,两个并行的 agent 都被分配了“write_file”工具,它们同时往同一个临时目录写结果文件,然后互相覆盖,最后拿到了一份两个章节混在一起的混乱文档。这个问题排查起来其实不难,但从根源上解决就只有两条路:要么给每个任务分配一个全局唯一的输出路径,要么把“写文件”改成“返回结构化内容”,由 agency 统一落盘。

我最终选择了第二种方案。因为让 agent 自己管理文件路径,本质上就是把架构职责下放给了大模型,而模型处理文件系统这种精确操作并不可靠。现在 framework 默认只允许极少数角色持有 write 类工具,其他角色一律通过返回值提交结果。

4.2 上下文越来越大,费用失控

跑一个 20 章的电子书生成项目,跑到第 12 章的时候单次调用已经要传 5 万多 token。我当时的第一反应是“换个更大的上下文窗口”,但后来仔细一想,这治标不治本。真正的解决办法还是前面提到的输出裁剪和按需传递。我专门做了一个“资料卡片”数据结构,把每一章写作需要的参考资料压缩成结构化卡片,而不是把原始链接和全文都塞进去。这样做之后,单章写作的上下文稳定控制在 8000 token 以内,费用直接下降了一个数量级。

4.3 重试逻辑导致的连环重复收费

早期我天真地给每个 agent 设置了 max_retries=5,想着“失败就多试几次”,结果某次调研类任务因为检索接口不稳定,连续重试了 5 次都失败,但前几次其实已经成功写入了部分结果。费用花了 6 份,任务却标记为 failed。后来我改成:写操作只允许执行一次,不自动重试;只有读操作(搜索、查库)才允许重试,而且重试前必须做幂等校验。这套规则说起来简单,但带来的成本节省是很直接的。

4.4 提示词里的“自由发挥”导致角色错位

写手 agent 经常越界去做审核员的事,审核员也偶尔忍不住“帮”写手改一段话。我一开始觉得这是模型不听话,后来反思才发现是我在 system_prompt 里没有把这些边界说清楚。现在每个角色的提示词里都有一段固定格式的“职责边界声明”,明确列出哪些操作绝对不允许做。同时我在工具层也做了限制,审核员根本不持有 write_file 工具,想越界也没有手段。

4.5 状态同步丢任务:没有唯一 ID 的教训

有一个版本我用的是可读性的任务名(比如“write_chapter_3”)作为主键,结果当系统支持并行执行时,两个完全相同的任务名互相覆盖了状态。这个问题直接导致整个流程跑完了,某几张卡片还是 pending 状态。从那以后我坚定不移地使用 UUID 作为 task_id,可读性的名称只放到描述字段里。永远不要让自然语言字符串充当系统里的唯一性约束,这是我学到的一条铁律。

5. 效果复盘与后续演进方向

5.1 一组可量化的对比数据

为了验证这套框架的价值,我用同一个内容生成流程做了一组对照:一边是传统的“单 Agent 长上下文”方案,一边是 agency-agents 的多 Agent 分工方案。任务是用同一批资料生成一份包含四个章节的产品分析报告。

对照组的结果是:首次生成约 18 分钟,API 总费用约 7 个计价单位,由于上下文过长导致中途重试了一次,最终报告里有两处事实错误没有被发现。agency-agents 版本的结果是:首次生成约 11 分钟,API 总费用约 5 个计价单位,全程没有出现因上下文超长导致的失败,并且因为专门配了一个审核 agent,两处事实问题在成稿前就被标记了。当然这个对比算不上严格意义上的科学实验,只能说明在类似规模的任务上,结构化协作并不会拖慢速度,反而在质量和稳定性上有优势。

5.2 当前局限:还做不到全自动无人值守

我必须坦白:这套系统不是万能的。它目前处理得最好的是“流程明确、步骤可拆解、结果可校验”的任务。一旦面对开放性强、目标模糊的任务,比如“帮我设计一套完整的架构方案”,拆解这一步本身就成了瓶颈。模型拆出来的子任务往往缺乏创造性连贯性,产出的结果像是拼凑出来的。另外,审核环节目前依赖模型互相审,它只能发现局部事实错误,对“整体表达好不好”这种主观质量几乎没有判断力。

所以现在的使用方式仍然保留着关键点人工确认。特别是在任务拆解完成之后,我都会把任务依赖图打印出来看一眼再去执行——这一步只需要几十秒,但能避免大量浪费。把拆解结果可视化出来检查,是我目前最推荐的一个使用习惯。

5.3 下一步:评审闭环和增量执行

接下来我的方向是给所有的 need_review 状态增加一个“评审反馈循环”。目前的审核是一个单次动作,审核员输出问题清单之后,修改后的文档不会再回到审核员手里确认一轮,这导致审核有时变成走过场。我计划把它做成最多两轮的闭环:修改完成后自动回到审核员,如果仍有严重问题则标记为人工处理。

另一个想做的事情是增量执行。现在每次重新跑流程都会从头开始,哪怕只有一部分资料变了。理想情况是 TaskCard 能记录每个子任务的输入指纹,当输入没有变化时,跳过执行直接复用之前的输出。这个优化一旦落地,对于迭代型的内容生产项目来说,省下的开销会非常可观。

最后说点个人体会。做 agency-agents 这几个月,最大的收获不是代码库本身,而是对“让多个大模型协作”这件事祛魅了。把所有功能塞进一个 Agent 里看起来很省事,但复杂任务一旦推进到某个规模,结构化的分工几乎是必然选择。而结构化的关键不在于设计多漂亮的抽象,而在于搞清楚每个角色到底能做什么、不能做什么,以及它们的交接物长什么样。这些道理放在人类组织里成立,放在 Agent 系统里同样成立。