
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(defaultNone, description父任务ID用于任务拆解溯源) role: str Field(description当前负责的agent角色) status: TaskStatus TaskStatus.PENDING input_payload: Dict[str, Any] Field(default_factorydict, description该角色的输入数据) output_payload: Optional[Dict[str, Any]] Field(defaultNone, description该角色的产出) error_info: Optional[str] Field(defaultNone, 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_retries5想着“失败就多试几次”结果某次调研类任务因为检索接口不稳定连续重试了 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 系统里同样成立。