生成ログを丸ごと残すのをやめました — 3層の保持設計とプロンプト指紋で追跡性だけを残す

# fingerprint.py
# Gemini API のリクエストから「同一性だけ」を取り出す
import hashlib
import json
import re
import unicodedata
from typing import Any
 
 
def _normalize_text(text: str) -> str:
    """表記ゆれを吸収する。全角/半角・連続空白・行末空白を揃える。"""
    text = unicodedata.normalize("NFKC", text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()
 
 
def _shape(text: str) -> dict[str, Any]:
    """本文を捨てても残しておきたい『形』の情報。"""
    return {
        "chars": len(text),
        "lines": text.count("\n") + 1,
        # 言語の当たりをつけるための粗い判定（保存するのは真偽値だけ）
        "has_cjk": bool(re.search(r"[぀-ヿ一-鿿]", text)),
        "has_latin": bool(re.search(r"[A-Za-z]", text)),
    }
 
 
def prompt_fingerprint(
    *,
    model: str,
    system_instruction: str,
    user_text: str,
    generation_config: dict[str, Any],
    tool_names: list[str] | None = None,
) -> dict[str, Any]:
    """本文を残さずに、同一性・構成・形を記録するための指紋を作る。
 
    含めるもの: モデル名 / system_instruction / user 入力 / 出力を左右する設定 / ツール名
    含めないもの: temperature 以外の非決定要因、リクエストID、タイムスタンプ、
                  ユーザーID（別フィールドで擬似ID化して持つ）
    """
    sys_norm = _normalize_text(system_instruction)
    user_norm = _normalize_text(user_text)
 
    # 出力に影響する設定だけを取り出して並び順を固定する
    cfg_keys = ("temperature", "topP", "topK", "maxOutputTokens", "responseMimeType")
    cfg = {k: generation_config.get(k) for k in cfg_keys if k in generation_config}
 
    material = json.dumps(
        {
            "model": model,
            "system": sys_norm,
            "user": user_norm,
            "config": cfg,
            "tools": sorted(tool_names or []),
        },
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    full = hashlib.sha256(material.encode("utf-8")).hexdigest()
 
    # system だけの指紋を別に持つ。プロンプト改訂の検知に効く
    sys_only = hashlib.sha256(sys_norm.encode("utf-8")).hexdigest()
 
    return {
        "fp": full[:16],           # リクエスト全体の指紋
        "fp_system": sys_only[:12],  # system_instruction の版
        "shape_user": _shape(user_norm),
        "config": cfg,
        "model": model,
    }

# logger.py
import random
import time
from datetime import datetime, timedelta, timezone
 
from google.cloud import firestore
 
db = firestore.Client()
 
TIER0_DAYS = 400
TIER1_DAYS = 14
SUCCESS_SAMPLE_RATE = 0.03  # 成功したリクエストの 3% だけ本文を残す
 
 
def _expires_at(days: int) -> datetime:
    return datetime.now(timezone.utc) + timedelta(days=days)
 
 
def log_generation(
    *,
    fingerprint: dict,
    usage: dict,          # {"promptTokens":..,"candidatesTokens":..,"totalTokens":..}
    finish_reason: str,
    latency_ms: int,
    ok: bool,
    request_body: str | None = None,
    response_text: str | None = None,
    task: str = "unknown",
) -> str:
    """Tier 0 は常に書き、本文は条件を満たしたときだけ添える。"""
    # 失敗は必ず、成功は抽選で本文を残す
    keep_body = (not ok) or (random.random() < SUCCESS_SAMPLE_RATE)
 
    doc = {
        "task": task,
        "ts": firestore.SERVER_TIMESTAMP,
        "fp": fingerprint["fp"],
        "fp_system": fingerprint["fp_system"],
        "model": fingerprint["model"],
        "config": fingerprint["config"],
        "shape_user": fingerprint["shape_user"],
        "usage": usage,
        "finish_reason": finish_reason,
        "latency_ms": latency_ms,
        "ok": ok,
        "tier": 1 if keep_body else 0,
        # TTL ポリシーが参照するフィールド
        "expires_at": _expires_at(TIER1_DAYS if keep_body else TIER0_DAYS),
    }
 
    if keep_body:
        doc["body"] = {
            "request": request_body,
            "response": response_text,
            # 本文が消えたあとも Tier 0 側から辿れるようにしておく
            "expires_at": _expires_at(TIER1_DAYS),
        }
 
    ref = db.collection("gen_logs").document()
    ref.set(doc)
    return ref.id

def log_generation_split(...) -> str:
    ref = db.collection("gen_logs").document()
    ref.set({...})  # Tier 0（expires_at = 400日後）
 
    if keep_body:
        db.collection("gen_log_bodies").document(ref.id).set({
            "request": request_body,
            "response": response_text,
            "expires_at": _expires_at(TIER1_DAYS),  # 14日後に自動削除
        })
    return ref.id

gcloud firestore fields ttls update expires_at \
  --collection-group=gen_logs --enable-ttl
 
gcloud firestore fields ttls update expires_at \
  --collection-group=gen_log_bodies --enable-ttl