AI Studio の開発者ログを本番の正本にしない — Interactions API を二層で観測する設計

import hashlib
import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Optional
 
 
@dataclass
class InteractionRecord:
    interaction_id: str
    prev_interaction_id: Optional[str]
    prompt_fingerprint: str   # 入力そのものは残さず、指紋だけを正本に置く
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: int
    status: str               # "ok" | "empty" | "error"
    created_at: float
 
 
def fingerprint(text: str) -> str:
    # PII を正本に残さないための最小化。突合には十分。
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
 
 
class ObservedClient:
    """Interactions API 呼び出しを薄く包み、正本ログへ1件ずつ追記する。"""
 
    def __init__(self, client, sink):
        self.client = client      # Gemini の Interactions クライアント
        self.sink = sink          # append_only(record: dict) を持つ格納先
 
    def run(self, prompt: str, *, model: str,
            prev_interaction_id: Optional[str] = None) -> dict:
        started = time.perf_counter()
        # クライアント側でも id を用意し、応答が空でも正本に痕跡を残す
        local_id = prev_interaction_id or str(uuid.uuid4())
        status = "error"
        usage = {"input_tokens": 0, "output_tokens": 0}
        server_id = local_id
        try:
            resp = self.client.interactions.create(
                model=model,
                input=prompt,
                previous_interaction_id=prev_interaction_id,
            )
            server_id = getattr(resp, "id", local_id)
            usage = {
                "input_tokens": resp.usage.input_tokens,
                "output_tokens": resp.usage.output_tokens,
            }
            status = "ok" if resp.output_text else "empty"
            return resp
        finally:
            # try/finally なので、例外時も必ず1件は残る
            rec = InteractionRecord(
                interaction_id=server_id,
                prev_interaction_id=prev_interaction_id,
                prompt_fingerprint=fingerprint(prompt),
                model=model,
                input_tokens=usage["input_tokens"],
                output_tokens=usage["output_tokens"],
                latency_ms=int((time.perf_counter() - started) * 1000),
                status=status,
                created_at=time.time(),
            )
            self.sink.append_only(json.dumps(asdict(rec)))

def reconcile(dashboard_ids: set[str], sink_ids: set[str]) -> dict:
    """速報レンズ側と正本側の interaction_id を突合し、ズレを返す。"""
    missing_in_sink = dashboard_ids - sink_ids   # 記録漏れ。最優先で調査
    missing_in_dashboard = sink_ids - dashboard_ids  # 提供側の保持切れ等
    return {
        "missing_in_sink": sorted(missing_in_sink),
        "missing_in_dashboard": sorted(missing_in_dashboard),
        "sink_coverage": (
            len(sink_ids & dashboard_ids) / len(dashboard_ids)
            if dashboard_ids else 1.0
        ),
    }