echo-agent 前身为 2025 年 11 月启动的个人助理项目 fubot，最初面向长期陪伴型个人智能体，围绕认知记忆、上下文延续、用户偏好沉淀、任务闭环与持续自我优化展开。随着真实场景迭代，项目逐步形成多入口接入、统一事件模型、消息总线、Agent Loop、多模型抽象、工具调用、MCP 接入、任务调度、权限审批、运行轨迹、长期记忆和受控自演进等能力。目前已支持微信、QQ、CLI、Gateway、Webhook、Cron 等入口，服务用户超过 20 万、累计下载超过 50 万，是面向长期运行、记忆增强和可持续成长智能体的开源 Agent Runtime。

项目地址：https://github.com/fuyuxiang/echo-agent

你可能遇到过这样的工程阶段：Agent 一开始只跑在命令行里，读一行输入，调一次模型，打印一段回复。很快，需求来了：接 Webhook、接 Gateway、接定时任务，再接聊天平台。

如果每接一个入口，就让它直接调用 AgentLoop；每多一个输出目标，又让 AgentLoop 直接调用对应平台 API，代码很快会变成一团平台分支。CLI、Webhook、Cron、Gateway、聊天平台都有自己的时间模型、消息结构和回复语义，核心 Agent 被这些细节拖着走。

本篇只讲一个点：用消息总线把通道和 Agent 核心解耦。

问题入口

单通道原型里，直接调用最省事。

text=input()

answer=awaitagent.run(text)

print(answer)

这个结构的问题不是“不优雅”，而是扩展到多通道以后没有稳定边界。Telegram 有 chat_id、reply_to、媒体文件和群组策略；Slack 有 channel、thread、reaction、typing；Webhook 可能来自任意内部系统；Cron 没有真实用户输入，却需要投递到原会话；Gateway 既有 HTTP 请求，也有 WebSocket 长连接。

如果 AgentLoop 直接理解这些差异，它就不再只是推理与执行核心，而会变成平台兼容层。

通道可以很多，但 Agent 核心只能有一套。否则每个平台都会长出自己的特殊路径，系统很快失去一致性。

消息总线做的事情很克制：在外部通道和内部 Agent 之间放一层稳定契约。通道只负责把平台原始事件翻译成统一的 InboundEvent；Agent 只订阅统一事件；Agent 输出统一的 OutboundEvent；再由通道适配器翻译回平台 API。

为了不停留在抽象层面，下面以 echo-agent 的实现为例。

通道边界

消息总线不是为了把代码绕远一层，而是为了同时解决三件事。

第一，空间解耦。新增平台时，只要实现通道适配器和出站 handler，不需要改 AgentLoop。Agent 面对的始终是统一事件。

第二，时间解耦。通道收到消息后可以快速入队，不必等待模型调用、工具执行、审批确认和最终回复全部完成。接收输入的时间、开始处理的时间、输出投递的时间可以错开。

第三，流量控制。队列长度、并发上限、session 限流、异常隔离和停止时的任务清理，都可以在总线层统一处理，而不是散落到每个通道。

直接调用和消息总线的差异，可以这样看：

设计	通道职责	Agent 职责	主要问题
直接调用	解析平台事件并调用 Agent	理解平台差异并回调通道	耦合高，难统一限流和观察
消息总线	翻译事件并发布到总线	只处理统一事件并发布输出	多一层契约，需要维护事件语义

这层契约的核心，就是 InboundEvent 和 OutboundEvent。

事件契约

事件总线的第一层价值不是队列，而是定义统一事件类型。echo-agent 中，所有通道输入最终都会变成 InboundEvent，所有 Agent 输出最终都会变成 OutboundEvent。

InboundEvent 关注“外部世界发生了什么”。它包含事件身份、来源通道、发送者、会话空间、内容块、时间戳、回复关系、线程关系、可选的 session 覆盖，以及平台扩展元数据。

这里最容易被低估的是 content。它不是单个字符串，而是 ContentBlock 列表。文本、图片、文件、音频、视频或混合内容，都可以进入同一个事件结构。text 属性把文本块拼接出来，media_urls 提取非文本媒体 URL。这样普通文本推理和视觉、文件类工具都能基于同一事件工作。

session_key 也很关键。默认情况下，它由 channel 和 chat_id 组成，例如 telegram:12345 或 cli:local。这让不同平台的会话天然形成上下文边界。session_key_override 则允许 Gateway、A2A 或内部系统显式控制会话归属。

OutboundEvent 关注“系统要如何表达”。它不只是回复文本，还包含目标 channel、chat_id、内容块、回复目标、编辑目标、任务 ID、工作流 ID、is_final 和 message_kind。

is_final 和 message_kind 对流式输出很重要。Agent 可以先发送 progress 或 tool hint，再发送最终回答。支持消息编辑的平台可以更新同一条消息，不支持的平台则可以只发送最终文本或做降级处理。

可以把事件契约压缩成下面这段伪代码：

@dataclass

classInboundEvent:

    event_id: str

    event_type: EventType

    channel: str

    sender_id: str

    chat_id: str

    content: list[ContentBlock]

    reply_to_id: str|None

    thread_id: str|None

    session_key_override: str|None

    metadata: dict

    gateway_metadata: dict

    @property

    defsession_key(self) ->str:

        returnself.session_key_overrideorf"{self.channel}:{self.chat_id}"

@dataclass

classOutboundEvent:

    channel: str

    chat_id: str

    content: list[ContentBlock]

    reply_to_id: str|None

    edit_message_id: str|None

    is_final: bool

    task_id: str|None

    workflow_id: str|None

    message_kind: str

    metadata: dict

事件契约比函数签名更稳定。函数签名通常服务一个调用方，事件契约服务多个生产者和消费者：CLI、Gateway、Cron、A2A、AgentLoop、测试替身、日志系统、WebSocket 广播都可能依赖它。

这也决定了事件模型的演进原则：新增字段优先可选，旧字段语义不能随意改，平台特有信息放进 metadata 或 gateway_metadata，核心字段只表达跨平台稳定概念。否则每接一个平台，事件模型都会被迫长出一组平台专用字段，统一事件的价值就被抵消了。

消息总线的价值不是“把消息放进队列”，而是用稳定事件模型把 Agent 的感知和表达从平台差异中解放出来。

总线机制

echo-agent 的 MessageBus 有几组核心结构：入站有界队列、按 channel 订阅的出站 handler、global outbound handler、入站订阅者、生命周期锁、并发 semaphore，以及可选的 SessionRateLimiter。

入站发布返回 bool。如果队列能在 5 秒内接受事件，返回 True；如果队列持续满，返回 False。这个返回值很重要，因为通道层可以据此记录失败、返回 HTTP 错误，或提示用户稍后再试，而不是假装消息已经被系统接受。

核心分发逻辑可以简化成这样：

asyncdefpublish_inbound(event):

    try:

        awaitwait_for(inbound_queue.put(event), timeout=5.0)

        returnTrue

    exceptTimeoutError:

        returnFalse

asyncdefdispatch_loop():

    whilerunning:

        event=awaitwait_for(inbound_queue.get(), timeout=1.0)

        ifrate_limiterandnotrate_limiter.try_acquire(event.session_key):

            awaitpublish_outbound(text_reply(

                channel=event.channel,

                chat_id=event.chat_id,

                text="请求过于频繁，请稍后再试。",

                reply_to_id=event.reply_to_id,

            ))

            continue

        awaitconcurrency_sem.acquire()

        task=create_task(dispatch_inbound_guarded(event))

        inflight.add(task)

        task.add_done_callback(inflight.discard)

asyncdefdispatch_inbound_guarded(event):

    try:

        awaitgather_handlers(event, return_exceptions=True)

    finally:

        concurrency_sem.release()

这里有两个细节值得注意。

第一，分发不是串行等待。旧式串行 dispatch 会让慢事件阻塞后续事件。当前实现会为每个入站事件创建任务，再用 semaphore 限制总体并发。测试中有一个场景专门验证：第一个事件的 handler 阻塞时，第二个事件仍然能开始处理。

第二，finally 必须释放 semaphore。否则某个 handler 抛异常后，并发额度会被永久占住，系统会慢慢耗尽处理能力。

生命周期也有边界。start 和 stop 由 lifecycle lock 保护，避免并发启动和停止造成状态错乱。停止时不仅取消 dispatch loop，还会取消所有 inflight inbound task，防止服务关闭后留下悬挂任务。

背压限流

Agent 处理一次消息可能很慢。模型调用要等 provider，工具执行可能访问网络或跑命令，审批可能等待人工确认。输入速度超过处理速度时，如果没有背压，队列会无限增长，最终把内存耗尽。

echo-agent 的背压不是一个开关，而是四层机制。

机制	控制什么	过载时的行为
max_queue_size	等待处理的事件总量	队列满后阻止继续无限堆积
publish timeout	通道入队等待时间	5 秒仍无法入队则返回 False
max_concurrency	同时处理的入站事件数	semaphore 限制并发任务
session rate limit	单个会话请求频率	生成限流出站提示

SessionRateLimiter 使用 token bucket。rpm 表示每分钟恢复多少 token，burst 表示短时间允许的突发请求数。每个 session_key 有自己的 bucket；rpm=0 表示关闭限流；不同 session 独立计数；token 会随时间恢复，但不会超过 burst。

它还用 OrderedDict 控制历史 session 数量，超过 max_sessions 时淘汰最旧 session。长期运行的服务不能让限流器为无限历史会话保留状态。

这个设计保护的是公平性。全局并发上限只能保护进程资源，不能防止一个用户或一个群聊占满处理能力。session 级限流把公平性下沉到会话维度：同一会话不能无限快速地向 Agent 发请求，不同会话仍可并发推进。

背压不是为了“消息永不丢失”。对 Agent 来说，过载时继续接受所有请求反而更危险，因为一次输入可能触发昂贵模型调用和真实副作用。更可靠的做法是明确表达“现在处理不过来”，把隐性崩溃变成可管理状态。

出站路由

入站事件从所有通道汇入 Agent，出站事件则从 Agent 分发到目标通道。publish_outbound 的流程很清楚：

1. 如果没有 global handler，也没有目标 channel handler，记录 warning 并返回。

2. 先调用所有 global outbound handlers。

3. 如果 event.metadata["_drop"] 为真，停止，不投递给具体 channel。

4. 找到目标 channel 的 handlers。

5. 并发调用这些 handlers，并使用 return_exceptions=True 隔离异常。

global handler 的价值在于观察。Gateway 可能需要把所有出站事件推给 WebSocket 客户端，日志系统可能要记录所有输出，测试也可以通过 global handler 捕获消息。

channel handler 的价值在于投递。Telegram 只处理发往 Telegram 的事件，Slack 只处理 Slack 事件。Agent 只需要设置 OutboundEvent.channel，总线就能把它路由到对应适配器。

_drop 是一个小但重要的语义：有些事件对系统观察者有意义，但不应该发给用户。比如内部进度、平台不支持的 tool hint，或者被通道管理器转换过的中间事件。global handler 仍然能看到，具体 channel 投递会被跳过。

异常隔离同样关键。一个日志 handler 失败，不应该影响真正的消息投递；某个 channel handler 失败，也不应该让同 channel 的其他 handler 一起失败。

生产可用性

判断一个消息总线是否接近生产可用，不要只看有没有队列。更硬的标准是看边界行为是否清楚。

工程项	可检查问题
事件契约	InboundEvent / OutboundEvent 字段是否稳定，是否覆盖文本、媒体、回复、线程、metadata
通道解耦	新增 CLI、Webhook、Cron、Gateway 或聊天平台时，是否不改 AgentLoop
背压	是否有有界队列、入队超时、并发上限和 session 限流
用户反馈	限流和队列满是否能给通道明确结果，而不是静默丢弃
异常隔离	坏 handler 是否不影响好 handler，global handler 是否不影响 channel 投递
生命周期	重复 start/stop、服务关闭、inflight task 是否有明确处理
可观察性	是否能通过 event_id、channel、chat_id、task_id、workflow_id 追踪一次事件流
测试覆盖	是否覆盖慢 handler、坏 handler、满队列、限流、_drop、无 handler、semaphore 释放

还有两个边界要提前想清楚。

第一，事件顺序和幂等性。通道可能重复投递 Webhook，网络重试可能让同一消息到达两次。事件应有稳定 event_id 或可推导身份；会产生副作用的工具还需要 idempotency key。消息总线负责分发，会话系统负责同一上下文内的顺序隔离，这也是下一篇要进入的主题。

第二，事件优先级。简单 FIFO 队列适合早期系统，但多用户、多通道、多任务场景下，取消任务、审批结果、系统关闭、健康告警，通常应该比普通消息更优先。echo-agent 当前消息总线提供队列、订阅和背压基础；未来可以进一步演进为更完整的运行时调度层。

小结

消息总线本身不做推理，也不调用模型，看起来不像“智能”模块。但如果事件丢失、重复投递、并发失控、输出路由错误，一个强模型也会表现成一个不可靠系统。

它解决的是 Agent 和外部世界之间的通道秩序：外部平台先变成统一 InboundEvent，Agent 只处理统一事件；Agent 输出统一 OutboundEvent，再由通道适配器投递到目标平台。队列、并发、限流、异常隔离和生命周期控制，让这个事件流在压力下仍有明确语义。

消息总线回答“事件进入哪里、如何分发”。但事件进入系统以后，还要回答另一个问题：这些事件属于哪个上下文，历史如何组织，同一会话中的并发如何隔离。下一篇就进入会话系统。

（全篇完）

---

本文为 echo-agent 设计笔记系列第 05 篇。项目源码已开源至 GitHub。如果你对工业级 Agent 的工程落地感兴趣，欢迎加入技术交流群参与日常讨论。下一篇我们将探讨 《Agent 会话系统：上下文、历史与并发隔离》，敬请期待。