hermes-agent：轻量级多智能体通信与编排框架实战

先说个背景。过去两年我一直在折腾多智能体协同相关的项目从最早的单体 Agent 调 API到后来把多个专用 Agent 拼在一起干活踩过不少坑。最让我头疼的其实不是模型本身的能力而是 Agent 之间的通信和协作方式——大家都是各自为政消息格式不统一任务状态没人管调试起来更是灾难。后来我自己动手写了一套框架取名 hermes-agent灵感来自希腊神话里的信使之神 Hermes核心就干一件事让多个 Agent 之间用统一、可靠、可追踪的方式对话和协作。这套东西不是什么大而全的平台它更像是一个轻量级的通信与编排中间层。如果你也在做多 Agent 应用或者准备从单体 Agent 往多智能体方向迁移这篇文章应该能给你一些可落地的参考。我会把设计思路、核心机制、完整实操流程和踩坑记录都摊开讲清楚代码也在文里照着重现一遍基本就能跑通。1. 为什么会写 hermes-agent多智能体场景下的现实痛点在聊框架本身之前我觉得有必要先把问题讲透。你只有真正理解了多 Agent 协作里缺的是什么才能明白 hermes-agent 里那些设计决策到底在解决什么问题。1.1 单体 Agent 的边界在哪里先看一个常见场景。你有一个 Agent它能帮你查天气、定闹钟、写周报看起来功能很全。但随着需求变多你会发现单体 Agent 的问题越来越明显提示词越写越长模型经常顾此失彼每加一个新能力都要重新测一遍旧功能有没有被影响更麻烦的是不同的任务对上下文的要求完全不一样让一个 Agent 记住所有任务的细节成本高得吓人。我在一个实际项目里做过对比。用一个单体 Agent 同时处理信息检索和内容撰写两类任务刚开始效果还行但任务复杂度上去之后检索时的长上下文会严重干扰撰写时对风格和结构的把握。后来我把这两类能力拆成两个独立 Agent各管一摊效果立刻好了很多。这就是最朴素的拆分逻辑不同职责、不同上下文、不同提示词策略的任务就不该挤在一个 Agent 里。但拆开之后新的问题马上就来了两个 Agent 之间怎么配合谁来协调它们的执行顺序它们的输出格式不统一怎么办这其实就是单体 Agent 的边界——你可以靠工程手段在内部做模块化但一旦涉及真正的多角色协同单体架构在复杂度和成本上都会很快触顶。1.2 多 Agent 协作到底缺什么多 Agent 协作说起来很好听但实际做起来你会发现自己缺的东西远比想象中多。我总结下来核心缺三样东西。第一缺的是统一的通信协议。Agent A 是 Python 写的Agent B 是 Node.js 写的Agent C 只是别人封装好的一个 HTTP 接口。它们之间怎么对话你总不能每次都写一套自定义的请求格式吧。没有统一的协议Agent 越多两两之间的适配成本就越高复杂度是指数级增长的。第二缺的是可靠的消息路由。Agent 之间通信不是简单的点对点很多时候是我把消息发出去谁感兴趣谁来处理。这就要有一套类似消息队列的机制能根据消息的类型、内容、目标去做路由。没有这层你就只能手写一堆 if-else 来调度代码很快就会烂掉。第三缺的是任务状态的共享与追踪。一个任务被拆成多个子任务分给不同的 Agent 之后你总得知道现在这个任务整体到哪一步了、每个子任务的结果是什么、中间有没有出错。没有统一的状态管理多 Agent 协作就成了一笔糊涂账出了问题你连从哪开始查都不知道。hermes-agent 就是冲着这三个痛点来的。它没有去重新发明模型能力也没有强行制定一套复杂的 Agent 开发规范它只做中间的通信与编排层。这就像你给一群各怀绝技的人配了一个信使——信使不负责干活但保证消息能准确、及时、可追溯地送到该去的人手里。2. hermes-agent 的核心设计拆解这一章我会把 hermes-agent 的几个关键设计点拆开讲。理解核心机制比单纯跑通 demo 重要得多因为只有理解了机制你才能在自己的业务场景里做出正确的取舍。2.1 消息协议Agent 之间的共同语言hermes-agent 里最基础的概念就是消息Message。每条消息不是简单的你一句我一句它有完整的结构化字段我直接贴一下核心定义# hermes_agent/message.py from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional import uuid dataclass class Message: id: str field(default_factorylambda: uuid.uuid4().hex) topic: str default # 消息主题用于路由 sender: str # 发送方 Agent 名称 recipient: Optional[str] None # 接收方None 表示广播 msg_type: str text # text / command / event / result payload: dict field(default_factorydict) # 主体内容 metadata: dict field(default_factorydict) # 链路追踪等元信息 timestamp: str field( default_factorylambda: datetime.now(timezone.utc).isoformat() ) correlation_id: Optional[str] None # 关联 ID用于追踪同一任务的多个消息 def to_dict(self) - dict: return self.__dict__.copy() classmethod def from_dict(cls, data: dict) - Message: return cls(**data)这里有几个字段我想单独拿出来说。topic是路由的核心依据。它有点像消息队列里的 Topic也有点像广播电台的频道。Agent 可以订阅自己感兴趣的 Topic然后只管接收这个 Topic 下的消息。比如订单 Agent 订阅order.created物流 Agent 订阅order.shipped互不干扰。recipient是可选的点对点目标。如果这条消息只想发给某个特定 Agent就填上对方的名字如果不填就按 Topic 广播给所有订阅者。这个设计兼顾了精准投递和发布订阅两种模式用起来非常灵活。correlation_id是我自己很得意的一个设计。多 Agent 协作时同一个业务请求会拆成好几条消息这些消息散落在不同 Agent 之间。有了correlation_id你就能把这一整串消息串起来查日志时按它一搜整个任务链路一目了然。这点在调试分布式系统式的 Agent 协作时真的能救命。2.2 路由与调度消息怎么送到该去的地方消息协议建好了接下来就是怎么路由。hermes-agent 的路由层实现得比较克制核心就是一个基于回调的注册表和一套分发逻辑# hermes_agent/broker.py import asyncio import logging from collections import defaultdict from typing import Callable, Awaitable from .message import Message logger logging.getLogger(hermes-agent) class AgentBroker: 轻量级消息代理负责注册、订阅、分发。 def __init__(self): self._agents {} # name - agent instance self._subscriptions defaultdict(set) # topic - set(agent_name) self._handlers {} # agent_name - callable def register_agent(self, agent, handler: Callable[[Message], Awaitable]): 注册 Agent 及其消息处理入口。 self._agents[agent.name] agent self._handlers[agent.name] handler logger.info(fAgent [{agent.name}] registered) def subscribe(self, agent_name: str, topic: str): 让某个 Agent 订阅指定 Topic。 self._subscriptions[topic].add(agent_name) logger.info(fAgent [{agent_name}] subscribed [{topic}]) def unsubscribe(self, agent_name: str, topic: str): self._subscriptions[topic].discard(agent_name) async def publish(self, message: Message): 把消息投递给目标。如果指定 recipient则点对点否则按 Topic 广播。 if message.recipient: if message.recipient not in self._handlers: raise ValueError(fRecipient {message.recipient} not found) await self._dispatch(message, message.recipient) return for agent_name in self._subscriptions.get(message.topic, set()): await self._dispatch(message, agent_name) async def _dispatch(self, message: Message, agent_name: str): handler self._handlers.get(agent_name) if not handler: logger.warning(fNo handler for agent {agent_name}) return try: await handler(message) except Exception: logger.exception(fAgent [{agent_name}] failed to process msg {message.id}) def route_stats(self) - dict: 返回路由统计方便排查。 return { agents: list(self._agents.keys()), subscriptions: {k: list(v) for k, v in self._subscriptions.items()}, }这套路由的设计思路是从消息队列里借鉴来的但不是完全照搬。我没有引入独立的消息中间件而是用一个进程内的异步 Broker 来承担路由职责。这样做的原因很简单在大部分多 Agent 应用里Agent 本来就是跑在同一个进程内的协程你没必要为了它们之间的通信再架一套 Redis 或 RabbitMQ那属于过度设计。但这也带来了一个取舍我后面会细说。如果未来你的 Agent 分布式部署进程内的 Broker 就不够用了那时候可以考虑把路由层替换成 Redis Stream 或 NATS。我在接口设计上特意把 Broker 做得薄一点点方便替换。2.3 记忆共享协作不是一次性的问答前面讲了消息怎么传但多 Agent 协作还有一个问题绕不开Agent 之间的记忆怎么共享。说白了Agent A 查到的信息Agent B 后续要用到那 A 就得把结果通过消息传给 B。但如果这个结果很大或者后续多个 Agent 都要用走消息通道就会很浪费。hermes-agent 做了一个非常轻量的共享记忆模块本质就是一个带命名空间的本地缓存# hermes_agent/memory.py import time from typing import Any, Optional class SharedMemory: 轻量级共享状态存储。 因为只是进程内共享所以实现得非常简单。 生产环境建议替换为 Redis 等外部存储。 def __init__(self): self._store {} self._ttl {} def put(self, key: str, value: Any, ttl: int 300): self._store[key] value if ttl 0: self._ttl[key] time.time() ttl def get(self, key: str) - Optional[Any]: expire_at self._ttl.get(key) if expire_at and time.time() expire_at: self._store.pop(key, None) self._ttl.pop(key, None) return None return self._store.get(key) def delete(self, key: str): self._store.pop(key, None) self._ttl.pop(key, None)这个模块的定位很明确它是给 Agent 之间交换中间结果用的不是给模型当长期记忆用的。你可以让 Agent A 把检索结果写到shared_memory.put(retrieve_result, data)然后 Agent B 处理完再从中取。这样优于把大段内容塞进消息体因为消息体一大会拖慢序列化和传输。不过我得提醒一句这种共享内存模式只在单进程部署时成立。如果你的 Agent 是跨机器部署的就不能这么干了得换 Redis 这类外部存储。hermes-agent 把接口封装好了替换成本不算高。3. 实操从零跑通一个多 Agent 任务流理论讲得再多不如实际跑一遍。这一章我带你完整走一遍装环境、起框架、定义两个 Agent、编排一个简单的检索-分析-总结流水线。3.1 环境准备与最小安装我的开发环境给你做个参考Python 3.10操作系统macOS 13Linux 和 Windows 同样能跑依赖pydantic、httpx、openai如果你要用大模型接口的话最小安装其实不需要安装任何额外的包因为 hermes-agent 核心代码就那两三个文件你可以直接把上一章的broker.py、message.py、memory.py保存到项目里然后开始写业务代码。我建议你先搭一个最小工程目录hermes-demo/ ├── hermes_agent/ │ ├── __init__.py │ ├── broker.py │ ├── message.py │ └── memory.py ├── agents/ │ ├── searcher.py │ ├── analyzer.py │ └── writer.py ├── main.py └── requirements.txt3.2 三分钟实现 Agent 注册与握手先定义两个最基础的 Agent。在 hermes-agent 里一个 Agent 就是一个对象它内部可以封装任意的能力——调大模型、调数据库、跑算法都行只要它对外能接收 Message 并返回一个结果。我写一个最简单的示例 Agent# agents/searcher.py from hermes_agent.message import Message class SearcherAgent: 模拟一个信息检索 Agent。 def __init__(self, name: str searcher): self.name name async def handle(self, message: Message): query message.payload.get(query, ) # 这里替换成你的真实检索逻辑比如向量数据库、搜索引擎 API 等 results [ {title: fresult_1_for_{query}, score: 0.95}, {title: fresult_2_for_{query}, score: 0.87}, {title: fresult_3_for_{query}, score: 0.72}, ] # 把结果发给下一个环节通过 Message 回复或者写入共享记忆 reply Message( topicsearch.completed, senderself.name, recipientmessage.sender, # 直接回给发起方 msg_typeresult, payload{ correlation_id: message.correlation_id, query: query, results: results, }, correlation_idmessage.correlation_id, ) return reply然后在main.py里注册它# main.py import asyncio from hermes_agent.broker import AgentBroker from hermes_agent.message import Message from agents.searcher import SearcherAgent async def main(): broker AgentBroker() searcher SearcherAgent() # 注册 Agent并绑定它的消息处理入口 broker.register_agent(searcher, searcher.handle) # 订阅模型如果消息 Topic 是 search.request就交给 searcher 处理 broker.subscribe(searcher, search.request) # 构造一条检索请求 msg Message( topicsearch.request, sendermain, payload{query: hermes-agent 使用教程}, ) await broker.publish(msg) print(route stats:, broker.route_stats()) if __name__ __main__: asyncio.run(main())跑一下如果看到route stats里 roster 有searcher基础链路就算通了。3.3 任务编排示例情报收集-分析-总结流水线单 Agent 跑通只是热身真实价值在于多 Agent 协作。我设计一个经典的流水线场景先收集素材再做分析最后写总结。这三个环节正好对应三个不同职责的 Agent。为了让流程更清晰我在消息里带上correlation_id这样可以把整个流程串起来# main_pipeline.py import asyncio from hermes_agent.broker import AgentBroker from hermes_agent.message import Message from agents.searcher import SearcherAgent from agents.analyzer import AnalyzerAgent from agents.writer import WriterAgent async def run_pipeline(broker: AgentBroker, query: str): # 1. 发起检索请求 search_msg Message( topicsearch.request, sendermain, payload{query: query}, correlation_idftask-{query[:10]}-{asyncio.time() if hasattr(asyncio, time) else 1}, ) # 2. 等检索完成回包 search_result await broker.publish_and_wait(search_msg, timeout10) # 3. 把检索结果发给分析 Agent analysis_msg Message( topicanalysis.request, sendermain, recipientanalyzer, payload{raw_data: search_result.payload[results]}, correlation_idsearch_msg.correlation_id, ) analysis_result await broker.publish_and_wait(analysis_msg, timeout15) # 4. 最后交给撰写 Agent 生成总结 writer_msg Message( topicwrite.request, sendermain, recipientwriter, payload{analysis: analysis_result.payload[analysis]}, correlation_idsearch_msg.correlation_id, ) final_result await broker.publish_and_wait(writer_msg, timeout15) return final_result.payload[article]你注意我这里调用了publish_and_wait这个是 hermes-agent 封装的一个同步等待接口。它背后的逻辑是当 Broker 把消息发给目标 Agent 后协程会等待目标 Agent 返回结果。为了支持这个模式我会在 Broker 内部维护一个 pending 的回执表每个correlation_id对应一个 futureAgent 返回时自动唤醒等待方。这个机制在同步编排任务流的时候非常好用代码读起来就像普通的同步调用但实际上底层是异步的。三个 Agent 的实现逻辑大体类似区别只在处理消息时的业务逻辑不同。AnalyzerAgent 会从原始素材里提取关键信息、生成分析结论WriterAgent 则负责把分析结论组织成长文。它们的消息类型可以统一用result也可以通过msg_type字段区分灵活处理就行。3.4 关键配置参数选择实操中你会遇到几个需要动手配置的参数我把我调参的经验列成一张表格参数建议值说明消息超时时间10~30 秒取决于你的 Agent 内部处理耗时如果调用了大模型接口尽量给足时间共享记忆 TTL300 秒中间结果保留 5 分钟足够了太长会占用内存重试次数2~3 次Agent 调用外部 API 失败时重试太长会导致雪崩并发限制按任务量进程内 Broker 不是无限并发的建议用 semaphore 控制同时运行的 Agent 数量日志级别DEBUG开发/ INFO生产多 Agent 联调强烈建议开 DEBUG能看到每条消息的路由情况超时这个参数我踩过坑。最早我把超时设成 5 秒结果每次调用大模型慢一点就超时整个流水线被中断。后来我把超时拆成两级网络请求超时短一点整体任务超时长一点这样既不会因为偶发网络抖动就崩又能及时暴露真死循环问题。4. 常见问题与排查技巧实录工具好不好用往往在踩坑的时候才能看出来。我把自己在这个框架上遇到的问题和排查思路整理成速查表方便你对照。4.1 Agent 之间消息不达路由表与 Topic 排查最常见的问题是消息发出去了但目标 Agent 没收到处理。我在日志里看到好多No handler for agent xxx警告基本就是订阅关系或者注册关系没配对。排查思路按顺序来打印broker.route_stats()看 Agent 是否注册成功、订阅的 Topic 是否正确。检查消息的topic和 Agent 订阅的topic是否完全一致——注意不要有空格或大小写差异。检查recipient字段是否正确。如果recipient写错了Broker 会直接抛ValueError不会静默丢弃。如果 Agent 内部异常Broker 会在_dispatch里 catch 到并打 ERROR 日志记得去看堆栈。这个排查思路我建议你固化下来遇到问题先走一遍能解决大部分路由问题。4.2 循环调用与消息风暴的防护多 Agent 协作里最容易出现也最危险的问题就是循环调用。比如 Agent A 处理完发消息给 BB 处理完又发消息给 A如果 A 再次触发同样的逻辑那永远也停不下来。进程内 Broker 没有 TTL 的概念所以这个问题只能靠应用层来防。我在框架里加了一个简单保护机制每条消息都带一个metadata.hop字段初始值是 0每次经过一个 Agent 处理并转发给下一个 Agent 时hop加 1。在 Broker 的publish入口检查如果hop超过最大跳数默认 10就拒绝转发并打错误日志。这能防住大多数死循环。还有一种消息风暴是广播造成的。某个 Topic 订阅了太多 Agent每条广播消息都会触发一轮全量调用。解决思路有两个一是尽量用点对点的recipient而不是广播二是给广播场景加节流阀同一个correlation_id的广播消息短时间内只允许处理一次。4.3 状态不同步记忆一致性的处理共享记忆模块在单进程下是同步的但有一个实际存在的坑多个 Agent 并发读取同一个 key 时可能读到旧值。这类问题比较隐蔽不会报错只会在结果层面出现看起来明显不对的现象。我的处理经验是对共享记忆的写操作尽量放在流水线的同步阶段完成比如 A 写完再通知 B 去读而不是 A 和 B 同时去写同一个 key。如果确实需要并发写不同 key那问题不大但并发写同一 key 一定要避免或者引入版本号读的时候带上版本校验。另外共享记忆一定要设置 TTL。早期我没设 TTL跑久了内存里堆积了大量不再使用的中间结果内存占用量涨得很快。300 秒 TTL 是个比较平衡的默认值如果单个结果特别大建议单独调小。4.4 性能优化序列化与批处理刚开始用 hermes-agent 时我总觉得框架开销大后来一分析才发现瓶颈不在框架而在消息体的序列化。有一次我让 Agent 直接把十页长的上下文塞进payloadBroker 每次分发都要重新处理这个巨大的 dict耗时暴涨。优化经验大对象不要放消息体里放共享记忆消息里只放 key。多条小消息能合并就合并减少路由次数。比如检索结果有 20 条就别一条一条发一次性发一个列表。如果你的 Agent 分布在多进程或多机器上务必换成更高效的序列化方案比如 protobuf不要用默认的 pickle 或 json。4.5 问题排查速查表我把上面几类问题整理成表格方便你快速定位现象可能原因处理方式消息发出后无任何日志订阅 Topic 不匹配检查 route_stats 中的订阅关系Agent 收到消息但不处理handler 内部异常被吞查看 ERROR 级别的完整堆栈完成任务后主流程卡住publish_and_wait 超时设太短调大整体超时时间同一任务反复执行存在循环调用检查 hop 计数设置最大跳数共享记忆读到旧值并发写同一 key增加版本号或调整串行逻辑内存持续增长共享记忆 TTL 未设置为所有 key 设置合理过期时间5. 真实使用心得与后续扩展最后这部分我不讲技术细节了聊聊我在实际项目里的真实体会以及这套框架适合用来做什么、不适合做什么。5.1 我在实际项目中踩过的坑第一个坑是过度抽象。我最早设计消息协议时想着一劳永逸地支持各种消息模式结果字段越加越多Agent 的处理逻辑反而变得特别绕。后来我砍掉一半字段只保留核心模型代码立刻清爽了。这个教训我一直记着框架设计要克制能覆盖 80% 场景就够了剩下 20% 交给使用者自己扩展。第二个坑是过早引入外部中间件。一开始我想当然地认为多 Agent 通信就应该用消息队列于是引进了 Redis Stream。结果发现单机场景下进程内协程通信比走网络快一个数量级而且部署和调试成本低太多。后来我保留了一个抽象的 Broker 接口单机用默认实现分布式再换消息队列这个灵活性帮我省了不少事。第三个坑跟模型调用有关。多个 Agent 同时调大模型接口时如果不做并发控制很容易触发接口限流。我在 Broker 外层加了一个简单的信号量semaphore限制同时调用模型接口的 Agent 数量这个问题就解决了。你如果也在这个框架里接大模型一定记得做这层保护。5.2 建议的适用边界与替代方案hermes-agent 这样的进程内轻量框架最适合的场景是你的 Agent 都部署在同一个服务里它们之间有协作关系但规模不会特别大比如几十个 Agent 以内。如果你的 Agent 要跨服务部署、需要持久化消息记录、或者有复杂的重试和补偿机制我不会建议你继续用这个框架的默认实现。这时候更合适的是成熟的消息基础设施比如 NATS、RabbitMQ或者自己封装一个基于数据库的任务表。hermes-agent 的价值在于帮你快速验证多 Agent 协作的逻辑而不是替你做一整套生产级基础设施。我在实际使用中最舒服的方式是用 hermes-agent 跑通业务逻辑等真正到了需要分布式扩展的时候再把 Broker 实现替换成消息中间件业务代码几乎不用动。这种先跑通、再加固的思路比一开始就上重型方案要务实的多。如果你也想尝试可以从一个小任务开始定义两个 Agent一个负责收集数据一个负责整理分析用 hermes-agent 把它们串起来。体验一下消息路由和协作到底是怎么运作的然后你会对多 Agent 应用有完全不一样的理解。