Python 实现 LLM 流式输出终极指南｜从原理到生产级代码全解析

一、什么是 LLM 流式输出？

1.1 传统模式 vs 流式模式

对比项	传统模式	流式模式
响应方式	等待全部生成完毕，一次性返回	边生成边返回，逐 token 推送
首字节时间 (TTFB)	5-15 秒（取决于内容长度）	0.2-0.5 秒
用户体验	❌ 长时间空白等待，感觉”卡死”	✅ 即时反馈，类似真人打字
技术实现	普通 HTTP 请求	SSE / WebSocket

1.2 SSE (Server-Sent Events) 协议解析

SSE 是一种基于 HTTP 的服务端推送技术，特点：

• 单向通信：服务端 → 客户端
• 基于文本：每条消息以 data: 开头，以 \n\n 结尾
• 自动重连：浏览器原生支持断线重连
• 轻量级：相比 WebSocket 更简单，无需握手升级

1.3 技术选型矩阵

方案	适用场景	复杂度	性能
OpenAI SDK 原生	快速原型、脚本工具	⭐	中
LangChain Streaming	链式调用、Agent	⭐⭐	中
FastAPI + SSE	Web API、生产部署	⭐⭐⭐	高
WebSocket	双向通信、实时协作	⭐⭐⭐⭐	最高

---

二、Python 实现 LLM 流式输出只需3步｜小白10分钟上手

Step 1: 环境准备 (2分钟)

pip install openai python-dotenv

创建 .env 文件配置 API 密钥：

# .env 文件
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://api.suxi.ai/v1  # 推荐使用速夕API中转，国内直连

Step 2: 最简流式代码 (3分钟)

from openai import OpenAI
from dotenv import load_dotenv
import os
load_dotenv()
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")  # 使用 api.suxi.ai 中转
)
# 🔥 核心：stream=True
response = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[{"role": "user", "content": "用Python写一个快速排序"}],
    stream=True  # 开启流式输出
)
# 逐 token 打印
for chunk in response:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

关键点解析：

• stream=True：告诉 API 以流式方式返回
• chunk.choices[0].delta.content：每个 chunk 只包含增量内容
• flush=True：强制刷新缓冲区，否则内容会等到换行才显示

Step 3: 完整响应收集 (5分钟)

实际项目中，我们通常需要边输出边收集完整响应：

def stream_chat(prompt: str) -> str:
    """流式输出并收集完整响应"""
    full_response = ""
    
    response = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    for chunk in response:
        content = chunk.choices[0].delta.content or ""
        print(content, end="", flush=True)  # 实时输出到控制台
        full_response += content  # 同时收集完整响应
    
    print()  # 换行
    return full_response
# 使用示例
result = stream_chat("解释什么是递归")
print(f"\n\n完整响应长度: {len(result)} 字符")

---

三、LangChain + FastAPI 企业级部署方案

3.1 LangChain 流式输出方案

如果你的项目使用 LangChain 框架，流式输出更加简单：

from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
llm = ChatOpenAI(
    model="gpt-4-turbo",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    openai_api_base="https://api.suxi.ai/v1"  # 速夕API中转
)
# 自动流式输出到控制台
response = llm.invoke("解释什么是微服务架构")

3.2 FastAPI + SSE 生产级架构

完整 FastAPI 服务端代码：

from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from openai import OpenAI
import os
app = FastAPI(title="LLM 流式输出 API")
# 允许跨域（开发环境）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://api.suxi.ai/v1"  # 速夕API中转
)
async def generate_stream(prompt: str):
    """异步生成器 - SSE 格式"""
    response = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    for chunk in response:
        content = chunk.choices[0].delta.content or ""
        if content:
            # SSE 格式：data: 内容\n\n
            yield f"data: {content}\n\n"
    
    yield "data: [DONE]\n\n"
@app.get("/chat/stream")
async def chat_stream(prompt: str = Query(..., description="用户输入的问题")):
    """流式聊天接口"""
    return StreamingResponse(
        generate_stream(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )
# 启动命令: uvicorn main:app --reload --host 0.0.0.0 --port 8000

前端调用示例 (JavaScript)：

const prompt = "用Python写一个快速排序";
const eventSource = new EventSource(`/chat/stream?prompt=${encodeURIComponent(prompt)}`);
const outputDiv = document.getElementById('output');
eventSource.onmessage = (event) => {
    if (event.data === '[DONE]') {
        eventSource.close();
        console.log('流式输出完成');
        return;
    }
    // 实时追加内容
    outputDiv.textContent += event.data;
};
eventSource.onerror = (error) => {
    console.error('SSE 连接错误:', error);
    eventSource.close();
};

3.3 高并发优化配置

配置项	推荐值	说明
uvicorn workers	CPU核心数 × 2	多进程并发处理
httpx timeout	120s	长文本生成需要更长超时
connection pool	100	连接池大小，避免频繁建连
max_retries	3	失败自动重试

# 生产环境启动（4个worker进程）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# 或使用 gunicorn + uvicorn worker
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

---

四、90%开发者踩过的5个性能陷阱

❌ 陷阱 1: 缓冲区阻塞 (最常见!)

❌ 错误写法	✅ 正确写法
print(content)	print(content, end="", flush=True)

原因：Python 默认使用行缓冲，不加 flush=True 会等到遇到换行符才输出。结果就是你以为没生效，其实内容在缓冲区里！

❌ 陷阱 2: 中文 Token 拼接乱码

# ❌ 错误：直接拼接可能导致 UTF-8 截断
full_text += chunk.content  # 可能出现乱码
# ✅ 正确：使用 buffer 处理不完整的 UTF-8 序列
buffer = b""
for chunk in response:
    if chunk.choices[0].delta.content:
        buffer += chunk.choices[0].delta.content.encode('utf-8')
        try:
            text = buffer.decode('utf-8')
            print(text, end="", flush=True)
            buffer = b""
        except UnicodeDecodeError:
            continue  # 等待下一个 chunk 补全

❌ 陷阱 3: 连接超时未处理

import httpx
# ❌ 错误：无超时设置，可能永久阻塞
client = OpenAI()
# ✅ 正确：设置合理超时
client = OpenAI(
    timeout=httpx.Timeout(
        60.0,        # 总超时 60 秒
        connect=10.0  # 连接超时 10 秒
    )
)

❌ 陷阱 4: 流未正确关闭导致内存泄漏

# ✅ 正确：使用 with 语句确保资源释放
with client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[{"role": "user", "content": prompt}],
    stream=True
) as response:
    for chunk in response:
        process(chunk)
# 自动关闭连接，避免内存泄漏

❌ 陷阱 5: 前端未处理 SSE 断连重连

// ✅ 正确：前端必须处理重连逻辑
let eventSource;
function connect(prompt) {
    eventSource = new EventSource(`/chat/stream?prompt=${encodeURIComponent(prompt)}`);
    
    eventSource.onmessage = (event) => {
        if (event.data === '[DONE]') {
            eventSource.close();
            return;
        }
        document.getElementById('output').textContent += event.data;
    };
    
    eventSource.onerror = () => {
        console.log('连接断开，3秒后重连...');
        eventSource.close();
        setTimeout(() => connect(prompt), 3000);  // 3秒后重连
    };
}

---

五、常见问题 FAQ

---

六、总结与下一步

核心要点回顾

---

 

参考资料

• OpenAI Streaming 官方文档
• MDN Server-Sent Events
• FastAPI StreamingResponse

---

📅 最后更新：2025年1月15日 | 版本 v2.0
本次更新：新增 LangChain 0.2 适配、GPT-4 Turbo 流式优化