Gemini API の空応答を finish_reason から逆引きする — 診断フロー・再試行分類・監視の実務

from google.genai import types
 
resp = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        max_output_tokens=4096,
        thinking_config=types.ThinkingConfig(thinking_budget=1024),
    ),
)
 
usage = resp.usage_metadata
print("thoughts:", getattr(usage, "thoughts_token_count", 0))
print("candidates:", getattr(usage, "candidates_token_count", 0))

resp = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=user_input,
)
 
if not resp.candidates:
    reason = getattr(resp.prompt_feedback, "block_reason", None)
    # 再試行しても結果は同じ。ユーザーに「入力を変えてほしい」と伝える
    raise InputBlockedError(f"prompt blocked: {reason}")

def safety_for(source: str):
    # 社内文書など信頼できる入力は緩く、生のユーザー入力は標準のまま
    threshold = "BLOCK_ONLY_HIGH" if source == "internal" else "BLOCK_MEDIUM_AND_ABOVE"
    return [
        types.SafetySetting(category=c, threshold=threshold)
        for c in (
            "HARM_CATEGORY_HARASSMENT",
            "HARM_CATEGORY_HATE_SPEECH",
            "HARM_CATEGORY_SEXUALLY_EXPLICIT",
            "HARM_CATEGORY_DANGEROUS_CONTENT",
        )
    ]

import random
import time
from dataclasses import dataclass, field
 
FAIL_FAST = {"SAFETY", "RECITATION", "PROHIBITED_CONTENT", "BLOCKLIST", "SPII", "LANGUAGE"}
RETRY_AS_IS = {"OTHER", "FINISH_REASON_UNSPECIFIED", "UNKNOWN"}
 
@dataclass
class Plan:
    action: str                     # "ok" / "retry" / "fix_and_retry" / "fail"
    fix: dict = field(default_factory=dict)
 
def classify(resp) -> Plan:
    if not resp.candidates:
        return Plan("fail")          # 入力ブロックは何度送っても同じ
    cand = resp.candidates[0]
    finish = str(cand.finish_reason or "UNKNOWN").split(".")[-1]
    parts = (cand.content.parts if cand.content else None) or []
    text = "".join((getattr(p, "text", "") or "") for p in parts)
    if text and finish == "STOP":
        return Plan("ok")
    if finish == "MAX_TOKENS":
        # 出力予算を倍にし、思考予算は絞って本文へ回す
        return Plan("fix_and_retry", fix={"grow_output": 2.0, "thinking_budget": 512})
    if finish in FAIL_FAST:
        return Plan("fail")
    return Plan("retry")             # OTHER 系のみ素直に再試行
 
def generate(prompt: str, model: str = "gemini-2.5-flash", max_attempts: int = 3) -> str:
    max_out = 2048
    thinking = 1024
    for attempt in range(max_attempts):
        resp = client.models.generate_content(
            model=model,
            contents=prompt,
            config=types.GenerateContentConfig(
                max_output_tokens=max_out,
                thinking_config=types.ThinkingConfig(thinking_budget=thinking),
            ),
        )
        plan = classify(resp)
        if plan.action == "ok":
            return resp.text
        if plan.action == "fail":
            raise RuntimeError(f"non-retryable: {triage(resp)}")
        if plan.action == "fix_and_retry":
            max_out = int(max_out * plan.fix["grow_output"])
            thinking = plan.fix["thinking_budget"]
        time.sleep((2 ** attempt) + random.random())   # ジッター付き指数バックオフ
    raise TimeoutError(f"retries exhausted for model={model}")

import { GoogleGenAI } from "@google/genai";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY });
 
const stream = await ai.models.generateContentStream({
  model: "gemini-2.5-flash",
  contents: prompt,
});
 
let text = "";
let finish: string | undefined;
 
for await (const chunk of stream) {
  text += chunk.text ?? "";
  finish = chunk.candidates?.[0]?.finishReason ?? finish;
}
 
if (!text) {
  console.error(`empty stream: finishReason=${finish ?? "never_arrived"}`);
}

logger.info(
    "gemini.response",
    extra={
        "triage": triage(resp),                     # ok / input_blocked:* / no_parts:* / non_text_parts:*
        "model": model,
        "thoughts_tokens": getattr(resp.usage_metadata, "thoughts_token_count", 0),
        "candidates_tokens": getattr(resp.usage_metadata, "candidates_token_count", 0),
    },
)