做AI应用的人,迟早会遇到一个问题:单体Agent真的够用吗?我最初也天真地以为,只要把一个足够强的大模型塞进一个Agent里,它就能包办从需求分析、内容撰写到代码检查、上线发布的所有事。后来在模拟项目X里试了一圈才发现,一个Agent同时身兼数职时,上下文会乱、角色会糊、指令偶尔还会自相矛盾。直到我把思路换成“agency-agents”——让多个分工明确的Agent组成一个虚拟机构,像团队一样各司其职地协作——整体效率和可控性才真正上了一个台阶。
这篇文章想跟你聊聊“agency-agents”这种多Agent协作架构的核心思路、角色设计、消息协议和调度机制,再给出一套可以直接复现的最小实现方案。无论你正在搭建智能客服、自动写作流水线、代码审查助手还是数据处理管道,只要系统中需要两个以上Agent配合,这篇文章的思路都能直接借鉴。
1. 核心思路拆解:为什么是“一个机构”而不是“一个超级Agent”
1.1 从“单Agent代管”到“Agent分工协作”
“agency-agents”这个命名很有意思,它是“agency”(机构/代理处)和“agents”(智能体)的合体。这个命名传递了一个很朴素的设计理念:当一个Agent做不了的事,你就“开一家公司”,让不同分工的Agent各司其职。
单体Agent的痛点,我实测下来主要有三个。第一个是上下文污染:让同一个Agent先做用户调研、再写文案、再审核合规,调研时的对话历史会残留到写作和审核阶段,模型经常被前面的信息带偏。第二个是角色混淆:一个人既当运动员又当裁判,指令冲突时模型会随机选一个立场,输出不稳定。第三个是工具权限无法隔离:如果这个Agent什么工具都能调,一旦提示词被诱导,风险面就是全部的。
而“agency-agents”用组织化的思路解决了这三个问题。它把一个任务拆解成多个可以由不同Agent并行或串行处理的子任务。每个Agent拥有独立的系统提示词、独立的工具集、独立的上下文窗口。它们之间通过统一的消息协议沟通,用类似“主管调度”的机制协调进度。
1.2 角色清单:先想清楚你的“公司”里需要哪些员工
我强烈建议,动手写代码之前,先把角色清单列出来。这一步省不了,想不清楚角色,后面全乱。
以我曾做过的一个“数据报告自动生成”场景为例,我设计了五个角色:
| 角色 | 职责 | 核心能力 |
|---|---|---|
| 主管Agent | 接收用户目标、拆解任务、分派给对应Agent、汇总结果 | 任务规划、编排调度 |
| 调研Agent | 抓取数据、搜索资料、整理结构化信息 | 网页检索、信息抽取 |
| 分析Agent | 基于调研结果做数据分析、提炼洞察 | 数据计算、图表生成 |
| 撰写Agent | 把分析结论写成流畅的报告正文 | 文本生成、结构组织 |
| 质检Agent | 检查报告的事实性错误、格式问题、合规风险 | 规则校验、提示词注入检测 |
每个角色边界清晰,不重叠。主管Agent不直接写报告,调研Agent不做分析,质检Agent不参与创作。这样做的最大好处是:你可以单独迭代任何一个角色的提示词或工具,而不会牵连其他角色。比如质检Agent的规则更新了,只需要重新部署那一个模块,其他Agent完全不需要动。
1.3 两个核心机制:任务编排与上下文隔离
角色定好之后,整个系统的地基就是两个机制:任务编排(Orchestration)和上下文隔离(Context Isolation)。
任务编排解决的是“谁先干、谁后干、干完交给谁”的问题。常用模式有三种:流水线模式(前一个Agent的输出是后一个Agent的输入)、主管-下属模式(一个主管Agent统一调度多个下属Agent)、网状模式(任意Agent之间可以互相通信)。我现在最推荐的做法,是从“流水线模式”起步,再逐步加入“主管-下属”的嵌套。不要一上来就用网格,那个模式调试起来真的会让人头秃。
上下文隔离解决的是“每个Agent应该看到什么、不应该看到什么”的问题。我的做法是为每个Agent维护一个独立的对话历史窗口,只在任务交接时传递结构化的任务摘要,而不是把原始对话记录全部塞过去。比如撰写Agent只需要接收调研Agent产出的“核心数据摘要”,不需要看调研Agent当时检索了哪些网页、使用的什么搜索关键词。这既节省了token费用,也避免了无关信息干扰生成质量。
其实你把“agency-agents”想成一个互联网公司的运作方式就通了:每个Agent是一个只有单一职责的独立岗位,主管Agent扮演项目经理,上下文隔离就是“跨部门只交接结论,不交接过程”。
2. 架构与协议设计:先把“团队规矩”定下来
2.1 最小消息协议:任务单、回执与终稿
多Agent系统里最重要的一件事,就是让每个Agent之间说的“话”格式统一。我踩过一个坑:一开始每个Agent返回纯文本,结果下游Agent解析了半天,各种格式兼容问题,最后气得我把协议全重写了。后来我总结出一套最小消息协议,结构非常简单,就三种消息:任务单、回执、终稿。
任务单的结构如下:
{ "task_id": "task_001", "assignee": "researcher_agent", "payload": { "instruction": "查找2024年AI Agent市场规模数据", "context": { "report_topic": "AI Agent行业趋势", "deadline": "2025-12-01" } }, "from": "supervisor_agent", "timestamp": "2025-07-20T10:00:00Z" }回执的结构如下:
{ "task_id": "task_001", "status": "success", "output": { "text": "2024年全球AI Agent市场规模约为XX亿美元", "structured_data": [ {"year": 2024, "value": "XX亿", "source": "某机构报告"} ] }, "meta": { "model": "gpt-4o-mini", "tokens_used": 1024, "latency_ms": 2300 } }所有Agent只认这三种消息。这个协议的好处有二。第一,机器可解析,后续要把日志接入监控、统计、成本核算系统都容易;第二,模型更好理解,因为指令非常显式,模型不需要猜测“这段文本要我做什么”,它只需要做填空题。
有一点要特别提醒:任务单里最好带上from字段,这样Agent可以在上下文中知道“是谁给我派的活”,有助于它理解任务背景。我之前去掉这个字段后发现,属下Agent偶尔会自言自语、忘掉自己的角色,加回来之后稳定不少。
2.2 调度策略:先规则后模型,能枚举的别让模型猜
多Agent系统的调度逻辑,我个人的经验是:能用规则代码解决的,就不要让模型做决策。你想想看,如果“任务A依赖任务B”这种关系也要模型来判断,那每跑一轮任务就要多调用一次模型,既慢又不稳定。
我的调度策略分成两层:第一层,用确定性代码处理依赖关系;第二层,用主管Agent处理模糊性任务分解。
比如在数据报告场景里,“调研Agent必须在分析Agent之前执行”、“分析Agent必须在撰写Agent之前执行”,这些都是确定性依赖,用代码写死就好,不需要主管Agent参与。主管Agent只负责做“用户想要什么”到“任务清单”之间的转换:
用户需求:写一篇关于AI Agent市场趋势的报告 主管Agent输出: 1. 调用调研Agent,查找近三年市场规模数据 2. 调用分析Agent,从数据中提炼增长趋势和驱动因素 3. 调用撰写Agent,基于分析结论生成报告正文 4. 调用质检Agent,检查报告事实与格式这个任务清单返回给调度器,调度器再根据清单里的依赖关系决定执行顺序。这样既保留了模型的灵活性,又避免了模型在执行层面的不可控。
2.3 工具挂载与权限控制:让每个Agent只用自己该用的工具
在多Agent架构里,工具挂载是一个很容易被忽视但特别关键的环节。我见过不少团队,把搜索、计算、发邮件、改数据库的工具全部授权给所有Agent,结果一个写文案的Agent因为上下文被注入了一段恶意指令,转头就去调了数据库接口,差点出事。
工具挂载的原则是“最小权限”。每个Agent只挂载它的职责范围的工具。比如:
- 调研Agent:搜索引擎API、网页抓取工具、RSS解析器
- 分析Agent:Python计算引擎、图表生成库
- 撰写Agent:文本编辑器接口(通常不需要额外工具,直接生成文本即可)
- 质检Agent:规则匹配器、合规检查API
在代码层面,可以用一个简单的注册表来管理工具权限:
TOOL_REGISTRY = { "researcher_agent": ["web_search", "web_fetch"], "analyst_agent": ["python_exec", "chart_render"], "writer_agent": [], "reviewer_agent": ["rule_checker"], }每个Agent在启动时只从注册表里加载自己允许的工具列表,这样即使Agent被提示词注入,它也无法调用不在白名单里的工具。这个设计看似简单,却能把系统的风险面控制在一个清晰边界内。
3. 从零实现一个可运行的多Agent协作Demo
3.1 环境准备与依赖选择
下面这部分是一份可以直接抄作业的最小实现。我用的是Python 3.10 + OpenAI兼容接口,因为目前大部分大模型服务商都提供OpenAI兼容的REST API,你只要换一下base_url和api_key就能切到不同模型,不需要改业务代码。
创建项目目录:
mkdir agency-agents-demo cd agency-agents-demo python -m venv venv source venv/bin/activate pip install openai python-dotenv根目录建一个.env文件,里面放模型服务的地址和密钥:
OPENAI_API_BASE=https://your-model-provider.example/v1 OPENAI_API_KEY=your-api-key-here MODEL_NAME=gpt-4o-mini这里有个经验之谈:先用便宜的小模型把流程跑通,再用更强的大模型替换。小模型速度快、成本低,可以快速暴露流程设计上的问题——比如消息协议字段不对、调度顺序有误——等流程稳定了,再换大模型提升单Agent的生成质量,这样迭代效率最高,也不会烧太多的钱。
3.2 定义Agent基类:用System Prompt固化“人设”
我们定义一个Agent基类,参数包括名字、系统提示词、允许使用的工具列表和模型名称。核心就一句话:“把system prompt和工具权限绑定在一起”。
import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() class Agent: def __init__(self, name, system_prompt, allowed_tools=None, model=None): self.name = name self.system_prompt = system_prompt self.allowed_tools = allowed_tools or [] self.model = model or os.getenv("MODEL_NAME") self.client = OpenAI( base_url=os.getenv("OPENAI_API_BASE"), api_key=os.getenv("OPENAI_API_KEY"), ) self.history = [] self.history.append({ "role": "system", "content": system_prompt }) def run(self, user_message): self.history.append({ "role": "user", "content": user_message }) response = self.client.chat.completions.create( model=self.model, messages=self.history, temperature=0.3, ) content = response.choices[0].message.content.strip() self.history.append({ "role": "assistant", "content": content }) return content这个基类看起来简单,但它已经完成了三件事:固定人设、维护上下文历史、调用大模型。有一点要注意:temperature=0.3。多Agent协作场景里,我一般不把温度调得太高,因为Agent之间必须严格按协议输出,只有质检Agent这种需要发散判断的角色,我才会把温度调到0.7。
3.3 创建五个角色Agent:像搭积木一样组装团队
有了基类,创建角色就是写system prompt的事。我建议每个system prompt都写清楚三件事:你是谁、你要输出什么格式、你有哪些禁忌。
supervisor = Agent( name="supervisor", system_prompt=( "你是一个项目主管。你的职责是理解用户的目标," "分解出可执行的子任务,并以JSON数组格式输出任务计划。" "每个任务必须包含:assignee字段(可选值为researcher/analyst/writer/reviewer)," "instruction字段(明确的任务描述)。" "禁忌:不要执行子任务本身,只输出任务计划。" ) ) researcher = Agent( name="researcher", system_prompt=( "你是一名调研专员。你擅长从给定的资料中提取信息。" "你的输出必须是JSON格式,包含data(array)和summary(string)两个字段。" "禁忌:不要对数据做趋势分析,不要写报告正文,只需要输出结构和摘要。" ) ) analyst = Agent( name="analyst", system_prompt=( "你是一名数据分析师。你擅长从结构化数据中发现趋势和规律。" "你的输出必须是markdown格式,包含:核心结论列表、关键数据点、驱动因素说明。" "禁忌:不要编造不存在的数据。" ) ) writer = Agent( name="writer", system_prompt=( "你是一名内容创作者。你擅长把分析结论转化为通俗、流畅的报告正文。" "你的输出是纯文本的markdown格式。" "禁忌:不允许补充分析部分没有提到的数据。" ) ) reviewer = Agent( name="reviewer", system_prompt=( "你是一名质量审核员。你负责检查报告是否符合要求。" "你的输出是JSON格式:passed(布尔值)、issues(发现问题列表)、suggestions(修改建议列表)。" "禁忌:不要重写报告,只输出问题和建议。" ) )写system prompt有几个细节需要经验积累。第一,一定要在结尾写明“不要做什么”,这比只写“你要做什么”更有效。第二,输出格式尽量用JSON这种机器可读的结构,方便下游处理。第三,每个角色的system prompt里不要塞太多背景故事,够用就行,否则模型会花大量token在“扮演”上,反而影响任务完成质量。
3.4 写一个轻量调度循环:把整个流程串起来
调度器的职责很简单:让各个Agent按顺序执行,把上一个Agent的输出作为下一个Agent的输入。先写一个最朴素的串行版本,后续要并行再优化。
import json def run_pipeline(user_request): # 1. 主管Agent拆解任务 plan_str = supervisor.run(user_request) plan = json.loads(plan_str) # 2. 按顺序执行 research_data = None for task in plan: if task["assignee"] == "researcher": research_data = researcher.run(task["instruction"]) elif task["assignee"] == "analyst": analysis = analyst.run(f"调研结果如下:\n{research_data}") elif task["assignee"] == "writer": report = writer.run(f"分析结果如下:\n{analysis}") elif task["assignee"] == "reviewer": review = reviewer.run(f"报告正文如下:\n{report}") # 3. 检查质检结果 review_result = json.loads(review) if not review_result["passed"]: print("报告需要修改:", review_result["suggestions"]) return report, review_result if __name__ == "__main__": request = "写一篇关于AI Agent市场规模与趋势的简要报告,重点说明2024年到2025年的变化。" report, review_result = run_pipeline(request) print("=== 最终报告 ===") print(report) print("=== 质检结果 ===") print(review_result)这个调度循环已经把“agency-agents”的核心跑通了:主管Agent负责计划,其余Agent按计划执行,质检Agent最后把关。
运行时你会发现,整个流程的稳定性很大程度上取决于第一环——主管Agent的任务分解。如果任务分解错了,后面再怎么调整也无济于事。所以我额外加了一步:在主管Agent输出计划之后,我用一段确定性代码做计划校验,比如检查assignee字段是否合法、instruction是否为空,如果计划非法就直接报错,避免把坏的计划执行到一半才暴露,浪费时间和token。
3.5 跑通案例:从需求到报告的全过程记录
用上面的代码跑一次后,我实话说,第一次大概率不会一次过。可能遇到输出JSON解析失败、质检Agent觉得数据不一致要求返工等状况。这是正常的,多Agent系统的调试本来就比单体调用要复杂。
我的建议是:在调度循环里加日志记录,打印每一步的Agent名称、任务内容、返回值摘要。调试时最怕的就是黑盒,一旦知道是哪个Agent在哪一步出了问题,修复起来就快多了。在真实项目里,这些日志还会打到监控系统里,用trace_id串联一整条执行链路,方便线上排查。
4. 常见问题与排查技巧实录
4.1 问题一:多个Agent相互“抢话”,任务没人接
这是我最早遇到的情况:主管Agent输出任务计划后,其他Agent不按计划执行,反而自己发起新任务,或者互相重复执行同一件事,搞得整个流程一团乱。
排查发现根因有两个。一个是system prompt里“边界”写得不够硬,Agent以为自己还能管别的事;另一个是消息格式不统一,有的Agent返回JSON、有的返回纯文本,下游无法判断当前应该执行到哪一步。
解决办法也很明确:一是把所有Agent的职责边界在最前面用加粗提示词强调“你只能做X”;二是所有Agent的输入输出都严格走消息协议,不让自由文本直接流转;三是在调度循环里显式判断当前阶段,只有当前阶段的Agent才被调用,其他Agent完全不激活。
这其实是用“流程确定性”来弥补“模型不确定性”,值得你在系统设计时反复琢磨。
4.2 问题二:上下文爆炸,对话越来越慢、费用越来越高
每个Agent维护自己的历史消息,如果单个Agent执行任务过多,它的上下文窗口会撑到极限。我在一个长文档场景里,写到第五次迭代时,writer_agent的上下文已经好几千行历史,请求延迟翻了快一倍,费用也涨得很厉害。
我的解决办法有两个。第一,在每次Agent执行完任务后,把历史中的中间过程做一次摘要压缩,只保留关键输出,比如用一句“已完成XX任务,关键结论是XX”替换掉之前的完整对话。第二,对于只执行一次的任务Agent,可以配置成“无状态模式”,不保留历史上下文,每次从干净的system prompt开始。这个“无状态模式”特别适合质检Agent和翻译Agent这种每次任务相对独立的角色,可以大幅降低token消耗。
4.3 问题三:工具调用参数总是填错
有一次我给分析Agent挂了Python执行工具,它需要传入一个JSON格式的data字段,结果模型反复把data字段的引号转义错了,连续报了三次错才成功。我检查了function calling的参数定义,发现是因为我的参数示例写得太抽象,模型不知道具体应该填什么。
后来我在工具定义的description里写了非常具体的填充示例,比如“data字段应传入上一步调研Agent的输出中raw_data的值,类型为数组,每个元素包含year和value两个字段”。模型看到这种提示,准确率一下就从60%提到了95%。function calling的参数说明要当成“给人的操作说明书”来写,而不是只给一个类型声明,这个经验在多Agent系统里尤其适用。
4.4 问题四:成本与并发,一个小型系统的真实开销
最后聊一下成本。多Agent系统处理一个任务,调用次数往往是单体Agent的5到10倍。我做过一个统计:一个需要“调研+分析+撰写+质检”的任务,假设每个Agent平均消耗2000 token,总消耗接近8000 token,其中还不包括来回返工的重试次数。
控制成本的关键手段,我复盘下来有三条。一是模型分层:简单Agent用便宜小模型(比如一个长上下文但便宜的模型),只有主管Agent和质检Agent用强模型。二是结果缓存:如果同一个用户请求的调研数据是重复的,可以直接复用之前调研Agent的产出,不需要重新调用。三是条件重试:质检Agent返回“需要修改”时,不要无限循环,设一个最大重试次数(我一般设3次),超过之后就返回人工介入再由人来决定怎么处理。
这四个问题,是任何一个真实的“agency-agents”项目都会遇到的。表面上看是技术问题,背后其实都是设计问题:职责边界不清、上下文管理不当、协议不够严格、成本意识缺失。把这几个方向理顺了,即使你用的是不同的模型、不同的编程语言,整个系统的健壮性都不会太差。
我自己在实际操作中反复体会到,multi-agent系统的复杂度不在于单个Agent的实现,而在于“组织方式”。你和模型打交道,本质上是在为模型设计一份清晰的组织章程,让它们各安其位、各司其职。如果你正在尝试把多个Agent组合起来做事,我建议你从最小的两个角色开始,跑通再扩,不要一上来就排十几个角色的“大公司”,不然你很快就会被各种莫名其妙的交互问题淹没。最后再分享一个长期有用的技巧:坚持在整个过程中记录每个Agent的关键决策日志,这些内容是你在系统出问题时的第一手排错依据,也会成为你不断优化角色设计的重要基础。