ストリーミング表示がローカルでは動くのに本番だと一括表示に戻るとき — 中間プロキシのバッファリングと沈黙する停止を計測する運用メモ

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from google import genai
import anyio
 
app = FastAPI()
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
async def sse_stream(prompt: str, request: Request):
    # コメント行(:)を1本流して経路をすぐ開かせる = TTFBを下げる
    yield ": open\n\n"
    loop_stream = await anyio.to_thread.run_sync(
        lambda: client.models.generate_content_stream(
            model="gemini-3.5-flash", contents=prompt
        )
    )
    last_beat = anyio.current_time()
    for chunk in loop_stream:
        # クライアントが離脱したら即やめる(無駄な生成と課金を止める)
        if await request.is_disconnected():
            break
        text = getattr(chunk, "text", "") or ""
        if text:
            yield f"data: {text}\n\n"
        # 15秒無音ならハートビートを送ってアイドル切断を防ぐ
        now = anyio.current_time()
        if now - last_beat > 15:
            yield ": ping\n\n"
            last_beat = now
    yield "event: done\ndata: [DONE]\n\n"
 
@app.get("/stream")
async def stream(prompt: str, request: Request):
    return StreamingResponse(
        sse_stream(prompt, request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
        },
    )

import hashlib, json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
 
# 本番は Redis 等。ここでは概念を示すためのインメモリ
_inflight: dict[str, str] = {}
 
def idem_key(prompt: str, cid: str) -> str:
    return hashlib.sha256(f"{cid}:{prompt}".encode()).hexdigest()[:16]
 
@app.get("/stream")
async def stream(prompt: str, cid: str, request: Request):
    key = idem_key(prompt, cid)
    if _inflight.get(key) == "running":
        # 再接続中の二重呼び出し。新規生成せず終了を伝える
        async def already():
            yield "event: dup\ndata: in-progress\n\n"
        return StreamingResponse(already(), media_type="text/event-stream")
    _inflight[key] = "running"
    async def guarded():
        try:
            async for evt in sse_stream(prompt, request):
                yield evt
        finally:
            _inflight.pop(key, None)
    return StreamingResponse(
        guarded(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache, no-transform", "X-Accel-Buffering": "no"},
    )

const cid = crypto.randomUUID();
const es = new EventSource(`/stream?cid=${cid}&prompt=${encodeURIComponent(prompt)}`);
 
es.onmessage = (e) => { output.textContent += e.data; };
es.addEventListener("done", () => es.close());   // ← 正常終了で必ず閉じる
es.onerror = () => es.close();                     // ← 自動再接続に任せない