56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
"""Structured JSON-lines logging of every model call (charter §10 — the seed and
|
|
the full prompt logged with every call; §4 — free eval/replay infrastructure).
|
|
|
|
One JSON object per line: everything needed to replay the call (canon_log +
|
|
messages + model + seed) against a candidate model. No secrets to redact — the
|
|
client never sends keys, the canon log carries no PII, and the prompt is exactly
|
|
what we want on record.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from typing import Callable
|
|
|
|
from . import config
|
|
|
|
|
|
def _default_write(line: str) -> None:
|
|
path = config.call_log_path()
|
|
if path:
|
|
with open(path, "a", encoding="utf-8") as handle:
|
|
handle.write(line + "\n")
|
|
else:
|
|
sys.stdout.write(line + "\n")
|
|
|
|
|
|
def record(
|
|
*,
|
|
role: str,
|
|
model: str,
|
|
options: dict,
|
|
messages: list[dict],
|
|
canon_log: dict,
|
|
ok: bool,
|
|
latency_ms: int,
|
|
response: str | None = None,
|
|
error: str | None = None,
|
|
write: Callable[[str], None] = _default_write,
|
|
) -> dict:
|
|
rec: dict = {
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
"role": role,
|
|
"model": model,
|
|
"options": options,
|
|
"messages": messages,
|
|
"canon_log": canon_log,
|
|
"ok": ok,
|
|
"latency_ms": latency_ms,
|
|
}
|
|
if ok:
|
|
rec["response"] = response
|
|
else:
|
|
rec["error"] = error
|
|
write(json.dumps(rec, ensure_ascii=False))
|
|
return rec
|