LLMの応答を全文受け取ってから小分けに送る「擬似ストリーミング」は、見た目こそ逐次表示ですが、最初の1文字が表示されるまでの待ち時間は全文の生成時間と変わりません。ユーザーは真っ白な画面をじっと見つめることになります。かといって、実ストリーミングへの移行はフロントもバックも巻き込む大改修になりそうで、ハードルが高く感じられます。

結論から言うと、フロントに送るイベントの仕様(どんなイベントを・どの順番で送るかの取り決め)を「常に token イベント列」として先に固定しておけば、擬似 → 実ストリーミングの移行はバックエンドだけの変更で完結します。 非対応プロバイダへのフォールバックも自然に組み込めます。

本記事では、実際に移行したときの設計判断とハマりどころをコード例つきで解説します。

こんな方に読んでほしい

• LLMチャットUIをこれから作る / 今まさに作っているエンジニア
• 擬似ストリーミングから実ストリーミングへの移行を検討している人
• LangChain・SSEの基礎はなんとなく分かる、くらいの人

ご注意:本記事のコードはすべて構造を示すための擬似コードです。監査・使用量計測・プロバイダ連携などの実装詳細は省略・抽象化しており、そのままでは動作しません。設計の考え方としてお読みください。特に監査ログは、実運用ではリクエスト処理をブロックしないよう非同期キューに投げる構成を推奨します(理由は実装①で触れます)。

用語ミニ解説(分かる方は読み飛ばしてください)

多くのチャット実装が最初に通る道:LLMの応答を全文受け取ってから刻んで送る、いわゆる擬似ストリーミングです。

移行前のバックエンドは、こんな形でトークンを送り出していました。

# Before: 全文を受け取ってから刻んで送る
response = await llm.ainvoke(messages)
for chunk in split_into_tokens(response.content):
    yield sse_event("token", {"delta": chunk})
yield sse_event("done", {...})

ここで伏線がひとつ。フロントは最初から「イベント列が届く」ことしか知らず、その刻みが本物のトークンか後から分割したものかは区別がつかない設計でした。フロントとの間で決めているイベント仕様はこれだけです。

done に全文は載せません(帯域と二重描画リスクを増やすだけ)。ミッドストリームの失敗は普通に起きるので、error まで仕様に含めて「途中まで出た本文の扱い」を決めておくのが実運用の肝です。なお、サーバープロセス自体が落ちた場合は done も error も届きません。そのケースはフロント側のタイムアウトで拾う話になるため、本記事では扱いません。

設計方針

1. フロントとのイベント仕様は不変:どちらの経路でも同じ start → token* → (done | error) が届く
2. 回帰ゼロ:ストリーミングできない経路は従来コードにそのまま落ちる
3. 横断関心の等価性:PIIマスク・監査・使用量計測は非ストリーム版と情報等価に保つ

「逐次らしさ」はフロントの演出ではなく、バックエンドがプロバイダのストリームを実際に中継しているかで決まる、と役割を明確化しました。擬似でも本物でも、フロントに届くイベントの形は同じです。フロントは何も知らないからこそ、1行も変えずに裏側を差し替えられます。

方針3の補足:本システムのPIIマスクは入力側のみです。出力側マスクが要件にあると、トークン境界でパターンが分断されるためバッファリングが必要になり、ストリーミングと素直に両立しません。移行前に要確認のポイントです。

全文確定型 execute_llm と対になる execute_llm_stream を用意します。入口の共通ミドルウェア(検証・入力側PIIマスク・添付初期化)は同じものを通し、本体だけ非同期ジェネレータにします。

# 擬似コード:構造のみ。監査・計測などの実装詳細は省略しています
async def execute_llm_stream(request) -> AsyncIterator[dict]:
    # 1. 能力判定(最初の yield より前に行う。理由は実装②)
    #    → ストリーミング非対応なら NotImplementedError を投げる

    # 2. 非ストリーム版と共通の前処理(検証・入力側PIIマスクなど)を通す

    try:
        # 3. プロバイダのストリームを逐次中継する
        #    受け取った増分を内部に蓄積しつつ {"kind": "text", "delta": ...} を yield
        # 4. 完走したら {"kind": "final", "usage": ...} を yield
        ...
    finally:
        # 5. 監査は完走・中断を問わず、ここで蓄積分を1回だけ記録する
        #    (中断時はその旨のフラグ付き。記録方法の詳細は省略)

ポイントは3つです。

• 内部イベントとSSEは別レイヤー:ファサードは text / final を返し、呼び出し側が token / done に変換します。SSEの語彙から切り離すとテストがHTTP抜きで書けます。
• 監査は完了後に1回だけです。ただしタブを閉じられると generator は GeneratorExit で閉じられ、ループの後ろは実行されません。finally に置いて中断フラグ付きで部分テキストを記録し、切断ターンの監査漏れを防ぎます。なお、この finally は GeneratorExit での終了時にも走るため、監査が非同期I/Oの場合は扱いに注意が必要です(finally 内で yield は不可。await 自体は可能ですが、確実に実行されるかは、フレームワークが切断時に generator をきちんと close してくれるかに依存します。FastAPI / Starlette の StreamingResponse はクライアント切断時に generator を close してくれますが、この挙動はフレームワークとバージョンに依存するので実機確認をおすすめします)。また、finally 内にブロッキングI/Oを置くのは避け、実運用では監査を非同期キューに積んでワーカー側で永続化する構成が安全です。
• usage も取れる:stream_usage=True + コールバックで、ストリーミングでもトークン数計測は諦めなくてOKです。

ツールなしなら chain.astream()、ツールあり(エージェント)なら astream_events(version="v2") から on_chat_model_stream イベントだけ抽出します。

# 擬似コード:公開APIの使い方のみ
async for event in agent.astream_events(inputs, version="v2"):
    if event["event"] == "on_chat_model_stream":
        # content は str とは限らないため、正規化してから増分として yield する

注意点は2つです。ひとつは、モデルからの増分の content は常に str とは限らず、プロバイダによってはコンテンツブロックのリストが返るので、正規化処理を1枚かませることです。もうひとつは知見として、ツール判断ターンではほぼ何も流れず、最終回答ターンでテキストが流れるため、ツール呼び出しJSONの垂れ流しは杞憂だったことです。

なお、astream_events のイベント体系はLangChainのバージョンで変わり得るため(本記事は version="v2" 前提)、最新の挙動は LangChain公式ドキュメント をご確認ください。

ストリーミング未対応のプロバイダに当たったとき、execute_llm_stream は NotImplementedError を投げます。これは失敗ではなく「私はストリーミングできません」という能力シグナルです。

呼び出し側のルールは「最初のトークンを emit する前なら、何も起きていないのと同じ」です。まだ1トークンも送っていなければ副作用ゼロなので、例外をキャッチして従来の非ストリーム経路へ透過フォールバックできます。このルールを成立させるために、(1) 能力判定はジェネレータの先頭(最初の yield より前)に置く、(2) 1トークンでも emit したらフォールバックしないことをコードで保証する、の2点を守ります。

# 擬似コード:外枠の構造のみ。start / done は経路に関わらずこの層が送出する
send("start")   # 必ず先頭で送出(仕様上 usage は欠落し得るので、final が来なくても done は閉じられる作りに)

emitted = False
try:
    # ストリーム版を購読し、text の増分を token として送出(1つでも送ったら emitted を立てる)
    # final が来たら usage を控えておく
    ...
except NotImplementedError:
    if emitted:
        raise  # ルール違反。上位の共通エラーハンドラが error イベントで閉じる
    # 未送出なら従来の非ストリーム経路へ透過フォールバックし、
    # 結果を刻んで token として送出する(監査はフォールバック先の経路が記録)
    ...

send("done")    # 正常系は必ず done で閉じる

start と done は経路に関わらずこの外枠が一手に送出するので、フロントにはどちらでも同一のイベント列が届きます。ミッドストリームの本物の失敗(プロバイダ切断など)はフォールバックせず、例外をそのまま上に投げます。投げられた例外は上位の共通エラーハンドラが受けて error イベントを送出するという役割分担なので、この外枠は「正常系は done で閉じる、失敗系は例外を投げれば error で閉じてもらえる」ことだけ守ればOKです。

ハマりどころ:async generator は遅延実行です。 関数を呼んだ瞬間には1行も実行されず、先頭の NotImplementedError が飛ぶのも最初の async for の1周目です。SSEレスポンスを返す前(ヘッダ決定段階)にストリーミング可否を知りたい場合は、ジェネレータとは別に先へプローブしておきます。

# 例: SSEを開く前に能力だけ先に確認しておきたい場合(擬似コード)
can_stream = provider.supports_streaming() and should_stream(turn)
# → can_stream に応じてレスポンスの組み立て方を決めてからストリームを開始する

監査の二重記録も自然に防げます。 finally 監査は本体が動き始めてから有効になるため、先頭の能力判定で例外になった場合は実行されず、フォールバック先が自分の監査を記録するだけです。「誰がログを書くか」は経路ごとに1箇所です。

全ターンを無条件にストリーミングしてはいけません。外部ツールを呼び出しながら応答を組み立てているターン(ツールイベントとテキストが交錯し、フロントの表示が壊れる)や、応答の全文から構造化された部分を抽出する必要があるターン(全文確定が前提になる)はNGです。そこで LLM呼び出しの前に should_stream を判定するゲートを置きました。「できるならやる」ではなく「このターンでやってよいか」を先に決める設計です。

DB書き込みの順序も再設計しました。start 送出のタイミングで assistant メッセージのDB行を空 content で作成し、完了時に確定本文で更新します。仕様上 start は必ずトークンより先に送ると決めているので、「start 時点で行が存在する」と揃えておけばフロントはメッセージIDの行を常に前提にできます(リロード復元で効きます)。「最初のトークン到着時」に作ると start との間に行がない瞬間が生まれるので、行作成は start に寄せるのがポイントです。

決まった delta 列を yield するフェイク(非同期ジェネレータのモック)を注入し、実LLMを叩かずにイベント仕様どおりに動くかを検証します。

async def fake_stream(*_):
    for d in ["こん", "にち", "は"]:
        yield {"kind": "text", "delta": d}
    yield {"kind": "final", "usage": {...}}

検証項目は次のとおりです:監査が1回だけ呼ばれること(完走・フォールバック両方)/途中で close しても interrupted=True で監査が残ること/イベント順序が start → token* → (done | error) を守ること/フォールバックは token 前の NotImplementedError のときだけ発動し、イベント列が非ストリーム経路と同一であること/emit 後の例外は error で閉じること。

この状態を先に作っておくと実装をTDDで進められ、経路が増えても安心です。

体感差は歴然でした。最初のトークンまでの時間が「全文生成時間」から「プロバイダの初動時間」に短縮され、長文回答ほど効きます。TTFT(Time To First Token)を移行前後で計測して数字で残すのがおすすめです。

この移行で得た学び

1. イベント仕様を先に「ストリーミング前提」で固定しておくと、差し替えがサーバー内で完結します。失敗系(error)まで仕様に含めるのが実運用の肝です。
2. 「非対応」は能力シグナルとして設計するとフォールバックが素直になります。ただし「最初のトークンより前」という境界は、コードで保証してこそルールになります。
3. ストリーミング化の本質は「どこでDB書き込み・監査・計測を1回やるか」の再設計です。切断・フォールバックまで含めて置き場所を決めます。トークンを流すこと自体はおまけです。

技術的に動いても、安心して本番に出せるかは別問題です。私たちは次の流れで段階的に切り替えました。

ロールアウトのチェックリスト

• ステージングで擬似/実の両経路をイベント列レベルで突き合わせる(仕様が同一である以上、差分はゼロのはず)
• フィーチャーフラグで実ストリーミングをターン単位にON/OFFできるようにする(should_stream ゲートがそのままスイッチになります)
• カナリアリリースで一部ユーザーから展開し、問題があればフラグOFFで即座に従来経路へ戻す(ロールバックがフラグ1つで済むのも、仕様を固定した恩恵です)

本番で見るべきメトリクス

• TTFT(移行効果そのもの)
• フォールバック率(想定外に高ければ能力判定かプロバイダ設定を疑う)
• error イベントで閉じた会話の割合(ミッドストリーム失敗の増減)
• 監査ログ未記録率(切断時の finally が想定どおり動いているかの検知)

あわせて、監査ログの保管期間と、部分テキストに含まれ得るPIIの取り扱いポリシーは、ストリーミング化とは独立に決めておくべき運用事項です。

• フロントには常に start → token* → (done | error) を届ける(仕様は不変)
• 非対応プロバイダは最初のトークンより前に NotImplementedError で能力シグナル → 透過フォールバック
• 監査・計測・DB書き込みは、切断やフォールバックも含めて「どこで1回やるか」を経路ごとに再設計

「うちのチャット、実は擬似ストリーミングかもしれない」と思った方は、まずフロントとのイベント仕様を確認してみてください。仕様がすでにトークン列前提なら、移行は思っているよりずっと近いですよ!