AI API流式输出(Streaming)实战：SSE与WebSocket完整实现指南

去年给客户做AI客服项目，上线第一天就被骂惨了。用户发一条消息，页面转圈10秒钟才显示完整回复。客户打电话过来：你们这AI是化石吗？反应这么慢？

我当时也懵了。接口明明调通了，AI也正常返回了，就是等得太久。后来查了一下，GPT-4o生成一段300字的回复本身就要8-10秒，但用户不知道啊，他们只看到一个转圈的loading，等得花儿都谢了。

后来加上Streaming（流式输出），第一个token 0.8秒就出来了，用户立刻感觉"系统在响应"。同样10秒的总生成时间，但体验完全不一样。这篇文章就是我踩完坑后的完整记录，从原理到代码到架构，全部真实可运行。

为什么AI需要Streaming

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

不像传统API，你发请求，它算完了给你完整结果。AI模型生成文字是一个token一个token往外蹦的。GPT-4o生成一段500字的回复，可能要调用几十甚至上百次推理步骤。

这就有了一个关键指标：TTFT（Time To First Token，首个Token响应时间）。用户发消息到看见第一个字的时间。

还有个有意思的点：用户对延迟的感知不是线性的。1秒以内感觉"即时"，1-3秒开始焦虑，3秒以上就开始骂街了。Streaming把峰值等待拆分成持续的小额等待，用户的心理等待感大幅降低。

三种实现方案对比

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

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

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

OpenAI GPT-4o Streaming API实战

先安装依赖：

pip install openai tiktoken sseclient-py

然后是完整的Python实现代码：

import os
from openai import OpenAI
import tiktoken
import sseclient
import requests

# 初始化客户端
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
            
            # 实时显示（实际项目可替换为WebSocket推送）
            print(delta, end="", flush=True)
        
        # 检查usage信息（流结束时返回）
        if hasattr(chunk, 'usage') and chunk.usage:
            print(f"\n\n总计消耗: {chunk.usage.completion_tokens} tokens")
    
    return full_response

# 使用示例
if __name__ == "__main__":
    messages = [
        {"role": "system", "content": "你是一个专业的技术顾问。"},
        {"role": "user", "content": "请解释什么是RESTful API，用通俗易懂的方式。"}
    ]
    
    import asyncio
    result = asyncio.run(stream_chat_openai(messages))

补充说明一下SSE的后端代理配置。很多人Streaming效果差，不是代码问题，是Nginx/网关的配置问题：

# Nginx SSE关键配置
location /api/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压缩
    gzip off;
}

踩坑记录：有一次我配置错了，把gzip on开着，结果SSE事件全被压缩成二进制流，前端EventSource解析全乱了。SSE场景下必须关掉gzip。

Claude Streaming API实战

Anthropic的Claude Streaming API实现类似，但返回格式稍有不同：

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完整实现
    Claude支持更好的结构化输出
    """
    full_response = ""
    input_tokens = 0
    output_tokens = 0
    
    with client.messages.stream(
        model="claude-opus-4-5",
        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)
        
        # 获取完整消息和usage
        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

# 使用示例
if __name__ == "__main__":
    messages = [
        {"role": "user", "content": "什么是向量数据库？为什么RAG需要它？"}
    ]
    result = stream_chat_claude(messages)

Claude有个独特优势：流式输出的同时支持实时usage统计。你可以边输出边显示已消耗的tokens，用户能直观看到"AI还在思考，还有多少输出"。

前端实现：打字机效果

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

先看EventSource方案（SSE）：

<!-- HTML结构 -->
<div id="chat-container">
    <div id="ai-response" class="typing-area"></div>
    <span class="cursor">|</span>
</div>

<style>
.typing-area {
    font-family: 'Noto Sans SC', sans-serif;
    line-height: 1.8;
    white-space: pre-wrap;
    word-break: break-word;
}

.cursor::after {
    content: '|';
    animation: blink 0.8s infinite;
    color: #00f0ff;
}

@keyframes blink {
    0%, 50% { opacity: 1; }
    51%, 100% { opacity: 0; }
}
</style>

<script>
class StreamingChat {
    constructor() {
        this.container = document.getElementById('ai-response');
        this.cursor = document.querySelector('.cursor');
        this.isStreaming = false;
    }

    async sendMessage(message) {
        // 清除之前的内容
        this.container.innerHTML = '';
        this.isStreaming = true;
        
        try {
            const response = await fetch('/api/chat/stream', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({message: message}),
            });
            
            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数据（格式: data: {"content": "xxx"}\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]') {
                            this.isStreaming = false;
                            this.cursor.style.display = 'none';
                            return;
                        }
                        
                        try {
                            const json = JSON.parse(data);
                            this.appendToken(json.content);
                        } catch (e) {
                            // 解析失败，忽略这行
                        }
                    }
                }
            }
        } catch (error) {
            console.error('Stream error:', error);
            this.container.innerHTML = '连接失败，请重试';
        }
        
        this.isStreaming = false;
        this.cursor.style.display = 'none';
    }

    appendToken(token) {
        // 创建新span并追加（方便后续高亮处理）
        const span = document.createElement('span');
        span.textContent = token;
        span.style.opacity = '0';
        this.container.appendChild(span);
        
        // 微小延迟后淡入，制造流畅打字效果
        requestAnimationFrame(() => {
            span.style.transition = 'opacity 0.05s';
            span.style.opacity = '1';
        });
    }
}
</script>

几点优化细节：

• 每个token单独一个span，方便后续实现代码高亮、链接点击等
• opacity 0到1的过渡动画，制造流畅感
• 闪烁光标提示"AI还在输出"
• buffer处理确保SSE事件不丢帧

背压问题处理

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

模型输出太快

有些模型每秒能吐几十个token，前端渲染跟不上，就会出现：

• CPU占用飙升
• 页面卡顿
• DOM节点爆炸（大量span堆积）

解决方案：服务端限流 + 前端批量渲染

# 服务端Token Bucket限流
import asyncio
from collections import deque
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
            
            # 等待补充token
            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 sinceStart = now - this.startTime;
            const sinceLastChunk = now - this.lastChunkTime;
            
            if (sinceLastChunk > this.chunkThreshold) {
                // 长时间没新chunk，可能卡住了
                this.onWarning?.('no-chunk', {
                    duration: sinceLastChunk,
                    total: sinceStart
                });
            }
            
            if (sinceStart > this.ttftThreshold && this.lastChunkTime === this.startTime) {
                // TTFT超时
                this.onTimeout?.('ttft-timeout');
                this.stop();
            }
        }, 1000);
    }

    recordChunk() {
        this.lastChunkTime = Date.now();
    }

    stop() {
        if (this.intervalId) {
            clearInterval(this.intervalId);
            this.intervalId = null;
        }
    }
}

生产级架构设计

小打小闹的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注入攻击。

攻击原理：恶意用户构造特殊的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):
        """设置期望的完整内容hash"""
        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 secure_stream_chat(messages):
    full_content = ""
    
    # 先获取完整内容（不流式传输）
    complete_response = await llm.generate_complete(messages)
    
    # 生成校验hash
    validator = SecureStreamValidator(SECRET_KEY)
    validator.set_expected_hash(complete_response)
    
    # 然后才流式输出
    async for token in llm.generate_stream(messages):
        full_content += token
        yield token
    
    # 流结束后校验
    if not validator.validate(full_content):
        raise SecurityError("内容完整性校验失败，可能遭到注入攻击")

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

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[系统提示：此回复已触发内容安全审核，仅供参考]
"

总结

Streaming是AI应用体验的标配，不是可选项。核心要点：

• AI模型逐token生成，Streaming让用户感知延迟降低80%
• 首选SSE方案，WebSocket用于需要双向通信的复杂场景
• Nginx必须关闭buffer、关闭gzip
• 前端用requestAnimationFrame实现流畅打字效果
• 处理背压：输出太快就限流，输出太慢就监控超时
• 注意Token注入安全风险和内容审核

如果你正在做AI产品的流式输出，希望这篇文章能帮你少踩坑。实际开发中遇到具体问题，欢迎来TokenNexus交流，我们收录了全球330+ AI API平台，有最新的价格、延迟和稳定性数据。