为什么必须流式

AI 生成一个完整回答可能需要 5-30 秒。如果等全部生成完再返回，用户体验极差。流式输出让用户看到"AI 正在打字"，将感知等待时间从 15 秒降到 0.5 秒。

SSE vs WebSocket

维度SSEWebSocket
协议HTTP独立协议
方向服务端→客户端（单向）双向
重连自动需手动实现
兼容性所有现代浏览器所有现代浏览器
复杂度低中
适用场景AI 回复流式输出实时协作、语音对话

结论：AI 对话场景用 SSE 足够，语音/多模态交互用 WebSocket。

SSE 实现

服务端（Python FastAPI）

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

app = FastAPI()

@app.post("/api/chat")
async def chat(request: dict):
    async def stream():
        # 调用 LLM 流式 API
        async for chunk in llm.stream(
            model="gpt-4o",
            messages=request["messages"]
        ):
            # SSE 格式：data: <content>\n\n
            content = chunk.choices[0].delta.content
            if content:
                yield f"data: {json.dumps({'content': content})}\n\n"
        
        # 发送结束标记
        yield f"data: {json.dumps({'done': True})}\n\n"
    
    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx 不缓冲
        }
    )

客户端（JavaScript）

class AIChat {
    async sendMessage(text) {
        const response = await fetch('/api/chat', {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                messages: [...this.history, { role: 'user', content: text }]
            })
        });
        
        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 数据
            const lines = buffer.split('\n\n');
            buffer = lines.pop(); // 保留不完整的行
            
            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    const data = JSON.parse(line.slice(6));
                    if (data.done) {
                        this.onComplete();
                    } else {
                        this.onToken(data.content);
                    }
                }
            }
        }
    }
    
    onToken(content) {
        document.getElementById('response').textContent += content;
    }
}

进阶：流式 + 工具调用

当 Agent 需要调用工具时，流式输出变得复杂：

async def agent_stream(query):
    # 阶段1：告知用户正在思考
    yield format_sse({"type": "status", "message": "正在思考..."})
    
    # 阶段2：LLM 决定调用工具
    tool_calls = await llm.plan(query)
    if tool_calls:
        for tc in tool_calls:
            yield format_sse({
                "type": "tool_call",
                "tool": tc.name,
                "args": tc.args
            })
        
        # 执行工具
        results = await execute_tools(tool_calls)
        yield format_sse({"type": "tool_result", "results": results})
    
    # 阶段3：基于工具结果流式输出最终回答
    async for chunk in llm.stream(final_prompt):
        yield format_sse({"type": "content", "content": chunk})
    
    yield format_sse({"type": "done"})

客户端根据事件类型渲染不同 UI：

eventSource.onmessage = (e) => {
    const data = JSON.parse(e.data);
    switch(data.type) {
        case 'status':
            showThinking(data.message);
            break;
        case 'tool_call':
            showToolBadge(data.tool);
            break;
        case 'content':
            appendContent(data.content);
            break;
        case 'done':
            finishResponse();
            break;
    }
};

体验优化技巧

1. 首字节时间（TTFB）优化

# 立即发送一个空格，让用户看到"开始回复了"
yield f"data: {json.dumps({'content': ' '})}\n\n"

# 然后再等 LLM 首个 token
async for chunk in llm.stream(prompt):
    yield format_sse({"content": chunk})

2. Markdown 流式渲染

// 边接收边渲染 Markdown
import { marked } from 'marked';

let rawContent = '';

function appendContent(chunk) {
    rawContent += chunk;
    // 每次增量渲染
    const html = marked.parse(rawContent);
    document.getElementById('response').innerHTML = html;
    
    // 代码块特殊处理：高亮
    document.querySelectorAll('pre code').forEach(el => {
        hljs.highlightElement(el);
    });
}

3. 打字机效果

// 平滑的打字机效果
const typeSpeed = 10; // ms per char

async function typewriter(text) {
    for (const char of text) {
        appendChar(char);
        await sleep(typeSpeed);
    }
}

4. 断线重连

class ResilientSSE {
    constructor(url) {
        this.url = url;
        this.lastEventId = null;
        this.connect();
    }
    
    connect() {
        this.eventSource = new EventSource(
            `${this.url}?last_id=${this.lastEventId}`
        );
        
        this.eventSource.onmessage = (e) => {
            this.lastEventId = e.lastEventId;
            this.handleMessage(e.data);
        };
        
        this.eventSource.onerror = () => {
            this.eventSource.close();
            setTimeout(() => this.connect(), 3000); // 3秒后重连
        };
    }
}

Nginx 配置

location /api/chat {
    proxy_pass http://backend;
    
    # 关键：关闭缓冲
    proxy_buffering off;
    proxy_cache off;
    
    # SSE 需要长连接
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
    
    # 超时设置
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
}

性能对比

方案首字节延迟内存/连接实现复杂度
非流式5-15s低低
SSE0.3-0.8s中中
WebSocket0.3-0.8s高高
SSE + 缓存0.1s中中高

结论

流式输出是 AI 应用的标配体验。选型建议：

  • 标准 AI 对话：SSE，简单可靠
  • 多模态交互：WebSocket，支持双向通信
  • 高并发场景：SSE + Nginx，水平扩展
  • 离线消息：WebSocket + 消息队列

关键实现要点：关闭所有层级的缓冲、做好断线重连、流式渲染 Markdown、控制首字节延迟在 1 秒内。

加入讨论

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

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