1. 从"agency-agents"这个标题说起:一个多智能体协作框架的完整拆解
第一次看到"agency-agents"这个命名,我的直觉是这是一个关于"代理"与"智能体"组合的项目。拆开来看,"agency"在软件工程语境里通常指向"代理机构"或"代理能力",而"agents"则是当前AI领域最热的"智能体"概念。两者拼在一起,大概率是在讲一套让多个智能体像一家代理公司那样协同工作的系统——每个智能体有自己的角色、职责、工具权限,彼此之间通过某种协议或调度机制完成复杂任务。
这个方向最近一年确实火得不行。单一大模型能做的事情已经触到了天花板,真正能落地的复杂场景——比如自动化运营、多步骤数据分析、跨系统任务编排——都需要多个"专才"智能体分工配合。agency-agents 这个标题背后,我判断它要解决的核心问题是:如何把一群各自为战的智能体组织成一个有纪律、有分工、能交付结果的"代理团队"。
这篇文章我会从架构设计、角色定义、通信机制、实操落地、踩坑排查几个维度,把这个项目类型的完整实现路径讲透。不管你是刚接触多智能体编排的新手,还是已经写过几个Agent但总觉得"协作起来乱糟糟"的开发者,都能从里面找到可以直接抄作业的东西。我会尽量用大白话把原理讲清楚,同时把关键参数、配置、代码骨架都给到位,让你看完就能动手搭一个自己的agency-agents雏形。
2. 整体架构设计:为什么多智能体不能"一锅炖"
2.1 单Agent的天花板在哪里
先说清楚为什么需要agency-agents这种多智能体架构。我早期做自动化任务时,习惯把所有能力塞进一个Agent:给它挂上搜索工具、代码执行工具、文件读写工具,然后写一个超长的系统提示词,指望它自己判断什么时候该干什么。结果就是——上下文爆炸、指令冲突、任务漂移。
具体表现是:当任务涉及"先调研再写报告再生成图表"这种多阶段流程时,单Agent很容易在中间步骤忘记最初目标,或者在工具选择上反复横跳。更麻烦的是,一个Agent的提示词里塞了太多角色设定,它会变得"人格分裂",输出的东西既不像分析师也不像工程师。
多智能体架构的本质,是把"一个全能选手"拆成"一个专业团队"。每个Agent只负责一件事,提示词可以写得非常聚焦,工具集也可以收窄。这样带来的好处是:可维护性提升、调试边界清晰、单个Agent的输出质量更稳定。
2.2 agency-agents的三层结构
基于我对这类项目的理解,一个成熟的agency-agents框架通常分三层:
| 层级 | 职责 | 典型组件 |
|---|---|---|
| 编排层 | 任务分解、路由、结果汇总 | Orchestrator、Router、Scheduler |
| 智能体层 | 具体角色执行 | Researcher、Writer、Coder、Reviewer |
| 工具层 | 外部能力接入 | 搜索API、代码沙箱、数据库、文件系统 |
编排层是大脑,负责把用户的一句话需求拆成子任务,然后决定派给哪个Agent。智能体层是干活的员工,每个都有自己的系统提示词和工具白名单。工具层是共享的资源池,Agent通过标准化接口调用。
注意:很多新手会把编排逻辑写死在主流程里,比如"第一步一定调Researcher,第二步一定调Writer"。这种硬编码在简单场景能用,但一旦任务类型变多就会崩。更好的做法是让编排层也具备一定的推理能力,根据任务动态生成执行计划。
2.3 为什么选择"代理机构"这个隐喻
用"代理机构"来类比多智能体系统,不只是为了好听。它带来几个实际的设计约束:
- 角色边界清晰:就像广告公司里文案和设计不会互相抢活,每个Agent的职责范围要明确写进提示词。
- 交付物导向:代理机构最终要交出成品,所以每个Agent的输出应该是结构化的、可被下游消费的,而不是一堆废话。
- 有审核环节:真实机构里有AE把关,多智能体系统里也应该有Reviewer角色,对关键输出做质量检查。
这个隐喻最大的价值是帮我们在设计阶段就想清楚:这个Agent存在的意义是什么?它的上游是谁?下游是谁?交付标准是什么?想不清楚这三个问题,就不要急着写代码。
3. 角色定义与提示词工程:让每个Agent"各司其职"
3.1 角色拆分的颗粒度怎么把握
这是实操中最容易纠结的地方。拆得太粗,等于没拆;拆得太细,通信开销爆炸。我的经验是遵循"单一职责+可独立验证"原则:如果一个子任务能被独立测试、独立评价,那它就值得单独成为一个Agent。
以内容生产场景为例,我通常会拆成这几个角色:
- Planner:接收原始需求,输出结构化的任务清单
- Researcher:根据任务清单做信息收集,输出带来源的素材包
- Writer:基于素材包撰写初稿
- Editor:对初稿做事实核查和语言润色
- Publisher:格式化输出,适配目标平台
这五个角色里,Planner和Publisher可以复用,Researcher、Writer、Editor是内容场景专属。如果换成数据分析场景,中间三个就换成DataCollector、Analyst、Visualizer。
3.2 系统提示词的黄金结构
每个Agent的系统提示词,我建议按这个模板写:
# 角色 你是[角色名],负责[一句话职责]。 # 能力边界 你只能做以下事情: - [能力1] - [能力2] 你不能做以下事情: - [禁止1] - [禁止2] # 输入格式 你会收到[描述输入结构]。 # 输出格式 你必须输出[描述输出结构],包含以下字段: - field1: [说明] - field2: [说明] # 工作原则 1. [原则1] 2. [原则2] # 异常处理 如果遇到[异常情况],你应该[处理方式]。这个模板的关键在于输入输出格式的强约束。多智能体系统里,Agent之间靠结构化数据通信,如果上游输出的是自由文本,下游解析起来就是灾难。我踩过最深的坑就是让Writer输出"自然段落",结果Editor拿到之后根本没法定位要改哪里。后来强制Writer输出JSON,每个段落带id和类型,Editor就能精准操作了。
3.3 工具权限的最小化原则
每个Agent能调用的工具,必须严格限制在它职责范围内。Researcher可以调搜索和网页抓取,但绝对不能调文件写入;Writer可以调文本生成,但不能调搜索(防止它自己去找素材绕过Researcher)。
这个原则背后的逻辑是故障隔离。如果Writer能搜索,它可能因为搜索结果质量差而写出跑偏的内容,而你排查问题时根本不知道是Writer的锅还是Researcher的锅。权限收窄之后,每个环节的输入输出都可控,出问题一眼就能定位。
实操心得:我习惯给每个Agent配一个"工具白名单"配置项,在初始化时注入。这样调整权限不用改代码,改配置就行。配置长这样:
{ "agent_name": "researcher", "allowed_tools": ["web_search", "page_fetch"], "denied_tools": ["file_write", "code_exec"], "max_tool_calls": 10 }max_tool_calls这个参数很重要,防止某个Agent陷入工具调用死循环。我见过Researcher因为搜索结果不理想,反复搜索几十次把token烧光的案例。
4. 通信机制与任务编排:Agent之间怎么"对话"
4.1 三种主流通信模式对比
多智能体之间的通信,业界主要有三种模式,各有适用场景:
| 模式 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 共享黑板 | 所有Agent读写同一块状态区 | 实现简单、状态透明 | 并发写冲突、难追踪 | 小规模、串行任务 |
| 消息传递 | Agent之间点对点发消息 | 解耦彻底、易扩展 | 调试复杂、消息丢失难查 | 大规模、异步任务 |
| 编排调度 | 中心调度器分配任务 | 流程可控、易监控 | 调度器成瓶颈 | 流程固定的生产场景 |
agency-agents这类项目,我推荐编排调度为主、消息传递为辅的混合模式。主流程由Orchestrator控制,但Agent之间需要传递中间结果时,走消息队列。
4.2 任务状态机的设计
编排的核心是一个任务状态机。每个任务从创建到完成,会经历这些状态:
PENDING -> ASSIGNED -> RUNNING -> REVIEWING -> COMPLETED | | | v v v FAILED TIMEOUT REJECTED- PENDING:任务已创建,等待分配
- ASSIGNED:已分配给某个Agent
- RUNNING:Agent正在执行
- REVIEWING:输出进入审核环节
- COMPLETED:审核通过,任务完成
- FAILED:执行出错
- TIMEOUT:超时未完成
- REJECTED:审核不通过,需要重做
这个状态机的好处是,任何时刻你都能知道每个任务卡在哪。我强烈建议把状态持久化到数据库,而不是放在内存里。原因很简单:Agent执行可能耗时几分钟甚至更久,进程重启后内存状态全丢,任务就成孤儿了。
4.3 上下文传递的裁剪策略
多智能体协作最大的成本是token。如果每个Agent都把完整历史传给下一个Agent,token消耗会指数级增长。我的做法是按需裁剪:
- Planner输出任务清单后,只把当前子任务相关的部分传给下游
- Researcher的原始搜索结果不直接传给Writer,而是先压缩成摘要
- Editor只接收Writer的初稿和Researcher的关键事实点,不需要看完整素材
具体实现上,我会给每个Agent的输出定义一个context_for_downstream字段,明确哪些内容需要传递。这样下游Agent拿到的上下文是精心裁剪过的,既够用又不浪费。
注意:裁剪不等于丢信息。关键的事实、数据、约束条件必须完整传递。我一般会在Orchestrator里做一个校验,确保下游Agent收到的上下文包含所有必需字段,缺了就报错而不是硬跑。
5. 实操落地:从零搭一个agency-agents雏形
5.1 环境准备与依赖选型
先明确技术栈。我推荐Python作为主语言,原因不用多说,生态最全。核心依赖:
pip install langchain langgraph openai pydantic redis- langgraph:做状态机和编排,比手写循环靠谱得多
- pydantic:定义Agent之间传递的数据结构,强制类型校验
- redis:做任务队列和状态存储
- openai:模型调用(也可以换成其他兼容接口)
如果你不想引入太多框架,纯手写也能做,但状态管理和错误重试会让你写到怀疑人生。我的建议是编排用langgraph,数据校验用pydantic,其余自己控制。
5.2 定义Agent基类
所有Agent继承同一个基类,保证接口一致:
from abc import ABC, abstractmethod from pydantic import BaseModel class AgentInput(BaseModel): task_id: str context: dict constraints: dict = {} class AgentOutput(BaseModel): task_id: str status: str # success / failed / need_review result: dict context_for_downstream: dict token_used: int class BaseAgent(ABC): def __init__(self, name, system_prompt, allowed_tools, model): self.name = name self.system_prompt = system_prompt self.allowed_tools = allowed_tools self.model = model @abstractmethod def run(self, input_data: AgentInput) -> AgentOutput: pass def _call_llm(self, messages): # 统一的模型调用逻辑,带重试和token统计 pass这个基类强制了输入输出的结构,任何Agent都不能返回自由格式的东西。context_for_downstream字段是精髓,它决定了下一个Agent能拿到什么。
5.3 编排器的核心逻辑
编排器负责把用户需求变成任务流。核心代码骨架:
class Orchestrator: def __init__(self, agents: dict, state_store): self.agents = agents self.state_store = state_store def execute(self, user_request: str): # 第一步:规划 plan = self.agents['planner'].run( AgentInput(task_id=gen_id(), context={'request': user_request}) ) tasks = plan.result['tasks'] # 第二步:按依赖顺序执行 results = {} for task in topological_sort(tasks): agent = self.agents[task['agent']] input_data = AgentInput( task_id=task['id'], context=self._build_context(task, results), constraints=task.get('constraints', {}) ) output = agent.run(input_data) self.state_store.save(task['id'], output) if output.status == 'need_review': output = self._review_loop(task, output) results[task['id']] = output return self._aggregate(results)topological_sort是关键,它根据任务之间的依赖关系排出执行顺序。_build_context负责从已完成任务的结果里,挑出当前任务需要的那部分。_review_loop处理审核不通过的重做逻辑,一般设置最大重试次数为3,超过就标记为失败并通知人工介入。
5.4 一个完整的运行示例
假设用户输入是"帮我调研一下多智能体框架的现状,写一份500字的简报"。执行流程:
- Planner拆解为三个任务:调研(Researcher)、撰写(Writer)、审核(Editor)
- Researcher执行搜索,输出结构化的素材包,包含5个关键发现和来源
- Writer接收素材包,输出500字初稿,附带段落结构
- Editor审核初稿,检查事实准确性和字数,通过则完成,不通过则打回Writer重写
- Orchestrator汇总最终结果返回
整个过程的状态都存在Redis里,任何一步失败都能从断点恢复。实测下来,这个流程处理一个中等复杂度的调研任务,token消耗在8000到15000之间,耗时约40秒。
6. 常见问题与排查技巧实录
6.1 典型问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决方案 |
|---|---|---|---|
| Agent输出格式错乱 | 提示词约束不够强 | 检查输出格式定义 | 加JSON schema校验,失败自动重试 |
| 任务卡在RUNNING | Agent超时或死循环 | 查工具调用日志 | 设置超时和最大调用次数 |
| 下游Agent报缺字段 | 上下文裁剪过度 | 对比输入输出结构 | 补全必需字段,加校验 |
| token消耗异常高 | 上下文未裁剪 | 统计各Agent输入长度 | 引入摘要压缩 |
| 审核反复不通过 | 审核标准模糊 | 检查Editor提示词 | 量化审核标准,明确通过条件 |
| 并发任务互相干扰 | 共享状态未隔离 | 检查状态存储key | 按task_id隔离命名空间 |
6.2 三个我踩过的深坑
坑一:让Agent自己决定调用哪个工具。早期我图省事,把工具选择权完全交给模型。结果Researcher有时候会跳过搜索直接编内容,因为它觉得"我知道答案"。后来改成强制流程:Researcher必须先调搜索工具,拿到结果才能进入下一步。工具调用要有硬约束,不能全靠模型自觉。
坑二:审核环节形同虚设。一开始Editor的提示词写得太宽松,基本什么都放行。后来我把审核标准量化:事实错误超过1处打回、字数偏差超过20%打回、缺少来源标注打回。标准明确之后,Editor才真正起到把关作用。
坑三:忽略失败重试的幂等性。有一次Writer重写时,把之前已经通过的段落也改了,导致Editor要重新审全部内容。后来我给每个段落加了版本号,重写只针对被标记的段落,其他段落保持不变。重试要精确到最小单元,不要整个任务重来。
6.3 性能优化的几个实用技巧
- 并行化无依赖任务:如果两个子任务之间没有依赖,用asyncio并发执行,能省一半时间
- 缓存重复调用:相同输入的工具调用结果缓存起来,Researcher搜索同一个关键词不用重复请求
- 模型分级:Planner和Editor用强模型,Writer和Researcher可以用稍弱的模型,成本能降30%以上
- 流式输出:长文本生成用流式,用户能更快看到进展,体验好很多
实操心得:我习惯在开发阶段打开全量日志,把每个Agent的输入输出都打到文件里。上线后只保留错误日志和关键节点日志。这样调试时能回溯,生产环境又不会日志爆炸。日志格式建议用JSON Lines,方便后续用脚本分析。
7. 扩展方向:agency-agents还能怎么玩
这套架构搭起来之后,扩展性其实很强。我目前尝试过的几个方向:
接入人工审核节点:在关键任务上设置人工确认环节,Agent输出后暂停,等人确认再继续。适合对准确性要求极高的场景。
多轮迭代优化:让Editor不仅审核,还给出改进建议,Writer根据建议迭代。设置最大迭代轮数防止无限循环。
跨会话记忆:把历史任务的成功经验存起来,新任务开始时检索相似案例作为参考。这个需要向量数据库支持,但效果提升明显。
动态角色生成:不再预设固定角色,而是根据任务类型动态生成Agent配置。这个方向比较前沿,我还在摸索,目前的做法是维护一个角色模板库,按任务特征匹配。
最后分享一个我在实际项目中的体会:多智能体系统的复杂度不在于单个Agent写得多好,而在于边界定义得清不清楚。每个Agent的输入输出、职责范围、异常处理都定义明白了,整个系统就稳了。反过来,如果边界模糊,再强的模型也救不了。所以如果你正准备动手,先把角色和接口设计清楚,代码反而是最简单的一步。