一个支持多消费者、多平台的实时流式输出设计

---

前言

在 AI 对话应用中，实时流式输出（Streaming Output）能显著提升用户体验——用户无需等待完整响应即可看到正在生成的内容。Hermes Agent 采用了一套精心设计的流式输出架构，支持同时向 CLI、TUI、第三方平台等多个目标推送内容。

本文将深入解析这一架构的设计思路和实现细节。

---

一、整体架构

Hermes 的流式输出采用回调机制 + 多消费者模式：

┌─────────────────────────────────────────────────────────────┐
│                   Agent (核心引擎)                          │
│                                                             │
│   run_conversation() → 流式调用 → stream_callback()        │
└───────────────────────────┼─────────────────────────────────┘
                            │
            ┌───────────────┼───────────────┐
            ▼               ▼               ▼
    ┌───────────────┐ ┌──────────────┐ ┌──────────────────┐
    │   CLI         │ │  TUI Gateway │ │  Gateway         │
    │   终端输出    │ │  Web UI 接口  │ │  多平台消费者    │
    └──────────────┘ └──────────────┘ └──────────────────┘
            │               │               │
            ▼               ▼               ▼
      本地终端显示     WebSocket推送    第三方平台编辑

核心思想：Agent 只需产生一次流式内容，通过回调分发到多个消费者。

---

二、核心组件详解

2.1 Agent 层 —— 内容生产者

Agent 是流式输出的起点，负责调用 LLM 的流式 API 并将内容通过回调分发。

工作流程：

1. 接收用户消息和回调函数
2. 调用流式 API，逐 token 获取响应
3. 每获取一个 token，调用一次回调
4. 响应结束时，调用回调并传入 None 表示结束

关键设计：

• 回调函数作为参数传入，实现解耦
• Agent 不需要知道有多少消费者
• 支持可中断的流式调用

2.2 CLI 消费者 —— 本地终端

CLI 是最直接的消费者，负责将流式内容输出到终端。

工作流程：

1. 接收增量文本（一个或多个 token）
2. 过滤推理/思考标签（如 <think> 等）
3. 进行行缓冲，逐行输出到终端
4. 使用 prompt_toolkit 实现兼容终端的实时打印

关键设计：

• 行缓冲：避免逐字符输出导致的闪烁
• 标签过滤：用户只看最终内容，不看推理过程
• KawaiiSpinner：加载动画提升体验

2.3 TUI Gateway —— Web UI 桥接

TUI Gateway 提供 Web/桌面 UI 的 JSON-RPC 接口。

工作流程：

1. 接收增量文本
2. 可选：实时 Markdown 渲染
3. 通过 WebSocket 发送 message.delta 事件
4. 前端收到事件后更新显示

关键设计：

• JSON-RPC 协议，支持多种前端
• 可选渲染字段，平衡性能和显示效果

2.4 Gateway Stream Consumer —— 多平台分发

这是支持多平台的核心组件，使用异步队列管理分发。

工作流程：

1. 接收增量文本，加入线程安全队列
2. 异步任务持续消费队列
3. 将内容分发到所有已注册的平台
4. 各平台使用自己的方式编辑消息（如 Telegram 的 edit_message_text）

关键设计：

• 线程安全队列：同步 Agent 和异步 UI 的桥接
• 统一接口：不同平台实现相同接口
• 自适应退避：API 调用失败时自动重试

---

三、多消费者设计

3.1 消费者列表

消费者	用途	输出方式
CLI	本地命令行	终端实时打印
TUI Gateway	Web/桌面 UI	WebSocket 推送
Telegram	Telegram 机器人	编辑消息 API
Discord	Discord 机器人	编辑消息 API
Slack	Slack 机器人	编辑消息 API

3.2 平台桥接机制

所有平台消费者都实现统一的接口，核心只有两个方法：

1. edit_message(delta)：编辑已发送的消息，追加新内容
2. finalize_message(text)：消息完成后调用，进行最终处理

以 Telegram 为例：

• 发送初始消息后获得 message_id
• 每次收到增量，调用 Telegram API 编辑该消息
• 由于 Telegram 限制，编辑间隔不能太频繁（1.5秒左右）

3.3 消息累积策略

由于各平台 API 的限制（如编辑间隔、字符数限制），消费者不会每收到一个 token 就调用 API，而是：

• 累积缓冲：收集一定量文本后再编辑
• 定时刷新：每隔固定时间（如 1.5 秒）编辑一次
• 光标提示：在消息末尾显示 ▉ 表示正在输入

---

四、事件协议

4.1 流式增量事件

{
    "type": "message.delta",
    "session_id": "会话ID",
    "payload": {
        "text": "正在分析...",
        "rendered": "<渲染后的文本>"    // 可选
    }
}

4.2 消息完成事件

{
    "type": "message.complete",
    "session_id": "会话ID",
    "payload": {
        "text": "完整响应内容",
        "reasoning": "推理过程",       // 可选
        "status": "complete"           // 或 interrupted/error
    }
}

4.3 推理块过滤

AI 响应中可能包含 <reasoning>...</reasoning> 等推理标记，这些对用户不可见，需要过滤：

• 打开标签（<think>）时开始过滤
• 关闭标签（``）时恢复输出
• 过滤后的内容不显示给用户

---

五、设计亮点

5.1 回调机制实现解耦

Agent  ──回调──▶  多个消费者

Agent 不需要知道有多少消费者，只需调用 stream_callback(text)。新增消费者只需注册回调，无需修改 Agent 代码。

5.2 线程安全队列

Agent 运行在独立线程，而 UI 需要在主线程更新。通过线程安全队列桥接：

Agent 线程  ──Queue.put()──▶  队列  ──Queue.get()──▶  UI 线程

5.3 自适应退避

当平台 API 调用失败（如频率限制）时，自动增加重试间隔：

第1次失败 → 等待 1.5 秒
第2次失败 → 等待 3 秒
第3次失败 → 等待 6 秒

5.4 Markdown 流式渲染

部分消费者支持实时 Markdown 渲染：

• 接收增量文本
• 逐步更新渲染结果
• 用户看到逐步格式化的内容

---

六、借鉴要点

如果要在其他项目中实现类似的流式输出架构，需要关注以下几点：

6.1 回调接口设计

定义统一的流式回调接口，参数只需一个增量文本，结束时应传入 None。

6.2 行缓冲输出

不要逐字符输出，应该累积到换行符或一定长度后再输出，避免终端闪烁。

6.3 多消费者管理

使用列表管理所有注册的消费者，广播式调用即可。

6.4 线程安全

如果涉及多线程，确保队列操作的原子性。

6.5 平台适配

不同平台有不同的 API 限制，需要针对性处理（如编辑间隔、消息长度）。

---

七、总结

Hermes Agent 的流式输出架构展示了如何优雅地处理实时内容分发：

特性	实现方式
解耦	回调机制 + 多消费者
线程安全	Queue 队列桥接
跨平台	统一接口 + 各自实现
容错	自适应退避
美观	行缓冲 + Markdown 渲染

这种设计使得同一个 AI 响应可以同时服务于 CLI、Web UI、第三方平台等多种场景，极大提升了架构的灵活性和可扩展性。

---

本文档基于 Hermes Agent 项目源码分析编写