为什么需要流式响应

LLM 生成一个完整回答可能需要 5-15 秒。如果等全部生成再返回，用户体验极差。流式响应让用户在第一个 token 生成时就能看到输出，体感延迟从"等待 10 秒"变成"等待 0.5 秒"。

非流式:  [████████████████████████████] 10s → 一次性返回
流式:    [█][█][█][█][█][█][█][█][█][█] 0.5s 首字 + 持续输出

三种流式协议对比

特性SSEWebSocketgRPC Stream
方向服务端→客户端双向双向
底层HTTP/1.1 或 HTTP/2HTTP 升级HTTP/2
浏览器支持原生 EventSource原生 WebSocket需要 gRPC-Web
自动重连内置需手动实现需手动实现
代理友好非常好一般差（需 HTTP/2）
适用场景LLM 流式输出实时对话 + 语音内部微服务

结论：面向用户的 LLM 流式输出，SSE 是首选。需要双向交互（如语音对话）用 WebSocket。内部服务间通信用 gRPC。

SSE 实现

后端：FastAPI + OpenAI Streaming

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import json
import asyncio

app = FastAPI()
client = AsyncOpenAI()

@app.post("/api/chat/stream")
async def chat_stream(request: dict):
    async def event_generator():
        try:
            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=request["messages"],
                stream=True,
                temperature=0.7,
            )

            async for chunk in stream:
                if chunk.choices[0].delta.content:
                    data = json.dumps({
                        "content": chunk.choices[0].delta.content,
                        "role": "assistant"
                    })
                    yield f"data: {data}\n\n"

            yield f"data: {json.dumps({'done': True})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # nginx: 禁用缓冲
        }
    )

前端：EventSource + ReadableStream

// 方案1: 使用 fetch + ReadableStream (支持 POST)
async function streamChat(messages: Message[]) {
  const response = await fetch('/api/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages }),
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split('\n\n');
    buffer = lines.pop() || '';

    for (const line of lines) {
      if (!line.startsWith('data: ')) continue;
      const data = JSON.parse(line.slice(6));

      if (data.done) {
        console.log('Stream completed');
        break;
      }
      if (data.error) {
        console.error('Stream error:', data.error);
        break;
      }
      // 追加到 UI
      appendToUI(data.content);
    }
  }
}

// 方案2: 使用 EventSource (仅支持 GET)
const evtSource = new EventSource('/api/chat/stream?prompt=hello');
evtSource.onmessage = (e) => {
  const data = JSON.parse(e.data);
  if (data.done) evtSource.close();
  else appendToUI(data.content);
};
evtSource.onerror = (e) => {
  console.error('SSE error, will auto-reconnect');
};

SSE 断线重连

SSE 原生支持自动重连，但需要服务端配合发送 retry 和 id 字段：

async def event_generator_with_recovery():
    last_id = 0
    async for chunk in stream:
        last_id += 1
        data = json.dumps({"content": chunk, "id": last_id})
        yield f"id: {last_id}\n"
        yield f"retry: 3000\n"  # 重连间隔 3s
        yield f"data: {data}\n\n"

// 前端处理重连和恢复
evtSource.addEventListener('open', () => {
  console.log('Connected, resuming from', lastEventId);
});

evtSource.onmessage = (e) => {
  const data = JSON.parse(e.data);
  lastEventId = data.id;
  // ...
};

WebSocket 实现

适用于需要双向实时通信的场景（如语音对话、实时纠错）：

# 后端: FastAPI WebSocket
from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/chat")
async def ws_chat(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            message = await ws.receive_json()

            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=message["messages"],
                stream=True,
            )

            async for chunk in stream:
                content = chunk.choices[0].delta.content
                if content:
                    await ws.send_json({"type": "token", "content": content})

            await ws.send_json({"type": "done"})

    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        await ws.send_json({"type": "error", "message": str(e)})

// 前端 WebSocket
class WSChatClient {
  private ws: WebSocket;
  private reconnectAttempts = 0;
  private maxReconnect = 5;

  constructor(url: string) {
    this.connect(url);
  }

  private connect(url: string) {
    this.ws = new WebSocket(url);

    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      console.log('WebSocket connected');
    };

    this.ws.onmessage = (e) => {
      const data = JSON.parse(e.data);
      if (data.type === 'token') appendToUI(data.content);
      if (data.type === 'done') console.log('Completed');
    };

    this.ws.onclose = () => {
      if (this.reconnectAttempts < this.maxReconnect) {
        const delay = Math.min(1000 * 2 ** this.reconnectAttempts, 30000);
        this.reconnectAttempts++;
        setTimeout(() => this.connect(url), delay);
      }
    };
  }

  send(messages: Message[]) {
    this.ws.send(JSON.stringify({ messages }));
  }
}

gRPC 流式实现

适用于内部微服务间的高性能通信：

// proto/llm.proto
syntax = "proto3";

service LLMService {
  rpc StreamChat (ChatRequest) returns (stream ChatResponse);
}

message ChatRequest {
  string model = 1;
  repeated Message messages = 2;
  float temperature = 3;
}

message Message {
  string role = 1;
  string content = 2;
}

message ChatResponse {
  string content = 1;
  bool done = 2;
}

# gRPC 服务端
import grpc
from concurrent import futures

class LLMServicer(llm_pb2_grpc.LLMServiceServicer):
    async def StreamChat(self, request, context):
        stream = await client.chat.completions.create(
            model=request.model,
            messages=[{"role": m.role, "content": m.content} for m in request.messages],
            stream=True,
        )
        async for chunk in stream:
            content = chunk.choices[0].delta.content or ""
            yield llm_pb2.ChatResponse(content=content, done=False)
        yield llm_pb2.ChatResponse(content="", done=True)

背压处理

当客户端消费速度慢于服务端生成速度时，需要背压机制防止内存溢出：

import asyncio

class BackpressureManager:
    def __init__(self, max_queue_size=100):
        self.queue = asyncio.Queue(maxsize=max_queue_size)
        self.dropped = 0

    async def produce(self, item):
        try:
            self.queue.put_nowait(item)
        except asyncio.QueueFull:
            self.dropped += 1
            # 策略1: 丢弃中间 token，保留首尾
            # 策略2: 等待消费者（可能增加延迟）
            # 策略3: 要求生产者减速

    async def consume(self):
        while True:
            item = await self.queue.get()
            yield item
            self.queue.task_done()

// 前端背压: 使用 ReadableStream 的 pull 控制
const stream = new ReadableStream({
  pull(controller) {
    // 只有当消费者请求数据时才读取
    return reader.read().then(({ done, value }) => {
      if (done) {
        controller.close();
        return;
      }
      controller.enqueue(value);
    });
  },
  // highWaterMark 控制缓冲区大小
}, { highWaterMark: 1 });

生产级最佳实践

1. Nginx 配置

location /api/chat/stream {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;           # 关键: 禁用缓冲
    proxy_cache off;
    proxy_read_timeout 300s;       # LLM 可能很慢
    chunked_transfer_encoding on;
}

2. 超时与取消

from fastapi import Request

@app.post("/api/chat/stream")
async def chat_stream(request: Request, body: dict):
    async def event_generator():
        try:
            stream = await client.chat.completions.create(...)
            async for chunk in stream:
                if await request.is_disconnected():
                    print("Client disconnected, stopping stream")
                    break
                yield format_sse(chunk)
        except asyncio.CancelledError:
            print("Stream cancelled")
            raise

    return StreamingResponse(event_generator(), ...)

3. 监控指标

# 关键指标
metrics = {
    "stream_first_token_latency": [],   # 首 token 延迟 (目标 < 500ms)
    "stream_total_duration": [],        # 总时长
    "stream_token_rate": [],            # token/秒
    "stream_disconnect_rate": [],       # 客户端断连率
    "stream_error_rate": [],            # 错误率
}

协议选择决策图

用户需要看到逐字输出?
├─ 是 → 单向输出?
│   ├─ 是 → SSE (最简单，浏览器原生支持)
│   └─ 否 → 需要双向交互?
│       ├─ 是 → WebSocket
│       └─ 否 → SSE + POST 组合
└─ 否 → 普通 HTTP JSON API

内部服务间通信?
├─ 需要流式 → gRPC Stream
└─ 不需要 → gRPC Unary 或 REST

总结

SSE 是 LLM 流式输出的最佳选择：浏览器原生支持、自动重连、代理友好、实现简单。WebSocket 适合需要双向通信的复杂场景。gRPC 适合内部高性能通信。无论选哪种，都要处理好背压、超时、取消和监控，才能扛住生产流量。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。