企业官网建设流程全解析

先说个背景。过去两年我一直在折腾多智能体协同相关的项目，从最早的单体 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_factory=lambda: 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_factory=dict) # 主体内容 metadata: dict = field(default_factory=dict) # 链路追踪等元信息 timestamp: str = field( default_factory=lambda: 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(f"Agent [{agent.name}] registered") def subscribe(self, agent_name: str, topic: str): """让某个 Agent 订阅指定 Topic。""" self._subscriptions[topic].add(agent_name) logger.info(f"Agent [{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(f"Recipient {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(f"No handler for agent {agent_name}") return try: await handler(message) except Exception: logger.exception(f"Agent [{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 13（Linux 和 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.txt

3.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": f"result_1_for_{query}", "score": 0.95}, {"title": f"result_2_for_{query}", "score": 0.87}, {"title": f"result_3_for_{query}", "score": 0.72}, ] # 把结果发给下一个环节：通过 Message 回复，或者写入共享记忆 reply = Message( topic="search.completed", sender=self.name, recipient=message.sender, # 直接回给发起方 msg_type="result", payload={ "correlation_id": message.correlation_id, "query": query, "results": results, }, correlation_id=message.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( topic="search.request", sender="main", 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( topic="search.request", sender="main", payload={"query": query}, correlation_id=f"task-{query[:10]}-{asyncio.time() if hasattr(asyncio, 'time') else '1'}", ) # 2. 等检索完成回包 search_result = await broker.publish_and_wait(search_msg, timeout=10) # 3. 把检索结果发给分析 Agent analysis_msg = Message( topic="analysis.request", sender="main", recipient="analyzer", payload={"raw_data": search_result.payload["results"]}, correlation_id=search_msg.correlation_id, ) analysis_result = await broker.publish_and_wait(analysis_msg, timeout=15) # 4. 最后交给撰写 Agent 生成总结 writer_msg = Message( topic="write.request", sender="main", recipient="writer", payload={"analysis": analysis_result.payload["analysis"]}, correlation_id=search_msg.correlation_id, ) final_result = await broker.publish_and_wait(writer_msg, timeout=15) return final_result.payload["article"]

你注意我这里调用了publish_and_wait，这个是 hermes-agent 封装的一个同步等待接口。它背后的逻辑是：当 Broker 把消息发给目标 Agent 后，协程会等待目标 Agent 返回结果。为了支持这个模式，我会在 Broker 内部维护一个 pending 的回执表，每个correlation_id对应一个 future，Agent 返回时自动唤醒等待方。这个机制在同步编排任务流的时候非常好用，代码读起来就像普通的同步调用，但实际上底层是异步的。

三个 Agent 的实现逻辑大体类似，区别只在处理消息时的业务逻辑不同。AnalyzerAgent 会从原始素材里提取关键信息、生成分析结论；WriterAgent 则负责把分析结论组织成长文。它们的消息类型可以统一用result，也可以通过msg_type字段区分，灵活处理就行。

3.4 关键配置参数选择

实操中你会遇到几个需要动手配置的参数，我把我调参的经验列成一张表格：

超时这个参数我踩过坑。最早我把超时设成 5 秒，结果每次调用大模型慢一点就超时，整个流水线被中断。后来我把超时拆成两级：网络请求超时短一点，整体任务超时长一点，这样既不会因为偶发网络抖动就崩，又能及时暴露真死循环问题。

4. 常见问题与排查技巧实录

工具好不好用，往往在踩坑的时候才能看出来。我把自己在这个框架上遇到的问题和排查思路整理成速查表，方便你对照。

4.1 Agent 之间消息不达：路由表与 Topic 排查

最常见的问题是：消息发出去了，但目标 Agent 没收到处理。我在日志里看到好多No handler for agent xxx警告，基本就是订阅关系或者注册关系没配对。

排查思路按顺序来：

1. 打印broker.route_stats()，看 Agent 是否注册成功、订阅的 Topic 是否正确。
2. 检查消息的topic和 Agent 订阅的topic是否完全一致——注意不要有空格或大小写差异。
3. 检查recipient字段是否正确。如果recipient写错了，Broker 会直接抛ValueError，不会静默丢弃。
4. 如果 Agent 内部异常，Broker 会在_dispatch里 catch 到并打 ERROR 日志，记得去看堆栈。

这个排查思路我建议你固化下来，遇到问题先走一遍，能解决大部分路由问题。

4.2 循环调用与消息风暴的防护

多 Agent 协作里，最容易出现也最危险的问题就是循环调用。比如 Agent A 处理完发消息给 B，B 处理完又发消息给 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 直接把十页长的上下文塞进payload，Broker 每次分发都要重新处理这个巨大的 dict，耗时暴涨。

优化经验：

• 大对象不要放消息体里，放共享记忆，消息里只放 key。
• 多条小消息能合并就合并，减少路由次数。比如检索结果有 20 条，就别一条一条发，一次性发一个列表。
• 如果你的 Agent 分布在多进程或多机器上，务必换成更高效的序列化方案，比如 protobuf，不要用默认的 pickle 或 json。

4.5 问题排查速查表

我把上面几类问题整理成表格，方便你快速定位：

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 应用有完全不一样的理解。