hermes-agent：多智能体协作的任务路由与消息分发中间层

如果你做过两个以上的 Agent 项目，大概会有一种感觉：模型变聪明了，但把多个模型拼在一起这件事，并没有变简单。最近我在梳理一套叫 hermes-agent 的多智能体调度层设计，它不做推理、不写 prompt，专门负责一件事——当一个任务进来时，判断应该交给哪个 Agent 处理，处理完再把结果送回该去的地方。

说白了，它是一个信使层。名字里的 Hermes 在神话里就是跑腿送信的神，agent 又点明了它的服务对象。不只服务 Agent，也能服务普通 API、工具函数、脚本任务。我把它理解为一套“任务路由 + 消息分发”的中间层，适合多 Agent 协作、自动化流水线，以及任何需要把一件事拆给多个执行者的场景。如果你正在被 Agent 之间互相调用的硬编码关系搞到头大，或者想让多个模型按职责分工而不是靠 prompt 硬怼，这篇文章应该能提供一套可落地的心法。

1. 项目全景：hermes-agent 要解决的不是“更聪明”，而是“不乱”

1.1 多智能体协作中最常见的四种混乱

先说一个诡异的现象：很多人一开始接触 Agent，习惯让一个“主 Agent”把所有事都干了——它自己拆任务、自己调工具、自己写总结。这种模式在一两个任务时没问题，一旦业务量上来，立刻会踩到四类问题。

第一种是点对点调用。Agent A 要调 B，B 要调 C，C 又要调回 A，形成调用环。代码层面看起来每个 Agent 都很干净，但任务真正跑起来，一个环节挂了，整个链条看不到问题到底出在哪。更难受的是，任何一个小改动都要牵连上下游。

第二种是上下文爆炸。A 处理完任务后，把全部对话历史传给 B，B 又追加一轮再传给 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 系统用 JSON，B 系统用 XML，C 系统干脆传一个 text 字段让下游自己解析。这就像公司里一半人用微信、一半人用邮件，互相都不知道去哪找对方。

hermes-agent 的消息体我建议固定几个核心字段：

这里有几个容易被忽略的细节。message_id 一定不能让下游自己生成，必须由 hermes-agent 统一分配，这样你才能做幂等处理。ttl 也很关键，没有超时机制的消息就是定时炸弹，一个 Agent 卡死会导致整条链路都卡住。callback 字段我建议一开始就预留，哪怕暂时用不上，因为等系统大了再回填协议字段，成本远高于一开始多写四个字节。

还有一个建议是加 version 字段。消息协议一定会演进，没有版本号，线上跑着老格式的调用方，你连兼容性补丁都不知道往哪里打。协议统一这件事看着不起眼，实际上它是整个信使层性价比最高的设计。

2.2 路由策略：从规则分流到语义分发

消息进来了，接下来是路由。路由是整个 hermes-agent 最核心的决策点，也是我和别人聊架构时被问最多的地方。其实路由策略就三条路，看你系统的特点选。

规则路由是最简单也最可靠的方式。根据消息的 type 字段直接匹配 Agent，比如 type=order 就发到订单 Agent，type=refund 就发到退款 Agent。如果业务方提交消息时 type 已经定好，那就不要折腾，直接用规则。它的优点是零延迟、零成本、结果完全确定，缺点是需要调用方在消息里带对字段。

语义路由是规则路由的补充，适合那些你没法控制上游的场景。比如用户发来的是一段自然语言，你需要判断该给客服、技术支持还是销售。常见做法是把每个 Agent 的能力描述用 embedding 模型向量化，任务进来后也做向量化，然后算相似度，取最匹配的 Agent。这种方式灵活，但要注意 embedding 模型的质量和计算耗时，而且一定要有阈值，相似度太低就别硬塞给某个 Agent，进兜底队列更稳妥。

混合路由是我个人最推荐的。先按规则走，规则能命中就直接分发；规则不命中的再用语义判断；两边都拿不准的就走 fallback Agent，或者交给人工处理。这样既保证了常规任务的确定性，又保留了长尾任务的处理能力。

不管用哪种路由，有一个观点我想强调：路由判定结果一定要留日志。谁分发的、根据什么规则分发的、命中还是兜底，这些信息比模型生成的回复内容更值得复盘。没有路由日志，后面一旦分发错，你只能靠猜。

2.3 任务生命周期与状态机设计

消息发出去之后，任务不是只有“成功”和“失败”两个状态。如果你的系统只有这两个状态，那你基本没法做分布式追踪。我建议把任务生命周期设计成一条清晰的状态链。

状态我用这几个：pending 表示任务已进入队列但还没有被消费者取走；dispatched 表示已经被分发到某个 Agent；running 表示 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_factory=dict) source: str = "api" priority: int = 0 callback: Optional[str] = None ttl: int = 300 max_attempts: int = 3 created_at: str = Field(default_factory=lambda: 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 作为内存队列，消费者常驻后台，从队列里取消息、路由、调用 Agent：

import asyncio from fastapi import FastAPI from pydantic import ValidationError import uuid app = FastAPI(title="hermes-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_to=order [succeeded] message_id=... result={'agent': 'order', 'status': 'succeeded', 'result': '订单处理完成：我的订单想退款'} [pending] message_id=... route_to=support [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 系统就能在这条线之上持续稳定地生长。