ストリーミング応答を読む ── AIが書いた「動くけど止まらない」チャット実装

氷河期世代のクラウドエンジニア、terralien です。ITエンジニアとしてデビューしたのはほんの数年前なので、コードを書くより「出てきたコードを読む」側にいる時間のほうが長いんですよね。

生成AIにチャット呼び出しを書かせると、だいたい10行くらいの綺麗なコードが返ってきます。動かすと、ちゃんと文字が流れてくる。ここで手を止めてマージしてしまうと、しばらくしてから「たまにリクエストが返ってこない」という報告が来ます。

この記事は、そのコードをマージする前に3分で読むための記事です。書けるようになる話ではありません。

この記事の数字はどこで測ったか

推測値は1つも書いていません。全部このMacの上で走らせた実測です。手元でも再現できるように、環境を先に出しておきます。

Ollama は OpenAI 互換のエンドポイントを持っているので、openai パッケージの base_url を差し替えるだけで同じコードが通ります。課金なしで、本物の SDK の挙動をそのまま測れるのがありがたいところです。以下のコードはモデル名と base_url を変えれば、そのまま本番の API にも向きます。

用途 ── ストリーミングが解決するのは「速さ」ではなく「待たされ方」

まず、何のための道具かを取り違えないことです。同じ入力（47都道府県の列挙・出力312トークン）を、stream=True と stream=False で通した結果がこれです。

全部揃う時刻は、むしろストリーミングのほうが 0.3 秒遅いんです。速くなっていません。変わったのは、読者が白い画面を見つめる時間が 5.65 秒から 0.48 秒になったこと、それだけです。

ここを外していると、そもそも監査するまでもありません。AIが書いたコードに stream=True が付いていて、それを受けるのが cron のバッチだったら、複雑さだけ増えて得るものがゼロです。

代表要素 ── 覚えるのは3つ

3つめは地味ですが、課金額を後から検証できるかどうかがここで決まります。付け忘れたまま本番に出ると、請求書と手元のログを突き合わせる手段がなくなります。

AIが書く例 ── 実際に書かせたもの

伝聞にしたくないので、手元の Qwen3.8:27b に「OpenAI の Python SDK でストリーミング応答を端末に流す関数を書いて」と頼みました。返ってきたのがこれです（一字も直していません）。

from openai import OpenAI

def stream_chat_response(prompt: str) -> None:
    client = OpenAI()
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

綺麗です。型ヒントまで付いています。そして動きます。手元で叩けば、ちゃんと文字が流れてきます。

この12行に、これから見る3つの穴が全部入っています。しかも chunk.choices[0] を無条件で見ているので、さっきの include_usage を足した瞬間に IndexError で落ちるおまけ付きです。

監査① ── timeout はストリームの総時間を止めません

ここが最大の穴です。AIが書いたコードに timeout を足せば直る、と思っていると外します。

同じクライアント（timeout=5.0）で、同じ入力を stream の有無だけ変えて投げました。

同じ 5 秒設定で、片方は止まり、片方は止まりません。もっと長い生成（1200トークン）にすると差は開いて、22.3 秒＝設定値の 4.5 倍を走り切ってエラーゼロでした。

なぜかを確かめるために、わざと途中で黙る SSE サーバを立てて、同じクライアントを向けました。

# ① 1チャンク送ったあと60秒だまるサーバ  → timeout は効くのか？
# ② 3秒おきにチャンクを送り続けるサーバ  → 合計60秒。timeout は効くのか？
c = OpenAI(base_url=f"http://127.0.0.1:{port}/v1", api_key="x",
           timeout=5.0, max_retries=0)
s = c.chat.completions.create(model="fake", messages=[...], stream=True)
for ch in s:
    ...

答えが出ました。timeout はチャンクとチャンクの「間隔」に効くのであって、ストリーム全体の長さには一切効きません。間隔が5秒を超えれば止まる。超えなければ、何分でも走ります。設定値の12倍まで見届けました。

止めたいなら、時計は自分で持つしかありません。

import time
DEADLINE = 30.0

t0 = time.monotonic()
with client.chat.completions.create(..., stream=True) as stream:   # ← with で閉じる
    for chunk in stream:
        if time.monotonic() - t0 > DEADLINE:
            break                       # with を抜けるので close される
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

chunk.choices and を足してあるのは、監査②の計測でこちらが実際に踏んだからです。

監査② ── 素朴なリトライは、実用ぶんの1.5倍近くを払わせます

「途中で切れたらリトライ」は正しいのですが、最初からやり直すリトライには値札が付いています。

312トークンの応答を 124 チャンク目（4割地点）で切って、素朴に最初から投げ直したときの請求を数えました。

出力の途中で捨てたぶんも、入力を読み直したぶんも、当然ながら課金対象です。成功率が下がるほど単価が上がる構造になっていて、しかもこれはログに「失敗」としか残らないので、請求書を見るまで気づきません。

読むときに見るのは、except の中身です。

監査③ ── break で抜けたストリームは、まだ喋っています

3つめは、コードを読んだだけでは絶対に気づけないやつです。

長い生成を始めて、8チャンクだけ読んで break で抜けました。そのあと別の軽いリクエストを投げて、生成速度を測っています。

break で抜けても、サーバ側の生成は止まっていません。裏で最後まで喋り続けていて、そのぶんの計算資源を後続のリクエストから奪っています。閉じた瞬間に元に戻るので、原因はこれで確定です。

Python の場合、変数がスコープを抜けてガベージコレクトされれば結果的に閉じます。「結果的に」で運用していい話ではないので、with を使うか .close() を明示するか、どちらかにします。AIが書いたコードには、まず入っていません。

隣で読むもの

ストリーミングの外側にも、同じ「動くけど間違っている」が並んでいます。

チートシート

読めるようになったか、ひとつだけ確認

Q. timeout=30 を設定したストリーミング呼び出しが、5分間返ってこないことはありますか？

あります。チャンクが30秒以内の間隔で届き続けている限り、timeout は一度も発火しません。実測では 5 秒設定のクライアントが 60.1 秒完走しました。総時間を止めたいなら、時計は自分で持つしかありません。

ここまで偉そうに書いてきましたが、白状すると、この記事の計測コードを書いているときに chunk.choices[0] で IndexError を出したのはこちらです。しかも直し方を調べようとして、AIに聞きました。同じ穴に落ちてから記事を書いているので、説得力があるのか無いのか、自分でもよく分かりません。