一晩のバッチで静かに落ちた数十件をどう拾うか — Gemini Batch API の行単位リトライ台帳

# 一時的とみなすコード（再投入で直る見込みがある）
TRANSIENT_CODES = {429, 500, 502, 503, 504}
 
# 恒久的とみなすコード（入力・権限・仕様起因で、再投入しても直らない）
PERMANENT_CODES = {400, 403, 404, 422}
 
 
def classify_error(error: dict) -> str:
    code = error.get("code")
    message = (error.get("message") or "").lower()
 
    if code in TRANSIENT_CODES:
        return "retryable"
    if code in PERMANENT_CODES:
        return "permanent"
 
    # safety ブロックはコードで表れないことがあるためメッセージでも判定
    if "safety" in message or "blocked" in message or "recitation" in message:
        return "permanent"
 
    # 判断がつかないものは、いったん retryable にして回数上限で頭打ちにする
    return "retryable"

import json
 
MAX_ATTEMPTS = 4  # 一時失敗をここまで再投入したら permanent に落とす
 
 
def reconcile(conn: sqlite3.Connection, output_path: str) -> dict:
    counts = {"succeeded": 0, "retryable": 0, "permanent": 0}
 
    with open(output_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            record = json.loads(line)
            key = record["key"]
 
            if "response" in record and "error" not in record:
                extracted = extract_structured(record["response"])
                if extracted is None:
                    # レスポンスは来たがスキーマ抽出に失敗 → 恒久扱い
                    _mark(conn, key, "permanent", "schema_extract_failed")
                    counts["permanent"] += 1
                else:
                    _mark(conn, key, "succeeded", None,
                          result=json.dumps(extracted, ensure_ascii=False))
                    counts["succeeded"] += 1
                continue
 
            # error 行の処理
            error = record.get("error", {"message": "unknown"})
            verdict = classify_error(error)
            row = conn.execute(
                "SELECT attempts FROM rows WHERE key = ?", (key,)
            ).fetchone()
            attempts = (row[0] if row else 0)
 
            if verdict == "retryable" and attempts + 1 >= MAX_ATTEMPTS:
                # 上限に達したら、これ以上投げない
                verdict = "permanent"
 
            _mark(conn, key, verdict, json.dumps(error, ensure_ascii=False))
            counts[verdict] += 1
 
    return counts
 
 
def _mark(conn, key, status, last_error, result=None):
    conn.execute(
        """
        UPDATE rows
           SET status = ?,
               last_error = ?,
               result = COALESCE(?, result),
               attempts = attempts + CASE WHEN ? = 'retryable' THEN 1 ELSE 0 END,
               updated_at = ?
         WHERE key = ?
        """,
        (status, last_error, result, status, time.time(), key),
    )
    conn.commit()

def build_retry_batch(conn: sqlite3.Connection, out_path: str) -> int:
    rows = conn.execute(
        "SELECT key, payload FROM rows WHERE status IN ('pending', 'retryable')"
    ).fetchall()
 
    with open(out_path, "w", encoding="utf-8") as f:
        for key, payload in rows:
            f.write(json.dumps(
                {"key": key, "request": json.loads(payload)},
                ensure_ascii=False,
            ) + "\n")
 
    return len(rows)

from google import genai
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
 
def submit_batch(src_path: str, model: str = "gemini-flash-latest") -> str:
    uploaded = client.files.upload(
        file=src_path,
        config={"mime_type": "application/jsonl"},
    )
    job = client.batches.create(model=model, src=uploaded.name)
    return job.name  # このジョブ名を台帳とは別のジョブ表に控えておく

from flask import Flask, request, abort
 
app = Flask(__name__)
 
 
@app.post("/gemini/batch-complete")
def on_batch_complete():
    # 署名検証は必須（詳細は割愛。共有秘密での HMAC を推奨）
    if not verify_signature(request):
        abort(401)
 
    job_name = request.json.get("name")
    job = client.batches.get(name=job_name)
    if job.state != "JOB_STATE_SUCCEEDED":
        # 失敗ジョブは別扱い。台帳の行はすべて pending のまま残る
        return ("noted", 200)
 
    output_path = download_output(job)   # dest から手元へ
    conn = open_ledger()
    counts = reconcile(conn, output_path)
    app.logger.info("reconciled: %s", counts)
    return ("ok", 200)