如果你做过两个以上的 Agent 项目大概会有一种感觉模型变聪明了但把多个模型拼在一起这件事并没有变简单。最近我在梳理一套叫 hermes-agent 的多智能体调度层设计它不做推理、不写 prompt专门负责一件事——当一个任务进来时判断应该交给哪个 Agent 处理处理完再把结果送回该去的地方。说白了它是一个信使层。名字里的 Hermes 在神话里就是跑腿送信的神agent 又点明了它的服务对象。不只服务 Agent也能服务普通 API、工具函数、脚本任务。我把它理解为一套“任务路由 消息分发”的中间层适合多 Agent 协作、自动化流水线以及任何需要把一件事拆给多个执行者的场景。如果你正在被 Agent 之间互相调用的硬编码关系搞到头大或者想让多个模型按职责分工而不是靠 prompt 硬怼这篇文章应该能提供一套可落地的心法。1. 项目全景hermes-agent 要解决的不是“更聪明”而是“不乱”1.1 多智能体协作中最常见的四种混乱先说一个诡异的现象很多人一开始接触 Agent习惯让一个“主 Agent”把所有事都干了——它自己拆任务、自己调工具、自己写总结。这种模式在一两个任务时没问题一旦业务量上来立刻会踩到四类问题。第一种是点对点调用。Agent A 要调 BB 要调 CC 又要调回 A形成调用环。代码层面看起来每个 Agent 都很干净但任务真正跑起来一个环节挂了整个链条看不到问题到底出在哪。更难受的是任何一个小改动都要牵连上下游。第二种是上下文爆炸。A 处理完任务后把全部对话历史传给 BB 又追加一轮再传给 C。本来每个 Agent 只需要关注自己那一段结果消息里塞满了无关内容模型的注意力被稀释输出质量肉眼可见地往下掉token 成本还翻着倍。第三种是责任错位。所有任务都往主 Agent 里塞指望大模型自己 decide。系统设计成什么样完全赌模型的即时发挥。业务规则少的时候没问题规则一多模型开始乱分明明是计费问题它分给客服 Agent明明是登录报错它又分给订单 Agent。第四种是不可观测。没有统一入口和统一日志任务失败后不知道谁处理的、处理到哪一步、为什么失败。想重放一个任务得靠人肉从各个 Agent 日志里拼时间线。如果你现在还在写 if agent_a then agent_b 这种硬编码或者把 Agent 注册表放在全局变量里到处 import那这些问题你大概率已经遇见过了。hermes-agent 的核心思路就是别让业务代码继续这样乱下去。1.2 做中间人而不是做大脑我最早设计这套消息层的时候第一反应是搞一个“超级指挥 Agent”让大模型去调度其他模型。后来实际做下去发现这是最贵也最难的方式模型调用的延迟高、费用高而且调度这件事偏偏需要确定性。所以在 hermes-agent 里我的选择是把调度逻辑下沉到代码层用一个轻量的中间人来接管“谁来做”这件事。模型只负责自己擅长的部分比如理解用户意图、生成回复、抽取信息而“这条消息应该发给谁”“超时了怎么办”“重试几次”全部由代码决定。这个设计本质上是在做关注点分离。你不需要让 Agent 知道消息来自哪里、后面还有多少步骤它只需要实现一个统一的处理接口接一个任务返回一个结果。消息往哪走是 hermes-agent 的事Agent 不感知全局链路也就不会被全局链路的复杂度拖垮。这样带来的直接收益有三个。第一Agent 之间彻底解耦你可以单独替换任何一个执行者不需要改其他模块第二链路可观测每个任务走到哪个节点都有记录出了问题能直接定位第三路由规则可以热更新不用为了改一个分发策略重新发布整个服务。2. 核心设计拆解消息、路由与任务生命周期2.1 消息协议让 Agent 之间说同一种话既然要做中间层第一件事就是定义消息格式。在我见过的失败项目里有一半以上是栽在消息体上A 系统用 JSONB 系统用 XMLC 系统干脆传一个 text 字段让下游自己解析。这就像公司里一半人用微信、一半人用邮件互相都不知道去哪找对方。hermes-agent 的消息体我建议固定几个核心字段字段说明message_id消息唯一 ID用于幂等和追踪task_id业务任务 ID一次业务流程可能拆成多条消息type任务类型路由的核心依据payload实际要处理的数据保持结构化和可序列化source消息来源方便回溯priority优先级高优任务可以插队callback处理完成后的回调地址或队列名ttl消息有效期超过即认定失败max_attempts最大重试次数created_at创建时间这里有几个容易被忽略的细节。message_id 一定不能让下游自己生成必须由 hermes-agent 统一分配这样你才能做幂等处理。ttl 也很关键没有超时机制的消息就是定时炸弹一个 Agent 卡死会导致整条链路都卡住。callback 字段我建议一开始就预留哪怕暂时用不上因为等系统大了再回填协议字段成本远高于一开始多写四个字节。还有一个建议是加 version 字段。消息协议一定会演进没有版本号线上跑着老格式的调用方你连兼容性补丁都不知道往哪里打。协议统一这件事看着不起眼实际上它是整个信使层性价比最高的设计。2.2 路由策略从规则分流到语义分发消息进来了接下来是路由。路由是整个 hermes-agent 最核心的决策点也是我和别人聊架构时被问最多的地方。其实路由策略就三条路看你系统的特点选。规则路由是最简单也最可靠的方式。根据消息的 type 字段直接匹配 Agent比如 typeorder 就发到订单 Agenttyperefund 就发到退款 Agent。如果业务方提交消息时 type 已经定好那就不要折腾直接用规则。它的优点是零延迟、零成本、结果完全确定缺点是需要调用方在消息里带对字段。语义路由是规则路由的补充适合那些你没法控制上游的场景。比如用户发来的是一段自然语言你需要判断该给客服、技术支持还是销售。常见做法是把每个 Agent 的能力描述用 embedding 模型向量化任务进来后也做向量化然后算相似度取最匹配的 Agent。这种方式灵活但要注意 embedding 模型的质量和计算耗时而且一定要有阈值相似度太低就别硬塞给某个 Agent进兜底队列更稳妥。混合路由是我个人最推荐的。先按规则走规则能命中就直接分发规则不命中的再用语义判断两边都拿不准的就走 fallback Agent或者交给人工处理。这样既保证了常规任务的确定性又保留了长尾任务的处理能力。不管用哪种路由有一个观点我想强调路由判定结果一定要留日志。谁分发的、根据什么规则分发的、命中还是兜底这些信息比模型生成的回复内容更值得复盘。没有路由日志后面一旦分发错你只能靠猜。2.3 任务生命周期与状态机设计消息发出去之后任务不是只有“成功”和“失败”两个状态。如果你的系统只有这两个状态那你基本没法做分布式追踪。我建议把任务生命周期设计成一条清晰的状态链。状态我用这几个pending 表示任务已进入队列但还没有被消费者取走dispatched 表示已经被分发到某个 Agentrunning 表示 Agent 正在处理succeeded 表示处理成功failed 表示业务失败timeout 表示超时dead 表示重试耗尽进入无人认领区。为什么非要一套状态机因为多 Agent 编排本质上是异步的你没法保证 Agent 处理完一定立刻拿到结果这时候状态就是唯一的真相。我在早期版本里偷懒只在日志里打了几行字结果排查线上问题时根本分不清任务是“还没跑”还是“跑挂了还在排队”。重试逻辑必须挂在状态机里。任务失败后区分一下是业务校验失败还是基础设施异常。业务失败很多时候重试也没用硬重试会放大损耗基础设施异常则可以重试但要设最大次数。我常用的参数是每次重试间隔指数退避1 秒、2 秒、4 秒这样递增最多试三次。加了 ttl 和 max_attempts 之后你的信使层就不会因为一个坏不了的 Agent 无限等待了。3. 实操落地从零搭一个最小可用的 hermes-agent 调度层3.1 先定技术栈别一上来就上 Celery我见过不少人在最开始就上 Celery RabbitMQ Flower配了一周发现一条业务都没接进来。这里我想说明一点初步验证阶段一套 FastAPI asyncio.Queue 进程内消费者完全够用。等任务量真的上来、多个服务需要跨进程通信了再把队列换成 Redis Streams 或 RabbitMQ调度层接口不用变。接下来的示例代码我直接用 Python 实现。Python 做这类胶水层最顺手FastAPI 提供入口pydantic 做消息校验asyncio 做并发调度。示例先定义两个 Agent一个处理订单问题一个处理技术报障。路由逻辑先做关键词规则后面可以替换成你需要的任意模型。3.2 消息体定义先写消息体模型from pydantic import BaseModel, Field from datetime import datetime, timezone from typing import Any, Dict, Optional class AgentMessage(BaseModel): message_id: Optional[str] None task_id: str type: str generic payload: Dict[str, Any] Field(default_factorydict) source: str api priority: int 0 callback: Optional[str] None ttl: int 300 max_attempts: int 3 created_at: str Field(default_factorylambda: datetime.now(timezone.utc).isoformat())这里有个小技巧message_id 我不强制调用方传如果没传就在入口处用 uuid4 生成这样可以保证整个系统消息 ID 唯一。ttl 默认 300 秒也就是说一条消息从进入队列到处理完成超过 5 分钟就认定超时。这个参数你按业务调我实际用下来 5 分钟在大多数内部场景里够用。3.3 实现 Agent 注册表Agent 的注册机制是让每个执行者实现一个接口然后注册到注册表里。示例里我先定义抽象基类from abc import ABC, abstractmethod class BaseAgent(ABC): name: str base abstractmethod async def handle(self, message: AgentMessage) - Dict[str, Any]: pass class OrderAgent(BaseAgent): name order async def handle(self, message: AgentMessage) - Dict[str, Any]: # 这里模拟业务处理实际场景里可以调用模型、数据库或第三方 API content message.payload.get(content, ) return {agent: self.name, status: succeeded, result: f订单处理完成{content}} class SupportAgent(BaseAgent): name support async def handle(self, message: AgentMessage) - Dict[str, Any]: content message.payload.get(content, ) return {agent: self.name, status: succeeded, result: f技术报障已受理{content}} class FallbackAgent(BaseAgent): name fallback async def handle(self, message: AgentMessage) - Dict[str, Any]: return {agent: self.name, status: succeeded, result: 无法自动分发转人工处理}你注意看每个 Agent 完全不知道自己的消息是从哪来的、后面还要去哪。它只做一件事收消息处理返回结构化的 dict。这就是解耦的边界。注册表我直接用字典保持简单AGENT_REGISTRY { agent.name: agent for agent in [OrderAgent(), SupportAgent(), FallbackAgent()] }真实项目中Agent 可能是独立服务注册表存的是服务名和地址分发时走 gRPC 或 HTTP。但抽象思路一样实现一个能拿到当前 Agent 执行入口的函数就行。3.4 路由器实现路由函数是信使层的决策核心。先用规则路由def route_message(message: AgentMessage) - str: content message.payload.get(content, ) if 订单 in content or 退款 in content or 下单 in content: return order if 报障 in content or 故障 in content or 连不上 in content or 报错 in content: return support return fallback这样太朴素了对不对可以加上一层语义兜底逻辑。真实场景中你可以把每个 Agent 的能力描述和用户内容分别做 embedding然后算相似度。示例里我用一个伪代码形态说明一下def semantic_match(content: str, agent_registry) - str: query_vec get_embedding(content) # 调用 embedding 模型 best_agent, best_score None, -1.0 for agent_name, desc in AGENT_DESCRIPTIONS.items(): desc_vec get_embedding(desc) score cosine_similarity(query_vec, desc_vec) if score best_score: best_agent, best_score agent_name, score if best_score 0.7: # 低于阈值就兜底 return fallback return best_agent真正的生产项目里我会选择先跑规则规则命中率不足时再启用语义避免每次请求都调一次 embedding 服务导致延迟飘高。混合路由的实际效果是覆盖长尾又不牺牲常规任务的响应速度。3.5 调度器与 FastAPI 入口现在把调度器和接口串起来。这里用 asyncio.Queue 作为内存队列消费者常驻后台从队列里取消息、路由、调用 Agentimport asyncio from fastapi import FastAPI from pydantic import ValidationError import uuid app FastAPI(titlehermes-agent) queue asyncio.Queue() results {} # 演示用实际项目应换 Redis 或数据库 async def worker_loop(): while True: message await queue.get() try: agent_name route_message(message) agent AGENT_REGISTRY[agent_name] # 模拟状态流转 print(f[pending] message_id{message.message_id} route_to{agent_name}) result await agent.handle(message) results[message.message_id] result print(f[succeeded] message_id{message.message_id} result{result}) except Exception as exc: results[message.message_id] {status: failed, error: str(exc)} print(f[failed] message_id{message.message_id} error{exc}) finally: queue.task_done() app.on_event(startup) async def startup(): asyncio.create_task(worker_loop()) app.post(/submit) async def submit(message_body: dict): message AgentMessage(**message_body) if not message.message_id: message.message_id str(uuid.uuid4()) await queue.put(message) return {status: accepted, message_id: message.message_id, queue_size: queue.qsize()} app.get(/result/{message_id}) async def get_result(message_id: str): if message_id in results: return results[message_id] return {status: pending}这个小系统的流程其实是调用方 POST /submit 提交任务接口生成 message_id把消息塞进队列worker 一直监听队列拿到消息后路由到对应 Agent处理完把结果写进内存里的 results 字典。外部可以通过 GET /result/{message_id} 查询结果。这套代码看起来短但它已经把信使层最骨干的东西都覆盖了统一消息入口、统一路由、统一状态记录。你可以在 worker_loop 里继续扩展状态流转、超时、重试等逻辑。3.6 联调演示把服务跑起来后用两个请求验证路由是不是正常分发。先提交一条订单问题curl -X POST http://127.0.0.1:8000/submit \ -H Content-Type: application/json \ -d {task_id: task-001, type: inquiry, payload: {content: 我的订单想退款}}再提交一条技术报障curl -X POST http://127.0.0.1:8000/submit \ -H Content-Type: application/json \ -d {task_id: task-002, type: inquiry, payload: {content: 系统登录一直报错连不上服务}}我在本地实测的输出长这样[pending] message_id... route_toorder [succeeded] message_id... result{agent: order, status: succeeded, result: 订单处理完成我的订单想退款} [pending] message_id... route_tosupport [succeeded] message_id... result{agent: support, status: succeeded, result: 技术报障已受理系统登录一直报错连不上服务}任务都能准确分到对应 Agent链路正常。到这里一个最小可用的 hermes-agent 就算跑通了。生产化的时候你把内存队列换成 Redis Streams把 results 换成数据库再基于 worker_loop 扩展状态机和重试逻辑就是一个非常靠谱的 Agent 编排底座。4. 踩坑实录Agent 编排的常见问题与排查方法4.1 任务提交后石沉大海这是在我实际排障中最常见的问题调用方说任务已经提交了服务端日志却什么都没有。排查顺序有讲究。先看入口通不通POST /submit 是否返回了 message_id如果接口直接报错可能是参数校验没过就要把 AgentMessage 的必填字段检查一遍。再看消费者在不在跑asyncio.create_task 在 startup 事件里创建后如果主进程异常退出队列里的任务就会一直堆积没有人消费。我踩过最典型的一次一个同事把 worker_loop 的启动写在了某个请求处理函数里新起了一个事件循环结果任务全进了旧队列新消费者永远拿不到。这类问题最好的解法是从一开始就统一用 FastAPI 的 startup 钩子启动消费任务不要自己在请求里偷偷开协程。4.2 Agent 返回的格式五花八门多个团队接入 Agent 时最难统一的是返回格式。有的返回 JSON 字符串有的返回 Python dict有的把结果写进一个超大字符串然后告诉你“你自己 parse 一下”。这种问题的根子不在 Agent而在入口契约太弱。破解方法是在消息协议里明确约定返回值结构必须有 status、必须有 result、必须能被 JSON 序列化。如果调用 LLM 接口最好让它做结构化输出而不是让下游去解析自由文本。我在项目里还会加一个解析兜底层如果 Agent 返回的是字符串先尝试用 json.loads 解析解析失败就把它原样放进 result同时把 status 标记为 degraded提醒下游这个结果可能不够可靠。整个链条因此不会因为一个格式问题直接断裂。4.3 路由老是分错人规则路由分错多半是关键词覆盖不全语义路由分错多半是相似度算法和 embedding 模型不太匹配业务。我的经验是路由判断结果一定要落到日志里。任务分发到哪个 Agent、命中的规则是什么、相似度打了几分这些信息都要存下来。这样即使分错了你也有数据可以复盘而不是每次都在猜“为什么这条消息进了订单 Agent”。另一个很有效的做法是给每个 Agent 配一段能力描述定期看实际分发统计不断拿真实数据去调描述文本。有没有发现当你这样做了之后路由的准确率是在持续变好的而不是发布完就听天由命。4.4 任务堆积导致延迟越来越高最典型的场景某一天活动的流量突然上来单个 worker 处理不过来队列里的任务堆积成才。我最早发现这个问题是看到 result 查询里大量请求都返回 pending才意识到队列已经排队排到几十秒开外了。后续我做了三件事。第一是按消息类型拆分队列订单、支持、其他各走各的队列避免一个慢 Agent 堵住整条链路。第二是给 worker 池扩容每个队列对应多个消费者协程这个是能直接提升吞吐的。第三是设置 cap队列积压超过一定数量就直接拒绝新提交告诉调用方稍后重试这叫背压。内存版 asyncio.Queue 毕竟有上限流量再大一点进程就该崩了。生产环境我会把队列换成 Redis Streams支持消费组、持久化和批量拉取稳定性完全不一样。5. 后续扩展与个人经验5.1 这个信使层还能长成什么样最小版本跑通只是开始。我接下来的扩展路径是这么规划的。第一是加可视化 Dashboard。每个 message_id 经过的路由节点、到达时间、处理耗时、结果状态全部可视化排障效率会高很多。你不用看十几万行日志打开面板输入 task_id 就能看到全链路时间线。第二是统一工具调用层。Agent 今天要调订单系统明天要调支付系统如果每个 Agent 自己记 API 地址和鉴权信息又变成新一轮硬编码。我的想法是把工具调用也注册到 hermes-agent 里Agent 只能通过工具层访问外部系统这样权限控制和审计都落在一个地方。第三是记忆与缓存。相似的问题问一次就够了做一个语义缓存层命中直接返回历史答案跨 Agent 共享的记忆可以用 Redis 存会话上下文不用再靠消息体传一大段历史记录。第四是插件机制。路由算法、Agent 注册、消息中间件都做成可插拔接口这样团队可以各自实现自己的策略不用长期共用一个主分支。5.2 关于 Agent 之间要不要直连我的最终看法这几年做多智能体项目我觉得最需要的不是“让 Agent 之间能直接对话”而是“让 Agent 之间的对话可以被管理”。纯直连在演示场景里很好看两个模型互相接话像真的一样但一碰生产环境超时、重试、权限、审计哪个你都跑不掉。所以我会选择把消息层做成中心化的信使执行端保持去中心化。这样既保留了分布式系统的可扩展性又让核心链路有了秩序。这里有个度的问题信使层不要设计得太重别一上来就是几十个表、十几个配置项。先接两个真实 Agent跑通整个链路再慢慢把复用能力抽象出来。我再分享一个很细微但很值得养成的习惯每次提交消息时把 message_id、task_id、route 三样东西绑在一条结构化日志里输出。这个习惯在初期看起来没多大价值等到你某一天需要回放一次线上事故会发现这三样东西能帮你省下至少半天排查时间。hermes-agent 这种信使型设计最终帮我解决的不是大模型的聪明问题而是系统里的责任边界问题哪些事该由模型判断哪些事该由代码判断分得清清楚楚。把这条线守住Agent 系统就能在这条线之上持续稳定地生长。