流式输出

用户问了一个问题，屏幕上什么都没有，等了三秒，一整段回答突然全部出现。同样是三秒，如果第一个字在半秒后就出现，然后一个字一个字地往外蹦，用户的感受会好很多，因为他知道程序在干活，而且可以边等边读。

这就是流式输出（streaming）。第 01 模块第 2 课讲过，模型本来就是一个词元一个词元生成的。流式输出只是把生成的每一小段立刻发给你，而不是攒齐了再一起发。

打开流式

在请求里加 stream=True，返回的就不再是一个完整的回答，而是一个可以用 for 循环遍历的数据流：

stream = client.chat.completions.create(
    model=MODEL,
    messages=[{"role": "user", "content": "用大约 200 字介绍 httpx 和 requests 的主要区别。"}],
    stream=True,
    extra_body={"thinking": {"type": "disabled"}},
)
for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

每个 chunk 是一小块数据，新生成的文字在 chunk.choices[0].delta.content 里。delta 的意思是"增量"：它只包含这一小块新增的内容，不是到目前为止的全部内容。所以要自己把它们拼起来。

print 的 end="" 让每块文字接在一起而不换行，flush=True 让它立刻显示在屏幕上，而不是等缓冲区满了再显示。少了 flush=True，你会看到文字一批一批地出现，失去了流式的效果。

实测：快了多少

code/03-llm-apps/streaming.py 用同一个问题，分别用流式和非流式各调用一次，记录第一个字出现的时间和全部完成的时间，开思考和不开思考各测一遍：

def streaming(thinking, show=False):
    start = time.time()
    first_content = None
    stream = client.chat.completions.create(
        model=MODEL, messages=QUESTION, stream=True,
        stream_options={"include_usage": True},  # 让最后一个数据块带上 usage
        extra_body={"thinking": {"type": "enabled" if thinking else "disabled"}},
    )
    usage = None
    for chunk in stream:
        if chunk.usage:
            usage = chunk.usage
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta
        if delta.content:
            if first_content is None:
                first_content = time.time() - start
            if show:
                print(delta.content, end="", flush=True)
    if show:
        print()
    return first_content, time.time() - start, usage.completion_tokens

我运行的结果（先打印了一遍流式输出的效果，这里只保留计时部分）：

不思考 非流式：第一个字 1.82 秒，全部完成 1.82 秒，输出 153 词元
不思考 流式：  第一个字 0.67 秒，全部完成 1.71 秒，输出 202 词元
开思考 非流式：第一个字 2.48 秒，全部完成 2.48 秒，输出 330 词元
开思考 流式：  第一个字 1.82 秒，全部完成 2.75 秒，输出 379 词元

不开思考时，流式输出的第一个字在 0.67 秒出现，非流式要等 1.82 秒才能看到任何东西。

注意"全部完成"的时间，两种方式差不多（1.71 秒对 1.82 秒，两次回答的长度也不一样）。流式输出不会让模型生成得更快，总时间不变，它改变的只是用户什么时候开始看到内容。回答越长，这个差别越明显：一段要生成 20 秒的长回答，非流式意味着用户盯着空白屏幕等 20 秒。

开思考时，流式的第一个字要 1.82 秒才出现。因为模型要先思考，思考完才开始写正式回答。思考内容也是流式返回的，在 delta.reasoning_content 里，如果你想让用户看到"正在思考……"，可以把它显示出来，或者只显示一个提示。

流式时怎么拿到 usage

非流式调用时，response.usage 直接告诉你用了多少词元。流式调用默认没有这个信息。

加上 stream_options={"include_usage": True}，服务器会在流的最后额外发一个数据块，里面有 usage，但 choices 是空列表。所以上面的代码里有 if not chunk.choices: continue，没有这一行，访问 chunk.choices[0] 就会报 IndexError。

流式时要注意的几件事

finish_reason 在最后一块里。流式时，前面的数据块 finish_reason 都是 None，只有最后一个带内容的块才会是 stop、length 这些值。要检查回答是否被截断，就在循环里记下它。

出错可能发生在中途。非流式调用要么成功要么失败。流式调用可能在输出了一半之后断开，用户已经看到了半段回答。这时重试会从头生成一段新的回答，和用户已经看到的前半段可能对不上。通常的处理办法是：在建立连接的阶段出错可以重试；已经开始输出后出错，就告诉用户"回答中断了"，让他决定要不要重新问。第 5 课的 RepoBot 就是这样做的。

要自己拼出完整的回答。多轮对话里，模型的回答要作为 assistant 消息存进历史。流式时没有一个现成的完整回答，要把所有 delta.content 拼起来。

把流式输出送到浏览器

命令行里 print 就行了。做成网页时，需要把模型的输出通过你的后端，实时转发给浏览器。最常用的办法是 SSE（Server-Sent Events）：一种浏览器原生支持的、服务器持续向浏览器推送消息的方式。它的格式非常简单，每条消息是一行 data: 内容，后面跟一个空行。

用 FastAPI 写一个最小的例子（code/03-llm-apps/streaming_web.py）：

import json
import os

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
from openai import AsyncOpenAI

client = AsyncOpenAI(  # 网页服务要同时应付很多请求，用异步客户端
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"),
)
MODEL = os.environ.get("LLM_MODEL", "deepseek-flash")
app = FastAPI()


@app.get("/chat")
async def chat(q: str):
    async def events():
        stream = await client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "user", "content": q}],
            stream=True,
            extra_body={"thinking": {"type": "disabled"}},
        )
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                # SSE 的格式：每条消息以 "data: " 开头，以空行结尾
                yield f"data: {json.dumps(chunk.choices[0].delta.content, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

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

几个要点：

• 用 AsyncOpenAI 而不是 OpenAI。网页服务要同时处理很多用户，同步的客户端在等模型回答时会卡住整个服务，异步客户端可以在等待时去处理别的请求。
• StreamingResponse 接收一个生成器，生成器每 yield 一次，就往浏览器发一段数据。
• 每段内容用 json.dumps 编码。模型的输出里可能有换行，而 SSE 用换行分隔消息，直接放进去会把格式弄乱，编码成 JSON 字符串就没有这个问题。
• 最后发一个 [DONE]，告诉浏览器结束了。

浏览器那边，用 EventSource 接收：

const source = new EventSource("/chat?q=" + encodeURIComponent(question));
source.onmessage = (e) => {
  if (e.data === "[DONE]") { source.close(); return; }
  out.textContent += JSON.parse(e.data);
};

完整的页面代码在 streaming_web.py 里。安装依赖并启动：

uv add fastapi uvicorn
uvicorn streaming_web:app --port 8000

用浏览器打开 http://127.0.0.1:8000 就能试。也可以用 curl 直接看 SSE 的原始数据，-N 参数让 curl 收到一段就显示一段：

curl -N "http://127.0.0.1:8000/chat?q=用一句话介绍httpx"

我运行时看到的开头几条：

data: "HTTP"

data: "X"

data: " "

data: "是一个"

data: "功能"

每条消息就是一两个词元。浏览器每收到一条，就把它追加到页面上。

EventSource 只能发 GET 请求，问题要放在网址里，长度有限，也不方便带上对话历史。真实项目里通常用 fetch 发 POST 请求，再读取返回的数据流。第 06 模块第 6 课部署 RepoBot 时会用这种写法。

什么时候不用流式

• 结果要交给程序处理。比如第 02 模块第 4 课的 JSON 提取，程序要拿到完整的 JSON 才能解析，流式没有意义。
• 后台批量任务。没有人在屏幕前等，流式只会让代码更复杂。

凡是有人在屏幕前等回答的地方，都应该用流式。

练习

1. 在 streaming.py 里把问题改成"写一篇 800 字的文章介绍 httpx"，再比较流式和非流式的首字时间。差距变大了吗？
2. 修改 streaming 函数，开思考时把 delta.reasoning_content 也用灰色或者别的标记打印出来，让用户看到模型"在想什么"。
3. 给 streaming_web.py 加一个功能：流结束时，额外发一条消息告诉浏览器这次用了多少词元（记得用 stream_options）。

自测

1. 流式输出能让模型更快地生成完整的回答吗？

不能。生成完整回答的总时间基本不变。流式输出改变的是用户看到第一个字的时间：内容一边生成一边显示，用户不用等全部生成完才看到东西。

2. 流式调用时加了 stream_options={"include_usage": True}，程序在 chunk.choices[0] 这一行报了 IndexError。为什么？

开启 include_usage 后，服务器会在流的最后发一个只包含 usage 的数据块，它的 choices 是空列表。访问 choices[0] 前要先判断 chunk.choices 是否为空。

3. 为什么网页后端要用 AsyncOpenAI，而不是 OpenAI？

同步客户端在等待模型回答时会阻塞，这段时间里服务器没法处理别的用户的请求。异步客户端在等待时可以切换去处理其他请求，一个进程就能同时服务很多用户。