流式输出

681 字约 2 分钟

domain/aiai/engineering

2026-07-24

1. 核心结论

流式输出（Streaming）让 AI 应用逐 token 返回结果，而非等待完整响应
流式输出显著改善用户体验，减少感知延迟
流式输出需要前端和后端协同支持
流式输出增加了错误处理的复杂度
不是所有场景都适合流式输出

2. 基础概念

Streaming：流式输出，逐步返回生成结果。

Server-Sent Events（SSE）：服务器推送事件，HTTP 长连接推送数据。

WebSocket：全双工通信协议，支持实时双向通信。

Token-by-Token：逐 token 返回。

First Token Latency：首 token 延迟，从请求到收到第一个 token 的时间。

Chunk：流式输出中的数据块。

Backpressure：背压，控制数据流速度防止溢出。

3. 工作原理

流式输出流程：

客户端                    服务端                    LLM API
  │                        │                        │
  │──── 请求 ──────────────►│                        │
  │                        │──── 请求 ──────────────►│
  │                        │                        │
  │                        │◄─── token 1 ────────────│
  │                        │──── SSE: token 1 ──────►│
  │◄─── SSE: token 1 ──────│                        │
  │                        │                        │
  │                        │◄─── token 2 ────────────│
  │                        │──── SSE: token 2 ──────►│
  │◄─── SSE: token 2 ──────│                        │
  │                        │                        │
  │                        │         ...             │
  │                        │                        │
  │                        │◄─── [DONE] ─────────────│
  │                        │──── SSE: [DONE] ───────►│
  │◄─── SSE: [DONE] ───────│                        │

实现方式：

SSE（推荐）：

基于 HTTP，兼容性好
单向推送，适合 AI 输出
自动重连
示例：

const eventSource = new EventSource('/api/chat');
eventSource.onmessage = (event) => {
  appendToUI(event.data);
};

WebSocket：
- 双向通信
- 适合交互式场景
- 需要额外处理连接管理
HTTP Chunked Transfer：
- 流式 HTTP 响应
- 简单但功能有限

后端实现（Python FastAPI）：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/api/chat")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for token in llm.generate_stream(request.prompt):
            yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

前端处理：

const response = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ prompt: 'Hello' })
});

const reader = response.body.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = new TextDecoder().decode(value);
  appendToUI(text);
}

4. 实战场景

聊天机器人实时回复
代码生成实时展示
文档摘要逐步显示
翻译实时输出
数据分析结果流式展示

5. 常见误区

不处理连接断开：网络问题导致中断
不处理错误：流式输出中错误难以处理
忽视首 token 延迟：流式不减少首 token 延迟
不适合的场景：需要完整结果才能处理的场景
不控制速率：大量并发连接消耗资源

6. 进阶方向

流式输出压缩
断线重连优化
流式输出缓存
多路流式合并
流式输出监控
自适应流式策略

7. 推荐资料

OpenAI Streaming - https://platform.openai.com/docs/api-reference/chat/create#chat-create-stream
Anthropic Streaming - https://docs.anthropic.com/en/api/messages-streaming
MDN Server-Sent Events - https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
FastAPI Streaming - https://fastapi.tiangolo.com/advanced/custom-response/#streamingresponse