| """Trace builder and saver for generation runs.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from uuid import uuid4 |
|
|
| from src.config import TRACE_DIR, get_runtime_settings, runtime_status |
| from src.models.schema import DiaryEntry, ObjectUnderstanding, PersonaEnvelope, TraceRecord |
| from src.traces.anonymizer import anonymize_text |
|
|
|
|
| def build_trace( |
| image_path: str | None, |
| description: str, |
| mode: str, |
| object_understanding: ObjectUnderstanding, |
| persona: PersonaEnvelope, |
| diary: DiaryEntry, |
| trace_id: str | None = None, |
| created_at: datetime | None = None, |
| model_runtime: dict[str, str] | None = None, |
| fallbacks: list[str] | None = None, |
| ) -> TraceRecord: |
| return TraceRecord( |
| trace_id=trace_id or uuid4().hex, |
| created_at=created_at or datetime.now(timezone.utc), |
| mode=mode, |
| input={ |
| "has_image": bool(image_path), |
| "image_filename": Path(image_path).name if image_path else None, |
| "description": anonymize_text(description), |
| }, |
| object_understanding=object_understanding, |
| persona=persona, |
| diary=diary, |
| model_runtime=model_runtime or runtime_status(get_runtime_settings()), |
| fallbacks=fallbacks if fallbacks is not None else ["mock-runtime"], |
| ) |
|
|
|
|
| def save_trace(trace: TraceRecord, trace_dir: Path = TRACE_DIR) -> str: |
| trace_dir.mkdir(parents=True, exist_ok=True) |
| path = trace_dir / f"{trace.trace_id}.json" |
| path.write_text( |
| json.dumps(trace.model_dump(mode="json"), ensure_ascii=False, indent=2), |
| encoding="utf-8", |
| ) |
| return str(path) |
|
|