1. 从agency-agents这个命名说起它到底指什么第一次看到agency-agents这个词很多人会愣一下——它不像某个具体框架的名字也不像某个库的官方命名更像是一个项目代号或者仓库名。我最初接触这个词是在翻一些开源项目的目录结构时当时第一反应是代理的代理后来结合实际项目经验才慢慢理清楚agency-agents 本质上描述的是一类代理编排层的设计模式即在多个智能代理agent之上再抽象出一层代理机构agency由这个机构统一调度、分配任务、管理生命周期。打个比方。如果把单个 agent 比作一个自由职业者——会写文案、会查资料、会调接口那 agency 就是一家接单公司。客户也就是上层业务系统不直接找自由职业者而是把需求丢给公司公司根据每个自由职业者的专长、当前负载、历史表现来派单最后把结果汇总交付。这个公司就是 agency它下面的员工就是 agents。为什么需要这么一层因为当你的系统里只有一两个 agent 时直接调用完全没问题但一旦 agent 数量上到十几个、任务类型横跨检索/生成/校验/执行硬编码的调用链会迅速变成一团乱麻。agency-agents 要解决的核心痛点就是让多代理协作从手写 if-else 调度进化到声明式编排。这套东西适合谁三类人最该关注。第一类是正在做 AI 应用的后端工程师手里已经有一堆 agent 但调度逻辑写得很痛苦第二类是技术负责人需要评估多代理架构到底值不值得上第三类是对 agent 生态感兴趣、想理解编排层这个概念的开发者。不管你是哪一类下面我会从设计动机、核心机制、落地步骤到踩坑经验一层层拆开讲。需要先说明一点agency-agents目前并不是一个像 LangChain 那样广为人知的标准化框架名它更多出现在一些自研项目的仓库命名和内部架构讨论中。所以本文讨论的是这一类编排模式的通用设计与实现思路具体代码会给出可复现的参考实现你可以直接套用到自己的项目里。2. 为什么裸调 agent 迟早会崩编排层存在的真实理由2.1 从三个 agent 到三十个 agent 的失控过程我拿一个真实场景举例。假设你要做一个自动生成行业周报的系统最初的设计很朴素Agent A负责联网检索本周行业新闻Agent B负责把新闻摘要成要点Agent C负责把要点组织成周报正文三个 agent串行调用代码大概五十行搞定。这个阶段你根本不需要什么 agency直接A.run() - B.run() - C.run()就完事了。但需求会膨胀。两周后产品说能不能加个事实核查于是有了 Agent D。再过一周能不能根据用户偏好调整语气有了 Agent E。能不能自动配图有了 Agent F。检索结果要按可信度排序——Agent G。等你反应过来系统里已经躺着十几个 agent而且它们之间的依赖关系不再是简单串行D 要在 B 之后、C 之前跑但如果 A 检索到的新闻少于 5 条D 要跳过E 的输入同时依赖 B 和用户配置F 只在 C 生成正文超过 800 字时才触发。这时候你的调度代码会变成什么样一大坨嵌套的 if-else加上各种状态标志位改一处牵动全身。这就是编排熵增——每加一个 agent复杂度不是线性增长而是组合式爆炸。2.2 agency 层解决的四个具体问题agency-agents 模式之所以值得单独抽象一层是因为它集中处理了四件脏活第一任务路由。上层只说我要一份周报agency 负责判断这个请求该拆成哪些子任务、分给哪些 agent。路由逻辑集中在一处而不是散落在业务代码里。第二生命周期管理。agent 的初始化、超时、重试、销毁这些如果每个调用点都写一遍代码会重复到令人崩溃。agency 统一托管。第三结果聚合与冲突消解。多个 agent 可能返回矛盾的结果比如 A 说某公司融资了D 核查说没这回事agency 需要一个仲裁策略来决定信谁。第四可观测性。当系统有二十个 agent 在跑你必须知道每个请求经过了哪些 agent、各花了多久、哪一步是瓶颈。agency 层天然是埋点的最佳位置。提示如果你现在的 agent 数量还少于 5 个且依赖关系是纯串行我的建议是先别上编排层。过早抽象和过晚抽象一样有害前者会让你为不存在的复杂度买单。2.3 一个反直觉的结论编排层不是越智能越好很多人在设计 agency 时第一反应是让 agency 自己用大模型来决定怎么调度。听起来很酷但实测下来往往是灾难。原因很简单调度决策需要的是确定性和可预测性而大模型的输出恰恰是概率性的。你让一个概率模型去决定这个任务该走哪条链路它偶尔会给你一个完全离谱的路径而且这种错误极难复现和调试。我的经验是agency 的调度逻辑应该是规则优先、模型兜底。80% 的常见请求走硬编码的规则路由快、稳、可测只有那些规则覆盖不到的边缘请求才交给模型去判断。这个比例可以根据业务调整但规则为主这个基调不能变。3. agency-agents 的核心机制拆解3.1 注册表agent 如何被 agency 认识agency 要调度 agent首先得知道有哪些 agent 可用、各自能干什么。这就需要一个注册表registry。最朴素的做法是一个字典key 是 agent 名字value 是 agent 实例加元数据class AgentRegistry: def __init__(self): self._agents {} def register(self, name, agent, capabilities, cost1.0): self._agents[name] { instance: agent, capabilities: set(capabilities), cost: cost, healthy: True, } def find_by_capability(self, capability): return [ name for name, meta in self._agents.items() if capability in meta[capabilities] and meta[healthy] ]这里有几个设计细节值得说。capabilities用集合而不是字符串是为了支持一个 agent 具备多种能力的场景比如一个 agent 既能做摘要又能做翻译。cost字段是我强烈建议加的——它代表调用这个 agent 的相对开销可以是 token 消耗、延迟、金钱成本agency 在路由时可以优先选低成本的 agent把贵的留给真正需要的任务。healthy标志位用于健康检查某个 agent 连续失败几次后自动摘除避免它拖垮整个链路。3.2 任务描述让 agency 看懂要干什么agency 收到一个请求后需要把它翻译成自己能理解的任务描述。这里我推荐用一个结构化的 Task 对象而不是裸字符串from dataclasses import dataclass, field dataclass class Task: goal: str # 任务目标自然语言 required_caps: list field(default_factorylist) # 需要的能力 context: dict field(default_factorydict) # 上下文数据 constraints: dict field(default_factorydict) # 约束如超时、预算 parent_id: str None # 父任务 ID用于追踪链路goal是给模型看的required_caps是给规则路由看的context携带上游 agent 的产出constraints里可以放max_latency、max_cost这类硬约束。把给人看的和给机器看的分开是让编排逻辑保持清晰的关键。3.3 调度策略规则路由与模型路由的混合调度是 agency 的心脏。我通常实现成两级第一级是能力匹配。根据 Task 的required_caps从注册表里筛出所有具备这些能力的 agent。这一步是纯确定性的不涉及任何模型调用。第二级是优先级排序。在候选 agent 里按什么顺序选常见策略有三种策略适用场景优点缺点成本优先预算敏感、任务量大省钱可能牺牲质量质量优先结果要求高效果好贵、慢负载均衡高并发吞吐高单次质量不稳定实际项目里我一般用加权打分score w1 * quality - w2 * cost - w3 * current_load权重根据业务调。这个打分函数就是 agency 的调度大脑它比让大模型自由发挥要可控得多。3.4 结果聚合当多个 agent 说法不一致聚合是容易被低估的一环。假设你让三个 agent 分别总结同一篇文章得到三段摘要怎么合成最终结果简单拼接会显得冗余直接选一个又浪费了另外两个的工作。我常用的策略是分层聚合先做去重把语义重复的句子合并再做投票多个 agent 都提到的要点权重更高最后做压缩用一个轻量模型把合并后的内容压成通顺的一段。如果 agent 之间出现事实性冲突则触发一个专门的仲裁 agent它的唯一职责就是判断哪个说法更可信判断依据可以是来源权威性、多个 agent 的一致性等。4. 从零搭一个最小可用的 agency-agents 系统4.1 环境与依赖准备这套东西对环境的依赖其实很轻。核心只需要 Python 3.9如果你要用异步并发asyncio是标准库自带的。模型调用部分看你用哪家我示例里会用一个抽象的LLMClient接口你替换成自己的实现即可。pip install pydantic # 用于数据校验可选但推荐我不建议一上来就引入重型框架。先用标准库把骨架搭起来跑通了再考虑要不要换。很多项目死在框架还没学明白需求已经变了。4.2 定义 agent 基类所有 agent 继承同一个基类保证接口一致这样 agency 才能统一调度class BaseAgent: name base capabilities set() async def run(self, task: Task) - dict: raise NotImplementedError async def health_check(self) - bool: return Truerun是异步的因为 agent 内部大概率要调模型或网络同步会拖垮并发。health_check默认返回 True有需要的 agent 可以重写比如检查自己的 API key 是否还有额度。4.3 实现一个具体的检索 agentclass SearchAgent(BaseAgent): name search capabilities {retrieve, web_search} def __init__(self, client): self.client client async def run(self, task: Task) - dict: query task.context.get(query, task.goal) results await self.client.search(query, top_k5) return { agent: self.name, results: results, count: len(results), }注意返回结构里带了agent字段这是为了后续追踪这个结果是谁产出的。别小看这个字段出问题时它能帮你快速定位是哪个 agent 的锅。4.4 把 agency 主循环写出来class Agency: def __init__(self, registry): self.registry registry async def execute(self, task: Task) - dict: # 1. 能力匹配 candidates [] for cap in task.required_caps: candidates.extend(self.registry.find_by_capability(cap)) candidates list(set(candidates)) if not candidates: raise ValueError(f没有 agent 能处理能力需求: {task.required_caps}) # 2. 打分排序 scored sorted( candidates, keylambda n: self._score(n, task), reverseTrue, ) # 3. 依次尝试失败则降级到下一个 last_error None for name in scored: agent self.registry._agents[name][instance] try: result await agent.run(task) return result except Exception as e: last_error e self.registry._agents[name][healthy] False continue raise RuntimeError(f所有候选 agent 均失败: {last_error}) def _score(self, name, task): meta self.registry._agents[name] return 1.0 / (meta[cost] 0.1)这个主循环虽然只有几十行但已经包含了 agency 的核心匹配、排序、降级重试。你可以在此基础上加超时控制、并发执行、结果缓存。4.5 跑通第一个端到端请求async def main(): registry AgentRegistry() registry.register(search, SearchAgent(client), {retrieve}) registry.register(summarize, SummaryAgent(client), {summarize}) agency Agency(registry) task Task( goal总结本周 AI 行业动态, required_caps[retrieve], context{query: AI industry news this week}, ) result await agency.execute(task) print(result) # asyncio.run(main())跑通这一步你就有了一个能用的最小系统。接下来所有的复杂度都是在这个骨架上长出来的。5. 实测中那些文档不会告诉你的坑5.1 超时设置全局超时和单 agent 超时是两回事我踩过最惨的一次坑是只设了全局超时没设单 agent 超时。结果某个 agent 卡在一个慢接口上全局超时是 60 秒它一个人就吃掉了 55 秒剩下的 agent 根本没时间跑整个请求超时失败。正确的做法是双层超时每个 agent 有自己的max_latency比如 10 秒agency 层再设一个总预算比如 60 秒。单 agent 超时就降级到下一个候选总预算超时才整体失败。import asyncio async def run_with_timeout(agent, task, timeout): try: return await asyncio.wait_for(agent.run(task), timeouttimeout) except asyncio.TimeoutError: raise RuntimeError(f{agent.name} 超时)5.2 重试的边界不是所有失败都值得重试新手容易犯的错是无脑重试。但失败分两类瞬时失败网络抖动、限流值得重试逻辑失败参数错误、能力不匹配重试一百次也没用。我的做法是在异常里区分错误类型只对瞬时失败做指数退避重试逻辑失败直接降级或报错。注意重试次数不要超过 3 次。超过 3 次还失败的基本可以判定不是抖动问题继续重试只会浪费预算和时间。5.3 上下文膨胀agent 之间传的数据会越滚越大多 agent 协作时每个 agent 的产出都会塞进context传给下游。跑几个 agent 之后context 可能已经几万 token下游 agent 光是读上下文就烧掉大量预算。解决办法是上下文裁剪每个 agent 只接收它真正需要的字段而不是整个 context 全量透传。可以在 Task 里加一个context_schema声明这个 agent 需要哪些 keyagency 在派发前做一次过滤。5.4 循环依赖A 等 BB 等 A当 agent 数量多了很容易不小心搞出循环依赖。比如 Agent A 需要 B 的结果来决策B 又需要 A 的结果来初始化。这种问题在运行时表现为死锁或无限循环。防御手段是在 Task 里维护一个visited_agents列表每次派发前检查目标 agent 是否已经在链路上出现过出现过就拒绝直接报错。这个检查成本极低但能救命。5.5 成本失控一个请求烧掉几十块这是最肉疼的坑。多 agent 系统里一个用户请求可能触发十几个 agent每个都调模型成本是单 agent 的十几倍。如果不加控制月底账单会让你怀疑人生。我的做法是给每个 Task 设max_cost预算agency 在每次派发前累加已消耗成本超预算就停止派发新 agent用已有结果做降级输出。宁可给用户一个不完整的结果也不要烧穿预算。6. 让 agency 从能用到好用的进阶思路6.1 结果缓存相同任务不要重复算很多请求其实是重复的。比如总结今天的新闻这种任务同一天内多个用户问结果应该是一样的。在 agency 层加一个缓存key 用goal required_caps的哈希命中就直接返回。缓存过期时间根据任务类型定新闻类可以短一点几小时知识类可以长一点几天。这一招能把成本降一大截。6.2 并发执行无依赖的 agent 并行跑如果两个 agent 之间没有数据依赖为什么要串行agency 完全可以并行派发。用asyncio.gather把无依赖的 agent 一起跑总耗时从求和变成取最大值。判断依赖关系可以靠 Task 里的depends_on字段显式声明也可以靠 agency 分析 context 的读写关系自动推断。显式声明更可靠我推荐前者。6.3 可观测性每个请求都要能复盘生产环境里你必须能回答这个请求经过了哪些 agent、各花了多久、哪一步失败。我的做法是给每个 Task 生成一个trace_id每个 agent 执行前后都打点记录agent_name、start_time、end_time、status、cost。这些日志汇总起来就能画出完整的调用链路。出问题时拿着trace_id一查一目了然。6.4 灰度与回滚新 agent 不要直接全量新加一个 agent 或者改了调度策略不要直接全量上线。先让 5% 的流量走新逻辑观察成功率、延迟、成本有没有异常没问题再逐步放量。agency 层天然适合做灰度因为所有请求都经过它加一个流量分流的判断就行。7. 关于 agency-agents 我个人的几点体会搭过多套多 agent 系统之后我最大的体会是编排层的价值不在于智能而在于可控。那些试图让大模型全权决定调度的方案demo 阶段很惊艳生产阶段很崩溃。真正扛住流量的系统调度逻辑往往是朴素的规则加打分模型只用在它真正擅长的地方——理解语义、生成内容、做模糊判断。另一个体会是别急着追求 agent 数量。我见过太多项目为了显得架构先进硬生生把一个大 agent 拆成五个小 agent结果协作开销比收益还大。判断标准很简单如果两个 agent 合并后代码更简单、延迟更低、效果不降那就该合并。agent 的粒度应该由任务的自然边界决定而不是由架构图好不好看决定。最后说个实操建议先把单 agent 做到极致再考虑编排。很多所谓的多 agent 需求其实一个设计良好的单 agent 加几个工具调用就能解决。编排层是解决复杂度的工具不是制造复杂度的理由。等你真的被多 agent 调度折磨到睡不着觉时再回头看 agency-agents 这套思路会更有共鸣。