| """
|
| trace_log.py — capture model prompt→response traces for later fine-tuning.
|
|
|
| Every real narration turn is appended as one JSON line to a dated file under
|
| the trace directory. Each record is already in the standard SFT chat shape
|
| (a `messages` list ending in the assistant's raw response), with extra
|
| `context` fields alongside for curating/filtering the dataset later:
|
|
|
| {"ts": "...", "model": "...", "context": {...},
|
| "messages": [{"role":"system",...},{"role":"user",...},
|
| {"role":"assistant","content": "<raw model output>"}]}
|
|
|
| To build a training set, read the lines and keep `record["messages"]`
|
| (drop the trailing assistant turn at inference time).
|
|
|
| Controls:
|
| ARS_FABULA_TRACE = "1"|"on" (default) to log, "0"|"off"|"false" to disable
|
| ARS_FABULA_TRACE_DIR = output directory (default: "logs/traces")
|
|
|
| Mock/scripted responses are NOT logged — the engine skips logging when the
|
| model reports `is_mock` (see run_turn), so tests and mock fallbacks don't
|
| pollute the dataset.
|
| """
|
| from __future__ import annotations
|
| import os
|
| import json
|
| from datetime import datetime, date
|
| from pathlib import Path
|
| from typing import Optional
|
|
|
|
|
| _DISABLED = {"0", "off", "false", "no", ""}
|
|
|
|
|
| def traces_enabled() -> bool:
|
| return os.getenv("ARS_FABULA_TRACE", "1").strip().lower() not in _DISABLED
|
|
|
|
|
| def _trace_dir() -> Path:
|
| return Path(os.getenv("ARS_FABULA_TRACE_DIR", "logs/traces"))
|
|
|
|
|
| def log_turn(messages: list[dict], response: str, *,
|
| model: str = "", context: Optional[dict] = None) -> None:
|
| """Append one prompt→response trace as a JSON line.
|
|
|
| Best-effort: any IO/serialization error is swallowed so a logging
|
| problem can never break a turn. `messages` is the prompt sent to the
|
| model (system + user); the assistant's raw `response` is appended so the
|
| stored `messages` list is directly usable for supervised fine-tuning.
|
| """
|
| if not traces_enabled():
|
| return
|
| try:
|
| rec = {
|
| "ts": datetime.now().isoformat(timespec="seconds"),
|
| "model": model,
|
| "context": context or {},
|
| "messages": list(messages) + [{"role": "assistant", "content": response}],
|
| }
|
| out_dir = _trace_dir()
|
| out_dir.mkdir(parents=True, exist_ok=True)
|
| path = out_dir / f"turns-{date.today().isoformat()}.jsonl"
|
| with open(path, "a", encoding="utf-8") as f:
|
| f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
| except Exception:
|
|
|
| pass
|
|
|
