可恢复流式传输架构

可恢复流式传输架构

学习目标

本章聚焦 LibreChat 的可恢复流式传输系统。你将了解：

• GenerationJobManager 的集中式作业管理
• IJobStore / IEventTransport 抽象层的设计
• 内存与 Redis 两种实现策略
• 多标签页/多设备恢复的工作原理

前置知识

• 流式传输架构

下文假设你已理解上述概念，直接聚焦 LibreChat 的具体实现。

---

项目实践

抽象层设计

LibreChat 的流式传输基于两个接口：

IJobStore:        管理生成作业的生命周期（创建、查询、更新状态）
IEventTransport:  将事件传输到客户端（SSE chunk、attachment 等）

这种分离允许独立替换存储和传输策略：

四种实现组合

JobStore	EventTransport	适用场景
InMemory	InMemory	开发环境、单实例部署
InMemory	Redis	开发调试（不推荐）
Redis	InMemory	不推荐（事件丢失）
Redis	Redis	生产环境、多实例部署

生产部署必须使用 Redis：内存实现不支持跨进程恢复——连接断开后无法找到之前的作业状态。

GenerationJobManager 的核心职责

GenerationJobManager:
  ├── createJob(streamId, config)    → 创建新作业
  ├── registerClient(streamId, res)  → 注册 SSE 响应
  ├── emitChunk(streamId, chunk)     → 向所有注册客户端发送事件
  ├── completeJob(streamId)          → 标记作业完成
  ├── abortJob(streamId)             → 中止作业
  └── getJobState(streamId)          → 查询作业状态

事件类型包括：

• chunk：LLM 输出的文本片段
• attachment：结构化附件（如 Memory Agent 的 artifact）
• error：错误事件

连接恢复流程

1. 客户端断线
2. 客户端重新连接，携带之前的 streamId
3. 服务器查找 JobStore 中的作业状态
4. 从断点处继续发送事件

多标签页场景：同一个 streamId 可以注册多个 SSE 响应（多个标签页），emitChunk 向所有已注册的客户端广播。

createStreamServices 工厂

根据配置创建适当的 JobStore 和 EventTransport 实例：

createStreamServices({ redis, jobStore, eventTransport }):
  if redis.enabled:
    return {
      jobStore: new RedisJobStore(redis.client),
      eventTransport: new RedisEventTransport(redis.pubsub),
    }
  else:
    return {
      jobStore: new InMemoryJobStore(),
      eventTransport: new InMemoryEventTransport(),
    }

---

问题与规避

问题 1：内存实现不支持跨进程恢复

规避：生产环境必须配置 Redis。createStreamServices 根据配置自动选择实现，部署配置通过环境变量控制。

问题 2：作业状态泄漏（已完成的作业未被清理）

规避：completeJob 和 abortJob 方法负责清理作业状态。Redis 实现使用 TTL 自动过期。内存实现需要定期清理（通常在应用重启时）。

问题 3：事件顺序不一致

规避：emitChunk 按调用顺序发送事件。Redis 的 Pub/Sub 保证单通道内的事件顺序。多通道场景需要额外的序列号机制。

---

设计取舍

SSE vs WebSocket

方案	优势	代价
SSE（LibreChat）	简单、浏览器原生、自动重连	仅服务端到客户端
WebSocket	双向通信	更复杂、需要自定义心跳

LibreChat 选择了 SSE 因为流式传输主要是服务端→客户端的单向数据流。WebSocket 的双向能力在此场景用不上。

---

参考来源

• SSE 规范
• Redis Pub/Sub