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   # 入力そのものは残さず、指紋だけを正本に置く
    feature: 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, feature: 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),
                feature=feature,
                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)))

import sqlite3
from dataclasses import asdict
 
DDL = """
CREATE TABLE IF NOT EXISTS interactions(
  interaction_id      TEXT PRIMARY KEY,
  prev_interaction_id TEXT,
  prompt_fingerprint  TEXT,
  feature             TEXT,          -- どの機能からの呼び出しか
  model               TEXT,
  input_tokens        INTEGER,
  output_tokens       INTEGER,
  latency_ms          INTEGER,
  status              TEXT,
  created_at          REAL
);
CREATE INDEX IF NOT EXISTS idx_created ON interactions(created_at);
CREATE INDEX IF NOT EXISTS idx_status  ON interactions(status);
"""
 
 
class SqliteSink:
    """ホット層の正本。INSERT OR IGNORE で再送・リトライを冪等に受ける。"""
 
    COLS = ("interaction_id", "prev_interaction_id", "prompt_fingerprint",
            "feature", "model", "input_tokens", "output_tokens",
            "latency_ms", "status", "created_at")
 
    def __init__(self, path: str):
        self.con = sqlite3.connect(path, check_same_thread=False)
        self.con.execute("PRAGMA journal_mode=WAL")
        self.con.executescript(DDL)
 
    def append_only(self, record) -> None:
        row = asdict(record) if not isinstance(record, dict) else record
        placeholders = ",".join(f":{c}" for c in self.COLS)
        self.con.execute(
            f"INSERT OR IGNORE INTO interactions VALUES({placeholders})",
            {c: row.get(c) for c in self.COLS},
        )
        self.con.commit()

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
        ),
    }

SELECT feature,
       model,
       COUNT(*)                              AS calls,
       SUM(input_tokens)                     AS in_tok,
       SUM(output_tokens)                    AS out_tok,
       ROUND(AVG(latency_ms))                AS avg_ms,
       ROUND(100.0 * SUM(status <> 'ok') / COUNT(*), 2) AS fail_pct
FROM interactions
WHERE created_at >= :period_start
GROUP BY feature, model
ORDER BY in_tok + out_tok DESC;