开始学习
课程 › 0基础Agent开发课 / API入门与模型调用 / 流式输出与SSE集成
— 11 min read

流式输出与SSE集成

等待 LLM 生成完整响应再显示，用户体验很差——特别是长文本需要等待 10-30 秒。流式输出（Streaming）让模型的回答逐 Token 出现，用户看到的是"打字机"效果，感知延迟从秒级降到毫秒级。

流式输出与 SSE 集成

等待 LLM 生成完整响应再显示，用户体验很差——特别是长文本需要等待 10-30 秒。流式输出（Streaming）让模型的回答逐 Token 出现，用户看到的是"打字机"效果，感知延迟从秒级降到毫秒级。

为什么需要这一篇：所有面向用户的 AI 应用，几乎都需要流式输出。等待完整响应再显示会让用户以为应用卡死。掌握流式输出，是现代 AI 应用开发的基本能力。

流式输出的工作原理

流式输出时序图
SSE 流式输出与非流式批量返回的时序对比

LLM 生成文本的方式是逐 Token 预测：预测第 1 个 Token，预测第 2 个 Token……每预测一个 Token 就可以立即发送，不需要等全部生成完。

这对应 HTTP（超文本传输协议，网页和 API 通信的基础协议）的 SSE（Server-Sent Events，服务器推送事件）协议：服务端持续推送数据，客户端实时接收，就像直播弹幕一样边生成边显示。

基础流式调用

python 
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

# 使用 stream=True 开启流式输出
stream = client.chat.completions.create(
    model="deepseek-chat",
    max_tokens=1024,
    messages=[{"role": "user", "content": "写一首关于编程的短诗"}],
    stream=True
)

for chunk in stream:
    text = chunk.choices[0].delta.content or ""
    print(text, end="", flush=True)  # flush=True 确保立即输出
print()  # 换行

flush=True 很重要：Python 的 print 默认有缓冲，不加 flush 可能看不到逐字输出效果。

获取流式响应的完整信息

如果需要在流结束后获取 token 用量等元数据，可以在最后一个 chunk 中读取：

python 
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

total_tokens = 0
collected_text = []

stream = client.chat.completions.create(
    model="deepseek-chat",
    max_tokens=1024,
    messages=[{"role": "user", "content": "你好"}],
    stream=True,
    stream_options={"include_usage": True}  # 开启后最后一个 chunk 会包含用量信息
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        text = chunk.choices[0].delta.content
        print(text, end="", flush=True)
        collected_text.append(text)
    # 最后一个 chunk 包含用量信息
    if chunk.usage:
        total_tokens = chunk.usage.total_tokens

print()
print(f"\n总 tokens: {total_tokens}")
full_response = "".join(collected_text)

异步流式调用

在 Web 服务中，通常需要异步处理（异步，即程序不需要等待某个操作完成就能继续处理其他任务，适合网络请求等需要等待的场景）：

python 
import asyncio
import os
from openai import AsyncOpenAI

async def stream_response(user_message: str):
    client = AsyncOpenAI(
        base_url="https://api.deepseek.com",
        api_key=os.environ.get("DEEPSEEK_API_KEY")
    )

    stream = await client.chat.completions.create(
        model="deepseek-chat",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
        stream=True
    )

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

asyncio.run(stream_response("解释一下什么是异步编程"))

FastAPI + SSE：在 Web 服务中实现流式输出

这是生产环境中最常见的用法：FastAPI（一个流行的 Python Web 框架，用来快速搭建后端服务）后端把 LLM 的流式输出通过 SSE 推送给前端。

python 
import os
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import json

app = FastAPI()
client = AsyncOpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

@app.post("/chat/stream")
async def chat_stream(request: dict):
    user_message = request.get("message", "")

    async def generate():
        stream = await client.chat.completions.create(
            model="deepseek-chat",
            max_tokens=1024,
            messages=[{"role": "user", "content": user_message}],
            stream=True
        )
        async for chunk in stream:
            text = chunk.choices[0].delta.content or ""
            if text:
                # SSE 格式：data: {json}\n\n
                yield f"data: {json.dumps({'text': text})}\n\n"

        # 发送结束信号
        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 禁用 Nginx（高性能 Web 服务器和反向代理）缓冲
        }
    )

前端接收 SSE

javascript 
// 原生 EventSource（只支持 GET）
const eventSource = new EventSource('/chat/stream');
eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.done) {
        eventSource.close();
    } else {
        document.getElementById('output').textContent += data.text;
    }
};

// fetch + ReadableStream（支持 POST，更灵活）
async function streamChat(message) {
    const response = await fetch('/chat/stream', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({message})
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

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

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = JSON.parse(line.slice(6));
                if (data.done) return;
                document.getElementById('output').textContent += data.text;
            }
        }
    }
}

流式输出的错误处理

流式输出中途可能出错，需要在生成器中处理：

python 
from openai import APIConnectionError, RateLimitError

async def generate_with_error_handling():
    try:
        stream = await client.chat.completions.create(
            model="deepseek-chat",
            max_tokens=1024,
            messages=[{"role": "user", "content": "你好"}],
            stream=True
        )
        async for chunk in stream:
            text = chunk.choices[0].delta.content or ""
            if text:
                yield f"data: {json.dumps({'text': text})}\n\n"
    except APIConnectionError:
        yield f"data: {json.dumps({'error': '网络连接失败'})}\n\n"
    except RateLimitError:
        yield f"data: {json.dumps({'error': '请求过于频繁，请稍后重试'})}\n\n"
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e)})}\n\n"
    finally:
        yield f"data: {json.dumps({'done': True})}\n\n"

流式输出与非流式的选择

场景 推荐方式
面向用户的对话界面 流式（更好的体验）
批量处理、后台任务 非流式（代码更简单）
需要完整响应后再处理（如解析 JSON） 非流式
长文本生成（文章、报告） 流式
工具调用（Function Calling，让模型调用外部程序的能力） 非流式（工具调用结果需要完整响应）

收集流式输出为完整文本

有时需要流式显示，但也需要完整文本做后续处理：

python 
collected_text = []

stream = client.chat.completions.create(
    model="deepseek-chat",
    max_tokens=1024,
    messages=[{"role": "user", "content": "你好"}],
    stream=True
)

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

full_response = "".join(collected_text)

流式输出是现代 AI 应用用户体验的基础。第 18 章的生产化部署会进一步讲解 SSE 的生产级实现，包括 Nginx 配置、超时处理、连接管理等工程细节。

常见错误和解决方法

错误现象 原因 解决方法
输出没有逐字出现，等了很久才一次性显示 没有加 flush=True，或者有中间缓冲层 print(text, end="", flush=True)
流式输出中途断开 网络超时或服务器端断开 添加超时重试逻辑
前端接收到数据但不显示 SSE 格式不对（缺少 data: 前缀或 \n\n 结尾） 确认格式：f"data: {json}\n\n"
工具调用时流式输出失效 Function Calling 需要完整响应才能解析 工具调用场景不使用流式输出

小结

场景 推荐方式 原因
面向用户的对话界面 流式 用户体验好，感知延迟低
批量处理、后台任务 非流式 代码更简单，不需要实时显示
需要解析完整 JSON 非流式 流式输出不能保证 JSON 在中间完整
工具调用（Function Calling） 非流式 工具调用需要完整响应后解析
←上一节多轮对话的消息历史管理下一节结构化输出与预填充技巧→
本页目录