Pythonでストリーミング応答を実装する（OpenAI SDK）

LLM の応答は、生成し終えるまで待つと数秒〜十数秒かかることがあります。ユーザーから見ると「固まっている」ように感じられ、離脱の原因になります。そこで使うのがストリーミングです。生成されたそばからトークンを画面に流せば、最初の文字が出るまでの待ち時間（TTFT）が短くなり、体感速度が大きく上がります。FastMetal は OpenAI 互換なので、stream=True を付けるだけで実装できます。

ストリーミングとは

通常のリクエストは、サーバーが応答をすべて生成し終えてからまとめて返します。一方ストリーミングでは、サーバーが生成され次第トークンを逐次送り返すため、クライアントは差分を受け取りながら表示できます。内部的には Server-Sent Events（SSE）という仕組みで、data: 行が連続して送られ、最後に data: [DONE] で終了を知らせます。OpenAI SDK はこの SSE のパースを肩代わりしてくれるので、私たちは差分を順番に受け取るだけで済みます。

Python の実装例

stream=True を渡すと、戻り値はレスポンスオブジェクトではなくイテレータになります。for で回し、各 chunk の choices[0].delta.content を取り出します。

from openai import OpenAI

client = OpenAI(
    base_url="https://api.fastmetal.ai/v1",
    api_key="sk-...",  # FastMetal の API キー
)

stream = client.chat.completions.create(
    model="anthropic-claude-haiku-4-5",
    messages=[{"role": "user", "content": "短い物語を書いて"}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)
print()

ポイントは次のとおりです。

• 各 chunk の choices[0].delta.content に差分テキストが入ります。
• 先頭（role だけのチャンク）や終端では content が None になるため、if delta: でガードします。
• print(..., flush=True) を付けると、バッファに溜めず即座に表示されます。

実運用で気をつけること

サンプルが動いても、本番では通信が途中で切れたり、応答が完了する前に止まったりします。最低限、次の点を押さえておきましょう。

try:
    stream = client.chat.completions.create(
        model="anthropic-claude-haiku-4-5",
        messages=[{"role": "user", "content": "短い物語を書いて"}],
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
except KeyboardInterrupt:
    stream.close()  # ユーザーが中断したら接続を閉じる
except Exception as e:
    print(f"\n[ストリーム中断] {e}")

• 中断・切断: ネットワーク断やタイムアウトで for の途中で例外が飛ぶことがあります。ここまで受け取った分を保持しつつ、エラーを握りつぶさずログに残します。
• 途中終了: ユーザーが生成を止めたい場合は stream.close() で接続を閉じれば、以降のトークン生成を打ち切れます。
• トークン課金は通常どおり: ストリーミングでも、実際に生成されたトークン分は通常リクエストと同じく課金されます。途中で閉じた場合は、その時点までに生成された分が対象です。「ストリームだから安い／無料」ということはありません。

Web アプリでチャット UI にする

ブラウザのチャット UI を作るときは、サーバー側で FastMetal の応答を受け取り、それを SSE としてフロントへ中継します。API キーをブラウザに晒さずに済むのが利点です。FastAPI なら StreamingResponse を使います。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI

app = FastAPI()
client = OpenAI(base_url="https://api.fastmetal.ai/v1", api_key="sk-...")

@app.get("/chat")
def chat(q: str):
    def event_stream():
        stream = client.chat.completions.create(
            model="anthropic-claude-haiku-4-5",
            messages=[{"role": "user", "content": q}],
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"  # SSE 形式で1チャンクずつ送る
        yield "data: [DONE]\n\n"

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

フロント側は EventSource で受け取り、onmessage で届いた差分を要素に追記していくだけです（[DONE] を受けたら閉じます）。実際にはテキストに改行が含まれると SSE がずれるため、本番では差分を JSON にエンコードして data: {"t": "..."} の形で送るのが安全です。

よくある質問

Q. ストリーミングと通常リクエストで料金は変わりますか？ A. 変わりません。どちらも実際に生成されたトークン量で課金されます。ストリーミングは「表示の出し方」が違うだけです。

Q. 途中で stream.close() した場合、残りのトークンも課金されますか？ A. 課金対象はその時点までに生成された分です。閉じた後に生成されないトークンは課金されません。

Q. どのモデルでもストリーミングできますか？ A. FastMetal の OpenAI 互換エンドポイントは stream=True に対応しています。利用できるモデル ID はモデルカタログで確認できます。

次のステップ

まずは手元のスクリプトで stream=True を試し、次にサーバー経由の SSE 中継へ進むのがおすすめです。対応モデルはモデルカタログ、エンドポイントの詳細やパラメータはドキュメントをご覧ください。