做AI应用落地这两年,我越来越觉得,单靠一个智能体打天下是不现实的。日常业务里真正的复杂任务,往往要同时处理资料检索、数据处理、方案生成、结果校验多个环节,让一个Agent全部扛下来,既容易上下文超限,也没法定位问题出在哪一环节。这就促使我去研究一种把多个智能体组织成“代理机构”形态的协作模式——也就是这次想聊的 agency-agents。这套模式的核心,是让多个Agent以类似中介机构的组织方式协同工作,由一个统一调度核心负责接单、派单和验收,每个Agent只负责自己最擅长的那段工序。它可以解决单个Agent能力边界有限、任务拆解不清晰、执行过程不可控的问题,也适合正在搭建AI工作流、想要落地多智能体协作系统的开发者借鉴参考。下面我会把这套方案的设计思路、模块职责、实操步骤和常见坑位尽量讲透。
1. 项目定位与设计思路拆解
1.1 为什么是“代理机构”而不是“自由市场”
如果让我一句话概括 agency-agents 的价值,就是“把无政府状态的多智能体变成一家有前台、有工位、有工长的公司”。很多人最初读这个词,会觉得它只是“Agent 的复数形式”,但真正把它当项目来做的时候,会发现重点全在 agency 这个限定词上。自由市场式的多智能体系统,比如让好几个 Agent 直接在同一个对话流里互相讲话、互相调用,在演示 Demo 时效果确实很好,因为模型本身有很强的归纳能力,听着听着就能自动分工。可一旦进入生产环境,问题就全冒出来了:没有明确的负责人,任务会来回踢皮球;上下文被多个 Agent 同时读写,经常出现信息串台;出了问题,你都不知道该看哪个环节的日志。
Agency 这种结构,本质上是把管理逻辑从模型推理中剥离出来。它假定模型不可靠,所以用一套确定性的调度规则来兜底:谁接收任务、谁拆分任务、谁执行、谁验收,都是预先配置好的,模型只负责在每个环节里做生成、判断或调用工具。这么做的好处非常直接——多智能体系统变成了一个可预测、可观测、可回滚的工作流,而不是一群模型在那里即兴表演。举个生活化的例子:装修房子你不会让水电工、木工、油漆工围在一起开会自由发挥,而是有一个工长安排工序、检查质量。agency-agents 里的调度核心就是这个工长,它不亲自干活,但整个工地的进度和质量都归它管。
1.2 Agency 层与 Agent 层如何分离
设计这个框架时,我们团队内部讨论最多的不是“用哪个大模型”,而是“Agent 和 Agency 的边界划在哪里”。最终确定的思路是:Agency 层负责“接单—派单—回收—质检—交付”,Agent 层只负责“执行单步任务并返回结构化结果”。这句话听起来像正确的废话,但真正落地时细节非常多。如果 Agency 层管得太细,比如连 Agent 内部的提示词都要统一模板化,那每个 Agent 就退化成了普通函数,大模型的灵活性和语义理解能力被废掉,这套系统也就没有存在的必要。反过来,如果 Agency 层管得太松,只负责把任务丢给第一个 Agent,那就又回到自由市场模式,问题一个都不会少。
所以比较合理的做法是:Agency 层只管任务生命周期、依赖关系、工具路由和结果校验这些“流程元数据”,不过问 Agent 具体用什么提示词、用什么模型、怎么思考。Agent 层则只向上暴露一个稳定的执行接口,类似run(task, context),返回(status, result, meta)。上层完全不关心内部实现,未来想替换模型、更换供应商,都只影响 Agent 本身,调度核心一行代码都不用改。我在实际项目中靠这个边界划分,把一个嵌在业务流程里的多智能体系统从“demo 能跑”推进到了“线上稳定”,改动成本主要都集中在 Agent 内部,外层几乎没有动过。
1.3 通信方式:消息总线 vs 直接调用
刚开始做原型时,我贪图省事,让调度核心直接调用每个 Agent 的方法。代码确实简洁,跑通只用了不到200行。但加了并发执行和失败重试之后,问题就来了:Agent A 调用 Agent B,B 又调用了 A,直接调用根本看不出依赖关系,日志混乱,一旦某个 Agent 超时,整个调用栈会纠缠在一起,排查问题要翻半天。后来我把通信方式改成了消息总线,才把这个坑填上。消息总线的第一个好处是解耦,每个 Agent 不关心结果是谁消费,只负责把自己的结果写到总线上,调度核心从总线拉取结果再决定下一步。
第二个好处是便于观测和审计,所有消息都有唯一 ID、时间戳、来源和去向,排查问题时直接按消息 ID 拉全链路日志就行。第三个好处是扩展性,后续想加新 Agent、做多服务分布式部署,都不用动调度的核心代码。当然,消息总线也有代价:多了一层序列化和路由,单次调用延迟会高几毫秒到几十毫秒。但对大多数基于大模型的应用来说,这个延迟完全在可接受范围内,因为模型推理本身就要几百毫秒甚至更久。需要强调的是,总线不一定要用重量级的消息队列,进程内用简单的发布订阅或事件队列也完全够用。我见过的很多半途而废的项目,不是技术选型不对,而是连“什么时候该用哪种通信方式”都没想清楚就盲目上重型中间件。
2. 核心模块与职责划分
2.1 调度中枢(Agency Core)的内部构造
调度中枢是整个框架的心脏。如果把各个 Agent 比作员工,它就是公司里的工长兼前台。我们内部习惯称它为 Agency Core,它至少包含四个组件:任务队列、Agent 注册表、状态存储、质检规则引擎。任务队列维护 pending、running、succeeded、failed 四种状态,我一般还会给它加优先级,让关键的时效性任务可以插队。队列入口处的任务结构体一定要包含task_id、parent_task_id、agent_type、input、retry_count、timeout这些字段。缺了parent_task_id,后面做依赖分析会非常痛苦,因为子任务和父任务之间一旦失去关联,多级任务链路就断掉了,查错都没法查。
Agent 注册表是一张记录了所有 Agent 元数据的表,每一条至少包含agent_name、role、endpoint、input_schema、output_schema、is_async。input_schema和output_schema不只是给人看的,调度核心要做结构校验,如果返回结果不符合声明的格式,直接判为失败并触发重试。这一步能拦截相当多模型幻觉导致的错误,比如模型把字符串字段写成了数组、漏掉了必填字段等。状态存储存放任务执行过程中的中间状态,最简单的做法是内存字典,但并发任务超过几十个之后,建议落到本地 SQLite 或 Redis。状态存储要支持按task_id查询完整轨迹,否则后面排查“某个任务为什么卡住”时,只能靠猜。
质检规则引擎是在任务完成后跑一遍确定性校验,比如输出是否为合法 JSON、是否包含必需字段、调用工具时是否传了必填参数。加上这层之后,整个系统的稳定性会有质的提升,因为大模型生成结果天生不稳定,完全指望提示词去控制是不现实的,必须用代码来兜底。这套“框架管确定性、模型管开放性”的思路,是 agency-agents 和其他纯提示词编排方案最不一样的地方。
2.2 角色型 Agent 与工具注册机制
在 agency-agents 里,Agent 不是一堆模型 API 的简单封装,而是按角色划分的能力单元。我常用这样一组角色基准配置:Planner 负责把复杂任务拆成有序的子任务清单,不做执行;Retriever 负责检索知识库、文档库、网络搜索结果,把候选信息返回给下游;Executor 负责执行具体动作,比如调用外部 API、写文件、生成图片;Evaluator 负责质检和对比,判断结果是否达到要求;Responder 负责把最终结果组织成用户友好的回答。每个角色可以对应多个具体 Agent 实例,比如 RetrieverA 查内部知识库、RetrieverB 查外部网页;角色是抽象,Agent 是具体实现。
工具注册机制是这套系统里最值得复用的部分。每个工具在注册时需要提供四类信息:名称全局唯一,建议用domain_action的命名风格,比如docs_search、data_query;描述用一句话说明用途,尽量短且包含关键词,因为这一步直接影响模型选工具的准确率;参数 Schema 用 JSON Schema 格式,声明每个参数的类型、必填、取值范围;权限级别分为read_only、write、admin,调度核心根据任务来源判断是否有资格调用。注册表建好后,模型在看到任务时会拿到工具清单,按描述匹配,所以工具描述别写太长,也别堆形容词,直接说“检索文档库中与关键词相关的条目,返回标题和摘要列表”就足够了。我在实践中发现,工具描述的字数、专业术语密度对选型准确率影响极大,这个细节后面会专门展开。
2.3 任务拆解与结果汇总的完整链路
完整链路可以分成六个阶段:接单、拆解、调度、回收、合并、质检与交付。用户提交主任务后,调度核心生成task_id并写入任务队列;Planner 把主任务拆成子任务清单,每个子任务标记依赖关系;调度核心按依赖关系并行派发给对应角色的 Agent,能并行的绝不串行,这是提升整体吞吐的关键;Agent 完成后,调度核心校验输出 Schema,写入状态存储;接着按依赖关系把子结果汇总为中间结果,传给评价器或下一步;最后 Evaluator 对照验收标准检查,不合格就触发重试或切换替代 Agent,最终由 Responder 输出结果。
这个链路里最容易出错的是第二步。Planner 拆出来的任务如果不是 DAG,就有可能在执行期出现循环依赖,所以我会让 Planner 输出一个明确的依赖列表,再由调度核心做一次拓扑排序,检测到环就立刻报错。这个实现不复杂,但能避免后续很多莫名其妙的卡顿。另外,结果汇总不能简单把所有子结果拼在一起。我的做法是:每个子任务输出结果除了正文,还必须附带一个context_relevant字段,汇总器按它做裁剪,避免把大量无关上下文丢给最终模型。实测这样做可以让最终回答的准确率提升不少,同时 token 消耗明显下降。很多团队跑到这一层就开始截图发朋友圈,但真正生产级的链路还需要考虑失败补偿、任务取消、结果回滚这些情况,越早设计越好。
3. 实操过程与核心环节实现
3.1 最小可运行骨架:从注册中心到首个任务
我直接给一个能跑起来的最小骨架思路,不要纠结具体的第三方框架,重点是理解结构。假设我们用 Python 写一个简化版,先定义任务消息结构:
@dataclass class TaskMessage: task_id: str parent_task_id: str | None agent_type: str input: dict status: str = "pending" retry_count: int = 0 timeout: int = 30 created_at: float = time.time()然后建一个 Agent 注册表,用字典存 Agent 的元数据和调用函数:
AGENT_REGISTRY = {} def register_agent(name, role, endpoint_func, input_schema): AGENT_REGISTRY[name] = { "role": role, "endpoint": endpoint_func, "input_schema": input_schema }调度核心简化版就是一个事件循环:从任务队列取任务,分发给注册表里对应的 endpoint,拿回结果后校验,写回状态存储。核心结构大致像下面这样:
class AgencyCore: def __init__(self, registry, queue, storage): self.registry = registry self.queue = queue self.storage = storage def dispatch(self, task: TaskMessage): agent = self.registry.get(task.agent_type) try: result = agent["endpoint"](task.input) self.storage.set(task.task_id, result) return {"status": "success", "result": result} except Exception as e: if task.retry_count < MAX_RETRY: task.retry_count += 1 self.queue.put(task) return {"status": "retry"} return {"status": "failed", "error": str(e)}不要小看这个不到100行的简化版,它已经把 agency-agents 的核心思想表达出来了:任务和 Agent 之间通过注册表解耦,调度核心只负责流程,不写业务逻辑。跑通之后你会很直观地看到,新增一个 Agent 只多了一行register_agent调用,业务方完全不用关心调度是怎么工作的。等到这一步跑顺了,再去引入现成的框架也不迟,因为你已经知道它内部大概率在做什么,排查问题也会有方向。
3.2 如何注册自定义 Agent 与业务工具
实际项目里,大多数人不会从零造轮子,而是在一个相对成熟的框架基础上注册自己的 Agent 和工具。这里模拟一个常见场景:做一个“用户反馈工单自动处理系统”,需要三个角色:读取工单、检索知识库、生成回复建议。注册检索工具的配置大概长这样:
tools: - name: knowledge_base_search description: 在内部知识库中检索与关键词相关的文档,返回标题和摘要列表 parameters: keyword: type: string required: true top_k: type: integer default: 5 permission: read_only注册 Agent 时,在配置里声明它使用哪些工具和模型,设置最大 token 数、温度参数。尤其要注意输入输出 Schema 要写得明确,比如输入字段包含question_text、channel,输出必须包含reply_draft、confidence、source_ids。这样调度核心才能在运行时校验结果是否符合预期,而不是等用户看了半天才发现输出是错的。这里有个扩展细节:channel字段会影响 Agent 的语气风格和回复长度策略,所以我在配置里给它单独开了映射表,不同渠道的回复模板可以复用。
我的经验是,自定义 Agent 上线前先做三轮“空跑测试”。第一轮用完全合规的输入,看基本路径是否通畅;第二轮故意传缺失参数,看校验逻辑是否触发;第三轮把工具的返回结果改成格式异常的数据,看 Agent 能不能识别并报错。这三轮过了,再放到正式链路里才不会天天救火。很多团队跳过这三轮,第一次上线就遇到线上数据格式五花八门,结果 Agent 直接崩溃,调度核心又没兜住,最后挨个救场。
3.3 编排规则、超时与失败重试策略配置
编排配置我建议单独放一个 YAML,不要散落在代码里。下面是一个简化示例:
workflow: name: ticket_handler steps: - step: intake agent: ticket_reader timeout: 15 retries: 2 - step: search_docs agent: knowledge_search depends_on: [intake] timeout: 30 retries: 3 - step: draft_reply agent: reply_generator depends_on: [search_docs] timeout: 60 retries: 1 fallback_agent: simple_rule_generator这里面有几个关键点。第一,timeout不能拍脑袋定,要看所调用的模型或服务的响应分布。一般做法是先跑一周日志,取 P95 响应时间,再加 20% 到 30% 的冗余。如果超时定太短,正常慢请求会被误杀;定太长则故障影响面变大。第二,retries要配合退避策略,我不推荐瞬时连续重试,因为模型服务一般都会有频率限制,连续重试大概率连续失败。用指数退避加抖动,比如第一次等2秒、第二次等4秒、第三次等8秒,可以显著提高重试成功率。第三,fallback_agent很实用,当一个 Agent 连续失败时,可以切换到更保守的规则型 Agent 兜底,比如从生成式回复降级到模板回复,保证业务流程不断。这个思想类似微服务里的降级策略,在 Agent 系统里同样适用。
4. 常见问题与排查技巧实录
4.1 共享上下文被覆盖:多智能体“串台”
我最早踩到的坑是上下文串台。当时所有 Agent 共用一个 context buffer,结果同时跑两个或多个子任务时,后完成的 Agent 会覆盖先完成的内容,最终输出里出现了另一个任务的数据。这个问题特别隐蔽,因为单任务调试完全正常,只有并发场景才偶发,很容易被归结为“模型状态不稳定”。后来我在状态存储里把每个task_id对应的 context 快照拉出来,对比写入时间戳,才发现同一个 context 被多个 task 同时写入,原因非常清晰——共享可变状态。
解决方案也很直接:每个子任务分配独立的 context buffer,父任务只传递裁剪后的摘要和引用信息,不让子任务直接操作全局上下文。这里给一个实操建议:无论用什么框架,都要在 Agent 接口设计里明确定义 context 的读写权限。我的默认规则是:子任务只读父任务提供的 context,写只能写自己的 result context,完成后由调度核心决定合并。这样既能保证上下文隔离,又能保证数据一致性。自打改了这条规则,类似问题再没有出现过。
4.2 互相等待导致的死锁
第二个坑是死锁。有一次线上任务全部卡住,状态长期停在running。我查了任务依赖表,发现四个 Agent 形成了一个循环:任务 A 在等任务 B,任务 B 在等任务 C,任务 C 在等任务 D,任务 D 又回过头等任务 A。原因是 Planner 在拆解时把依赖关系搞成了环,而调度核心没有做检查。解决办法分两层:第一层在拆解阶段,让 Planner 必须输出严格的 DAG,调度核心拿到依赖列表后先做拓扑排序,一旦发现环就直接让任务失败,返回明确提示,而不是让它卡死。第二层是加一个全局看门狗,每个任务从入队开始记录时间,超过阈值就自动回收并标记timed_out,调度层根据情况决定重试或降级。
看门狗可以用一个简单的定时扫描线程,也可以借助 Redis 的过期键机制,我建议两步同时做,双保险。排查死锁还有个技巧:看状态存储里的waiting_dependencies字段。如果某任务长时间显示 waiting,而它依赖的任务状态又不是 success,那你就顺着依赖链往上找,最终源头通常就是循环起点。日志里这个字段一定要记录完整,缺了它,遇到死锁就只能靠肉眼翻代码碰运气了。
4.3 工具选择偏置:模型总爱挑冷门的
第三个比较隐蔽的问题是工具选择偏置。模型在选择工具时并不完全按语义相关性,描述长短、工具在列表里的顺序都会带来偏差。我测试时发现,同样功能的两个工具,一个描述写了20个字,另一个写了200个字,模型大概率选长的那个,哪怕长的那个并不合适;排在工具列表后面的工具被选中的概率也会明显下降。优化手段主要有三个:一是工具描述标准化,所有工具用统一的“用途+参数要求+输出格式”模板,长度控制在50字以内;二是对工具列表排序,把高频工具放前面,把相似工具分组,避免模型在雷同描述上反复纠结;三是给每个工具加一个use_when字段,说明适合场景和不适合场景,模型决策准确率会明显提升。
我实测用这三种手段组合,工具命中率能回升到95%以上。如果你发现自己写的 Agent 经常调用到莫名其妙的工具,先别急着换模型,检查一下工具注册表比什么都管用。工具注册表就是给模型看的一本说明书,说明书写得乱,模型自然乱选。这一点很多人容易忽略,因为单看每个工具都觉得没什么问题,合在一起互相干扰就出来了。
4.4 问题速查表与排查顺序
下面这几种是 agency-agents 项目里出现频率最高的问题,按从启动到运行的顺序整理成表:
| 现象 | 可能原因 | 优先排查点 |
|---|---|---|
| 任务一直 pending | 调度核心没启动或队列阻塞 | 检查事件循环、队列消费者线程 |
| 任务超时频繁 | timeout 设置过短或服务响应慢 | 看 P95 响应时间日志,调整超时 |
| 输出为非法 JSON | 模型返回被截断或提示词约束不足 | 增加输出格式校验,开启重试 |
| 上下文串台 | 共享 context buffer | 改为每任务独立 context |
| 任务相互等待 | 依赖图成环 | 拓扑排序检测 DAG |
| 工具选择错误 | 描述过长或排序不合理 | 标准化描述并优化排序 |
| 并发一高就卡 | 状态存储热点锁竞争 | 使用带锁的安全容器或换 Redis |
我的排查顺序一般是:先看任务状态,定位是 pending 还是 running;再看消息链路日志,按task_id拉全链路;最后检查 Agent 的输入输出 Schema 校验,很多问题在第一层校验就会暴露。别一上来就怀疑模型能力,以我的经验,90% 的故障其实出在框架层,模型只是按规定办事,问题往往出在给它传的数据和约束上。
5. 影响范围与应用场景延伸
5.1 真正适合引入 agency-agents 的业务场景
我不会推荐所有项目都用 agency-agents,因为有些场景用单 Agent 反而更划算。但凡是需要多数据源、多步骤、需要质检闭环的场景,agency-agents 的优势就很明显。第一类典型场景是客服工单自动处理:工单读取、知识库检索、答案生成、敏感词检查,每一步都能对应一个 Agent,最终输出还要过 Evaluator,比单个 Agent 把所有事情做完稳定得多。第二类是数据分析报告生成:数据查询、指标计算、图表生成、文字总结用不同 Agent,中间结果可以人工介入检查,避免模型直接对着原始数据编造结论。第三类是知识库问答与企业内部文档审核:多文档交叉验证、结论溯源、引用标注,这些都需要明确的流程拆解和结果聚合。
这几个场景的共同点是“容错率低、步骤固定、要求可追溯”。如果业务是开放式的闲聊或创意生成,你其实不需要 agency-agents,一个模型调用就够,加了这套反而增加延迟和成本。判断标准可以很简单:如果任务能一句话说清楚,就只用一个 Agent;如果需要一张流程图才能说清楚,再考虑引入多智能体编排。
5.2 从单体进程到多服务集群的演进路线
很多团队一开始都是在单进程里跑通,Agent 之间的消息走进程内事件队列,这样最省事。但业务量上来之后,单进程会碰到几个瓶颈:并发 Agent 数量受限,模型调用集中在一个进程会打满网络连接,出故障时整个流程一起挂掉。演进路线我建议分三步走。第一步,把状态存储从内存迁移到 Redis 或数据库,这是最容易做的,收益也最大。第二步,把 Agent 进程拆成独立服务,每个 Agent 暴露 HTTP 或 gRPC 接口,调度核心不再直接调用函数,而是调用远程服务。第三步,引入任务级负载均衡和水平扩展,调度核心不关心具体哪台机器在处理,只按注册表里的服务地址分发任务。
走到第三步要注意三个问题:幂等性、状态一致性、服务发现。Agent 服务的接口必须能幂等重试,否则消息重复投递会导致重复执行;状态存储要具备跨节点一致性,至少达到最终一致;服务地址不能硬编码进注册表,要用动态服务注册,否则扩容机器就要改代码。很多团队在第二步和第三步之间卡很久,其实核心问题不在技术,而在一开始没有留出足够的接口抽象空间。如果你在设计初期就把消息结构、状态存储、注册表都做成可插拔的,后续拆服务几乎不用改业务代码。
5.3 安全边界、权限隔离与审计日志设计
多智能体系统一旦进入生产,安全问题就要认真对待,尤其是工具调用权限。一个 Agent 能调用什么工具,不能只看它的描述里写什么,而要在框架层面硬性限制。我做权限隔离时用三个维度:角色、工具权限、数据可见范围。角色决定它可以申请哪些工具;工具权限分read_only、write、admin三级;数据可见范围则规定它能访问哪些业务域。比如检索类 Agent 只能读知识库,绝不能拿到写数据库工具的权限;内容生成 Agent 可以看到用户脱敏后的信息,但看不到完整身份证号这些敏感字段。
审计日志建议设计成不可篡改的追加结构,每条日志记录消息 ID、任务 ID、调用时间、Agent 名称、输入摘要、输出摘要、token 消耗。注意不要记录全量输入输出,否则敏感信息会雪崩式扩散;记摘要和哈希,需要时再回查原始数据。另外,工具调用必须做二次确认机制,凡是write或admin级别的工具,调度核心在正式执行前要停留一次,等待人工审批信号或严格的条件检查通过后再放行。这个机制在业务里非常必要,因为模型幻觉很难彻底消除,给写操作加一道人工闸门,能避免很多灾难性的自动化事故。
说实话,我自己在折腾 agency-agents 的过程中最大的收获,不是写出了多少行代码,而是学会了一种“把不确定交给模型、把确定留给框架”的思维方式。大模型负责理解、生成、判断这些开放性问题,而流程、校验、重试、权限这些确定性问题,全部交给调度核心和配置系统。每次踩坑之后我都会发现,问题几乎都出在“该框架管的交给了模型”,或者反过来“该模型管的框架却死板限制”。希望这组经验能帮你在做多智能体协作方案时少走弯路。如果后面有空,我会把工具注册表设计和调度核心的状态机单独拿出来再细化聊聊。