做AI应用的人,最近应该都被一个词刷屏了:Agent。我自己手上也压着不少活儿——整理调研报告、批量生成产品文案、把碎片信息整理成结构化文档。一开始我尝试用一个大模型挂一堆工具硬怼,效果始终差口气,不是输出格式飘忽不定,就是中间某一步出错整条链路跟着崩。后来我把思路换了换:与其让一个模型干所有事,不如让一群Agent像一家数字代理机构那样分工协作。这就是我搭agency-agents这套小框架的起因。
说白了,agency-agents不是一个现成的商用平台,而是我在实际业务里总结出的一套多Agent协作模式:有接单的、有做调研的、有写稿的、有审稿的,每个角色只干自己最擅长的一件事,通过任务队列传递半成品,最后由一个调度者汇总交付。它解决的核心问题是单模型上下文窗口有限、责任边界模糊、过程不可追踪这几件事。这篇文章我尽量讲得实在一点,包括角色怎么拆、任务怎么传、编排逻辑怎么写、运行时会踩哪些坑,希望给正在折腾多Agent的人一点参考。
1. 项目第一眼:agency-agents到底解决什么问题
1.1 为什么我宁愿养一支Agent队伍,也不去调一个大模型
在我把业务拆成多Agent之前,所有任务都走同一个入口:把完整需求丢给一个能力很强的旗舰模型。这种模式看着简单,用久了问题非常明显。
第一个问题是上下文互相污染。比如让模型先检索资料、再写分析、最后做排版,它会把检索阶段的原始网页内容全记在上下文里,等写到后半段时注意力早就被垃圾信息带偏了。我试过在提示词里反复强调“只关注结构化摘要”,但效果维持不了几次,模型一旦在长上下文里漂了,输出质量就断崖式下跌。
第二个问题是排错像大海捞针。单模型链路一旦出错,你根本不知道是检索环节没取到关键信息,还是理解环节把需求解读歪了,还是生成环节格式崩了。整条链路是一个不透明的黑盒,出了问题只能整锅重煮,费用和时间全搭进去。
第三个问题是模型能力分配不合理。一个任务里往往只有20%的步骤需要顶级推理能力,剩下80%都是信息抽取、格式化、关键词匹配这种体力活。结果我用旗舰模型把体力活也干了,成本高昂,速度还慢。多Agent的好处就在于可以按角色分配不同档位的模型:便宜快速的模型负责检索和抽取,能力强的模型只负责判断和写作,整体成本能压下来,稳定性反而提上去。
1.2 这个项目适合谁、解决哪些常见痛点
如果你手头是下面这几类事情,多Agent这套玩法大概率对你有用。
第一类是长周期调研任务。比如“整理过去三年某个行业的政策变化并输出对比报告”,单模型要做完得反复续写,中间任何一次走神都会让前后口径不一致。拆成多个Agent后,有人专门检索政策原文,有人负责做时间线梳理,有人负责写对比分析,各环节独立运行再合并,质量好控制得多。
第二类是批量内容生产。产品文案、营销脚本、社媒短贴这类任务,特点是想清楚规则之后重复执行。我用一个“策略Agent”把写作规范拆成模板字段,再用多个“执行Agent”并发跑不同产品线,最后统一格式,效率比原来单线程调模型高一截。
第三类是信息整理与结构化。比如会议纪要转行动项、用户反馈分类打标、竞品信息抽取建表,这些活儿本质上需要“先理解再提炼”,但既不复杂也不需要创造力。用指定角色的Agent加上一套严格的输出Schema,稳定性和准确率都能做到让人放心。
当然,多Agent不是万能的,后面第五部分我会详细讲它引入的新麻烦。但如果你已经在单模型方案里感受到“什么都干等于什么都干不漂亮”,那这套模式值得一试。
2. 整体架构:让一群Agent像“数字代理机构”一样干活
2.1 角色拆解:这些Agent怎么分工
我设计角色的时候参考了现实中广告公司和策划团队的分工方式,因为项目名叫agency-agents,本质上就是一支数字代理团队。每个角色都是一个独立的Agent实例,有自己专属的System Prompt、工具权限和输出规范。
协调者(Coordinator)是整个队伍的领队,它不干活,只负责拆任务和派单。它拿到需求后,会把大目标拆成若干子任务,判断哪些可以并行、哪些有依赖关系,然后分发给下面的角色。编辑者(Editor)负责查漏补缺,它会对交付成果做一轮质量审查,发现逻辑漏洞、格式问题或者信息缺失就打回重做。研究员(Researcher)负责检索信息,它主要调用搜索和网页读取工具,输出结构化的事实清单而不是整段文章。分析者(Analyst)负责做数据处理,把研究员拿回来的事实整理成图表、对比表、趋势摘要。写作者(Writer)负责内容生成,把分析结果落成符合要求的文案、报告或者推广稿。
这里的核心原则是:每个角色只对输入的一个局部目标负责,输出结果会被下一个角色的输入Schema严格限制。用交通来类比,不是一个人从出发地一脚油门开到目的地,而是机场、高铁、出租车各管一段,每一站都有明确的交接文件。这样做的直接好处是出问题能定位到具体某段路线,而且可以单独优化某个环节,不用整条链路一起改。
2.2 任务编排:线性流程、并行分支还是黑板架构
多Agent跑起来之后,角色之间到底按什么顺序协作,是架构上最需要想清楚的一件事。我试过三种模式,各有各的适用场景。
线性流程最直观:A干完传给B,B干完传给C。适合步骤之间强依赖、中间结果必须按顺序加工的任务,比如“提取数据—生成图表—写入报告”。缺点是慢,一个环节卡住后面全等。
并行分支适合多个独立子任务同时推进。协调者一次派出去五个研究员分别查五个方向,等他们全部回来后分析者再统一汇总。这种模式能把整体耗时压到接近单个最慢节点的耗时,是我在处理调研类任务时使用最频繁的模式。
黑板架构灵活性强,所有Agent共享一块“黑板”,谁发现新信息就往黑板上写,谁需要信息就从黑板上读。这种模式适合探索性很强的任务,比如产品头脑风暴、多轮信息补充,但实现起来要处理读写冲突和消息版本问题,对于我这种小团队来说成本偏高。
我在agency-agents里最终采用的是混合编排:主干用线性流程保证交付节奏,宽泛的调研和检索环节用并行分支提速度,特殊情况下允许Agent向协调者请求补充信息,形成小范围的黑板交互。这样既稳又灵活。
2.3 选型背后的成本账:多Agent最怕什么
很多人一听多Agent就兴奋,但我被现实毒打过几次,这里必须先泼一盆冷水:多Agent最怕的是成本失控和链路失控。
链路失控指的是Agent之间互相传递信息时,格式稍微偏一点,下游就解析失败,然后触发重试,重试又产生新的输出,可能把上游结果覆盖掉。我在早期版本里直接用函数返回值传参,Agent多起来之后整个程序栈纠缠在一起,排查问题非常痛苦。后来我才改成基于消息队列的异步解耦,每个Agent只从队列里取任务、往队列里放结果,角色之间完全不直接调用。这是整个架构稳定下来的关键转折点。
成本失控则发生在上下文重复传递上。想象一下,一条任务链上有五个Agent,每个Agent都把完整的原始资料带在身上,五个Agent加起来可能吃掉五倍的输入token。我的解决办法是三层:第一,传递消息时只传结构化摘要和必要字段,不传原始长文本;第二,让下游Agent按需通过“引用指针”去公共存储里取完整文档,而不是默认都塞进上下文;第三,不同角色按任务难度分配不同档位的模型,检索类用便宜的轻量模型,写作和判断类再用旗舰模型。
这里送大家一句话:多Agent架构的收益来自“专业化分工”,但如果分工的代价是信息重复搬运,收益很快会被成本吃掉。所以设计消息结构时,时刻问自己一个问题:这个阶段真的需要把这么长的内容传给下一个角色吗?大多数时候答案是不需要。
3. 从零搭一个可跑的agency-agents:核心模块与实操
3.1 环境准备与依赖
我自己的实现方案比较轻,没有引入特别重的框架,主要依赖的是Python生态里的几件常规武器。Python版本建议3.10以上,因为我用到了不少类型标注特性。大模型调用层我做了统一封装,方便切换不同服务商的API,底层其实就是一个自定义的ChatClient。
进程内的任务队列我用了Python标准库queue加asyncio组合。任务量再大一点的话,可以换成Redis作为消息中间件,让不同角色的Agent跑在不同的进程甚至不同的机器上。但刚开始玩的时候不建议上分布式,单机多进程足以验证架构,等真有必要再迁移也不迟。
数据存储方面,我用了SQLite存任务快照和审计日志。每个Agent收到的输入、产生的输出、耗时、模型调用的token数都会落库。这套日志系统在调试多Agent时帮了大忙,后面讲问题排查时你们会看到它的价值。依赖清单大致包括:pydantic做数据模型校验、openai或其他SDK做模型调用、jinja2做提示词模板渲染、sqlite3做任务审计。
提示:这里其实有一个很关键的架构取舍。我刻意没有依赖现成的Agent框架,原因是我需要在消息传递和重试策略上有完全的控制权。对于学习项目,自己从零搭一遍会让你理解每个环节到底在干什么,后续再用任何框架都事半功倍。
3.2 最小化Agent基类与角色注册
我实现的核心是Agent基类,所有角色都从这个基类派生。基类需要包含几个关键要素:角色名称、模型配置、系统提示词、可用的工具列表、输入输出Schema、以及最重要的run方法。下面是一个极简实现的核心部分,你可以当作骨架来扩展。
from abc import ABC, abstractmethod from typing import Any from pydantic import BaseModel class AgentInput(BaseModel): task_id: str instruction: str payload: dict[str, Any] # 上游产物,只放必要字段 class AgentOutput(BaseModel): task_id: str ok: bool data: dict[str, Any] error: str | None = None tokens: int = 0 class BaseAgent(ABC): role: str = "base" system_prompt: str = "" model_name: str = "fast-model" def __init__(self, client, tools: dict[str, Any] | None = None): self.client = client self.tools = tools if tools else {} @abstractmethod def process(self, inp: AgentInput) -> AgentOutput: """子类实现自己的核心逻辑""" ... def run(self, inp: AgentInput) -> AgentOutput: # 统一埋点:日志、耗时、token统计都在这里 start = time.time() try: result = self.process(inp) except Exception as e: result = AgentOutput(task_id=inp.task_id, ok=False, data={}, error=str(e)) self._log(inp, result, time.time() - start) return result每个实际角色只需要继承BaseAgent,重写process方法,然后在角色注册表里登记即可。比如研究员Agent的核心逻辑,就是拿着指令调用搜索工具,把搜索结果清洗成事实清单然后返回:
class ResearcherAgent(BaseAgent): role = "researcher" system_prompt = "你是一名信息检索专家。只输出结构化事实清单,不要推测,不要写结论性的文字。" model_name = "fast-model" def process(self, inp: AgentInput) -> AgentOutput: query = inp.payload["query"] raw_results = self.tools["web_search"](query) facts = self._convert_to_facts(raw_results) return AgentOutput(task_id=inp.task_id, ok=True, data={"facts": facts})这个设计看着简单,真正帮我省心的是两个点。第一,所有角色共用一套run方法,意味着任何Agent的错误都会被统一捕获、统一落库,不会因为某个人写代码时忘了加try就静默失败。第二,输入输出都用pydantic约束了Schema,上游传来的数据不对时本地报错,而不是把脏数据带进更下游。
3.3 任务总线与结果传递
接下来是任务总线的设计。我在系统里定义了三类消息:TaskMessage(派任务)、ResultMessage(交结果)、RequestMessage(请求协同)。三类消息共享一个队列,通过任务ID关联。
from dataclasses import dataclass, field @dataclass class TaskMessage: task_id: str assignee: str # 角色名 instruction: str # 任务指令 payload: dict # 必要字段,不含冗长原文 meta: dict = field(default_factory=dict) @dataclass class ResultMessage: task_id: str assignee: str ok: bool data: dict error: str | None = None在任务总线上,我硬性规定了一条铁律:同一个task_id的所有消息放在同一个流式批次里,并且带有序号。这个设计是为了防止多个Agent并发跑同一任务时结果互相覆盖。每个任务从创建到最终交付都有一个唯一生命周期,状态机只允许按照“待处理—处理中—已完成—已打回—已归档”这几个状态流转。
实际测试中我发现,如果在任务ID上不加区分、只靠角色名找结果,几轮并发跑下来一定会出现串数据的问题。加了task_id和序号之后,即使两个任务的内容一模一样,消息也不会混。这条经验是我在排查一个“两次运行结果互相污染”的bug时总结出来的,强烈建议各位一上来就把任务ID当成头等公民对待。
3.4 编排器:把角色串成一条生产线
有了角色和任务总线,剩下的核心组件就是一个编排器。它的职责是接收原始需求,生成一张DAG(有向无环图),然后按节点依赖关系调度Agent执行。DAG的结构是动态的,协调者Agent会根据需求内容决定拆成几个节点、哪些节点并行。
class Orchestrator: def __init__(self, agents: dict[str, BaseAgent], queue): self.agents = agents self.queue = queue def run(self, task_graph: dict) -> dict: """task_graph: {'nodes': [...], 'edges': [[上游id, 下游id], ...]}""" pending = {node["id"]: node for node in task_graph["nodes"]} results = {} ready = [n for n in pending.values() if not self._has_unfinished_upstream(n, task_graph["edges"], results)] while ready: # 并发执行所有ready节点 for node in ready: agent = self.agents[node["role"]] inp = AgentInput(task_id=node["id"], instruction=node["instruction"], payload=node["payload"]) output = agent.run(inp) results[node["id"]] = output # 根据edges找出下一批ready节点 ready = self._next_ready_nodes(pending, task_graph["edges"], results) return results这个实现虽然简化了很多,但核心思路足够参考。实际生产版本里还加了并发控制信号量、节点超时设置和失败重试策略。其中一个比较反直觉的经验是:不要把重试逻辑塞进单个Agent里,而应该在编排器里统一处理。原因是Agent内部重试很难知道是上游数据问题还是自己的问题,编排器层面能看到全局状态,判断是打回上游还是让当前节点重跑更准确。
我还做了一个很有用的可视化辅助:每次DAG执行完成后,把节点执行顺序、耗时、token消耗都转成一条日志链。调试阶段我经常盯着这条链看,哪个节点耗时异常、哪个节点结果为空、哪个节点反复打回,一眼就能定位。
3.5 质量校验与成本控制钩子
为了让多Agent产出稳定,我在流水线末尾和关键节点上加了质量关卡。最实用的方法是用一个独立的校验Agent扮演“第二双眼睛”,它不参与内容生成,只负责对照标准检查交付物。检查项包括:必填字段是否完整、格式是否符合Schema、内容里是否有明显矛盾或者幻觉、是否偏离了原始指令。
这个校验Agent我用的是“双模型交叉”策略。执行Agent用旗舰模型,校验Agent用另一个来源的模型,两个模型的错误模式往往不重叠。实践下来,交叉验证能抓出不少单模型内部察觉不到的问题。当然,纯自动校验还是覆盖不了所有主观质量指标,所以我在流程里保留了人工介入的接口:返回结果里带一个is_confident字段,置信度低于阈值时自动把任务挂起到人工审核队列。
成本控制钩子方面,我在编排器里维护了一个token计数器,每次Agent调用模型都会上报模型名、输入token数和输出token数。我设了三道防线:第一道是单节点token上限,超过直接终止并告警;第二道是单任务全局token预算,用完了自动降级到轻量模型从头跑;第三道是每日总预算,防止某个异常流程半夜里疯狂烧钱。有一次我离开电脑前忘了关测试任务,结果第二天看账单发现跑了四百万token,从那以后这三道防线成了标配。
4. 跑起来之后:实测过程与典型效果
4.1 一个具体的任务实例:行业趋势调研报告
为了验证这套agency-agents到底靠不靠谱,我拿一个典型的调研任务做了次端到端测试,任务是“整理某行业过去一年的融资趋势,分析主要方向变化,输出一份面向非专业读者的简报”。
任务进入系统后,协调者先做任务分解。它把大需求拆成了五个节点:研究员A查融资事件总量和季度分布,研究员B查热门细分赛道变化,研究员C查头部项目的业务方向,分析者汇总三个研究员的事实清单并做趋势归纳,写作者参考分析结果生成简报,最后由编辑者审核格式和逻辑。
执行时研究员A、B、C三个节点是并行跑的,整个任务的实际执行时间主要取决于最慢的节点。单模型跑类似任务大概要十五分钟,这套多Agent流水线最终用了四分钟出头。token消耗反而不高,因为研究员用轻量模型抽取事实、写作者用旗舰模型润色,两个成本大头被分开了。
产出质量比单模型让我放心得多。核心数据都标注了来源,引用的事实清单和分析结论分离,哪里存疑可以追溯。编辑者打回了一次,原因是简报里的季度趋势描述和数据表对不上,研究员A重新核验后修正了数据,整个修正过程只重放了相关子任务,没有让其他模块跟着返工。这一点是单模型链路做不到的。
4.2 观测到的收益:稳定性、可追溯性与可控性
跑了一段时间后,我对这套多Agent架构的收益认识更清晰了,大致可以总结成三点。
稳定性上,因为每个环节的输出都会被下一个环节的Schema强制约束,跑几十个任务下来格式崩掉的概率明显降低。早期单模型方案要时不时调整提示词才能维持输出格式,现在每个角色只维护一种输出格式,维护成本大幅下降。
可追溯性是附赠的大礼。单模型链路里,结果错了就只能重来,现在每个Agent每一步都有入参出参记录,错在哪一步、是谁造成的,可视化日志里清清楚楚。有一次写作者生成的文案数据口径有问题,我顺着日志查到是分析者交付的数据把同比和环比搞混了,只修改了分析者的提示词就解决了问题,不用动其他代码。
可控性体现在可以按需置换角色。前端文案风格要变,我只改写作者的系统提示词;检索策略要换,我只替换研究员的工具配置。因为角色之间是解耦的,改一个角色不影响整条链路。这个特点在实际协作开发中价值很大,多个同事可以各自维护自己负责的角色模块,互不干扰。
5. 常见问题与排查技巧实录
5.1 上下文污染:角色之间互相“剧透”怎么办
多Agent跑起来之后最常遇到的问题之一就是上下文污染。表现是下游Agent的输出里混进了上游的内部思考内容,或者明明只需要摘要,却把原始资料整段搬进了上下文。
我排查后发现主要原因有两个。第一,我在消息传递时图省事,把上游返回的完整结果直接塞进了下游的payload。解决方法是严格定义每条消息的payload字段,只允许传结构化摘要,原始长文档放在公共存储里,需要时按引用ID读取。第二,系统提示词里没有明确告诉下游Agent“你只应该基于本次传入的信息做判断”,导致模型自由发挥。我在每个提示词模板末尾加了一句固定的话:“只基于本消息中提供的资料作答,不要猜测未提供的信息。”虽然看着简单,但实测下来对减少越权引用效果很大。
5.2 死循环与超时:Agent任务怎么兜底
多Agent系统比单模型更容易出现死循环,因为多了一个“互相触发”的可能。比如编辑者发现格式不对就触发重写,写作者交回的格式又触发编辑者新一轮打回,两边互踢皮球,任务永远走不到归档状态。
我用的兜底方案分三层。第一层是最大迭代次数,每个任务在编排器里都有max_loop参数,超过一定轮次强制终止,并且自动降级为人工处理。第二层是超时设置,每个Agent调用模型都设置了硬超时,超时后自动走失败分支而不是无限等待。第三层是“打回收敛”机制,编辑者每次打回时必须注明具体原因,同一个原因打回超过两次就不再自动重试,直接把问题和历史记录一起提交给人审。
这里有个反反复复踩出来的经验:多Agent系统最怕的不是出错,而是出错后无法终止。所以在我这个架构里,任何循环都要有明确的退出条件,宁可错杀也不要让它无限烧token。
5.3 解析失败与格式漂移:输出不稳定怎么治
大模型输出格式漂移是长跑任务里的顽疾。同一套提示词,跑十次可能有一两次输出的JSON格式不合法。我试过增加few-shot示例、要求模型先输出一个“格式化思考”再输出正式结果,都没彻底解决问题。
最有效的方案是在编排器里加一个万能兜底:如果Agent返回的内容无法通过pydantic解析,程序自动做一次“修复式调用”。修复式调用的提示词是“下面是一个模型的输出,它没有满足JSON格式要求,请只输出修复后的严格JSON,不要添加任何解释”,然后用更低的温度参数跑一遍。这个策略把解析失败率从之前的大概百分之五降到了千分之一以下。
另外一个管用的技巧是要求每个Agent的最终输出固定以“```json”开头并保证标志性的代码块关闭,在所有角色之间统一成同一种格式约定,比给每个角色单独设置格式要求更容易维持稳定。系统的一致性偏好在这里帮了大忙。
5.4 成本失控:一个任务烧掉几百万token的教训
前面提到我吃过一次四百万token的亏,那个场景其实是测试任务触发了小概率路径。正常流程跑完了但编辑者连续打回,打回后写作者重跑,重跑的prompt里带了越来越多的历史记录,上下文越滚越大,最后几个版本的输入token数都超过了二十万。
这次教训让我做了两处调整。第一,消息传递必须做截断和摘要,运行中的中间产物在写入下游payload之前都会做一次“瘦身”,把大段内容替换成要点加引用指针。第二,每个Agent的输入做了上限保护,超过阈值就启动分段处理,把一个大任务拆成多个小任务,而不是让上下文无限膨胀。
现在我在每次运行结束后都会看一眼审计表里的token消耗排行,哪个节点消耗最多就优先优化哪里。实际跑下来基本能做到:同一类任务,多Agent架构比单模型方案节省三到五成token成本,同时结果稳定性更好。
我在实际使用中发现,多Agent这套玩法的精髓不在于Agent数量越多越好,而在于把任务切成合理的颗粒度,让每个角色在一个狭窄的范围内做到稳定。贪多嚼不烂,一上来就排十个角色,编排和排查成本会远远超过收益。如果你准备动手尝试,我的建议是先从一个稳定的单Agent开始,把它打磨到你在日志里能看清它的每一步,再开始复制这个角色。角色多了之后你会感谢当初留了审计字段。这套agency-agents目前还在迭代,下一步我打算把召回和记忆模块做成可插拔,让不同任务能直接换掉检索策略,省得每次都在提示词里挤牙膏。折腾Agent这条路挺有意思,但也确实是一场和不确定性打交道的持久战。