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()

import gzip
import json
import sqlite3
import time
 
COLD_COLS = ("interaction_id", "prompt_fingerprint", "feature",
             "input_tokens", "output_tokens", "status", "created_at")
HOT_DAYS = 90
 
 
def demote_to_cold(db_path: str, cold_path: str, now: float | None = None) -> int:
    """90日を過ぎた行を射影して gzip へ追記し、ホット層から落とす。"""
    now = now or time.time()
    cutoff = now - HOT_DAYS * 86400
    con = sqlite3.connect(db_path)
    rows = con.execute(
        f"SELECT {','.join(COLD_COLS)} FROM interactions WHERE created_at < ?",
        (cutoff,),
    ).fetchall()
    if not rows:
        con.close()
        return 0
    # 追記モードで開く。既存のコールドファイルは読み直さない
    with gzip.open(cold_path, "at", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(dict(zip(COLD_COLS, r)), ensure_ascii=False) + "\n")
    con.execute("DELETE FROM interactions WHERE created_at < ?", (cutoff,))
    con.commit()
    con.execute("VACUUM")   # DELETE だけでは実ファイルは縮まない
    con.close()
    return len(rows)

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

import contextvars
import functools
import logging
 
log = logging.getLogger("interactions.guard")
_inside_wrapper = contextvars.ContextVar("inside_wrapper", default=False)
 
 
def install_bypass_guard(client, *, strict: bool = False) -> None:
    """生クライアントの直接呼び出しを検出する。ObservedClient 経由なら通す。"""
    raw = client.interactions.create
 
    @functools.wraps(raw)
    def guarded(*args, **kwargs):
        if not _inside_wrapper.get():
            msg = "ObservedClient を経由しない Interactions 呼び出しです"
            if strict:
                raise RuntimeError(msg)
            log.warning("%s (model=%s)", msg, kwargs.get("model"))
        return raw(*args, **kwargs)
 
    client.interactions.create = guarded

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;