AI API流式响应（SSE）实战指南：从原理到前后端完整实现

from openai import OpenAI

client = OpenAI(api_key="sk-your-key")

# stream=True 就是开启流式响应的关键
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "解释一下量子计算的基本原理"}],
    stream=True
)

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

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]

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-your-key")

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "用简单的语言解释相对论"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

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\n');
        buffer = lines.pop(); // 最后一段可能不完整，留着下次拼

        for (const line of lines) {
            const dataLines = line
                .split('\n')
                .filter(l => l.startsWith('data: '));

            for (const dataLine of dataLines) {
                const data = dataLine.slice(6);

                if (data === '[DONE]') {
                    console.log('流结束');
                    return;
                }

                try {
                    const parsed = JSON.parse(data);
                    const content = parsed.choices?.[0]?.delta?.content;
                    if (content) {
                        // 把内容追加到页面上
                        appendToChat(content);
                    }
                } catch (e) {
                    // JSON解析失败，可能是不完整的数据，忽略
                }
            }
        }
    }
}

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

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI(api_key="sk-your-key")

async def generate_stream(user_message: str):
    """生成SSE流"""
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}],
        stream=True
    )

    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            data = json.dumps({
                "choices": [{
                    "delta": {"content": chunk.choices[0].delta.content}
                }]
            }, ensure_ascii=False)
            # SSE格式：data: xxx\n\n
            yield f"data: {data}\n\n"

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

@app.post("/api/chat")
async def chat(request: dict):
    return StreamingResponse(
        generate_stream(request.get("message", "")),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 禁止Nginx缓冲
        }
    )

async def generate_claude_stream(user_message: str):
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    ) as stream:
        for event in stream:
            if event.type == "content_block_delta":
                text = event.delta.text
                data = json.dumps(
                    {"content": text},
                    ensure_ascii=False
                )
                yield f"data: {data}\n\n"

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

location /api/ {
    proxy_pass http://127.0.0.1:8000;
    proxy_buffering off;           # 关闭缓冲
    proxy_cache off;               # 关闭缓存
    proxy_set_header Connection '';  # 清除Connection头
    chunked_transfer_encoding on;  # 启用分块传输
}

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"