1. 从“agency-agents”这个标题说起它到底在解决什么问题第一次看到“agency-agents”这个组合词我脑子里跳出来的第一反应是这大概率不是一个单纯的工具库而是一套围绕“代理”和“代理机构”之间关系做文章的东西。拆开看agency 在软件语境里通常指“代理机构、代理服务、代理层”而 agents 则是“代理者、执行体、智能体”。两个词叠在一起指向的其实是同一件事的两面谁在代表谁执行动作以及这些动作如何被组织、调度和约束。我在实际项目里接触过不少类似命名的系统它们通常出现在三种场景里。第一种是任务分发类系统一个中心节点把任务拆给多个执行体执行体各自完成一部分工作再汇总第二种是权限与身份代理类系统某个主体需要以另一个主体的身份去访问资源中间需要一个可信的代理层来转换和校验第三种是近两年特别火的智能体协作类系统多个 agent 各自有专长通过一个 agency 层来协调它们的调用顺序、上下文传递和结果合并。这三种场景虽然领域不同但底层要解决的问题高度一致解耦、编排、可观测。“agency-agents”这个标题之所以值得单独拿出来讲是因为它天然带着一种架构味道。它不是“agent”单数也不是“agencies”复数而是 agency 和 agents 并列暗示着代理层和执行体是分离的两层。这种分离不是随便设计的它直接决定了系统后续能不能扩展、能不能排查问题、能不能在某个执行体挂掉的时候不影响整体。我见过太多项目一开始把调度逻辑和执行逻辑揉在一起等到执行体数量从 3 个涨到 30 个的时候代码就变成了一团乱麻改一处崩三处。所以这篇内容我打算按“一个真实项目从零搭建到稳定运行”的思路来写把 agency-agents 这类系统的设计思路、核心细节、实操步骤和踩坑经验完整地摊开。不管你是刚接触这个概念的新手还是已经做过类似系统但总觉得哪里别扭的老手应该都能从里面找到能直接抄作业的部分。我会尽量少讲空泛的概念多讲“为什么这么选”“这一步到底在干什么”“出问题了先看哪里”。2. 整体架构设计为什么要把 agency 和 agents 拆开2.1 核心思路调度层与执行层的职责边界任何一套 agency-agents 系统最核心的设计决策就是划清调度层和执行层的边界。我个人的经验是这条边界画得好不好直接决定了系统后期维护成本是线性增长还是指数增长。调度层agency只做四件事接收请求、决定由谁执行、把上下文传下去、把结果收回来。执行层agents也只做四件事接收任务、执行具体逻辑、返回结果、上报状态。除此之外的任何逻辑都要慎重考虑放在哪一层。为什么这么强调边界因为一旦调度层开始关心“这个任务具体怎么算”它就会和某个具体执行体绑死换一个执行体就得改调度代码反过来一旦执行层开始关心“我这个结果要发给谁”它就会依赖全局拓扑单独测试都测不了。我踩过最典型的一个坑是早期把重试逻辑写在了执行体里结果每个执行体都要自己维护一套重试计数和退避策略后来想统一改成指数退避改了七八个地方还漏了一个线上直接出现重试风暴。正确的做法是重试、超时、熔断、限流这些横切关注点全部放在 agency 层执行体只管“给我任务我就干干完就返回干不动就报错”。这样执行体可以做得非常薄薄到可以用不同语言、不同框架实现只要遵守统一的输入输出契约就行。这也是 agency-agents 这种命名方式背后真正的价值agency 是稳定的、集中的、可观测的agents 是可替换的、分布式的、无状态的。2.2 方案选型同步调用、消息队列还是事件驱动确定了分层之后下一个要拍板的就是 agency 和 agents 之间的通信方式。我实际用过三种各有各的适用场景没有绝对的好坏只有合不合适。同步调用最简单agency 直接通过 HTTP 或 RPC 调 agent拿到结果就返回。优点是链路短、调试方便、结果实时缺点是 agency 会被慢 agent 拖住一个 agent 卡住整个请求就卡住。我一般只在 agent 数量少、任务耗时短、对实时性要求高的场景用这种方式比如内部管理后台的几个校验 agent。消息队列是折中方案agency 把任务丢进队列agent 消费后把结果写回另一个队列agency 再异步收结果。优点是解耦彻底、天然支持削峰填谷、agent 可以水平扩展缺点是链路变长、调试麻烦、需要额外维护队列的可靠性。任务量大、允许秒级延迟的场景我优先选这个。事件驱动最灵活agency 只负责发事件谁关心谁订阅agent 之间也可以互相触发。优点是扩展性最强加一个新 agent 不用改 agency缺点是链路最难追踪一个请求可能触发十几个事件出问题的时候排查起来非常痛苦。我一般只在业务逻辑本身就高度事件化的场景用比如风控、监控告警这类。下面这张表是我自己总结的选型对照实际决策时基本照着看就行通信方式适用 agent 数量典型延迟调试难度扩展性我的推荐场景同步调用3 到 10 个毫秒级低弱内部校验、实时查询消息队列10 到 100 个秒级中中批量任务、异步处理事件驱动100 个以上秒到分钟级高强风控、监控、复杂编排2.3 状态管理无状态执行体加外部状态存储执行体到底要不要保存状态这个问题我纠结过很久。早期为了图方便让 agent 在内存里缓存一些上下文结果一扩容就出问题同一个用户的两次请求落到不同实例上第二次读不到第一次的缓存行为完全不一致。后来统一改成执行体完全无状态所有需要跨请求保留的数据都放到外部存储问题才彻底解决。具体做法是agency 在派发任务时把这次任务需要的全部上下文打包成一个 task context 传下去agent 执行完只返回结果不保留任何东西。如果确实需要跨任务共享状态比如会话信息、用户偏好就放到 Redis 或数据库里用统一的 key 规则去读写。这样做的好处是 agent 可以随时重启、随时扩容、随时替换对上层完全透明。注意无状态不等于不缓存。agent 内部可以缓存一些只读的、与请求无关的数据比如配置、字典、模板这些缓存不会因为实例切换而不一致。判断标准很简单这份数据是否和“当前这个请求”绑定。绑定就不能放内存不绑定就可以。3. 核心细节解析agency 层到底要做哪些事3.1 任务路由怎么决定把任务派给谁任务路由是 agency 层最核心的逻辑也是最容易写歪的地方。我见过两种极端一种是硬编码 if-else任务类型 A 就调 agent1类型 B 就调 agent2加一个类型就得改代码重新发版另一种是过度设计搞了一套规则引擎结果规则比业务代码还多没人看得懂。我的经验是路由逻辑要数据驱动但不要过度抽象。具体做法是维护一张路由表用配置的方式描述“什么条件下派给谁”。最简单的形式就是一个 JSON 或 YAML 文件key 是任务类型或匹配条件value 是 agent 标识和调用参数。agency 启动时加载这张表运行时查表决定路由。这样加一个新 agent 只需要改配置不用动代码。等路由条件复杂到配置表达不了的时候再考虑引入规则引擎也不迟。路由还要考虑负载均衡。如果同一个任务类型有多个 agent 实例agency 要决定派给哪一个。轮询最简单但没考虑实例的实际负载按连接数或响应时间加权更合理但需要 agent 上报指标。我一般先用轮询等出现明显的负载不均再换成加权。这里有个细节加权算法一定要有兜底某个 agent 上报的指标异常比如响应时间突然变成 0时不能让它吸走所有流量要有最小权重保护。3.2 上下文传递任务信息怎么完整带到执行体上下文传递看起来简单实际上坑特别多。最常见的问题是上下文太大一个请求带了几百 KB 的 JSON网络传输和序列化都成了瓶颈。我的做法是上下文分层必传的核心字段任务 ID、类型、超时时间、追踪 ID直接放在请求头或消息属性里体积小、解析快业务数据放在 body 里按需传递能引用就不内联。另一个坑是上下文污染。agent A 处理完往上下文里塞了一些中间结果agent B 拿到后误以为是原始输入逻辑就错了。解决办法是给上下文加命名空间每个 agent 只能读写自己命名空间下的数据跨 agent 共享的数据必须由 agency 显式声明。这样虽然麻烦一点但能避免大量隐蔽的 bug。还有一点容易被忽略上下文的版本兼容。agency 和 agent 可能不是同时升级的新 agency 传的字段老 agent 不认识或者老 agency 传的字段新 agent 已经废弃了。我的做法是上下文里带一个 schema 版本号agent 启动时声明自己支持的版本范围agency 路由时做一次兼容性检查不匹配就拒绝派发并告警。这个机制在灰度发布的时候特别有用。3.3 结果聚合多个 agent 的结果怎么合并当一个任务需要多个 agent 协作完成时结果聚合就成了 agency 的另一个核心职责。聚合策略取决于业务语义常见的有几种全部成功才算成功、多数成功就算成功、取第一个成功的结果、按优先级取最高优先级的结果。这些策略没有优劣关键是要在 agency 层统一实现不要让 agent 自己判断。聚合还要处理部分失败的情况。比如 5 个 agent 里 3 个成功 2 个失败是整体失败还是返回部分结果我的经验是默认整体失败但保留部分结果供排查。因为部分成功的结果往往是不完整的直接返回给上层可能导致更隐蔽的问题。如果业务确实需要部分结果就单独定义一个“允许部分成功”的任务类型明确标注出来让调用方自己决定怎么处理。实操心得结果聚合的时候一定要记录每个 agent 的耗时和状态哪怕最终结果用不上。这些数据在排查“为什么这次请求特别慢”的时候是救命稻草。我一般会把它们写进追踪系统按任务 ID 就能查到完整的调用链。4. 实操过程从零搭一套可运行的 agency-agents4.1 环境准备与目录结构假设我们要搭一套最小可用的 agency-agents 系统用 Python 实现通信走消息队列。先规划目录结构这一步别偷懒结构清晰后面省很多事agency-agents/ ├── agency/ │ ├── __init__.py │ ├── router.py # 路由逻辑 │ ├── dispatcher.py # 任务派发 │ ├── aggregator.py # 结果聚合 │ └── config/ │ └── routes.yaml # 路由配置 ├── agents/ │ ├── base.py # 执行体基类 │ ├── agent_a.py │ └── agent_b.py ├── common/ │ ├── context.py # 上下文定义 │ └── protocol.py # 通信协议 └── tests/依赖方面消息队列我用的是 Redis 的 Stream 功能轻量、够用、部署简单。如果任务量特别大再换专业队列。序列化用 JSON虽然比二进制慢一点但可读性好排查问题方便太多。追踪用 OpenTelemetry标准化不用自己造轮子。4.2 定义统一的通信协议协议是 agency 和 agents 之间的契约必须先定好。我用一个简单的数据结构# common/protocol.py from dataclasses import dataclass, field from typing import Any, Dict, Optional dataclass class TaskContext: task_id: str task_type: str schema_version: str 1.0 trace_id: str timeout_ms: int 5000 payload: Dict[str, Any] field(default_factorydict) namespace: Dict[str, Dict] field(default_factorydict) dataclass class TaskResult: task_id: str agent_id: str status: str # success / failed / timeout data: Optional[Dict] None error: Optional[str] None elapsed_ms: int 0这里有几个设计点值得说明。schema_version是为了兼容性检查前面提过。namespace是为了隔离不同 agent 写入的数据每个 agent 只能写自己 key 下的内容。elapsed_ms由 agent 自己上报agency 也会记录一份两边对不上就说明有网络延迟或时钟问题。4.3 实现路由与派发路由配置用 YAML直观好改# agency/config/routes.yaml routes: - task_type: text_classify agents: [agent_a, agent_b] strategy: round_robin timeout_ms: 3000 - task_type: image_process agents: [agent_c] strategy: single timeout_ms: 10000派发逻辑的核心是查表加负载均衡# agency/dispatcher.py import yaml import itertools from common.protocol import TaskContext class Dispatcher: def __init__(self, config_path): with open(config_path) as f: self.routes yaml.safe_load(f)[routes] self._counters {} def _pick_agent(self, route): agents route[agents] if route[strategy] single: return agents[0] # 轮询 idx self._counters.get(route[task_type], 0) self._counters[route[task_type]] (idx 1) % len(agents) return agents[idx] def dispatch(self, ctx: TaskContext): route next( (r for r in self.routes if r[task_type] ctx.task_type), None ) if not route: raise ValueError(fno route for {ctx.task_type}) agent_id self._pick_agent(route) ctx.timeout_ms route[timeout_ms] return agent_id, ctx这段代码故意写得简单实际项目里还要加上指标上报、异常处理、配置热更新。但核心逻辑就这么多路由的本质就是查表加选实例不要想复杂。4.4 执行体的标准写法执行体继承一个基类基类负责和队列交互、上报状态、处理超时子类只写业务逻辑# agents/base.py import time from common.protocol import TaskContext, TaskResult class BaseAgent: agent_id base def handle(self, ctx: TaskContext) - dict: raise NotImplementedError def run(self, ctx: TaskContext) - TaskResult: start time.time() try: data self.handle(ctx) return TaskResult( task_idctx.task_id, agent_idself.agent_id, statussuccess, datadata, elapsed_msint((time.time() - start) * 1000) ) except Exception as e: return TaskResult( task_idctx.task_id, agent_idself.agent_id, statusfailed, errorstr(e), elapsed_msint((time.time() - start) * 1000) )子类只需要实现handle方法# agents/agent_a.py from agents.base import BaseAgent class AgentA(BaseAgent): agent_id agent_a def handle(self, ctx): text ctx.payload.get(text, ) # 这里写具体的分类逻辑 return {label: positive, confidence: 0.92}这种写法的好处是业务逻辑和框架逻辑彻底分离。想加一个新 agent复制一个文件改handle就行不用关心队列怎么连、结果怎么发、超时怎么处理。我团队里新人上手基本半天就能写出第一个可用的 agent。4.5 结果聚合与返回聚合器根据任务类型选择策略默认全部成功才算成功# agency/aggregator.py class Aggregator: def aggregate(self, results): if not results: return {status: failed, error: no results} failed [r for r in results if r.status ! success] if failed: return { status: failed, error: f{len(failed)} agents failed, partial: [r.data for r in results if r.status success] } merged {} for r in results: if r.data: merged.update(r.data) return {status: success, data: merged}这里特意保留了partial字段前面说过部分结果对排查问题很有价值。实际返回给上层的时候可以不带但日志里一定要有。5. 常见问题与排查技巧实录5.1 任务丢失消息发出去了但 agent 没收到这是消息队列方案最常见的问题原因通常有三个。第一是队列的持久化没开服务重启消息就没了第二是 agent 的消费确认机制有问题消息被取走但处理失败后没有重新入队第三是队列满了被丢弃或者设置了过期时间消息自动过期。排查顺序我一般是先看队列的积压指标如果积压为 0 但任务确实没执行那就是消息根本没进去或者被消费了没确认如果积压持续增长那就是 agent 消费能力不够或者卡住了。确认机制我强烈建议用手动确认处理成功才 ack失败就 nack 重新入队配合最大重试次数防止死循环。注意重新入队的消息一定要带重试计数超过阈值就进死信队列不要无限重试。我见过一个 bug 导致某条消息重试了几十万次把整个队列都堵死了。5.2 结果不一致同一个任务两次执行结果不同如果 agent 是无状态的同一个输入两次执行结果应该一致。出现不一致通常是这几个原因agent 内部用了随机数或时间戳但没固定种子agent 依赖了外部可变数据比如数据库里的最新值上下文传递过程中字段丢失或顺序变化导致逻辑分支不同。排查方法是把两次执行的完整上下文和结果都打出来对比差异点就是问题所在。我一般会在 agency 层加一个“重放”功能把历史任务的上下文原样再发一次对比结果。这个功能在排查偶发问题时特别好用。5.3 性能瓶颈任务越积越多性能问题先定位瓶颈在哪一层。如果 agency 的 CPU 高那是路由或聚合逻辑太重考虑把聚合下推到 agent 或者用更高效的数据结构如果队列积压但 agent 的 CPU 不高那是 agent 在等 IO考虑加并发或优化外部调用如果 agent 的 CPU 打满那就是计算密集只能加实例或优化算法。我遇到过一个隐蔽的瓶颈agency 每次派发任务都要重新加载路由配置配置大了之后光解析 YAML 就占了不少 CPU。改成启动时加载一次、变更时热更新就好了。任何“每次请求都做一遍”的重复工作都要怀疑一下能不能缓存。5.4 常见问题速查表现象可能原因排查动作解决方向任务丢失队列未持久化 / 未确认查队列积压和确认日志开持久化改手动确认结果不一致随机性 / 外部依赖对比两次上下文和结果固定种子隔离外部依赖任务积压消费能力不足看 agent CPU 和队列长度加实例优化 IO超时频繁超时设置过短看各 agent 耗时分布调超时拆分大任务内存泄漏上下文未释放看 agent 内存曲线检查缓存和引用5.5 几个我踩过的坑第一个坑是超时时间设得太统一。所有任务都设 5 秒结果图片处理类任务经常超时文本类任务又浪费了等待时间。后来改成按任务类型配置超时图片类给 30 秒文本类给 2 秒超时率立刻降下来了。第二个坑是日志打太多。为了排查问题每个 agent 把完整上下文都打进日志结果日志量暴涨磁盘几天就满了还拖慢了 agent。后来改成只打关键字段加追踪 ID需要详细信息的时候按追踪 ID 去追踪系统查。第三个坑是没有做优雅停机。agent 收到停止信号直接退出正在处理的任务就丢了。后来加了信号处理收到停止信号后先停止消费新消息把当前任务处理完再退出配合队列的重新入队机制基本做到零丢失。6. 扩展方向这套架构还能怎么用agency-agents 这套分层思路不只适用于任务分发很多场景都能套。比如做数据管道agency 负责调度和监控agents 负责各种数据转换加一个新的转换逻辑就是加一个 agent。比如做自动化测试agency 负责编排测试步骤agents 负责各种检查项测试报告由 agency 统一汇总。再比如做内容审核agency 负责路由和聚合agents 各自负责一种审核维度命中规则就返回。我个人的体会是这套架构最大的价值不是某个具体实现而是“调度与执行分离”这个思路本身。一旦你习惯了把横切关注点集中到 agency 层把业务逻辑下沉到 agents 层很多原本纠缠不清的问题会突然变得清晰。代码是这样团队分工其实也是这样有人负责协调和兜底有人负责专注执行各司其职整体才稳。最后分享一个小技巧如果你现在手上的系统还没有明确的分层可以先从“把重试逻辑抽出来”开始。找一个重复实现重试的地方把它统一到一个中间层你会立刻感受到解耦带来的好处。等这个中间层慢慢长大它自然就变成了你的 agency 层。架构不是一开始就设计出来的很多时候是长出来的。
