流式輸出

使用者問了一個問題，螢幕上什麼都沒有，等了三秒，一整段回答突然全部出現。同樣是三秒，如果第一個字在半秒後就出現，然後一個字一個字地往外蹦，使用者的感受會好很多，因為他知道程式在幹活，而且可以邊等邊讀。

這就是流式輸出（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？

同步客戶端在等待模型回答時會阻塞，這段時間裡伺服器沒法處理別的使用者的請求。非同步客戶端在等待時可以切換去處理其他請求，一個程序就能同時服務很多使用者。