流式输出的断线续传机制

1.1 SSE (Server-Sent Events) 协议

SSE 是 HTTP 长连接协议：客户端发一个 GET/POST 请求，服务端持续推送 text/event-stream 格式的数据，直到结束或断开。

// 客户端发起 POST 请求
POST /v1/chat/completions
Content-Type: application/json
Accept: text/event-stream

{ "stream": true, "messages": [...] }

// 服务端持续推送 SSE 事件（逐 token）
data: {"id":"req_abc", "delta":{"content":"春"}}

data: {"id":"req_abc", "delta":{"content":"眠"}}

data: {"id":"req_abc", "delta":{"content":"不"}}

data: {"id":"req_abc", "delta":{"content":"觉"}}

data: {"id":"req_abc", "delta":{"content":"晓"}}

data: [DONE]

1.2 关键概念：请求级有状态

一次 SSE 连接对应 一个 LLM 推理会话。服务端在内存中保存了：

• 当前已生成的 token 列表
• KV Cache（注意力机制的中间状态，加速后续生成）
• 用户原始 messages 上下文

这个会话状态只在连接存活期间有效。连接一断，服务端通常会立即清理。

3.1 架构总览

LLM 推理引擎和消息推送解耦 —— 推理不断，消息堆积在 Redis，客户端随时来取

4.1 第一步：建立会话

用户发起提问时，服务端不仅要启动 LLM 推理，还要创建一个持久化会话：

// 服务端处理用户提问
func HandleChat(userId, conversationId, messages) {

    // 1. 生成唯一的会话 ID
    sessionId := UUID()

    // 2. 在 Redis 中创建会话状态
    redis.HSet("session:"+sessionId, {
        "status":         "generating",      // 状态：生成中
        "conversation_id": conversationId,    // 关联的对话 ID
        "user_id":         userId,              // 用户 ID
        "messages":         json(messages),     // 原始请求 messages
        "created_at":       now(),
        "total_tokens":     0,
        "last_seq":         0,                  // 最后一条消息序号
    })

    // 3. 创建一个 Redis List 用于堆积生成的 token
    //    key: session:{id}:tokens

    // 4. 异步启动 LLM 推理（不阻塞 HTTP 响应）
    go startLLMInference(sessionId, messages)

    // 5. 立即返回 sessionId 给客户端
    return { "session_id": sessionId }
}

4.2 第二步：推理进程写 Token 到 Redis 队列

LLM 推理在独立 goroutine / 进程中进行，每生成一个 token 就写入 Redis：

func startLLMInference(sessionId, messages) {

    stream := llm.ChatCompletionStream(messages)

    seq := 0

    for chunk := range stream {

        seq++

        // 每个 token 封装成带序号的消息
        tokenMsg := json({
            "seq":     seq,                    // 消息序号，用于断点续传定位
            "content": chunk.Delta.Content,   // token 文本内容
            "ts":      now(),                 // 时间戳
        })

        // 写入 Redis List — 尾部追加
        redis.RPush("session:"+sessionId+":tokens", tokenMsg)

        // 更新最后序号
        redis.HSet("session:"+sessionId, "last_seq", seq)

        // 发布到 Redis Pub/Sub，通知推送层有新 token
        redis.Publish("channel:session:"+sessionId, tokenMsg)
    }

    // 推理完成，标记会话状态
    redis.HSet("session:"+sessionId, "status", "completed")
    redis.Publish("channel:session:"+sessionId, "__DONE__")
}

4.3 第三步：实时推送（正常流程）

客户端正常在线时，通过 SSE / WebSocket 订阅 Redis Pub/Sub，实时收到每个 token：

// 推送层 —— 订阅 Redis 频道，转发给客户端
func streamTokens(sessionId) {
    // 建立 SSE 连接，设置 Content-Type: text/event-stream

    pubsub := redis.Subscribe("channel:session:"+sessionId)

    for msg := range pubsub.Channel() {
        if msg.Payload == "__DONE__" {
            writeSSE("data: [DONE]\n\n")
            break
        }
        writeSSE("data: " + msg.Payload + "\n\n")
        flush()
    }
}