企业官网建设流程全解析

1. 项目缘起：Hermes这个名字不是随便起的

做AI Agent相关开发的朋友应该都有一个共同感受：单个Agent的demo跑通相对容易，真正麻烦的是让多个Agent协作起来。模型调用、Prompt、工具函数这些单点能力其实都很好堆，但Agent和Agent之间怎么传消息、任务怎么拆解、状态怎么同步、出问题怎么排查，我翻了大量开源项目，要么是把重心全砸在模型侧，要么是绑定某家厂商的生态，始终没找到一个足够轻量、又能自托管的编排框架。

hermes-agent这个项目，就是冲着这个空档去的。项目名借用希腊神话里的信使赫尔墨斯——神界传话跑腿的角色，我觉得这个名字特别贴切：Agent协作的本质就是“找对人、传对话、办对事”。项目做成之后，它解决的核心问题有三个：一是统一Agent之间的消息协议，让不同语言、不同框架写的Agent能互相听懂；二是提供一套声明式的任务编排方式，用一份配置文件就能描述多Agent流水线；三是给每一条消息都带上链路追踪信息，方便我们在多Agent场景里快速定位“到底是谁出问题了”。

这个项目适合谁参考？如果你是刚接触多Agent开发，想找一个不绑云服务、能本地跑通全链路的编排方案；或者你已经在做Agent业务，但被消息格式混乱、任务状态难维护的问题折磨过，那这份实践记录应该能帮到你。就算你是后端工程师，想了解“多服务协作模式在AI场景下长什么样”，也可以当个技术案例看看。下面我把设计与实现细节完整拆开讲。

2. 核心设计与技术选型：先想清楚再写代码

2.1 多Agent协作里最常见的三座大山

动工之前，我先花了一周时间梳理现有方案。看得越多越确定，多Agent协作最大的三个坑分别是：协议、状态、可观测性。

先说协议。很多Agent框架为了省事，让Agent之间直接调函数。短期看确实快，但Agent一旦拆成独立进程、不同团队维护、甚至跨语言实现，函数调用就成了耦合炸弹——改一个签名，下游全崩。更合理的做法是Agent之间只通过消息通信，谁产的字段、谁消费字段，用协议来约束。

再说状态。一个任务在Agent链路上跑，中间经过了三四个节点，每个节点耗时还不同。如果没有统一的状态管理，光靠日志拼出“当前跑到第几步了”会非常痛苦。多个Agent都在往同一个任务里写结果时，如果不用任务状态机来做约束，很容易出现数据互相覆盖。

最后是可观测性。单Agent调模型，出了问题看一轮日志基本能定位。多Agent链路上出了问题，你面对的是多个进程、多段日志、还有异步消息，没有trace_id串起来的话，排查成本会成倍放大。这也是我后来坚持在消息协议里内置trace_id和span_id的原因。

2.2 设计原则：统一、声明式、可审计

hermes-agent一共就三条设计原则，后面所有模块都是围绕这三条来的。

第一，一切交互皆消息。Agent之间不能直接调用对方的内部方法。理论上任何能读写消息队列的语言都可以接入，不要求大家用同一个SDK。

第二，任务编排用声明式配置，不写胶水代码。把“谁先做、谁后做、结果传给谁”固化成一份YAML文件，业务同学也能看懂整个流程，而不是只有写代码的人能维护。

第三，默认追踪、默认审计。每一条进入系统的消息都必须携带链路信息，每一个Agent动作都记录到审计日志里。为了可观测性多花这一点存储成本，在线上排障的时候能省回来十倍不止。

2.3 技术选型背后的取舍

核心框架用了FastAPI + Uvicorn，原因很简单：Python生态里跟AI项目集成最顺，异步支持好，Agent里的IO操作（调模型、查数据库、等外部API）天然适合asyncio来跑。消息中间件选了Redis，用它的Pub/Sub做即时消息路由，用Redis Stream做持久化消息队列。Persistence选了SQLite起步，数据量大了之后可以平滑切换到PostgreSQL。

有一个取舍值得展开说一下：为什么不用WebSocket让Agent直接互联？因为WebSocket方案要求两端都保持长连接，Agent一多就是一张网，连接管理本身就是巨大负担；而且WebSocket连接一旦断开，中间的消息就丢了，没有重试机制。Redis Pub/Sub天然就是松耦合的发布订阅模型，Agent只管发、只管收，不需要关心对方是谁。配合Redis Stream还能把没消费掉的消息留着，等Agent恢复后再补消费，这一点在生产环境非常关键。

3. 核心模块拆解与关键实现

3.1 统一消息协议：Agent之间说什么语言

所有Agent之间的通信，都走我定义的一套JSON消息结构。字段不多，但每一个都是踩过坑之后才加上的。

{ "msg_id": "a1b2c3d4e5f6", "trace_id": "tr-20250101-0001", "span_id": "sp-crawler-002", "scene": "daily_report", "sender": "agent.crawler", "recipient": "agent.summarizer", "msg_type": "task_request", "priority": 1, "payload": { "source_urls": ["https://example.com/news/1"], "keyword": "AI Agent" }, "context": { "tenant": "internal", "deadline": "2025-01-01T12:00:00Z" }, "created_at": "2025-01-01T10:00:00Z" }

这里有几个字段后来被证明特别值钱。msg_id是全局唯一标识，主要用于幂等——消费者处理完消息后如果消费确认丢了，重新消费时能认出这是同一件事，不会重复干活。trace_id是整个任务链路的根编号，一个完整任务的每条消息都带同一串trace_id，排查时只要按trace_id查日志，整个链路的所有环节就全拉出来了。span_id标记当前是哪一个环节，parent_span_id可以拼接出完整调用链。context里的deadline是我后来补的，因为发现有些任务会在某个环节卡住，没有截止时间约束的话会一直占着队列资源。

消息类型我分成了四类：task_request是下发任务，task_result是回传结果，tool_invoke是请求调用某个工具，event_report是上报状态事件。一开始我图省事只设计了前两种，结果工具调用环节的消息没法归类，只能硬塞进task_request的payload里，排查日志的时候非常别扭。后来补上tool_invoke和event_report，整个模型才完整。

3.2 消息路由与任务编排：Hermes怎么当信使

消息路由这块，我实现了一个轻量的Router服务。Router本身不干活，只负责“看信、送信”。它维护一张路由表，键是Agent名称，值是这个Agent当前监听的Topic。发布者把消息发到Router，Router检查recipient字段，把消息丢到对应的Topic里。这样做的好处是发布者完全不需要知道接收者的物理位置，Agent实例扩容、迁移都不影响已有通信。

任务编排是另一个核心组件，叫Pipeline Engine。它消费开发者写好的流程定义文件，按部就班地推进任务。比如一个“日报生成”流程，定义为三个Step：爬取新闻、生成摘要、发送通知。每个Step声明自己由哪个Agent执行，输入从哪里来，执行完把结果输出到哪里。

pipeline: id: daily_report_flow name: 每日日报自动生成 version: "1.0" steps: - id: fetch_news agent: crawler action: fetch_news_urls inputs: keywords: ["hermes-agent", "AI Agent"] next: summarize - id: summarize agent: summarizer action: summarize_news inputs_from: fetch_news.output.urls next: notify - id: notify agent: notifier action: send_to_feishu inputs_from: summarize.output.summary

这个配置文件就是整个业务逻辑的“单向链表”。Engine启动后，把文件读进来构建成DAG，然后从第一个Step开始跑。每个Step跑完后，Engine把上个Step的输出字段映射成下个Step的输入字段，再生成一条新的task_request放进消息队列。整个流程里没有手写胶水代码，改流程顺序、加删Step，都只需要改YAML文件。

这里有一个State Store设计值得单独说。每个Pipeline执行实例都有一个状态机，流转是这样：pending -> running -> success / failed / timeout。状态存到SQLite里，每次状态变更都带时间戳。之所以非要引入状态机而不是简单地用“已完成Step数”来表示进度，是因为流程里可能有并行分支——多个Agent同时干活，各自完成后都往State Store里汇报，状态机要把每个分支的完成情况合并起来，全部完成才推进到下个阶段。

3.3 工具接入层与安全边界

Agent本身最擅长的还是决策和编排，具体“干活”往往靠工具。hermes-agent里有一个工具注册中心，Agent通过声明式的方式把外部能力挂进来。工具可以是读HTTP接口、执行SQL查询、调用内部RPC，任何可编程调用的能力都行。

from hermes_agent import BaseAgent, tool class NewsCrawler(BaseAgent): name = "crawler" @tool(name="fetch_urls", schema={ "type": "object", "properties": { "keywords": {"type": "array", "items": {"type": "string"}} }, "required": ["keywords"] }) async def fetch_urls(self, keywords: list[str]) -> list[str]: # 这里请求搜索接口，获取匹配的新闻链接 return await search_api.query(keywords)

工具接入层里有一个做得比较坚决的设计：Agent不能随意调用所有工具，每个Agent有一份独立的工具权限清单。权限在Agent注册的时候明确指定，运行期不能动态修改。这样做是被一次线上事故逼出来的——某个Agent的任务被Prompt注入，它拿着一个本不该有的数据库查询工具去遍历了全表，虽然没造成真实损失，但给我吓出一身冷汗。从那以后，工具的授权逻辑一律走配置，不走运行时判断。

安全边界这块还有一层考虑：任何来自外部的内容都默认不可信。搜索结果、用户输入、上游Agent传过来的文本，Agent在处理前都要经过一次内容清洗，防止把不可信数据直接拼进Prompt里执行。

3.4 人机协作节点：给自动化留一个刹车

全自动链路看起来很爽，但真正跑业务的时候你会发现，有些环节必须让真人拍板。比如“自动发送对外邮件”这个Step，直接让Agent发出去，万一内容有误就是事故。所以我做了一个人类审批节点（Human Approval Node）。

- id: review_before_send agent: human_approver action: review_summary inputs_from: summarize.output.summary timeout: 3600 on_timeout: cancel_pipeline

任务跑到这个节点时，Agent会把待审内容主动推到相关人的聊天工具里，这个人点击“通过”或“驳回”，审批结果写成一条task_result消息回流给Engine，Engine根据结果决定继续还是终止。审批节点本质上是一个“特殊的Agent”，它的消息协议跟其他Agent完全一样，不同的只是执行者是人而不是模型。

这个模块做好之后，整个系统的可用性上了一个台阶。业务方开始愿意把真实流程接进来，因为他们知道关键节点上人是可以介入的，不再是一个黑盒自动跑到底。对Agent应用来说，我认为“人机协同”不是过渡方案，而是长期必要的能力——至少在涉及钱、对外发声、删除数据这三类高风险操作上，必须保留人工卡口。

4. 实操记录：把一个多智能体应用跑起来

4.1 五分钟初始化环境

说了这么多设计，直接看怎么跑起来。环境要求很简单：一台装了Docker的Linux机器，Python 3.11+。项目目录下直接执行：

git clone https://github.com/yourname/hermes-agent.git cd hermes-agent cp .env.example .env docker compose up -d redis pip install -e .

docker compose那一步先把Redis拉起来，因为所有消息流转的基础设施就是它。装好之后启动Router和Pipeline Engine两个核心进程：

hermes router start --config config/router.yaml hermes engine start --config config/engine.yaml

看到两个进程都打印出“ready”就算起成功了。这时候系统处于“空转”状态，还没有任何Agent注册进来，但只要往Router里发一条注册请求，它就能感知到。

4.2 写第一个Agent

用一个最简单的EchoAgent来验证链路。这个Agent什么实际业务都不做，收到消息后把payload原样返回。

from hermes_agent import BaseAgent, AgentContext class EchoAgent(BaseAgent): name = "echo" async def handle_message(self, message: dict, ctx: AgentContext): self.logger.info(f"received: {message}") return message["payload"]

写完保存为echo_agent.py，在另一个终端里跑：

hermes agent run echo_agent:EchoAgent --name echo

Agent启动后会自动向Router发送注册请求，把自己挂到名为echo的Topic上。万事俱备后，往Router发一条测试消息：

hermes message send --recipient echo --payload '{"hello": "world"}'

如果能在Agent终端看到收到的消息内容，说明“Router -> Redis -> Agent”这条通路是通的。我到现在都会保留这个EchoAgent，每次改了Router的代码先拿它回归一遍，心里才踏实。

4.3 用一份编排文件串联三个Agent

EchoAgent只是验证了通路，真正体现价值的还是多Agent串联。做一个能自动爬新闻、生成摘要、发到飞书群的小流程，来演示Pipeline Engine怎么工作。

除了刚才的crawler，再写一个summarizer和notifier。summarizer收到URL列表后，逐个抓取正文并调用大模型接口做摘要：

class Summarizer(BaseAgent): name = "summarizer" async def handle_message(self, message, ctx): urls = message["payload"]["urls"] results = [] for url in urls: content = await http_client.get(url) summary = await llm_client.chat( prompt=f"请用三句话概括以下内容：{content[:2000]}" ) results.append({"url": url, "summary": summary}) return {"summaries": results}

notifier负责把摘要内容组装成一条飞书机器人消息：

class Notifier(BaseAgent): name = "notifier" async def handle_message(self, message, ctx): summaries = message["payload"]["summaries"] text = "\n\n".join([f"- {s['url']}\n{s['summary']}" for s in summaries]) await feishu_webhook.send(webhook_url, {"msg_type": "text", "content": {"text": text}}) return {"status": "sent"}

三个Agent都启动后，手动触发一次Pipeline：

hermes pipeline run --file examples/daily_report.yaml

触发后可以观察Engine的日志输出。正常顺序是：Engine把fetch_news消息发给crawler，crawler返回URL列表；Engine把summarize消息发给summarizer，summarizer返回摘要；Engine把notify消息发给notifier，notifier返回发送结果；最后Engine把整个Pipeline状态更新为success。整个过程不需要写任何针对这三个Agent的胶水代码，全部由配置文件驱动。

4.4 可观测性：怎么看链路与日志

运行过程中最常用的排查工具是这条命令：

hermes trace query --trace-id tr-20250101-0001

输出会按时间线把这条trace涉及的所有消息、节点、耗时打出来，一眼就能看出哪个环节最慢、哪个环节出了错。有个很实用的细节：每个Agent在启动时都会自动初始化结构化日志，输出格式是JSON，包含trace_id、span_id、agent_name。这样日志系统里按trace_id一搜，就能看到整个链路日志，不用再写复杂的grep正则去拼上下文。

5. 常见问题与排查实录

5.1 消息发出去了，但Agent没反应

这是被问得最多的一个问题。排查思路固定三步走。第一步，确认Agent是否真的注册成功。Router启动时会打印已注册的Agent列表，或者用hermes agent list查一下。如果列表里没有，大概率是Agent注册的Topic名跟消息recipient对不上。第二步，查Redis队列里有没有积压。执行redis-cli llen hermes:queue:crawler，如果数量一直涨，说明消费端处理不过来，去查Agent的日志看有没有异常。第三步，看消费确认是否成功。消费者处理完消息后必须向Redis Stream发送ACK，如果ACK逻辑没触发，消息会被重复消费或者直接留在待确认列表里。

5.2 任务一直Pending，不往下走

这个问题通常出在条件依赖上。Pipeline Engine里有一个规则：下游Step的执行条件是上游所有输入都ready。如果上游Agent返回的结果字段名跟配置文件里inputs_from对不上，Engine就一直等不到“就绪”信号，状态自然卡在pending。排查方法很直接，看Engine日志里有没有“missing input field”之类的警告，或者用trace query查一下上游Step实际返回了哪些字段。我自己有一次就是因为上游返回的是output.urls，配置文件里写的是output.url，少了个s，整个流程卡了十分钟才反应过来。

5.3 工具调用超时与鉴权失败

Agent调用外部工具，最容易翻车的是网络超时和凭证过期。默认HTTP客户端超时时间我设置成了30秒，但有些外部接口天生又慢又爱超时。建议把超时做成可配置项，不同工具给不同阈值。鉴权失败出现得也挺频繁，原因是工具凭证在Agent进程里做了缓存，外部系统密码轮转后，Agent还在用旧token。我后来在工具调用返回401时强制清缓存再加一次重试，能解决大部分问题。

5.4 问题排查速查表

5.5 几个值得注意的设计坑

有一些更隐蔽的坑，很难靠查文档发现，这里一并分享了。

消费端要做好幂等。消息队列能做到at-least-once，但不保证exactly-once。Agent处理消息之前，先查一下msg_id是不是已经处理过了，避免重复执行产生重复结果。

消息体要控制大小。Redis Stream单条消息过大时，内存和序列化开销都会显著上升。像新闻全文这种大字段，不应该放进消息体，而是存到对象存储里，消息里只带一个引用ID。

异步任务要设置deadline。这一点在最开始的消息协议里已经设计了context.deadline字段。任何任务节点执行时间超过deadline，Engine会终止整条Pipeline并告警，防止一个卡住的任务拖死整条链路。

6. 写在最后：一点真实体会

hermes-agent从立项到跑通第一个多Agent流程，前后折腾了大概一个月。踩坑最多的不是模型侧，反而是消息协议和状态管理这些看起来“不那么AI”的模块。我个人最深的体会是：做多Agent系统，本质上是做分布式系统，一切以前在分布式环境中踩过的坑——消息丢失、重复消费、链路追踪、超时控制——在这里一个都不会少，只会因为每个节点背后都挂着一个不确定性的模型而变得更刺激。

如果你正准备做自己的多Agent项目，我的建议是先别急着上框架，花两天时间想明白消息协议和任务状态模型这两个底座。协议和状态搞定了，后面加Agent就像往插线板上插电器一样简单；这两个没想清楚，Agent越多，系统越接近失控边缘。hermes-agent目前还在持续迭代中，下一步我计划加入更细粒度的并行分支支持、在线可视化DAG编辑，以及更多现成的Agent模板。如果你也在做类似的事，欢迎在评论区聊聊你的方案，或者给它提issue一起完善。