第2章流式响应¶

学习目标¶

理解流式响应（streaming）的作用与底层 SSE 协议
掌握 OpenAI SDK 中 stream=True 的同步与异步用法
能够解析 Responses API 流式事件（event stream）
学会在 Web 服务中实现非阻塞的实时聊天
识别流式处理中的常见陷阱并应用最佳实践

问题导入¶

想象你在构建一个面向 C 端的 AI 写作助手。用户点击"生成"后，如果让屏幕静止 8 秒再一次性吐出 2000 字，体验灾难性地糟糕——用户会以为应用卡死。人类对延迟的容忍度与"是否看到进度"强相关：首字延迟控制在 1 秒内、后续字逐个流出，主观等待感会大幅下降。这正是流式响应存在的意义。

OpenAI 的流式响应基于 Server-Sent Events（SSE） 协议，服务端在生成 token 的同时持续推送增量片段，客户端边接收边渲染。对 Agent 来说，流式不仅能改善 UX，还能让你更早地触发后续逻辑——例如检测到模型开始输出工具调用 JSON 时立即预热下游服务。本章将带你从协议层到工程实现完整理解流式响应。

核心概念¶

1. SSE 协议¶

Server-Sent Events 是 HTTP 长连接上的一种单向推送协议，响应头为 Content-Type: text/event-stream。每个事件以 data: 开头，事件之间用空行分隔。OpenAI 在每个 chunk 中推送 JSON，包含增量文本、usage 统计、结束信号等。SSE 相比 WebSocket 更轻量，适合"服务端推、客户端收"的单向场景。

2. 流式事件的类型¶

Responses API 在流式模式下会推送多种事件类型，常见的有：

response.output_text.delta：增量文本片段，最常消费的事件。
response.output_text.done：单条 message 的文本结束。
response.function_call_arguments.delta：工具调用参数的增量。
response.completed：整个响应结束，携带完整 response 对象。
response.error：流中出现的错误。

通过事件类型可以精确控制渲染逻辑，例如只在文本事件时更新 UI，在工具调用事件时执行副作用。

3. 同步流与异步流¶

SDK 提供两种消费方式：

同步流：with client.responses.stream(...) as stream，使用 for event in stream 迭代，适合脚本与简单服务。
异步流：async with await client.responses.create(..., stream=True) as stream，配合 async for，适合 FastAPI、aiohttp 等异步框架，能在等待 token 期间不阻塞事件循环。

4. 非阻塞对话的本质¶

流式并不改变模型生成速度，而是把"等待 + 一次性返回"变成"边生成边返回"。在 Web 服务中，真正的非阻塞需要：① 使用异步 SDK 避免线程阻塞；② 把 token 通过异步队列或 SSE 转发给前端；③ 在前端用 fetch 的 ReadableStream 或 EventSource 接收。

5. 增量拼接与状态管理¶

流式片段需要按顺序拼接成完整文本。SDK 的 stream.on("output_text.delta", ...) 回调或 Stream 对象的 .text 属性可自动累积。但自定义逻辑（如按句子分段、敏感词过滤）需要自己维护一个缓冲区，注意处理跨 chunk 的边界。

原理图¶

sequenceDiagram
    participant Client as 客户端
    participant Server as API 服务端
    participant Model as 模型

    Client->>Server: POST /v1/responses (stream=true)
    Server-->>Client: HTTP 200, Content-Type: text/event-stream
    loop 逐 token 生成
        Model->>Server: 产出 token
        Server-->>Client: data: {"type":"response.output_text.delta","delta":"你"}
    end
    Server-->>Client: data: {"type":"response.completed"}
    Note over Client: 客户端拼接 delta<br/>渲染并执行副作用

图：SSE 流式响应的 token 推送过程

示例代码¶

import os
import asyncio
from openai import OpenAI, AsyncOpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
async_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))


# 1. 同步流：最简逐字打印
def stream_sync(prompt: str) -> str:
    full_text = ""
    with client.responses.stream(
        model="gpt-4.1",
        input=prompt,
    ) as stream:
        for event in stream:
            # 只关心文本增量
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
                full_text += event.delta
        print()  # 换行
    return full_text


# 2. 高级用法：使用事件回调
def stream_with_callbacks(prompt: str) -> str:
    collected = []

    def on_text(event):
        collected.append(event.delta)

    def on_completed(event):
        print(f"\n[完成，共 {len(event.response.output_text)} 字]")

    with client.responses.stream(
        model="gpt-4.1",
        input=prompt,
    ) as stream:
        stream.on("output_text.delta", on_text)
        stream.on("response.completed", on_completed)
        stream.until_done()

    return "".join(collected)


# 3. 异步流：适配 FastAPI / aiohttp
async def stream_async(prompt: str) -> str:
    full_text = ""
    async with await async_client.responses.create(
        model="gpt-4.1",
        input=prompt,
        stream=True,
    ) as stream:
        async for event in stream:
            if event.type == "response.output_text.delta":
                full_text += event.delta
                # 此处可改为 yield event.delta 转发给前端
    return full_text


# 4. 在 FastAPI 中转发 SSE（伪代码示例）
"""
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
async def chat(q: str):
    async def gen():
        async with await async_client.responses.create(
            model="gpt-4.1", input=q, stream=True
        ) as stream:
            async for event in stream:
                if event.type == "response.output_text.delta":
                    yield f"data: {event.delta}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(gen(), media_type="text/event-stream")
"""


if __name__ == "__main__":
    print("=== 同步流 ===")
    stream_sync("用 50 字介绍 SSE 协议。")

    print("\n=== 异步流 ===")
    result = asyncio.run(stream_async("用 50 字介绍流式响应的优势。"))
    print(result)

企业案例¶

案例 1：实时 AI 客服的"打字机"体验¶

某电商平台在客服窗口接入流式响应，首字延迟从原来的 4.2 秒降至 0.9 秒，用户中途放弃率下降 18%。关键技术点：前端用 EventSource 监听 SSE，后端用 FastAPI 异步流转发，并在 function_call_arguments.delta 事件触发时预热订单查询服务。

案例 2：长文档生成的进度反馈¶

写作类应用在生成长篇内容时，利用流式事件实时更新"已生成字数 / 预估总字数"进度条，并在每段结束时自动保存草稿，避免因网络中断丢失内容。

案例 3：Agent 工具调用的实时观测¶

Agent 平台在调试模式下展示流式事件序列，开发者能直观看到"模型思考 → 输出工具调用 JSON → 等待结果 → 继续生成文本"的全过程，显著提升调试效率。

常见问题¶

Q1：流式响应会降低总耗时吗？ 不会显著降低。流式是把"一次性返回"变为"分段返回"，首字延迟降低但整体耗时接近。真正的提速应通过模型选型、Prompt 优化、缓存实现。

Q2：流中途断开怎么办？ SSE 连接可能因网络抖动中断。建议：① 客户端保存已接收的 delta；② 断开后用 previous_response_id 续接；③ 关键业务保留幂等键，避免重复扣费。

Q3：为什么我的异步流会阻塞事件循环？ 检查是否误用了同步 client。异步代码必须使用 AsyncOpenAI 实例，且 responses.create(stream=True) 返回的是协程，需要 await。

Q4：如何过滤敏感词？ 在 output_text.delta 回调中维护缓冲区，检测到敏感词时用 *** 替换再推给前端；注意敏感词可能跨 chunk，需做窗口匹配。

Q5：流式模式下能拿到 usage 吗？ 能。response.completed 事件携带完整的 response 对象，其中 usage 包含 input/output tokens，可用于计费统计。

面试题¶

1. SSE 与 WebSocket 的区别？何时选 SSE？ 答：SSE 是基于 HTTP 的单向服务端推送，自动重连、文本协议、更简单；WebSocket 是双向全双工、二进制支持。OpenAI 流式只需服务端推 token，SSE 足够且更轻量。

2. 在 FastAPI 中如何把 OpenAI 流式响应转发给浏览器？ 答：用 StreamingResponse 包装一个异步生成器，生成器内部用 AsyncOpenAI 的 responses.create(stream=True) 迭代 delta，按 data: {text}\n\n 格式 yield，设置 media_type="text/event-stream"。

3. 流式事件 response.output_text.delta 与 response.function_call_arguments.delta 有何不同？ 答：前者是模型输出给用户的文本增量；后者是模型在调用工具时输出工具参数 JSON 的增量，Agent 据此可提前预热下游服务。

4. 为什么异步流必须用 AsyncOpenAI？ 答：同步 OpenAI 的 responses.stream 是阻塞调用，会卡住 asyncio 事件循环；AsyncOpenAI 基于 httpx 异步实现，能在等待 token 期间让出控制权。

5. 如何在流式过程中获取 token 用量？ 答：监听 response.completed 事件，从 event.response.usage 读取 input_tokens 与 output_tokens，注意推理模型还有 reasoning_tokens。

练习题¶

用同步流实现一个"逐字打印 + 总耗时统计"的命令行工具，并在结束时打印 token 用量。
用 FastAPI 实现一个 /chat 接口，把 OpenAI 流式响应通过 SSE 转发给前端，要求断开连接时能优雅取消上游请求。
扩展示例代码，在流式过程中检测到""时切换为代码块渲染模式，输出""时切回普通模式。

小结¶

流式响应把"等待"变为"渐进式反馈"，是 LLM 应用的体验基石。本章覆盖了 SSE 协议、事件类型、同步与异步消费、Web 转发与异常处理。关键要点：异步框架必须用 AsyncOpenAI；按事件类型分流逻辑；做好断线续接与 usage 统计。下一章我们将进入函数调用，让模型不仅能"说"，还能"做"。

第2章 流式响应¶