1. 从agency-agents这个名字说起一个被低估的架构模式第一次看到agency-agents这个组合词很多人会下意识地把它理解成代理的代理或者某个具体框架的名字。但如果你在分布式系统、任务调度或者多智能体协作领域待过一段时间就会意识到这个词背后指向的其实是一种非常经典却又经常被做歪的架构思路——把代理和执行体这两层职责彻底拆开让代理层只负责决策与编排让执行体只负责干活。我在过去几年里接触过不少号称智能调度自动化协作的系统发现一个普遍现象大部分项目在早期为了快速跑通流程会把决策逻辑和执行逻辑揉在一个模块里。刚开始看起来很方便一个函数里既判断条件又调用工具代码量少、上手快。但等到业务复杂度上来之后这套东西就会变成一团乱麻——你想改一个调度策略结果发现要动十几个执行函数你想替换一个执行后端结果发现决策层里到处硬编码了执行细节。agency-agents 这个模式的价值恰恰在于它从一开始就把这两件事分清楚了。那它到底解决什么问题简单说它解决的是谁来决定做什么和谁来做之间的耦合问题。代理层agency负责理解目标、拆解任务、选择执行体、汇总结果执行体agents负责接收明确指令、完成具体动作、返回结构化结果。两者之间通过一套约定好的协议通信互不关心对方内部怎么实现。这套思路在任务编排、自动化流水线、多角色协作系统里都能直接套用。这篇文章适合谁看如果你正在设计一个需要动态调度多个执行单元的系统或者你手里已经有一个越改越乱的自动化项目再或者你只是对多智能体协作的架构模式感兴趣那接下来的内容应该能给你一些可以直接抄作业的思路。我会从核心概念拆解开始讲到协议设计、执行体管理、调度策略最后落到实际落地时最容易踩的几个坑。2. 代理层与执行体到底该怎么切分职责2.1 代理层不是更聪明的执行体很多人做架构设计时容易犯一个错误把代理层当成高级执行体来写。具体表现就是代理层里开始出现具体的业务逻辑比如如果用户要查天气就调用天气接口这种判断。一旦这么写代理层就退化成了一堆 if-else 的集合失去了它本该有的通用性。代理层的核心职责应该只有四件事目标解析、任务拆解、执行体选择、结果聚合。它不应该知道天气接口怎么调它只需要知道有一个执行体声称自己能处理天气查询我把查询请求交给它就行。这个边界一旦模糊整个架构的扩展性就会迅速崩塌。我见过一个比较典型的反例某团队做了一个自动化运维系统代理层里硬编码了如果是数据库告警就走A流程如果是应用告警就走B流程。后来他们想接入一个新的监控源发现代理层里到处是针对旧监控源的假设改起来牵一发动全身。正确的做法应该是让执行体自己声明我能处理哪类告警代理层只负责根据声明做匹配。2.2 执行体的最小接口约定执行体这一侧最重要的设计决策是接口的最小化。一个执行体对外暴露的东西越少它内部的变化就越不会影响到代理层。根据我的经验一个执行体至少需要暴露三个信息能力声明、输入契约、输出契约。能力声明用来说明我能做什么通常是一个标签集合或者能力描述符。输入契约定义我需要什么参数输出契约定义我会返回什么结构。这三样东西定清楚之后代理层就可以完全基于契约来做调度不需要关心执行体内部是用什么语言写的、部署在哪里、用的是哪种数据库。这里有个实操细节值得注意能力声明不要写得太细。我见过有人把能力声明写成了一棵巨大的分类树结果代理层做匹配时反而变得极其复杂。比较好的做法是用扁平化的标签体系每个执行体声明三到五个核心标签代理层用简单的集合运算就能完成匹配。2.3 为什么不能把两者合并有人可能会问既然代理层和执行体要通信那为什么不干脆合并成一个模块省得来回传数据这个问题我在早期也纠结过后来想明白了合并的代价不是性能而是演化能力。当两者合并时任何一方的变更都会直接影响另一方。你想换一个执行后端得改代理逻辑你想调整调度策略得动执行代码。而在拆分的架构下代理层可以独立演化——今天用规则引擎做调度明天换成基于优先级的队列执行体完全不用动。执行体也可以独立演化——今天用本地进程明天换成远程服务代理层也完全不用动。这种独立演化能力在系统规模小的时候看不出价值但一旦执行体数量超过十个、调度规则超过二十条合并架构的维护成本就会指数级上升。所以我的建议是哪怕一开始只有两三个执行体也要把这两层分开因为后期重构的成本远高于前期多写一点胶水代码的成本。3. 代理与执行体之间的通信协议设计3.1 消息格式的选择结构化优先代理层和执行体之间的通信最忌讳的就是用自然语言或者非结构化的文本。我见过一些项目为了灵活让代理层直接给执行体发一段描述性文字执行体自己解析。这种做法在演示阶段看起来很酷但一旦进入生产环境解析失败、歧义理解、格式漂移这些问题就会层出不穷。比较稳妥的做法是用结构化消息通常是 JSON 或者类似的键值结构。消息里至少包含这几个字段任务标识、能力标签、输入参数、超时设置、回调地址。任务标识用于追踪和去重能力标签用于执行体自检输入参数是具体数据超时设置防止执行体卡死回调地址用于异步返回结果。{ task_id: task-20240501-001, capability: data.transform, params: { source: raw_input, format: csv, target_format: json }, timeout_ms: 30000, callback: agency://result/task-20240501-001 }这个格式看起来简单但每个字段都有讲究。task_id的生成规则要保证全局唯一通常用时间戳加随机后缀capability要和执行体声明的标签完全一致不能有模糊匹配timeout_ms要根据执行体的历史表现来设定不能一刀切。3.2 同步与异步的取舍通信模式上同步和异步各有适用场景。同步模式适合短任务、强依赖的场景比如代理层需要拿到执行结果才能继续下一步。异步模式适合长任务、弱依赖的场景比如批量数据处理、耗时计算。我的经验是默认走异步只在明确需要阻塞等待时才用同步。原因是异步模式对代理层的压力更小代理层不需要为每个任务维持一个等待线程只需要在收到回调时处理结果即可。而且异步模式天然支持超时和重试执行体挂了也不会把代理层拖死。但异步模式有个坑回调地址的管理。如果代理层是分布式的回调地址必须能路由到正确的实例。常见的做法是用一个共享的消息队列作为回调通道代理层从队列里消费结果而不是直接暴露 HTTP 回调地址。这样代理层的实例可以随意扩缩容不用担心回调丢失。3.3 错误码与重试语义协议设计里最容易被忽视但又最重要的部分是错误码体系。很多项目只定义了成功和失败两种状态结果代理层无法区分执行体暂时不可用和执行体明确拒绝任务这两种情况导致重试策略要么过于激进要么过于保守。我建议至少定义这几类错误码可重试错误如执行体过载、网络抖动、不可重试错误如参数格式错误、能力不匹配、部分成功如批量任务中部分失败、超时。代理层根据错误码决定是重试、降级还是直接失败。重试语义也要明确重试次数、重试间隔、退避策略。我通常用指数退避第一次重试等 1 秒第二次等 2 秒第三次等 4 秒最多重试三次。超过三次就标记为失败进入死信队列人工处理。这个策略在大多数场景下都能平衡成功率和系统压力。4. 执行体的注册、发现与健康管理4.1 注册机制主动上报还是被动探测执行体怎么让代理层知道自己存在两种常见方式主动注册和被动探测。主动注册是执行体启动时向注册中心上报自己的能力和地址被动探测是代理层定期扫描已知的执行体列表检查它们是否存活。主动注册的优点是实时性好执行体一启动就能被代理层发现缺点是需要一个可靠的注册中心而且执行体下线时如果没来得及注销会留下脏数据。被动探测的优点是实现简单不需要额外的注册中心缺点是发现新执行体有延迟而且扫描频率不好把握。我的建议是两者结合执行体启动时主动注册同时代理层定期做健康检查来清理失效条目。注册中心可以用一个简单的键值存储来实现不需要引入太重的组件。健康检查用轻量级的探针比如执行体暴露一个/health接口代理层每隔 30 秒调一次连续三次失败就标记为不可用。4.2 能力标签的版本管理执行体的能力标签不是一成不变的。今天一个执行体只支持 CSV 转 JSON明天可能就支持 CSV 转 XML 了。如果代理层缓存了旧的能力列表就会错过新能力。所以能力标签需要版本管理。比较简单的做法是给每个能力标签加一个版本号执行体注册时上报当前版本。代理层在调度时检查版本是否匹配不匹配就重新拉取能力列表。版本号可以用时间戳或者递增整数关键是变更时要能体现出来。另一个细节是能力标签的废弃。如果某个执行体不再支持某个能力不能直接删掉标签否则代理层可能还在往这个执行体派发任务。正确的做法是先把标签标记为废弃代理层停止向该执行体派发对应任务等一段时间后再彻底移除。4.3 健康检查的粒度与频率健康检查的粒度是个需要权衡的问题。检查太粗执行体内部某个子模块挂了但整体还响应代理层会继续派发任务导致失败检查太细每次检查都要消耗执行体的资源高频检查反而影响正常任务执行。我的经验是分两层进程级健康检查和能力级健康检查。进程级检查只判断执行体是否存活频率可以高一些比如 10 秒一次能力级检查判断具体能力是否可用频率低一些比如 60 秒一次。代理层在派发任务前先看进程级状态如果进程级正常但能力级异常就把任务派给其他执行体。健康检查的失败处理也要设计好。连续失败几次才标记为不可用我通常用三次避免因为偶发网络抖动误判。标记为不可用后多久重新尝试我通常用 30 秒给执行体恢复的时间。恢复后是否立即接收任务我建议先进入预热状态接收少量任务验证稳定性再恢复正常流量。5. 调度策略从简单轮询到优先级队列5.1 轮询与随机简单但不够用最简单的调度策略是轮询代理层维护一个执行体列表每次派发任务时按顺序选下一个。这种策略实现简单负载均衡效果也还行。但它的前提是所有执行体的处理能力相同而且所有任务的优先级相同。一旦这两个前提不成立轮询就会出问题。随机策略比轮询稍微好一点至少不会因为列表顺序导致某些执行体总是先被选中。但随机策略的问题是不可控——你可能连续多次选中同一个执行体导致它过载而其他执行体闲着。这两种策略适合执行体数量少、任务同质化程度高的场景。一旦执行体数量超过十个或者任务有轻重缓急之分就需要更复杂的调度策略。5.2 基于负载的调度基于负载的调度核心思想是把任务派给当前最闲的执行体。执行体定期上报自己的负载指标比如当前队列长度、CPU 使用率、内存占用等。代理层根据这些指标计算一个综合负载分数选分数最低的执行体派发任务。这个策略的关键是负载指标的选择和权重。队列长度是最直接的指标但不同执行体的队列处理速度不同单纯看队列长度可能不公平。CPU 和内存指标能反映执行体的资源压力但采集频率和精度会影响准确性。我的做法是用队列长度为主CPU 和内存为辅权重分别是 0.6、0.2、0.2然后根据实际效果微调。另一个细节是负载指标的时效性。如果执行体每 60 秒才上报一次负载代理层拿到的可能是过时数据。所以负载上报频率要高一些比如 10 秒一次同时代理层在派发任务时做本地计数把已派发但未完成的任务也算进负载里。5.3 优先级队列与抢占当任务有优先级之分时就需要优先级队列。高优先级任务排在前面低优先级任务排在后面。但这里有个问题如果低优先级任务已经派发出去了高优先级任务来了怎么办这就涉及到抢占。抢占的实现比较复杂因为执行体可能已经执行了一半强行中断会导致状态不一致。我的建议是只在任务粒度足够小的时候才支持抢占比如每个任务执行时间不超过 5 秒。如果任务执行时间长就不要抢占而是让高优先级任务排队等待同时给低优先级任务设置较短的超时让它尽快让出资源。优先级队列的另一个坑是优先级反转。低优先级任务占着执行体不放高优先级任务一直等结果低优先级任务因为资源不足执行得更慢。解决方法是给每个优先级设置配额低优先级任务最多占用一定比例的执行体保证高优先级任务总有资源可用。6. 结果聚合与状态追踪的实操细节6.1 任务状态的完整生命周期一个任务从代理层发出到最终完成中间会经历多个状态。如果状态管理没做好出了问题就很难排查。我通常定义这几个状态已创建、已派发、执行中、已完成、已失败、已超时、已取消。每个状态之间的转换都要有明确的触发条件。比如已派发到执行中是执行体确认接收任务后触发执行中到已完成是执行体返回成功结果后触发。代理层要记录每次状态转换的时间戳这样排查问题时可以清楚地看到任务在哪个环节卡住了。状态数据要持久化不能只放在内存里。我见过一些项目为了性能把状态放在内存缓存里结果代理层重启后所有任务状态丢失正在执行的任务变成了孤儿。持久化可以用关系数据库也可以用键值存储关键是写入要可靠不能因为性能问题丢数据。6.2 部分失败的处理策略批量任务经常出现部分成功部分失败的情况。比如派发了 100 个子任务80 个成功20 个失败。这时候代理层怎么处理直接标记整体失败显然不合理因为 80 个成功的成果就浪费了直接标记整体成功也不对因为 20 个失败的任务需要处理。我的做法是引入部分成功状态同时记录成功和失败的子任务列表。代理层根据业务需求决定后续动作如果业务允许容忍部分失败就直接返回部分成功的结果如果业务要求全部成功就对失败的子任务进行重试重试后仍然失败的进入人工处理队列。这里有个实操技巧失败子任务的重试要单独调度不要和原始任务混在一起。因为失败原因可能各不相同有的需要换执行体有的需要修改参数混在一起重试会导致策略不清晰。6.3 结果聚合的幂等性代理层在聚合结果时必须保证幂等性。也就是说同一个任务的结果被处理多次最终状态应该是一致的。这个要求看起来简单但在异步回调的场景下很容易出问题——执行体可能因为网络问题重复发送回调代理层如果没做去重就会把同一个结果处理两次。实现幂等性的常见方法是用任务标识做去重。代理层维护一个已处理任务标识的集合收到回调时先检查标识是否已存在存在就忽略不存在就处理并记录。这个集合可以用布隆过滤器来节省空间也可以用数据库的唯一索引来保证。另一个细节是结果聚合的顺序。如果多个子任务的结果需要按特定顺序合并代理层要保证聚合时顺序正确。我通常给每个子任务加一个序号聚合时按序号排序避免因为回调到达顺序不同导致结果错乱。7. 落地时最容易踩的五个坑7.1 代理层变成了瓶颈代理层负责调度和聚合如果执行体数量多、任务量大代理层很容易成为性能瓶颈。我见过一个项目代理层用单线程处理所有调度请求结果任务量一上来调度延迟从毫秒级涨到了秒级。解决办法是代理层水平扩展。但代理层扩展有个前提状态要外置。如果代理层把任务状态放在本地内存扩展后不同实例的状态就不一致了。所以状态必须放在共享存储里代理层实例变成无状态的这样才能随意扩缩容。另一个优化点是调度决策的缓存。如果执行体的能力列表和负载指标变化不频繁代理层可以缓存这些数据减少对注册中心和监控系统的查询。缓存的有效期根据实际情况设定通常 5 到 10 秒比较合适。7.2 执行体的假死问题执行体进程还在健康检查也正常但实际已经无法处理任务了——这就是假死。常见原因包括线程池满了、死锁、依赖的外部服务挂了但没做超时处理等。假死的执行体会导致任务派发过去后一直没结果代理层等到超时才发现问题。应对假死超时机制是最后一道防线。代理层给每个任务设置合理的超时时间超时后标记任务失败并重新调度。同时执行体内部也要有自我保护机制比如线程池满了就拒绝新任务而不是让任务排队等到天荒地老。更主动的做法是执行体定期上报心跳和负载代理层如果发现某个执行体心跳正常但负载指标异常比如队列长度持续增长但完成数不增长就主动把它标记为可疑减少派发给它的任务量。7.3 能力标签的漂移能力标签漂移是指执行体实际能做的事情和它声明的能力不一致。比如执行体声明支持图片压缩但实际上只支持 JPEG 格式遇到 PNG 就报错。这种问题在演示阶段不容易发现因为演示用的数据往往是最常见的格式。解决办法是在注册时做能力验证。代理层或注册中心在收到执行体的能力声明后发送一个测试任务验证能力是否真的可用。测试任务用各种边界情况的数据比如不同格式、不同大小、不同编码。验证通过才允许执行体接收正式任务。另一个做法是运行时监控能力成功率。代理层统计每个执行体每个能力的成功率和失败率如果某个能力的失败率超过阈值就自动降低该执行体的权重同时告警通知人工检查。7.4 重试导致的重复执行重试是保证可靠性的重要手段但重试也可能导致重复执行。比如执行体实际上已经完成了任务但返回结果时网络断了代理层以为失败就重试结果任务被执行了两次。对于有副作用的操作比如扣款、发邮件重复执行的后果很严重。解决办法是执行体侧做幂等。执行体收到任务时先检查任务标识是否已处理过处理过就直接返回上次的结果不再执行。这要求执行体持久化已处理任务的标识和结果增加了执行体的复杂度但对于有副作用的操作是必须的。如果执行体无法做幂等代理层就要谨慎重试。只对明确可重试的错误码进行重试对于结果未知的情况比如超时先查询执行体的状态再决定是否重试。7.5 监控与可观测性不足很多项目在功能跑通后就急着上线监控和可观测性做得很少。结果线上出问题时只能靠日志一点点翻排查效率极低。agency-agents 这种架构涉及多个组件没有好的可观测性几乎无法运维。我建议至少做三件事指标采集、链路追踪、日志聚合。指标包括任务派发量、成功率、平均延迟、执行体负载等链路追踪记录一个任务从代理层到执行体再回到代理层的完整路径日志聚合把各个组件的日志集中存储方便搜索和关联。指标要设置合理的告警阈值。比如任务失败率超过 5% 告警执行体不可用数量超过 10% 告警调度延迟超过 1 秒告警。告警不要太多太杂否则会让人麻木关键告警反而被忽略。8. 一个可复现的最小实现思路8.1 技术选型与目录结构如果你想自己动手实现一个 agency-agents 的最小版本我建议从最简单的技术栈开始。代理层用一个 HTTP 服务执行体也用 HTTP 服务两者之间用 JSON 通信。存储用 SQLite 或者 Redis前者适合单机后者适合分布式。不需要一上来就上消息队列、服务网格这些重组件。目录结构可以这样组织agency/放代理层代码agents/放执行体代码shared/放共享的协议定义和工具函数。代理层里再分scheduler/、registry/、aggregator/三个模块分别对应调度、注册、聚合。执行体里分handler/、reporter/两个模块分别对应任务处理和状态上报。# shared/protocol.py from dataclasses import dataclass from typing import Dict, Any dataclass class TaskMessage: task_id: str capability: str params: Dict[str, Any] timeout_ms: int callback: str dataclass class TaskResult: task_id: str status: str # success, failed, timeout data: Dict[str, Any] error_code: str error_msg: str 这个协议定义虽然简单但已经包含了前面说的所有关键字段。实际使用时可以根据需要扩展比如加优先级、加重试次数、加追踪标识。8.2 代理层的核心循环代理层的核心循环做三件事接收任务、调度任务、处理结果。接收任务可以是一个 HTTP 接口调度任务是一个后台线程处理结果可以是另一个 HTTP 接口或者消息队列消费者。调度任务的逻辑是从待调度队列里取出任务根据能力标签找到可用的执行体根据负载指标选一个发送任务消息记录任务状态。如果发送失败根据错误类型决定是否重试。# agency/scheduler.py def schedule_loop(): while True: task pending_queue.pop() if not task: sleep(0.1) continue agents registry.find_by_capability(task.capability) if not agents: mark_failed(task, no_agent_available) continue agent select_by_load(agents) try: send_task(agent, task) mark_dispatched(task, agent) except SendError as e: if e.retryable: pending_queue.push(task) else: mark_failed(task, str(e))这个循环看起来简单但每个步骤都有细节可以打磨。比如select_by_load可以用加权轮询也可以用最小负载优先send_task要处理超时和连接失败mark_dispatched要持久化状态。8.3 执行体的任务处理框架执行体这一侧核心是注册能力、接收任务、执行任务、上报结果。注册能力在启动时完成接收任务用一个 HTTP 接口执行任务根据能力标签路由到具体的处理函数上报结果用 HTTP 回调或者消息队列。# agents/handler.py CAPABILITIES { data.transform: handle_transform, data.validate: handle_validate, } def handle_task(task: TaskMessage) - TaskResult: handler CAPABILITIES.get(task.capability) if not handler: return TaskResult(task.task_id, failed, {}, capability_not_found) try: result handler(task.params) return TaskResult(task.task_id, success, result) except Exception as e: return TaskResult(task.task_id, failed, {}, execution_error, str(e))这个框架的关键是异常处理要全面。处理函数可能抛出各种异常执行体要捕获所有异常并转换成标准的错误码不能让异常直接冒泡导致进程崩溃。同时执行体要记录每个任务的执行日志方便排查问题。8.4 从最小实现到生产可用的差距最小实现跑通后距离生产可用还有不少差距。持久化是最先要补的任务状态、执行体注册信息、能力列表都要持久化不能放在内存里。并发控制也要加代理层和执行体都要能同时处理多个请求。超时和重试要完善不能只靠默认值。监控和告警要接入否则出了问题不知道。我的建议是分阶段推进第一阶段跑通基本流程第二阶段加持久化和并发第三阶段加监控和告警第四阶段做压力和故障测试。每个阶段都确保稳定后再进入下一阶段不要想着一步到位。9. 关于这套架构的一些个人体会agency-agents 这个模式我用了大概两年多最大的体会是它的价值不在于技术有多先进而在于它强迫你把决策和执行这两件事想清楚。很多系统出问题根源就是这两件事混在一起导致改一处动全身。另一个体会是协议设计比代码实现更重要。协议定好了代理层和执行体可以用不同的语言写、部署在不同的环境、由不同的团队维护。协议没定好哪怕用同一种语言写在一个仓库里也会互相牵制。最后分享一个小技巧在代理层加一个模拟执行体它不真正执行任务只是按配置的延迟和成功率返回模拟结果。这个模拟执行体在开发和测试阶段非常有用可以在没有真实执行体的情况下验证代理层的调度逻辑、超时处理、重试策略。等真实执行体接入后再用模拟执行体做压力测试比直接用真实执行体安全得多。这套东西说到底就是一层窗户纸捅破了之后你会发现很多系统都可以用这个思路来重构。关键是要忍住先跑通再说的冲动在早期就把边界划清楚。前期多花一天设计协议后期可能省下一周的排查时间。