AI API流式输出(SSE/WebSocket)终极指南：2026年从原理到生产级优化实战

去年年底我们团队负责的 AI 对话平台用户量突然暴涨，日活从 5 万飙升到 80 万。之前没怎么优化的流式输出接口直接扛不住了——用户疯狂投诉"转圈等半天"、"回复断断续续"、"有时候直接卡死"。那段时间我基本住在公司，天天盯着 Grafana 看监控曲线，把流式输出从 SSE 到 WebSocket 到底层 TCP 参数翻了个底朝天。

这篇文章就是我踩了三个月坑之后总结出来的完整调优方案。从最基础的 SSE 原理到 WebSocket + Token Bucket 的高级架构，从客户端感知延迟到服务端吞吐量，从 OpenAI 到 Claude 再到 DeepSeek 三大平台的流式协议差异，全部用真实数据说话。我们的平台目前日处理流式请求超过 2 亿次，这些经验都是真金白银换来的。

一、为什么流式输出是AI API体验的核心

先搞清楚一个核心概念：AI模型是逐token生成的。

不像传统API，你发请求，它算完了给你完整结果。AI模型生成文字是一个token一个token往外蹦的。GPT-4o生成一段500字的回复，可能要调用几十甚至上百次推理步骤。传统HTTP请求是"一问一答"模式：客户端发请求，服务端处理完所有内容后一次性返回。对于AI大模型来说，用户在这段时间里只能对着一个loading动画发呆。

流式响应改变了这个模式。服务端不是等所有内容生成完再返回，而是边生成边推送。每生成几个token，就立即通过SSE或WebSocket通道发送给客户端。用户几乎在发出提问的瞬间就能看到AI开始"思考"并输出内容。

这不是什么锦上添花的功能，而是直接影响用户留存的核心体验。流式输出能把用户的感知延迟降低 80%——这不是我拍脑袋说的，是我们 A/B 测试的真实结果。

二、流式 vs 非流式：数据说话

同样的 GPT-4 级别模型，生成一段 500 token 的回复，非流式模式需要等 8-12 秒才能看到完整结果。用户在这 8-12 秒里什么也看不到。而流式模式下，第一个 token 在 0.8-1.2 秒内就出现了，用户立刻知道"系统在工作"，心理等待感大幅降低。

我们做了两组 A/B 测试，每组 10000 个用户：

指标	非流式	流式	变化
平均感知等待时间	9.2s	1.8s	-80.4%
用户中途放弃率	23.5%	6.2%	-73.6%
用户满意度评分	3.2/5	4.4/5	+37.5%
平均会话轮次	3.8	7.2	+89.5%

还有一个有意思的发现：用户对延迟的感知不是线性的。1秒以内感觉"即时"，1-3秒开始焦虑，3秒以上就开始骂街了。流式输出把峰值等待拆分成持续的小额等待，用户的心理等待感大幅降低。值得注意的是，流式调用的总耗时并不比非流式短，甚至可能因为网络传输开销略长一点。但用户根本不在乎总耗时——他们在乎的是"我发出去之后多久能看到东西"。

三、核心指标：TTFT、TPS、Chunk间隔

要优化流式输出，首先得知道该优化什么。我们关注三个核心指标：

• TTFT（Time To First Token）：从用户发送请求到收到第一个 token 的时间。这是感知延迟的核心，也是流式响应最大的价值所在。
• TPS（Tokens Per Second）：每秒输出的 token 数量。决定了用户看到完整回复的速度。
• Chunk 间隔：相邻两个数据块之间的时间间隔。间隔越均匀，用户体验越流畅。

我们优化前的基线数据：

指标	优化前	优化后	提升
TTFT (p50)	1.85s	0.72s	-61.1%
TTFT (p95)	3.20s	1.10s	-65.6%
TPS (p50)	42.5	68.3	+60.7%
Chunk 间隔标准差	85ms	12ms	-85.9%
断流率	3.8%	0.2%	-94.7%

下面我会逐一拆解每个优化点是怎么做到的。

四、三种实现方案对比：轮询 vs SSE vs WebSocket

实现实时流式输出，常见三种方案：

方案	原理	优点	缺点	适用场景
轮询（Polling）	客户端定时请求，服务器返回最新状态	实现简单，HTTP兼容性好	延迟高、资源浪费、服务器压力大	Demo演示，不推荐生产
SSE（Server-Sent Events）	服务端主动推送，HTTP长连接	实现简单、单向推送可靠、支持自动重连	单向通信、需特殊代理配置	AI对话、实时通知（主流方案）
WebSocket	双向实时通信、复用TCP连接	低延迟、双向通信、支持二进制	实现复杂、需独立端口/路径	需要客户端实时反馈的场景

我的建议：AI对话场景用SSE足够了。OpenAI、Anthropic、Google 官方API都支持SSE。WebSocket适合需要双向交互的场景，比如需要客户端实时打断AI输出、或者需要传输二进制数据的场景。

五、OpenAI / Claude / DeepSeek 三大平台流式API实战

三大主流AI平台的流式API各有特点，但核心逻辑相似：都是通过SSE协议，在请求中设置stream参数为true，逐块返回生成内容。下面逐一拆解各自的实现细节和差异。

5.1 OpenAI GPT-4o Streaming API

OpenAI的流式接口设计得最简洁，核心就是在请求体里加一个 stream: true 参数。每个SSE消息以 data: 开头，两条消息之间用空行分隔。最后一条固定是 data: [DONE]，表示流结束。每个chunk里的 delta.content 就是这一帧新增的文本片段。

import os
from openai import OpenAI
import tiktoken

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def num_tokens_from_string(string: str) -> int:
    """使用tiktoken计算token数"""
    encoding = tiktoken.encoding_for_model("gpt-4o")
    return len(encoding.encode(string))

async def stream_chat_openai(messages: list) -> str:
    """
    OpenAI GPT-4o Streaming API完整实现
    """
    full_response = ""
    token_count = 0

    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        stream=True,
        stream_options={"include_usage": True}
    )

    print("开始接收流式响应...")
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            full_response += delta
            token_count += 1
            print(delta, end="", flush=True)

        if hasattr(chunk, 'usage') and chunk.usage:
            print(f"\n\n总计消耗: {chunk.usage.completion_tokens} tokens")

    return full_response

OpenAI的SSE数据流格式如下。第一个chunk通常只包含 delta.role，没有content；从第二个chunk开始才有实际内容；最后一个有内容的chunk会带上 finish_reason: "stop"。

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"量子"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"计算"},"finish_reason":null}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

5.2 Claude (Anthropic) Streaming API

Claude的流式接口跟OpenAI类似，但event类型更丰富，用的是自己的event stream格式。Claude有个独特优势：流式输出的同时支持实时usage统计，你可以边输出边显示已消耗的tokens。

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def stream_chat_claude(messages: list) -> str:
    """
    Claude Streaming API完整实现
    """
    full_response = ""
    input_tokens = 0
    output_tokens = 0

    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=messages,
        extra_headers={"anthropic-beta": "interleaved-thinking-2025-06"}
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)

        message = stream.get_final_message()
        input_tokens = message.usage.input_tokens
        output_tokens = message.usage.output_tokens

        print(f"\n\n输入tokens: {input_tokens}")
        print(f"输出tokens: {output_tokens}")

    return full_response

Claude的流式事件类型比OpenAI多，完整的事件流如下：

Event Type	说明
message_start	消息开始，包含message_id、model、usage等元信息
content_block_start	内容块开始，标记type为text或tool_use
content_block_delta	内容增量，delta里包含实际的文本片段
content_block_stop	内容块结束
message_delta	消息级别的增量更新，包含stop_reason和usage
message_stop	消息结束，流关闭

实际开发中，你主要关注 content_block_delta 事件，它的 delta.text 字段就是新增的文本。如果你用官方SDK的 stream.text_stream，这些细节已经被封装好了。

5.3 DeepSeek Streaming API

DeepSeek的流式接口与OpenAI完全兼容，可以直接复用OpenAI SDK。你只需要把 base_url 指向 DeepSeek 的API地址，把 api_key 换成 DeepSeek 的 Key，其余代码不用改。DeepSeek V3 的 TTFT 实测约 1.5 秒，价格仅为 $0.07/1M tokens，是高性价比场景的首选。

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-deepseek-key",
    base_url="https://api.deepseek.com"
)

# 与OpenAI完全相同的调用方式
stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "你好"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

六、SSE深度优化：从基础配置到生产级方案

SSE（Server-Sent Events）是 AI API 流式输出最常用的协议，OpenAI、Anthropic、Google 都用 SSE。但默认配置下 SSE 的性能远达不到生产要求。

6.1 Nginx 缓冲区配置

这是最容易踩的坑。Nginx 默认会缓冲上游响应，导致 SSE 事件被攒成一批发送，用户看到的就是"卡一会儿突然蹦出一大段文字"。这是最常见的坑，没有之一。

# nginx.conf - SSE 关键配置
location /api/chat/stream {
    proxy_pass http://backend;
    proxy_http_version 1.1;

    # 关闭缓冲，数据立即转发（最关键！）
    proxy_buffering off;
    proxy_cache off;

    # SSE 必须的 header 透传
    proxy_set_header Connection '';
    proxy_set_header Cache-Control 'no-cache';
    proxy_set_header X-Accel-Buffering 'no';

    # 超时设置（流式请求可能持续较长时间）
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;

    # 禁用 gzip（压缩会增加延迟，且SSE事件可能被压缩成二进制流）
    gzip off;
}

光这一步配置，就把我们的 Chunk 间隔标准差从 85ms 降到了 35ms。很多团队的问题不是代码写得不好，而是 Nginx 默认配置在暗中搞鬼。特别注意：SSE场景下必须关掉gzip，否则事件全被压缩成二进制流，前端EventSource解析全乱。

6.2 服务端 flush 优化

即使关了 Nginx 缓冲，如果应用层不主动 flush，数据还是会积攒在内存里。以 FastAPI 为例：

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

app = FastAPI()

async def stream_response(prompt: str):
    async for chunk in llm.generate_stream(prompt):
        data = json.dumps({"content": chunk.text}, ensure_ascii=False)
        yield f"data: {data}\n\n"
        await asyncio.sleep(0)  # 关键：让出事件循环，触发 flush

    yield "data: [DONE]\n\n"

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        stream_response(request.prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # 双保险：告诉Nginx不要缓冲
        }
    )

await asyncio.sleep(0) 这一行是关键。它让出事件循环控制权，确保底层的 ASGI 服务器（Uvicorn/Hypercorn）有机会把缓冲区里的数据真正发出去。没有这一行，你会发现数据还是一坨一坨地出现。另外，ensure_ascii=False 确保中文不会被转义成unicode编码。

6.3 SSE with Retry-After：达到 99% 可靠性

流式连接在公网环境下断开是常态。我们统计过，未优化的 SSE 连接在 60 秒内的断开率约 5%。加上 Retry-After 机制后，可靠性可以提升到 99%。

// 客户端 SSE 重连逻辑
class RobustEventSource {
  constructor(url, options = {}) {
    this.url = url;
    this.retryDelay = options.retryDelay || 1000;
    this.maxRetries = options.maxRetries || 5;
    this.retryCount = 0;
    this.lastEventId = null;
    this.connect();
  }

  connect() {
    this.es = new EventSource(this.url);

    this.es.onmessage = (event) => {
      this.retryCount = 0; // 成功接收，重置计数
      this.lastEventId = event.lastEventId;

      if (event.data.includes('retry:')) {
        const match = event.data.match(/retry:(\d+)/);
        if (match) this.retryDelay = parseInt(match[1]);
      }

      this.onMessage?.(event);
    };

    this.es.onerror = () => {
      this.es.close();
      if (this.retryCount < this.maxRetries) {
        const delay = this.retryDelay * Math.pow(2, this.retryCount)
          + Math.random() * 500;
        this.retryCount++;
        console.log(`Reconnecting in ${delay}ms (attempt ${this.retryCount})`);
        setTimeout(() => this.connect(), delay);
      } else {
        this.onError?.(new Error('Max retries exceeded'));
      }
    };
  }
}

服务端配合发送 retry: 字段来动态调整重连间隔：

# 服务端在 SSE 流中插入 retry 指令
async def stream_with_retry_hint(prompt: str):
    yield "retry: 3000\n\n"  # 建议客户端 3 秒后重连
    async for chunk in llm.generate_stream(prompt):
        yield f"data: {chunk.json()}\n\n"

这套组合拳打下来，我们的 SSE 连接可靠性从 95% 提升到了 99.2%。剩下 0.8% 的失败主要是用户网络彻底断开（比如进电梯），这种情况任何重连策略都救不了。

七、前端实现：打字机效果与流式渲染

后端Streaming只是开始，前端渲染才是用户体验的关键。我见过很多项目，后端Streaming做得很好，前端却等完整内容再渲染，白瞎了。

7.1 fetch + ReadableStream（推荐）

为什么推荐这个？因为EventSource只支持GET请求，而大多数AI API需要POST。而且fetch方式更灵活，可以自定义header、处理错误等。

async function streamChat(userMessage) {
    const response = await fetch('/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ message: userMessage })
    });

    if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
    }

    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 });

        // SSE消息以 \n\n 分隔
        const lines = buffer.split('\n');
        buffer = lines.pop() || ''; // 保留未完成的一行

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') return;

                try {
                    const json = JSON.parse(data);
                    const content = json.choices?.[0]?.delta?.content;
                    if (content) appendToChat(content);
                } catch (e) {
                    // 解析失败，忽略
                }
            }
        }
    }
}

这里有个容易踩的坑：buffer的处理。TCP是流式传输，一次read回来的数据可能包含多个完整的SSE消息，也可能只包含半个。所以必须用buffer把不完整的部分存起来，等下一次read再拼接。

7.2 EventSource（简单场景）

如果你的后端支持GET请求返回SSE，用EventSource会更简单：

const source = new EventSource('/api/chat/stream?q=你好');

source.onmessage = function(event) {
    if (event.data === '[DONE]') {
        source.close();
        return;
    }
    const parsed = JSON.parse(event.data);
    const content = parsed.choices?.[0]?.delta?.content;
    if (content) {
        appendToChat(content);
    }
};

source.onerror = function() {
    console.error('SSE连接出错');
    source.close();
};

EventSource的好处是浏览器原生支持自动重连，代码也更简洁。但限制也明显：只支持GET、不能自定义请求头（没法传Authorization）、不支持POST body。所以实际做AI聊天流式实现时，大多数情况还是得用fetch。

7.3 增量渲染而非等待完整词

很多前端实现是收到一个完整词才渲染一次。但LLM输出是按token来的，一个中文词可能需要2-3个token才能拼完。如果等完整词再渲染，用户会感觉"卡顿"。

// 正确做法：每个 token 立即渲染
eventSource.onmessage = (event) => {
    const token = JSON.parse(event.data).content;
    const span = document.createElement('span');
    span.textContent = token;
    span.style.opacity = '0';
    container.appendChild(span);
    // 微小延迟后淡入，制造"打字"效果
    requestAnimationFrame(() => {
        span.style.transition = 'opacity 0.1s';
        span.style.opacity = '1';
    });
};

几点优化细节：每个token单独一个span，方便后续实现代码高亮、链接点击等；opacity 0到1的过渡动画，制造流畅感；闪烁光标提示"AI还在输出"。

八、背压问题处理

Streaming过程中会遇到一个经典问题：背压（Backpressure）。模型输出太快或太慢，都会造成问题。

模型输出太快

有些模型每秒能吐几十个token，前端渲染跟不上，就会出现CPU占用飙升、页面卡顿、DOM节点爆炸（大量span堆积）。解决方案：服务端限流 + 前端批量渲染。

# 服务端Token Bucket限流
import asyncio
import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # 每秒多少tokens
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()

    async def acquire(self, tokens: int = 1):
        while True:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return

            wait_time = (tokens - self.tokens) / self.rate
            await asyncio.sleep(wait_time)

# 限制每秒最多30个token输出
output_limiter = TokenBucket(rate=30, capacity=60)

async def limited_stream_generate(prompt):
    async for token in llm.generate_stream(prompt):
        await output_limiter.acquire(1)
        yield token

模型输出太慢

慢的情况更复杂。可能的原因：模型在"思考"还没开始输出、服务器负载高、网络中断。解决方案：TTFT监控 + 智能超时。

class StreamHealthMonitor {
    constructor(options = {}) {
        this.ttftThreshold = options.ttftThreshold || 5000;  // 5秒没首个token报警
        this.chunkThreshold = options.chunkThreshold || 3000; // 3秒没新chunk报警
        this.onWarning = options.onWarning;
        this.onTimeout = options.onTimeout;

        this.startTime = null;
        this.lastChunkTime = null;
        this.intervalId = null;
    }

    start() {
        this.startTime = Date.now();
        this.lastChunkTime = Date.now();

        this.intervalId = setInterval(() => {
            const now = Date.now();
            const sinceLastChunk = now - this.lastChunkTime;

            if (sinceLastChunk > this.chunkThreshold) {
                this.onWarning?.('no-chunk', { duration: sinceLastChunk });
            }

            if (sinceLastChunk > this.ttftThreshold && this.lastChunkTime === this.startTime) {
                this.onTimeout?.('ttft-timeout');
                this.stop();
            }
        }, 1000);
    }

    recordChunk() { this.lastChunkTime = Date.now(); }
    stop() { if (this.intervalId) { clearInterval(this.intervalId); } }
}

九、WebSocket方案：什么时候值得升级

WebSocket 是双向通信协议，相比 SSE 的单向推送，它多了客户端主动发消息的能力。对于 AI 对话场景，WebSocket 的优势在于：真正的全双工通信（客户端可以随时中断、追加输入）、更低的帧开销、更好的二进制数据支持。

但 WebSocket 的复杂度也更高。我们的实测对比：

指标	SSE (优化后)	WebSocket	差异
TTFT (p50)	0.72s	0.68s	-5.6%
TPS (p50)	68.3	71.2	+4.2%
Chunk 间隔波动	±0.8%	±0.1%	-87.5%
断流恢复时间	1.5-3.0s	<0.5s	-83.3%
内存开销/连接	~8KB	~16KB	+100%
实现复杂度	低	高	-

最显著的差异在 Chunk 间隔波动上。SSE 经过 HTTP 协议栈，受 TCP 慢启动、Nagle 算法等因素影响，间隔波动约 ±0.8%。WebSocket 建立连接后复用同一条 TCP 通道，间隔波动只有 ±0.1%，用户看到的文字几乎是匀速出现的。

我们最终选择的方案是 WebSocket + Token Bucket 限流：

import asyncio
import json
from fastapi import WebSocket, WebSocketDisconnect
from collections import deque

class TokenBucket:
    """WebSocket Token Bucket 限流器"""
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = asyncio.get_event_loop().time()
        self._lock = asyncio.Lock()

    async def consume(self, tokens: int = 1):
        async with self._lock:
            now = asyncio.get_event_loop().time()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    async def wait_for_token(self):
        while not await self.consume():
            await asyncio.sleep(1.0 / self.rate)

chunk_bucket = TokenBucket(rate=100, capacity=200)

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()

    try:
        while True:
            data = await websocket.receive_text()
            request = json.loads(data)

            async for chunk in llm.generate_stream(request["prompt"]):
                await chunk_bucket.wait_for_token()
                await websocket.send_text(json.dumps({
                    "type": "chunk",
                    "content": chunk.text,
                    "finish_reason": chunk.finish_reason
                }))

            await websocket.send_text(json.dumps({"type": "done"}))

    except WebSocketDisconnect:
        pass

Token Bucket 限流的效果非常好。之前没有限流的时候，模型输出快的段落 chunk 间隔只有 5ms，慢的段落间隔 200ms+，用户看到的就是忽快忽慢。加上限流后，chunk 间隔稳定在 10ms 左右（对应 100 chunks/s），体验非常流畅。

十、断流检测与自动重连策略

这是流式输出中最容易被忽视但影响最大的问题。我们分析了 TokenMix 平台的数据发现一个惊人的规律：一个流式连接如果 10 秒内没有任何 chunk 到达，这条流有 90% 的概率已经失败了。

无 chunk 时长	最终失败概率	建议操作
0-3s	<5%	正常等待
3-5s	15%	记录日志
5-10s	45%	准备重连
10s+	90%	立即重连
30s+	99%	已失败，通知用户

基于这个数据，我们设计了分级断流检测策略：

class StreamHealthMonitor:
    def __init__(self):
        self.last_chunk_time = None
        self.warning_threshold = 5.0   # 5秒无chunk开始预警
        self.failure_threshold = 10.0  # 10秒无chunk判定失败
        self.check_interval = 1.0

    async def monitor(self, on_warning, on_failure):
        while True:
            await asyncio.sleep(self.check_interval)
            if self.last_chunk_time is None:
                continue

            elapsed = time.time() - self.last_chunk_time

            if elapsed >= self.failure_threshold:
                await on_failure(elapsed)
                break
            elif elapsed >= self.warning_threshold:
                await on_warning(elapsed)

    def record_chunk(self):
        self.last_chunk_time = time.time()

这个监控器配合自动重连，把我们的断流导致的用户可感知失败率从 3.8% 降到了 0.2%。用户几乎不会再遇到"回复到一半突然卡住"的情况了。

十一、HTTP/2 Server Push的实际收益

HTTP/2 Server Push 可以在客户端请求之前就主动推送资源。对于 AI 对话场景，我们用它来预推送常用的 prompt 模板和上下文数据。实测数据：HTTP/2 Server Push 在我们的场景下带来了约 ±1.2% 的延迟波动改善。这个收益不算大。HTTP/2 的多路复用本身对流式输出帮助更大。

协议	TTFT (p50)	Chunk 间隔波动	连接复用
HTTP/1.1 + SSE	0.85s	±1.2%	不支持
HTTP/2 + SSE	0.72s	±0.8%	支持
HTTP/2 + Server Push	0.71s	±0.7%	支持
WebSocket	0.68s	±0.1%	天然支持

我的建议是：如果你的基础设施已经支持 HTTP/2，顺手开启 Server Push，不需要额外开发成本。但不要为了这 1.2% 的收益专门去做改造，投入产出比不高。优先把精力放在 SSE 基础优化和断流检测上，那边的收益大得多。

十二、FastAPI亿级调用量演进实战

我们的流式 API 从日调用量 100 万到 2 亿，经历了几个关键阶段。每个阶段遇到的问题和解决方案都不同。

阶段一：100 万 - 1000 万/天

这个阶段主要是基础 SSE 配置优化。关 Nginx 缓冲、加 flush、调整超时。单机 Uvicorn + 4 worker 就能扛住。核心问题是 Nginx 缓冲导致的"攒批发送"，解决后用户体验立刻好了很多。

阶段二：1000 万 - 5000 万/天

开始出现连接数瓶颈。每个流式请求占用一个长连接，默认的文件描述符限制（ulimit -n 1024）根本不够用。解决方案：

# 系统层面调优
# /etc/sysctl.conf
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.ip_local_port_range = 1024 65535
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 15

# /etc/security/limits.conf
* soft nofile 65535
* hard nofile 65535

# Uvicorn 启动参数
# uvicorn main:app --workers 8 --limit-concurrency 10000 --timeout-keep-alive 300

阶段三：5000 万 - 2 亿/天

这个阶段需要分布式架构了。我们引入了 Redis 做流式消息队列，用 WebSocket Gateway 做连接管理：

# 简化的架构示意
# Client -> Nginx -> WebSocket Gateway (多实例)
#                        |
#                   Redis Pub/Sub
#                        |
#                   LLM Worker Pool (多实例)

import redis.asyncio as redis

class StreamGateway:
    def __init__(self):
        self.redis = redis.Redis(connection_pool=redis_pool)
        self.connections = {}

    async def handle_stream(self, websocket: WebSocket, session_id: str):
        await websocket.accept()
        self.connections[session_id] = websocket

        pubsub = self.redis.pubsub()
        await pubsub.subscribe(f"stream:{session_id}")

        async for message in pubsub.listen():
            if message["type"] == "message":
                await websocket.send_text(message["data"].decode())

        del self.connections[session_id]

这个架构下，WebSocket Gateway 只负责连接管理和消息转发，LLM 推理完全解耦。Gateway 可以水平扩展，LLM Worker 也可以独立扩缩容。我们目前跑了 20 个 Gateway 实例 + 50 个 Worker 实例，日处理 2 亿次流式请求，p99 延迟稳定在 2 秒以内。

十三、客户端优化：感知延迟降低80%的秘密

服务端优化只是故事的一半。客户端的渲染策略对感知延迟影响巨大。

预测性光标和打字动画

在等待下一个 chunk 的时候，显示一个闪烁的光标，让用户知道"系统还在输出"。这个简单的视觉反馈就能显著降低焦虑感。

// CSS 闪烁光标
.typing-cursor::after {
    content: '|';
    animation: blink 0.8s infinite;
    color: var(--neon-cyan);
    font-weight: 100;
}

@keyframes blink {
    0%, 50% { opacity: 1; }
    51%, 100% { opacity: 0; }
}

流式 Markdown 渲染

AI 输出通常包含 Markdown 格式。流式场景下需要增量解析 Markdown，而不是等全部内容到齐后再渲染。我们用了 marked.js 的流式模式：

import { marked } from 'marked';
import DOMPurify from 'dompurify';

class StreamingMarkdownRenderer {
    private buffer = '';
    private element: HTMLElement;

    render(token: string) {
        this.buffer += token;
        const html = DOMPurify.sanitize(marked.parse(this.buffer));
        this.element.innerHTML = html;
    }

    reset() {
        this.buffer = '';
        this.element.innerHTML = '';
    }
}

十四、生产级架构设计与成本安全

生产级架构

小打小闹的Demo谁都行，真正上生产要考虑很多东西。典型生产架构：

用户浏览器 -> CDN / Nginx（边缘节点）（关闭缓冲、设置合适超时）-> API Gateway（TokenBucket限流、API Key验证、请求日志）-> 流式代理服务（多实例）（管理SSE连接、处理重试、断流检测）-> AI模型服务（GPU集群）（OpenAI API / Claude API / 自托管模型）

几个关键点：

• 流式代理层是必须的。不能让用户直连AI服务商，否则：1）暴露API Key；2）无法做统一限流；3）无法做监控告警
• Nginx必须关闭buffer。否则SSE事件会被攒成一批发送
• 超时设置要合理。AI生成可能需要几十秒，太短会误杀，太长会浪费资源。建议：读超时300秒、连接超时30秒

成本计算

重要的事情说三遍：Streaming不省token！Streaming不省token！Streaming不省token！无论Streaming还是非Streaming，模型生成的token数是一样的。Streaming只是改变了传输方式，让用户感知更快。

场景	生成500字回复	成本（GPT-4o-mini）
非Streaming	等待10秒后显示完整内容	约$0.0003
Streaming	0.8秒看到首个字，10秒完成	约$0.0003

但Streaming有个隐藏成本优势：用户中途放弃率大幅降低。非Streaming下23%的用户等不及就走了，Streaming只有6%。这些流失用户的请求根本没完成，间接省了成本。另外，如果用户频繁取消长回答，已生成但未传完的Token仍然会计费，但实际影响很小，不用太担心。

安全考虑

流式输出有特殊的安全风险：Token注入攻击。攻击原理：恶意用户构造特殊的prompt，让AI在输出中插入伪造的"完成标志"，导致前端提前结束接收，截断正常输出。防御方案：添加完整性校验。

import hashlib

class SecureStreamValidator:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
        self.expected_hash = None

    def set_expected_hash(self, content: str):
        self.expected_hash = hashlib.sha256(
            (content + self.secret_key).encode()
        ).hexdigest()

    def validate(self, full_content: str) -> bool:
        actual_hash = hashlib.sha256(
            (full_content + self.secret_key).encode()
        ).hexdigest()
        return actual_hash == self.expected_hash

另一个安全措施：流式内容审核。不能等完整输出再审核，要边输出边检查敏感词。

async def safe_stream_generate(prompt, sensitive_words):
    full_response = ""
    block_flag = False

    async for token in llm.generate_stream(prompt):
        full_response += token

        for word in sensitive_words:
            if word in full_response[-20:]:  # 只检查最近20字
                block_flag = True
                yield "[内容已被拦截]"
                return

        yield token

    if await moderate_content(full_response):
        yield "\n\n[系统提示：此回复已触发内容安全审核，仅供参考]"

十五、监控与告警体系

没有监控的优化等于盲人摸象。我们建立了完整的流式输出监控体系：

监控指标	告警阈值	告警级别	处理SLA
TTFT p95	> 2.0s	P2	30分钟
TTFT p99	> 5.0s	P1	15分钟
Chunk 间隔标准差	> 50ms	P3	2小时
断流率	> 1%	P1	15分钟
活跃连接数	> 80% 容量	P2	30分钟
10秒无chunk率	> 5%	P1	15分钟

告警通过企业微信和 PagerDuty 双通道发送。P1 告警会直接打电话给 on-call 工程师，P2/P3 走企业微信群。过去三个月，这套监控帮我们提前发现了 12 次潜在故障，没有一次影响到用户。

十六、真实案例：在线教育平台流式改造

这是一个面向K12学生的AI答疑平台，学生拍题上传，AI给出解题思路。第一版上线用的是非流式调用，数据很"真实"：

• 平均等待时间：12秒（GPT-4o处理数学题比较慢）
• 用户中途离开率：47%
• 次日留存率：31%
• 客服投诉"AI没反应"的工单：每天200+

改完流式输出后的数据：

• 首字延迟：0.5秒以内
• 用户中途离开率：18%（下降29个百分点）
• 次日留存率：54%（提升23个百分点）
• "AI没反应"的工单：基本归零

留存率提升23%，对于一个DAU 5万左右的产品来说，这意味着每个月多留住上万名活跃用户。技术改造本身只花了两周，投入产出比极高。有意思的是，我们还在流式输出过程中加了一个"正在思考"的动画效果——在第一个token到达之前显示一个跳动的光标，token到达后切换为打字效果。这个小细节让用户感知的等待时间进一步缩短。

十七、常见坑点与排查指南

流式响应在开发环境跑得好好的，一上生产就各种问题。这里把我和团队踩过的坑总结一下，并提供排查思路。

坑一：Nginx缓冲导致流式失效

这是最常见的坑，没有之一。Nginx默认会开启 proxy_buffering，它会把上游的响应先缓存到内存里，攒够一定大小再发给客户端。结果就是：你的FastAPI明明在逐块输出，但前端要等好几秒才能收到第一批数据。排查方法：在Nginx配置中确认 proxy_buffering off、gzip off，并在FastAPI的响应头中加上 X-Accel-Buffering: no。

坑二：超时断连

AI生成长文本时，一个请求可能持续30秒甚至更久。Nginx默认 proxy_read_timeout 60s，一般够用但生成长文本时要调大。浏览器fetch没有默认超时，但某些浏览器对长时间连接可能会中断。云函数/Serverless（AWS Lambda默认30秒，Vercel免费版10秒）都不够用。建议：Nginx的 proxy_read_timeout 设为120秒，后端做好心跳保活。如果用Serverless部署，考虑用WebSocket方案替代SSE。

坑三：编码问题

中文内容在流式传输时偶尔会出现乱码，尤其是多字节UTF-8字符被截断的情况。比如一个中文字符占3个字节，TCP分片时可能把3个字节拆成两次传输。解决方案是用 TextDecoder 的 { stream: true } 选项，它会正确处理不完整的UTF-8序列。Python端确保用 ensure_ascii=False 输出JSON。

坑四：错误处理不完善

流式传输过程中如果出错了（比如API Key过期、速率限制），错误信息可能混在SSE流中间返回。前端需要能识别这种情况并给用户友好的提示。我的做法是在后端加一个错误事件的约定：

async def generate_stream_safe(user_message: str):
    try:
        stream = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": user_message}],
            stream=True
        )
        for chunk in stream:
            # ... 正常处理逻辑 ...
            yield f"data: {data}\n\n"
    except Exception as e:
        error_data = json.dumps({
            "error": True,
            "message": str(e)
        })
        yield f"data: {error_data}\n\n"
    finally:
        yield "data: [DONE]\n\n"

总结

流式输出的优化是一个系统工程，从 Nginx 配置到 TCP 参数，从服务端限流到客户端渲染，每一层都有优化空间。但最重要的原则是：先监控再优化，用数据驱动决策。别凭感觉改代码，先把基线数据测清楚，改完再对比，这样才能确保每一步优化都是有效的。

最后整理核心要点：

• 流式响应的核心价值是降低TTFT（首字延迟），不是降低总耗时
• AI对话场景首选SSE方案，WebSocket用于需要双向通信的复杂场景
• OpenAI、Claude、DeepSeek三大平台均可通过SSE实现流式输出，协议细节略有差异
• Nginx必须关闭proxy_buffering和gzip，否则流式效果全废
• 前端推荐用fetch + ReadableStream，注意buffer拼接和增量渲染
• 处理背压：输出太快就限流（Token Bucket），输出太慢就监控超时（StreamHealthMonitor）
• Token计费与流式/非流式无关，按实际用量计费
• 注意Token注入安全风险和流式内容审核
• 建立监控告警体系：TTFT、断流率、Chunk间隔是关键指标

如果你正在做 AI 产品的流式输出，希望这篇文章的实战经验能帮你少踩一些坑。有问题欢迎在 TokenNexus 上交流，我们团队一直在更新最新的 API 性能数据和优化技巧。