File size: 3,754 Bytes
eff511c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | from __future__ import annotations
import json
import uuid
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from config import CLARIFICATION_LOG_PATH, USAGE_LOG_PATH
def record_usage_event(
question: str,
answer: str,
status: str,
clause_eval: dict[str, Any] | None = None,
) -> None:
clause_eval = clause_eval or {}
record = {
"event_id": str(uuid.uuid4()),
"created_at": datetime.now(timezone.utc).isoformat(),
"question": (question or "").strip(),
"status": status,
"answer_chars": len(answer or ""),
"document_ids": clause_eval.get("document_ids", [])[:8],
"source_ids": clause_eval.get("source_ids", [])[:8],
"scores": clause_eval.get("scores", [])[:8],
"top_source": (clause_eval.get("source_ids", []) or [""])[0] if clause_eval.get("source_ids") else "",
"top_document": (clause_eval.get("document_ids", []) or [""])[0] if clause_eval.get("document_ids") else "",
}
_append_jsonl(USAGE_LOG_PATH, record)
def read_usage_events(limit: int | None = None) -> list[dict[str, Any]]:
records = _read_jsonl(USAGE_LOG_PATH)
if limit is None:
return records
return records[-limit:]
def read_clarification_events(limit: int | None = None) -> list[dict[str, Any]]:
records = _read_jsonl(CLARIFICATION_LOG_PATH)
if limit is None:
return records
return records[-limit:]
def record_clarification_event(
event_type: str,
source_question: str = "",
selected_key: str = "",
resolved_question: str = "",
options: dict[str, str] | None = None,
raw_message: str = "",
) -> None:
record = {
"event_id": str(uuid.uuid4()),
"created_at": datetime.now(timezone.utc).isoformat(),
"event_type": event_type,
"source_question": (source_question or "").strip(),
"selected_key": str(selected_key or ""),
"resolved_question": (resolved_question or "").strip(),
"raw_message": (raw_message or "").strip(),
"options": options or {},
}
_append_jsonl(CLARIFICATION_LOG_PATH, record)
def usage_summary() -> dict[str, Any]:
records = read_usage_events()
statuses = Counter(str(record.get("status", "unknown")) for record in records)
top_sources = Counter(str(record.get("top_source", "")) for record in records if record.get("top_source"))
top_documents = Counter(str(record.get("top_document", "")) for record in records if record.get("top_document"))
recent = records[-10:]
return {
"total_events": len(records),
"status_counts": dict(statuses.most_common()),
"top_sources": top_sources.most_common(8),
"top_documents": top_documents.most_common(8),
"recent": recent,
}
def _append_jsonl(path: Path, record: dict[str, Any]) -> None:
try:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as file:
file.write(json.dumps(record, ensure_ascii=False) + "\n")
except Exception:
pass
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
if not path.exists():
return []
records = []
try:
with path.open("r", encoding="utf-8") as file:
for line in file:
line = line.strip()
if not line:
continue
try:
records.append(json.loads(line))
except json.JSONDecodeError:
continue
except Exception:
return []
return records
|