去年我在做自动化流程改造时,碰到一个特别现实的问题:单独一个 AI 模型再聪明,也只能按提示词死板地执行单线程任务。真正要做的内容调研、报告生成、代码审查这类活儿,往往需要好几道工序来回配合。后来我把目光落在一个叫 agency-agents 的项目设计思路上——它不是某个特定产品的名字,而是一套把多个 AI 智能体组合成“数字化小团队”的工程模式。简单说,就是让每个智能体扮演一个业务角色,比如情报收集、方案起草、质量检查,然后通过统一的消息通道协作,最终输出一份完整结果。
我现在写这篇文章,就是想把这段实践完整沉淀下来:从概念拆解、架构选型,到一个能跑起来的最小系统,再到真实场景里的踩坑与排查。无论你是刚接触智能体开发的新手,还是已经写过几个 Agent Demo、想往工程化方向走的开发者,这份笔记应该都能给你一条可以照抄的路线。
1. 什么是 agency-agents:先搞清楚概念边界
1.1 名字拆开看:Agency 和 Agents 分别意味着什么
第一次看到 agency-agents 这个名字,大多数人会愣一下。Agency 在英文里有“代理机构”“中介”的意思,Agents 又跟中文“代理”撞车,两个词叠在一起特别容易绕晕。
我自己的理解是:这里的 Agency,重点不是“代理”的接口含义,而是“组织、机构、能动性”的混合体。它强调的是一群 Agent 出于共同目标聚在一起,各自承担职责,像一个微型组织那样运转。Agents 则指那些具备自主决策能力的智能体,它们内部通常有一个大语言模型作为推理内核,负责理解目标、拆解动作、判断要不要调用工具、生成最终内容。
所以 agency-agents 合在一起,最精准的翻译应该是“机构化智能体群体”或者“多智能体协作组织”。它回答的问题是:如果一个人形单影只,那我们就拉一队人,用工程手段让它们像公司部门一样配合。
1.2 它到底解决了什么问题
单智能体和多智能体之间,差的不是数量,而是能力边界。
第一个问题是任务延续性差。单次对话里,模型生成完一段内容之后,就不会主动往下推进了。它没有办法“过一会儿再回来检查结果”,更不可能在发现素材不足时自行去补一轮搜索。多智能体结构可以做到:规划者派活、执行者干活、检查者验收,不合格就退回重做,直到通过。
第二个问题是复杂度超过单提示词的承载能力。你把十步流程写进一个提示词里,模型大概率会在中间某一步开始丢三落四。但如果你把这些步骤分配给三个不同角色,每个角色只关心自己那一段,出错的概率就小得多。这跟软件工程里“模块职责单一”是一个道理。
第三个问题是结果无人把关。单一模型直接生成报告,经常出现数据过时、引用错误、逻辑跳跃。多智能体结构里专门安排一个质检角色,等于给流程加了一道人工评审的替代品,不依赖用户自己去发现所有问题。
你甚至可以把它想象成开一家小型内容公司:一个人既采访又写作还校对,效率低质量差;但如果分成记者、编辑、校对三个人,每个人都聚焦在自己的专业环节上,整体质量会立刻上一个台阶。
1.3 它适合用在哪些场景里
我现在判断一个需求要不要上多智能体,就看三个特征:是否有明确分工、是否有反复迭代、是否有独立验收动作。
比较典型的有四类场景:
- 内容调研与报告生成:先收集资料,再整理结构,然后撰写全文,最后核对事实。
- 客户工单处理:先理解用户诉求,再查知识库,答复前做风险判断,必要时转人工。
- 代码修复流程:先读异常栈,再定位代码位置,给出修改建议,然后执行测试验证。
- 数据分析流水线:先写清洗脚本,再跑统计分析,生成图表说明,最后输出业务结论。
这些场景有一个共同点:只靠一次模型调用远远不够,必须靠多条消息链路把任务层层推进,最终得到一个经过校验的产出物。
2. 设计一个能跑起来的 agency-agents 框架:核心模块选型
2.1 先决定编排方式:中心化、去中心化还是混合模式
开始写代码之前,我建议你先定拓扑结构。这不是理论洁癖,而是直接影响后续消息怎么转发、出问题怎么排查。
中心化编排是最容易上手的。系统里有一个主控 Agent,它负责任务分配、状态跟踪、结果回收,其他 Agent 收到指令后执行,跑完就把结果交回主控。优点是流程透明,好调试;缺点是主控会成为瓶颈,所有消息都要经过它中转。
去中心化编排则是每个 Agent 都直接给别的 Agent 发消息,靠消息总线或者共享黑板来协作。它的扩展性好,适合多角色自由讨论的场景,但调试难度高,很容易出现消息环回和状态不一致。
混合模式一般是这样:常态下主控负责调度,但遇到特定分支时,允许子 Agent 之间临时直接交互,比如研究员和数据分析师私聊获取某个指标,然后把信息带回主流程。我个人的建议是:第一个版本别去做去中心化,先老老实实用中心化,等你对协作模式有把握了,再放开其他路线。
2.2 把角色定义成“部门”而不是“工种”
在设计角色时,我的一个经验是:不要只考虑“这个 Agent 做什么动作”,要考虑“它代表哪个部门来做事”。
规划者岗位的性质,本质上是项目经理。它不负责具体的信息检索,也不负责最终撰稿,只负责把一个大目标拆成可执行小任务,并判断哪些任务可以并行。
执行者更像是业务骨干。例如内容策划、代码实现、数据分析、资料整理。这类 Agent 是实际产生产出的部分,需要至少掌握一个外部工具,比如搜索接口、代码执行环境或者数据库查询。
检查者负责质量审核。它的提示词应该明确列出验收标准,告诉它什么情况可以放行,什么情况需要退回修改。它的反馈不能只说“不行”,必须指出具体问题,并给出修正建议。
汇总者类似于编辑。它要把各执行者交付的内容拼成最终成品,处理格式、去重、润色,形成一个完整一致的输出。
把角色设计成部门还有一个额外的好处:升级时你只需要替换某个部门的能力,不需要推翻整个流程。比如想换执行策略,改一个 Agent 内部提示词就行,其他部分完全不受影响。
2.3 消息协议:让智能体之间说同一种语言
多智能体最忌讳的事情就是各说各话。一个 Agent 回复纯文本,另一个回复 JSON,主控还得临时解析各种稀奇古怪的格式。所以第一件事,就是定义一套统一的消息协议。
我常用的消息结构长这样:
{ "id": "msg_001", "type": "task", "from": "planner", "to": "researcher", "status": "pending", "payload": { "task": "收集过去三个月智能体开发框架的更新动态", "deadline": "2025-03-01", "context_ids": ["req_01"] } }字段一定要固定。id 是唯一标识,用于追踪;type 决定消息类别,是普通任务、审核反馈还是完成通知;from 和 to 决定路由方向;status 表示消息当前状态;payload 里放实际内容。
用 Python 定义的话,可以写成 dataclass:
from dataclasses import dataclass, field from typing import Any, Dict @dataclass class Message: msg_id: str msg_type: str sender: str receiver: str status: str = "pending" payload: Dict[str, Any] = field(default_factory=dict)统一协议之后,主控只管像邮局一样把消息从 A 路由到 B,不需要理解消息内部的具体业务含义。所有 Agent 也都用同一套对象处理入站和出站消息,逻辑变得非常干净。
2.4 工具和权限边界:不能把所有钥匙都交给智能体
设计多智能体系统时,最容易忽略但又最容易出事的地方,就是工具权限。
每个执行型 Agent 都会绑定若干工具,例如搜索引擎接口、内部文档库、数据库连接器、代码执行沙箱。但你要清楚:模型本身不知道哪些操作是安全的,它只是根据上下文推断“该用什么工具”。
我的建议是三条铁律:
- 只给最小必要权限。查询类 Agent 只给只读接口,写操作必须单独分配给特定角色,不能人人都有写库能力。
- 高风险操作必须加确认环节。比如删除数据库记录、发送邮件、修改生产配置,这类操作应该返回一个待确认状态,由主控或者人工审批后再执行。
- 工具结果要做校验。工具返回的内容不能直接当成事实,需要由检查者做二次确认。
这套权限设计,看着像限制了智能体的自由度,实际恰恰相反。边界清晰之后,智能体才能真正发挥能力,因为我们可以放心地把更复杂的任务交给它。
3. 手把手实现一个最小可用的 agency-agents 系统
3.1 环境和项目结构
这一节我们直接动手。我假设你本地已经有了 Python 3.10 以上的环境,并且可以在命令行里安装依赖。除了标准库之外,我们还需要一个能调用大模型接口的客户端,你可以根据自己实际情况选型。
项目目录我建议这样安排:
agency_agents_demo/ ├── main.py ├── agents/ │ ├── base.py │ ├── planner.py │ └── researcher.py ├── core/ │ ├── message.py │ └── scheduler.py └── tools/ └── search_stub.py把代码拆成 agents、core、tools 三个目录,长期维护起来会很舒服。agents 目录放角色逻辑,core 目录放调度和消息,tools 目录放外部工具接口。
3.2 先写消息对象和引擎基底
core/message.py 里就放前面提到的 Message 类。然后我们再补一个 MessageQueue,用来做简单流转:
from collections import deque class MessageQueue: def __init__(self): self.queue = deque() self.waiting_map = {} def publish(self, msg: Message): self.queue.append(msg) def consume(self) -> Message | None: if self.queue: return self.queue.popleft() return None def register_reply(self, msg: Message): self.waiting_map[msg.msg_id] = msg注意到这里我还留了一个 waiting_map,用途是记录某个请求有没有返回结果,方便排查丢失消息。
3.3 定义智能体基类
agents/base.py 里的职责很简单:接收消息、调用模型、执行工具、返回消息。
from abc import ABC, abstractmethod from typing import Dict from core.message import Message class BaseAgent(ABC): def __init__(self, name: str, system_prompt: str): self.name = name self.system_prompt = system_prompt def build_prompt(self, task: str) -> str: return f"{self.system_prompt}\n\n任务:{task}" @abstractmethod def handle(self, msg: Message, context: Dict) -> Message: pass这里有一点需要解释:我没有把模型调用直接写死在基类里。不同角色可能希望用不同参数,比如规划者用温度低一点的配置,保证稳定;创作者用适度随机的配置,保证多样性。所以模型调用放到子类里实现更合适。
3.4 实现规划者智能体和执行者智能体
规划者的核心工作,是把一个用户需求拆成多个子任务,然后分配出去。在这个最小示例里,为了让事情可复现,我先不依赖真正的大模型调用,而是用一个简化的规则实现:
from agents.base import BaseAgent from core.message import Message class PlannerAgent(BaseAgent): def handle(self, msg: Message, context: Dict) -> Message: user_request = msg.payload.get("task", "") subtasks = [ {"type": "research", "task": f"收集关于「{user_request}」的基础资料"}, {"type": "draft", "task": f"基于资料撰写初稿"}, ] return Message( msg_id="plan_1", msg_type="task_batch", sender=self.name, receiver="scheduler", status="ok", payload={"subtasks": subtasks} )执行者我们写一个 Researcher,负责搜索资料。这里工具调用我用的是一个 search_stub,只返回固定文本,方便演示:
class ResearcherAgent(BaseAgent): def handle(self, msg: Message, context: Dict) -> Message: task = msg.payload.get("task", "") result = search_stub.query(task) return Message( msg_id="res_1", msg_type="result", sender=self.name, receiver="planner", status="ok", payload={"research": result} )实际项目中,research 部分应该换成真实的搜索 API。但你注意这个骨架只需要改工具实现,不需要改 Agent 的协作逻辑,这就体现出分层的好处。
3.5 调度器:把消息在智能体之间转运
调度器是这个小系统的“邮局”。它从队列里取消息,根据消息里的 receiver 字段找到对应 Agent,把消息交给它处理,然后把返回结果再入队。一个简化版本是这样的:
class Scheduler: def __init__(self, agents: Dict[str, BaseAgent]): self.agents = agents self.queue = MessageQueue() def run(self, root_msg: Message, max_rounds: int = 10) -> str: self.queue.publish(root_msg) final_result = "" for _ in range(max_rounds): msg = self.queue.consume() if msg is None: break agent = self.agents.get(msg.receiver) if agent is None: continue response = agent.handle(msg, context={}) if response.receiver == "scheduler": # 如果是需要继续分配的子任务,就再入队 for sub in response.payload.get("subtasks", []): self.queue.publish(Message( msg_id=sub["task"], msg_type="task", sender="planner", receiver="researcher", payload=sub )) else: final_result = response.payload return final_result注意我这里用的是循环上限 max_rounds,这个非常关键。真实系统里防死循环要靠这个上限兜底,否则一旦 Agent 之间来回转发,程序就永远不会停下来。
3.6 跑起来看效果
组合主程序:
from agents.planner import PlannerAgent from agents.researcher import ResearcherAgent from core.scheduler import Scheduler agents = { "planner": PlannerAgent(name="planner", system_prompt="你是一个项目规划者。"), "researcher": ResearcherAgent(name="researcher", system_prompt="你是一个资料研究员。"), } scheduler = Scheduler(agents) root = Message( msg_id="root_1", msg_type="user_request", sender="user", receiver="planner", payload={"task": "新一代智能体框架的市场动态"} ) result = scheduler.run(root, max_rounds=10) print(result)运行后,你会看到任务从 user 到 planner,被拆成两个子任务,然后每个子任务又转给 researcher,最后由 researcher 返回搜索结果。虽然这个结果目前还很简陋,但完整链路的雏形已经出来了,后续所有高级功能都是在这个骨架上长出来的。
4. 真实场景:用三个智能体完成一次内容调研
4.1 把需求拆成可执行任务
理论说多了容易飘,我们把前面这套骨架压到一个真实用例里:做一个“智能体框架选型”的调研报告,目标读者是技术负责人。
原始需求拆出来是这样:
- 收集智能体框架的社区活跃度,包括 GitHub Star 增长、Discord 讨论量。
- 对比各框架的编排能力,是否支持多智能体协作。
- 输出一份选型建议,包含优缺点和适用场景。
这三个点互相之间有依赖,但又有明显边界。社区数据是一个方向,能力对比是另一个方向,最后选型建议要结合前面两个结果。正好对应执行者、分析者、汇总者三种角色。
4.2 角色和工具映射表
我给出一个工具与角色的映射,你照着搭即可:
| 角色 | 需要完成的职责 | 需要的工具 | 数量配置 |
|---|---|---|---|
| 规划者 | 拆解任务、设置优先级、调派资源 | 无外部工具,只靠模型推理 | 1 |
| 调研员 | 采集社区数据、整理原始事实 | 搜索接口、GitHub API、社区归档 | 2 |
| 分析员 | 对比框架能力、识别优劣 | 文档库检索、表格处理 | 1 |
| 审查员 | 核对数据、检查逻辑、退回修改 | 文件读取、元数据校验 | 1 |
| 汇总者 | 合并结果、润色语言、输出终稿 | 文档模板、格式化工具 | 1 |
注意我在这里给调研员配了 2 个实例。并不是所有 Agent 都只能有一个,有些角色因为工作量大,是可以多实例并行的。这也是多智能体系统比单模型有优势的地方。
4.3 任务流转过程
真实的流转过程比前面 demo 要丰富很多。我用文字描述一遍完整链路:
第一步,用户把需求提交给规划者。规划者判断这个问题可以拆成三条并行分支,于是发出三条 task 消息,分别给两个调研员和一个分析员。
第二步,调研员收到消息后各自去查数据。一个查社区增长,一个查框架文档,然后分别把结果发回规划者。规划者收集齐后,再发给分析员做能力对比。
第三步,分析员把对比结果整理成带结论的结构化数据,发给审查员。审查员检查发现某个框架的数据缺失,于是回了一条 feedback 消息给分析员,要求补数据。
第四步,分析员重新调用工具补齐,再次发给审查员,这次通过。审查员把通过结果发给汇总者。
第五步,汇总者把调研结果、分析结论、审查意见合并,形成最终报告。
这个流程里的每个步骤都留下了消息记录,所以哪怕中途出错,我们也能从日志里看到卡在哪个环节。
4.4 结果组装与验收
多智能体系统的最后一步,一定不能省略。汇总者输出初稿之后,要再过一道程序化检查,比如:
- 每个论证是否都能追溯到原始调研数据。
- 是否包含至少一个“不推荐某框架”的明确理由。
- 字数是否在目标范围内。
- 是否带有可执行的下一步建议。
我建议把这类验收判断也交给一个专门做质检的 Agent,或者写成一个规则检查器。不要把验收动作交给汇总者自己,否则等于让运动员自己做裁判,系统很容易陷入“我写得很好”的自我感觉良好里。
5. 常见问题与排查技巧实录
5.1 智能体陷入死循环,消息来回转
这是我见过最多的问题。两个智能体你一句我一句,消息永远停不下来。原因多数是:A 给 B 发了一条消息,B 不知道该不该把结果回传给 A,于是又回了一条“请确认”,A 又把它当成新任务继续处理。
排查方法有三个重点。第一,设置最大轮数,像 demo 里那样用一个 max_rounds 参数兜底。第二,在消息结构里带上“链路追踪号”,从根消息继承下来,这样能快速圈定一条循环链路。第三,在 Agent 提示词里明确写清楚“除非收到明确的修正指令,否则不要主动发起新一轮任务”。这句话看着不起眼,实际非常有用。
另一个技巧是给消息做哈希去重。每个 Agent 在处理消息之前,可以把消息内容摘要比对一遍,如果发现最近已经处理过相似内容,就直接忽略。
5.2 上下文越来越长,成本失控
多智能体系统跑的时间越长,每个 Agent 积累的上下文越大。一个大语言模型每次调用都要重新处理全部上下文,费用随 Token 数线性上涨,最后可能比人工做还贵。
解决思路不是“清理上下文”,而是“各人只管各家事”。每个 Agent 只携带自己完成当前任务所需的最小上下文,比如调研员只需要它的查询词和最新结果,不需要知道整个项目的来龙去脉。规划者需要全局视图,但规划者上下文也不会太长,因为规划者只关注任务列表,不需要读懂每份调研原文。
我习惯给每个任务块单独建立“工作记忆”文件,任务结束后自动归档,Agent 启动时不加载历史,只在必要时通过检索拿到相关片段。
5.3 工具调用失败,任务卡住不动
工具调用崩溃几乎是必然发生的。搜索引擎接口超时、API 密钥过期、返回格式不符合预期,任何一个出问题,都会让 Agent 停止推进。
处理办法分三层。第一层是重试:发现失败时,让 Agent 稍等片刻再调用一次。第二层是降级:如果接口超时,可以暂时用缓存数据或更简单的模拟数据顶上,并在报告里标注“此处数据来自备用源”。第三层是转人工:连续失败超过指定次数后,把消息状态标记为 escalation,由人工介入处理。
这里的核心经验是:工具调用失败不能只记录日志,必须让系统有能力改变后续行为。如果一条路走不通,Agent 要能换一条路走完任务,而不是单纯崩溃退出。
5.4 并发写同一份文件,结果互相覆盖
我在一次实践里遇到两个调研员同时写同一个结果文件,后写的把先写的覆盖了,整份报告丢了三分之一。排查的时候发现消息队列正常,任务也没有死循环,就是并发写文件没做控制。
解决办法几种:一是给写操作加锁,同一时刻只允许一个 Agent 写文件;二是用版本号控制,每个 Agent 写之前带上自己的版本号,提交时发现版本陈旧就要重新拉取合并;三是尽量把所有写操作都收敛到汇总者一个角色上,其他执行者只返回消息,不直接落盘。
第三种方式最简单,推荐你先用这个。收敛写权限,本质上跟前面说权限设计是一致的。
5.5 效果评估:一个报告的好坏靠什么判定
多智能体系统最容易被质疑的一点是:结果到底靠谱吗?我的做法是建立两层评估。
客观层看结构化指标:任务完成率、消息丢失率、平均轮数、工具调用成功率。这些指标都该有日志埋点,用一次跑批任务就能统计出来。
主观层看内容质量:每个报告都要人来做简易盲评,可以从信息准确性、逻辑连贯性、可操作性三个维度打分。初期效果不稳定很正常,关键是分数要能反馈回提示词和流程设计里。
我自己的经验是:先跑五到十次“黄金测试集”,也就是固定几个已知答案的问题,用同一套配置跑十遍,看结果波动多大。如果波动大到不可接受,就说明角色分工或提示词还要继续收敛。
6. 从 demo 到可用系统:边界与扩展建议
6.1 安全与合规底线要先划清楚
demo 阶段怎么折腾都行,但真要把 agency-agents 用到生产环境,第一条就是安全边界。
数据脱敏必须提前做。用户输入里可能带手机号、邮箱、内部系统路径,这些信息一旦被 Agent 塞进无关工具,就会造成泄露。我通常在入口处加一个脱敏模块,把所有可识别信息替换成占位符,等最终生成结果后再还原。
权限最小化不能只在文档里写。实际配置工具时,每个 Agent 的 API Key 都要单独创建,不能共用一个全局超级权限账号。比如调研员只需要只读接口,给它写权限等于埋雷。
人工审批环节不能省。涉及对外发布、删除数据、修改生产配置的操作,至少要有一个“人工确认”的步骤卡在中间。哪怕这个步骤只是点一下按钮,都能在很大程度上挡住误操作。
6.2 成本和质量控制
多智能体系统看起来炫酷,但如果不控制成本,改造成本会很惊人。我做过一次估算,一个三智能体的调研任务,如果每个 Agent 都调用一次大模型,合计 Token 数是单模型直接生成的 4 倍左右。也就是说,你必须换掉“什么都交给大模型”的思路。
省钱的办法有几条:
- 简单任务不要上多智能体,一个正则匹配能解决的问题别绕弯。
- 能用规则地方不要调用模型,比如格式转换、字段校验、去重识别。
- 模型选型分级,规划者用更便宜的模型,执行者用能力更强的模型,生成类任务用最高规格模型。
- 对结果做缓存,相同或相似任务不再重复跑全套流程。
质量控制上,核心是建立“结果版本”思维。每个任务都要有追踪 ID,每个结果都要标明由哪个 Agent 产出、基于什么材料。这样一发现问题可以快速回滚到特定版本。
6.3 几个值得继续深挖的方向
agency-agents 这套模式落地到一定程度后,你会发现可扩展的空间还很大。
一个方向是意图路由。用户输入进来,先由一个轻量分类模型判断该走哪条智能体链路,而不是所有需求都进同一套流程。比如“帮我看一下报错日志”和“帮我梳理数据库表结构”,走完全不同的协作链条。
另一个方向是人工反馈闭环。加入人类在环机制,在关键节点上允许人工修改 Agent 的产出,然后把修改结果作为反馈数据,回灌到提示词优化里。这也是一种低成本的数据飞轮。
还有一个方向是可观测性平台。多智能体的消息流转其实很像分布式系统,日志、链路追踪、指标监控缺一不可。如果你把这个系统当正式服务来运维,建议早点接上结构化的监控面板,不然跑半个月之后你根本不知道是哪条消息出问题。
我在实际操作中最大的体会是:agency-agents 的成功不在于智能体数量多,而在于边界清晰。每个智能体知道自己能干什么、不能干什么,消息该怎么传、工具该用到什么程度,这些设计清晰了,哪怕底层模型能力一般,整体效果也不会差。反过来,如果你把所有复杂度都堆在一起,再强的模型也救不了混乱的协作流程。
后面我会继续把这个最小框架扩展成一份可以复用在不同业务场景的基础设施,如果你也在折腾类似的东西,建议从今天这个三智能体骨架开始试,跑通一次闭环之后,你对整体的理解立刻会上一个台阶。