Gemini API の平均レイテンシは速いのに、たまに固まる — p95/p99 を守る運用メモ

# pip install google-genai
import time, json, math
from google import genai
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
def timed_stream(model: str, prompt: str, request_id: str):
    """1リクエストのレイテンシ内訳を構造化ログとして1行吐く。
    後段でp50/p95/p99に畳むことを前提に、生の数値だけを残す。"""
    t0 = time.perf_counter()
    t_first = None
    out_tokens = 0
    status = "ok"
    try:
        stream = client.models.generate_content_stream(model=model, contents=prompt)
        for chunk in stream:
            if t_first is None:
                t_first = time.perf_counter()
            if chunk.text:
                out_tokens += len(chunk.text)
    except Exception as e:
        status = type(e).__name__
    t_end = time.perf_counter()
 
    rec = {
        "request_id": request_id,
        "model": model,
        "ttft_ms": round((t_first - t0) * 1000) if t_first else None,
        "e2e_ms": round((t_end - t0) * 1000),
        "out_chars": out_tokens,
        "status": status,
    }
    print(json.dumps(rec, ensure_ascii=False))  # 構造化ログ基盤へ
    return rec

def percentiles(values, ps=(50, 95, 99)):
    """ソート済み配列から分位点を線形補間で求める。
    依存を増やさず、ログを流し込んだ直後の点検に使う。"""
    xs = sorted(v for v in values if v is not None)
    if not xs:
        return {}
    out = {}
    for p in ps:
        k = (len(xs) - 1) * (p / 100)
        lo, hi = math.floor(k), math.ceil(k)
        out[f"p{p}"] = round(xs[lo] + (xs[hi] - xs[lo]) * (k - lo))
    return out
 
# 例: ttfts = [行ログから集めた ttft_ms のリスト]
# print(percentiles(ttfts))  ->  {'p50': 480, 'p95': 1700, 'p99': 4200}

import asyncio
from google import genai
 
aclient = genai.Client(api_key="YOUR_GEMINI_API_KEY").aio
 
async def first_token_within(model, prompt, budget_s):
    """budget_s 以内に最初のトークンが来たらそのストリームを返す。
    来なければ TimeoutError を投げ、呼び出し側でフォールバックさせる。"""
    stream = await aclient.models.generate_content_stream(model=model, contents=prompt)
    agen = stream.__aiter__()
    first = await asyncio.wait_for(agen.__anext__(), timeout=budget_s)
    return first, agen
 
async def answer(prompt):
    try:
        first, rest = await first_token_within("gemini-2.5-flash", prompt, 1.2)
    except (asyncio.TimeoutError, StopAsyncIteration):
        # 予算超過: 速い構成に逃がす（Thinking無効 + 軽量モデル）
        first, rest = await first_token_within("gemini-2.5-flash-lite", prompt, 2.0)
    yield first.text
    async for chunk in rest:
        if chunk.text:
            yield chunk.text