Gemini API の月額5万円を8,400円まで下げた運用設計 — キャッシュ・モデル使い分け・Batch API の組み合わせ方

from google import genai
from google.genai import types
 
client = genai.Client()  # GEMINI_API_KEY は環境変数から読み込まれます
 
with open("style_guide.md", encoding="utf-8") as fh:
    style_guide = fh.read()  # 約3万トークンの固定リファレンス
 
def summarize(daily_input: str) -> str:
    response = client.models.generate_content(
        model="gemini-3.5-flash",
        # 固定部分を先頭、可変部分を末尾に。この順序が暗黙キャッシュの鍵です
        contents=[style_guide, daily_input],
    )
    meta = response.usage_metadata
    print(
        f"input={meta.prompt_token_count} "
        f"cached={meta.cached_content_token_count} "
        f"output={meta.candidates_token_count}"
    )
    return response.text

cache = client.caches.create(
    model="gemini-3.5-flash",
    config=types.CreateCachedContentConfig(
        display_name="site-reference-corpus",
        system_instruction="あなたは技術記事の編集を支援するアシスタントです。",
        contents=[style_guide],
        ttl="3600s",  # 1時間。処理をこの窓に集約して使い切ります
    ),
)
 
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="この題材を当サイトの文体方針に沿って要約してください。\n\n" + daily_input,
    config=types.GenerateContentConfig(cached_content=cache.name),
)

ROUTING_RULES = {
    # タスク種別: (モデル, 根拠)
    "tagging":        ("gemini-3.5-flash", "分類はFlashで品質差が出なかった"),
    "short_summary":  ("gemini-3.5-flash", "300字以内の要約はFlashで十分"),
    "long_analysis":  ("gemini-3.1-pro",   "複数文書の突き合わせはProが安定"),
    "code_review":    ("gemini-3.1-pro",   "誤検出の手戻りがFlashだと高くつく"),
}
 
def pick_model(task_type: str) -> str:
    model, _reason = ROUTING_RULES.get(task_type, ("gemini-3.5-flash", "既定"))
    return model
 
def run_task(task_type: str, prompt: str):
    return client.models.generate_content(
        model=pick_model(task_type),
        contents=prompt,
    )

import json
 
# JSONL でリクエストを並べる
requests = [
    {
        "key": f"summary-{i}",
        "request": {
            "contents": [{"parts": [{"text": text}], "role": "user"}],
        },
    }
    for i, text in enumerate(daily_texts)
]
 
with open("batch_input.jsonl", "w", encoding="utf-8") as fh:
    for r in requests:
        fh.write(json.dumps(r, ensure_ascii=False) + "\n")
 
uploaded = client.files.upload(
    file="batch_input.jsonl",
    config=types.UploadFileConfig(mime_type="jsonl"),
)
 
job = client.batches.create(
    model="gemini-3.5-flash",
    src=uploaded.name,
    config={"display_name": "nightly-summaries"},
)
print(job.name, job.state)

from pydantic import BaseModel
 
class ArticleMeta(BaseModel):
    issue: str
    sentiment: str
    priority: str
 
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=mail_body,
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=ArticleMeta,
    ),
)
meta = ArticleMeta.model_validate_json(response.text)

import json
import time
from pathlib import Path
 
LOG_PATH = Path("logs/gemini_usage.jsonl")
LOG_PATH.parent.mkdir(exist_ok=True)
 
def generate_logged(model: str, task: str, **kwargs):
    started = time.time()
    response = client.models.generate_content(model=model, **kwargs)
    meta = response.usage_metadata
    record = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "model": model,
        "task": task,
        "input_tokens": meta.prompt_token_count,
        "cached_tokens": meta.cached_content_token_count or 0,
        "output_tokens": meta.candidates_token_count,
        "latency_sec": round(time.time() - started, 2),
    }
    with LOG_PATH.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    return response
 
def weekly_report():
    totals: dict[str, dict[str, int]] = {}
    with LOG_PATH.open(encoding="utf-8") as fh:
        for line in fh:
            r = json.loads(line)
            t = totals.setdefault(r["model"], {"calls": 0, "in": 0, "cached": 0, "out": 0})
            t["calls"] += 1
            t["in"] += r["input_tokens"]
            t["cached"] += r["cached_tokens"]
            t["out"] += r["output_tokens"]
    for model, t in totals.items():
        hit = t["cached"] / t["in"] * 100 if t["in"] else 0
        print(f"{model}: {t['calls']}回 入力{t['in']:,} (キャッシュ率{hit:.0f}%) 出力{t['out']:,}")