---
title: "Streaming LLM Responses with Python: Complete API Guide 2026"
description: "用 Python 实现 LLM 流式响应的完整指南，包含可直接用于生产环境的代码。"
keyword: "streaming llm api python tutorial 2026"
date: 2026-02-01
---

Streaming LLM Responses with Python：完整 API 指南 2026

3 个关键数字：

• 首字节延迟（TTFT）：流式响应可将用户感知延迟从平均 8-12 秒降低至 < 400ms（取决于模型和网络）
• 成本：流式传输本身不改变 token 计费，但可通过提前中断节省 20-40% 的无效 token 消耗
• 实现复杂度：从零开始实现基础流式响应只需约 15 行 Python 代码

---

目录

1. 前置条件
2. 认证与基础配置
3. 基础实现：OpenAI 兼容接口
4. 进阶实现：BentoML Generator 方案
5. 生产级实现：AWS Lambda + Amazon Bedrock
6. 错误处理
7. API 参数参考表
8. 性能与成本对比
9. 何时不应使用流式响应
10. 结论

---

1. 前置条件 {#prerequisites}

账号要求

• OpenAI API Key（用于基础示例）：platform.openai.com
• AWS 账号（用于 Serverless 示例）：需要开通 Amazon Bedrock 权限
• Python 3.10+（asyncio 的 async for 语法需要此版本）

安装依赖

# 核心依赖
pip install openai>=1.12.0 httpx>=0.27.0

# BentoML 方案（可选）
pip install bentoml>=1.2.0

# AWS 方案（可选）
pip install boto3>=1.34.0 awscli>=2.15.0

# 工具库
pip install python-dotenv>=1.0.0 tiktoken>=0.6.0

环境变量配置

创建 .env 文件：

OPENAI_API_KEY=sk-your-key-here
AWS_ACCESS_KEY_ID=your-aws-key
AWS_SECRET_ACCESS_KEY=your-aws-secret
AWS_DEFAULT_REGION=us-east-1
BEDROCK_MODEL_ID=anthropic.claude-3-5-sonnet-20241022-v2:0

---

2. 认证与基础配置 {#auth-setup}

在开始写任何流式逻辑之前，先把客户端初始化做对。一个常见错误是在每次请求时重新初始化 OpenAI() 客户端——这会导致连接池被反复重建，增加约 50-100ms 的额外延迟。

# config.py
# 把客户端初始化放在模块级别，利用连接池复用

import os
from openai import OpenAI, AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

# 同步客户端：用于脚本和简单服务
sync_client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    # timeout 设置针对流式请求要更宽松
    # 非流式请求通常设 30s，流式请求要等模型全部生成完
    timeout=120.0,
    max_retries=2,  # 内置重试，针对 529 和 503 错误
)

# 异步客户端：用于 FastAPI / asyncio 环境
async_client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=120.0,
    max_retries=2,
)

# 用于测试连接是否正常
def verify_connection() -> bool:
    try:
        models = sync_client.models.list()
        print(f"[OK] Connected. Available models: {len(list(models.data))}")
        return True
    except Exception as e:
        print(f"[FAIL] Connection error: {e}")
        return False

if __name__ == "__main__":
    verify_connection()

---

3. 基础实现：OpenAI 兼容接口 {#basic-implementation}

3.1 最简流式实现（同步）

下面这段代码展示了流式响应的核心机制：stream=True 让 API 返回一个迭代器，每次 yield 一个 ChatCompletionChunk 对象。

# basic_stream.py
# 这是最简单的可运行流式示例

from config import sync_client

def stream_completion(prompt: str, model: str = "gpt-4o") -> None:
    """
    直接打印流式输出到终端。
    适合：CLI 工具、脚本调试
    不适合：需要收集完整响应的场景
    """
    
    # stream=True 是关键参数
    # 它改变了 API 的返回类型：从 ChatCompletion 变为 Stream[ChatCompletionChunk]
    with sync_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=1024,
        temperature=0.7,
    ) as stream:
        for chunk in stream:
            # chunk.choices[0].delta.content 可能是 None（第一个和最后一个 chunk）
            content = chunk.choices[0].delta.content
            if content is not None:
                # end="" 避免换行，flush=True 强制立即输出到终端
                print(content, end="", flush=True)
        
        # 流结束后打印换行
        print()

if __name__ == "__main__":
    stream_completion("用三句话解释量子纠缠")

3.2 收集完整响应（同步）

实际生产中，通常需要记录完整的响应内容用于日志或后续处理：

# collect_stream.py
# 同时流式显示 + 收集完整文本

import time
from dataclasses import dataclass
from config import sync_client

@dataclass
class StreamResult:
    full_text: str
    chunk_count: int
    time_to_first_token: float  # 秒
    total_time: float           # 秒

def stream_and_collect(
    prompt: str,
    model: str = "gpt-4o",
    print_live: bool = True,
) -> StreamResult:
    """
    流式显示的同时，收集完整响应和性能指标。
    time_to_first_token 是衡量用户体验最重要的指标。
    """
    
    chunks = []
    start_time = time.perf_counter()
    first_token_time = None
    
    with sync_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=2048,
        temperature=0.7,
    ) as stream:
        for chunk in stream:
            content = chunk.choices[0].delta.content
            if content is not None:
                # 记录第一个 token 的时间戳
                if first_token_time is None:
                    first_token_time = time.perf_counter()
                
                chunks.append(content)
                
                if print_live:
                    print(content, end="", flush=True)
    
    if print_live:
        print()
    
    end_time = time.perf_counter()
    
    return StreamResult(
        full_text="".join(chunks),
        chunk_count=len(chunks),
        time_to_first_token=first_token_time - start_time if first_token_time else 0,
        total_time=end_time - start_time,
    )

if __name__ == "__main__":
    result = stream_and_collect("写一首关于 Python 的俳句")
    print(f"\n--- 性能指标 ---")
    print(f"首字节延迟 (TTFT): {result.time_to_first_token:.3f}s")
    print(f"总耗时: {result.total_time:.3f}s")
    print(f"Chunk 数量: {result.chunk_count}")
    print(f"完整文本长度: {len(result.full_text)} 字符")

3.3 异步流式实现（AsyncIO）

在 FastAPI 或其他异步框架中，必须使用异步版本，否则会阻塞事件循环：

# async_stream.py
# 用于 FastAPI / aiohttp 等异步环境

import asyncio
from typing import AsyncGenerator
from config import async_client

async def async_stream_generator(
    prompt: str,
    model: str = "gpt-4o",
    max_tokens: int = 1024,
) -> AsyncGenerator[str, None]:
    """
    返回异步生成器，适合作为 FastAPI StreamingResponse 的数据源。
    
    使用方式：
        async for chunk in async_stream_generator("你好"):
            print(chunk, end="", flush=True)
    """
    
    async with async_client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        max_tokens=max_tokens,
    ) as stream:
        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content is not None:
                yield content

# FastAPI 集成示例（需要 pip install fastapi uvicorn）
# from fastapi import FastAPI
# from fastapi.responses import StreamingResponse
#
# app = FastAPI()
#
# @app.post("/stream")
# async def stream_endpoint(prompt: str):
#     # StreamingResponse 会把每个 yield 的 chunk 立即发送给客户端
#     return StreamingResponse(
#         async_stream_generator(prompt),
#         media_type="text/plain",
#     )

async def main():
    print("开始流式输出：")
    async for chunk in async_stream_generator("解释 async/await 的工作原理"):
        print(chunk, end="", flush=True)
    print()

if __name__ == "__main__":
    asyncio.run(main())

---

4. 进阶实现：BentoML Generator 方案 {#bentoml-implementation}

BentoML 官方文档 支持通过 Python generator 实现 LLM 输出流式传输，适合需要自托管模型的场景。

BentoML 的流式方案优势在于：可以将任意 Python 函数（包括本地 Transformers 模型）包装成流式 HTTP 端点，而不依赖 OpenAI 格式的 API。

# bento_service.py
# BentoML >= 1.2.0 的流式服务示例

import bentoml
from typing import Generator

# 这个示例使用 OpenAI 客户端作为后端，但同样的模式适用于
# 任何本地 Transformers 模型（替换 generate_from_model 函数即可）
@bentoml.service(
    resources={"cpu": "2"},
    traffic={"timeout": 120},
)
class LLMStreamingService:
    
    def __init__(self):
        # 在实际的本地模型场景中，这里加载 tokenizer 和 model
        # self.tokenizer = AutoTokenizer.from_pretrained("...")
        # self.model = AutoModelForCausalLM.from_pretrained("...")
        
        # 此示例用 OpenAI 客户端演示 Generator 接口
        import os
        from openai import OpenAI
        self.client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    
    @bentoml.api
    def generate(self, prompt: str) -> Generator[str, None, None]:
        """
        返回 Generator[str, None, None] 告诉 BentoML 这是流式端点。
        BentoML 会自动将每个 yield 的字符串作为 HTTP chunk 发送。
        """
        
        with self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
            max_tokens=1024,
        ) as stream:
            for chunk in stream:
                content = chunk.choices[0].delta.content
                if content is not None:
                    # 每次 yield 会立即通过 HTTP 发送给调用方
                    yield content

# 启动命令：
# bentoml serve bento_service:LLMStreamingService --port 3000
#
# 调用示例（curl）：
# curl -X POST http://localhost:3000/generate \
#   -H "Content-Type: application/json" \
#   -d '{"prompt": "解释机器学习"}' \
#   --no-buffer

---

5. 生产级实现：AWS Lambda + Amazon Bedrock {#aws-lambda-implementation}

基于 AWS Compute Blog 的 Serverless 流式方案，适合需要弹性扩展且不想维护服务器的场景。

架构：客户端 → API Gateway (WebSocket) → Lambda → Bedrock

# lambda_handler.py
# 部署到 AWS Lambda，通过 API Gateway WebSocket 实现流式响应
# 依赖：boto3（Lambda 环境内置，无需额外安装）

import json
import boto3
import os

# Bedrock 客户端 - Lambda 执行环境中复用（在 handler 外初始化）
bedrock_client = boto3.client(
    "bedrock-runtime",
    region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
)

def lambda_handler(event, context):
    """
    AWS Lambda handler for WebSocket streaming.
    
    event 结构来自 API Gateway WebSocket：
    {
        "requestContext": {"connectionId": "...", "domainName": "...", "stage": "..."},
        "body": '{"prompt": "你好"}'
    }
    """
    
    # 从 event 中提取 WebSocket 连接信息
    connection_id = event["requestContext"]["connectionId"]
    domain = event["requestContext"]["domainName"]
    stage = event["requestContext"]["stage"]
    
    # 解析请求体获取 prompt
    body = json.loads(event.get("body", "{}"))
    prompt = body.get("prompt", "")
    
    if not prompt:
        return {"statusCode": 400, "body": "Missing prompt"}
    
    # API Gateway Management API 客户端，用于向 WebSocket 客户端推送消息
    # endpoint_url 格式是固定的
    apigw_client = boto3.client(
        "apigatewaymanagementapi",
        endpoint_url=f"https://{domain}/{stage}",
    )
    
    # Bedrock invoke_model_with_response_stream 返回流式响应
    model_id = os.environ.get("BEDROCK_MODEL_ID", "anthropic.claude-3-5-sonnet-20241022-v2:0")
    
    # Anthropic Claude 的请求格式（Messages API）
    request_body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "messages": [
            {"role": "user", "content": prompt}
        ],
    })
    
    try:
        response = bedrock_client.invoke_model_with_response_stream(
            modelId=model_id,
            body=request_body,
            contentType="application/json",
        )
        
        # 遍历流式 chunk，逐个推送给 WebSocket 客户端
        for event_chunk in response["body"]:
            chunk_data = json.loads(event_chunk["chunk"]["bytes"])
            
            # Claude 的流式响应格式：delta.type == "content_block_delta"
            if chunk_data.get("type") == "content_block_delta":
                text = chunk_data.get("delta", {}).get("text", "")
                if text:
                    # 通过 WebSocket 向客户端推送 chunk
                    apigw_client.post_to_connection(
                        ConnectionId=connection_id,
                        Data=json.dumps({"type": "chunk", "text": text}).encode("utf-8"),
                    )
        
        # 发送结束信号
        apigw_client.post_to_connection(
            ConnectionId=connection_id,
            Data=json.dumps({"type": "done"}).encode("utf-8"),
        )
        
    except bedrock_client.exceptions.ModelNotReadyException as e:
        # Bedrock 特有错误：模型还在预热
        apigw_client.post_to_connection(
            ConnectionId=connection_id,
            Data=json.dumps({"type": "error", "code": "MODEL_NOT_READY", "message": str(e)}).encode("utf-8"),
        )
        return {"statusCode": 503, "body": "Model not ready"}
    
    return {"statusCode": 200, "body": "Stream complete"}

---

6. 错误处理 {#error-handling}

流式请求的错误处理比普通请求复杂，因为错误可能发生在流的中途——此时你已经发送了部分响应给客户端。

错误码 / 异常类型	触发场景	解决方案
openai.APIConnectionError	网络中断，TCP 连接被切断	指数退避重试，最多 3 次
openai.RateLimitError (429)	超过 TPM/RPM 限制	检查 retry-after header，等待后重试
openai.APIStatusError (500/503)	OpenAI 服务端问题	退避重试；记录 request_id 用于排查
openai.APITimeoutError	超过 timeout 设置	对长文本生成适当增大 timeout
StopIteration / 空 chunk	流正常结束，但代码未处理 None content	检查 if content is not None
Bedrock ModelNotReadyException	Bedrock 模型冷启动	等待 10-30s 后重试，或使用预置吞吐量
Bedrock ThrottlingException	超过 Bedrock 并发限制	减小并发数，申请更高配额

# error_handling.py
# 生产级错误处理包装器

import time
import openai
from typing import Generator, Optional
from config import sync_client

def robust_stream(
    prompt: str,
    model: str = "gpt-4o",
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> Generator[str, None, None]:
    """
    带指数退避重试的流式生成器。
    
    注意：重试逻辑在流开始之前生效。
    一旦开始 yield chunk，中途断流需要在调用方处理。
    """
    
    last_exception: Optional[Exception] = None
    
    for attempt in range(max_retries):
        try:
            with sync_client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                stream=True,
                max_tokens=1024,
                timeout=120.0,
            ) as stream:
                for chunk in stream:
                    content = chunk.choices[0].delta.content
                    if content is not None:
                        yield content
            
            # 正常完成，退出重试循环
            return
        
        except openai.RateLimitError as e:
            # 429 错误：需要等待
            # openai SDK 会在 headers 里告诉你要等多久
            retry_after = float(e.response.headers.get("retry-after", base_delay * (2 ** attempt)))
            print(f"[WARN] Rate limited. Waiting {retry_after:.1f}s before retry {attempt + 1}/{max_retries}")
            time.sleep(retry_after)
            last_exception = e
        
        except openai.APIConnectionError as e:
            # 网络问题：指数退避
            delay = base_delay * (2 ** attempt)
            print(f"[WARN] Connection error: {e}. Retrying in {delay:.1f}s ({attempt + 1}/{max_retries})")
            time.sleep(delay)
            last_exception = e
        
        except openai.APIStatusError as e:
            if e.status_code in (500, 502, 503, 529):
                # 服务端临时错误：可重试
                delay = base_delay * (2 ** attempt)
                print(f"[WARN] Server error {e.status_code}. Request ID: {e.request_id}. Retrying in {delay:.1f}s")
                time.sleep(delay)
                last_exception = e
            else:
                # 4xx 客户端错误：不重试（重试也没用）
                print(f"[ERROR] Client error {e.status_code}: {e.message}")
                raise


> **提示：** 如果你需要在同一个项目中使用多个 AI 模型，[AtlasCloud](https://www.atlascloud.ai?ref=JPM683) 提供统一 API 接入 300+ 模型（Kling、Flux、Seedance、Claude、GPT 等），一个 key 全部搞定。新用户首次充值享 25% 赠送（最高 $100）。