ストリーミング出力

ユーザーが質問をしても画面には何も出ず、3 秒待ったところで回答全体が一気に現れる。同じ 3 秒でも、最初の文字が 0.5 秒後に現れ、それから 1 文字ずつ出てくるなら、ユーザーの感じ方はずっとよくなります。プログラムが働いていることがわかりますし、待ちながら読むこともできるからです。

これがストリーミング出力（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 で十分です。Web ページにするなら、モデルの出力をバックエンド経由でリアルタイムにブラウザへ転送する必要があります。最もよく使われる方法は SSE（Server-Sent Events）です。ブラウザが標準で対応している、サーバーからブラウザへメッセージを送り続ける仕組みです。形式はとても簡単で、各メッセージは data: 内容 の 1 行で、その後に空行が続きます。

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")

要点がいくつかあります。

• OpenAI ではなく AsyncOpenAI を使う。Web サービスは多くのユーザーを同時にさばく必要があり、同期クライアントはモデルの回答を待つあいだサービス全体を止めてしまいます。非同期クライアントなら、待つあいだに別のリクエストを処理できます。
• 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: "功能"

各メッセージは 1～2 トークンです。ブラウザは 1 件受け取るたびに、それをページに追加します。

EventSource は GET リクエストしか送れないので、質問は URL に入れることになり、長さに制限があるうえ、会話の履歴も付けにくくなります。実際のプロジェクトではふつう fetch で POST リクエストを送り、返ってくるデータのストリームを読み取ります。モジュール 06 第 6 課で RepoBot をデプロイするときはこの書き方を使います。

ストリーミングを使わないとき

• 結果をプログラムで処理する。たとえばモジュール 02 第 4 課の JSON 抽出では、プログラムは完全な JSON を受け取らないと解析できないので、ストリーミングには意味がありません。
• バックグラウンドのバッチ処理。画面の前で待っている人がいないので、ストリーミングはコードを複雑にするだけです。

画面の前で誰かが回答を待っている場面では、必ずストリーミングを使うべきです。

練習問題

1. streaming.py の質問を「httpx を紹介する 800 字の文章を書いて」に変え、ストリーミングと非ストリーミングで最初の文字が出るまでの時間をもう一度比べてください。差は大きくなりましたか。
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. Web のバックエンドで OpenAI ではなく AsyncOpenAI を使うのはなぜですか？

同期クライアントはモデルの回答を待つあいだブロックし、その間サーバーは他のユーザーのリクエストを処理できません。非同期クライアントなら待つあいだに別のリクエストの処理に切り替えられるので、一つのプロセスで多くのユーザーに同時に対応できます。