Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """Provider-neutral model usage telemetry for SmartPaper benchmarks. | |
| A3-R5 intentionally keeps this module independent from product entitlement, | |
| payment, quota, and production routing. It normalizes provider usage metadata | |
| and emits append-only benchmark records without retaining prompts or secrets. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import asdict, dataclass, field | |
| from datetime import datetime, timezone | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| import threading | |
| import uuid | |
| from typing import Any, Mapping | |
| def _utc_now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _safe_int(value: Any) -> int: | |
| try: | |
| return max(0, int(value or 0)) | |
| except (TypeError, ValueError): | |
| return 0 | |
| def _safe_text(value: Any, limit: int = 240) -> str: | |
| text = str(value or "").replace("\r", " ").replace("\n", " ").strip() | |
| return text[:limit] | |
| def sanitize_metadata(value: Any, *, depth: int = 0) -> Any: | |
| """Return a bounded, secret-safe JSON-compatible metadata value.""" | |
| if depth > 4: | |
| return "<max-depth>" | |
| if value is None or isinstance(value, (bool, int, float)): | |
| return value | |
| if isinstance(value, str): | |
| lowered = value.lower() | |
| if any(marker in lowered for marker in ("bearer ", "api_key", "api-key", "x-goog-api-key")): | |
| return "<redacted>" | |
| return value[:500] | |
| if isinstance(value, Mapping): | |
| clean = {} | |
| for key, item in list(value.items())[:40]: | |
| name = str(key) | |
| lowered = name.lower() | |
| if any(marker in lowered for marker in ("key", "token", "authorization", "secret", "credential", "prompt", "manuscript")): | |
| clean[name] = "<redacted>" | |
| else: | |
| clean[name] = sanitize_metadata(item, depth=depth + 1) | |
| return clean | |
| if isinstance(value, (list, tuple)): | |
| return [sanitize_metadata(item, depth=depth + 1) for item in list(value)[:40]] | |
| return _safe_text(value, 500) | |
| def prompt_identity(system_prompt: Any = "", user_prompt: Any = "") -> str: | |
| """Hash prompt identity without storing the prompt itself.""" | |
| payload = (str(system_prompt or "") + "\n<SEP>\n" + str(user_prompt or "")).encode("utf-8") | |
| return hashlib.sha256(payload).hexdigest() | |
| class ModelUsageEvent: | |
| provider: str | |
| model: str | |
| status: str | |
| event_id: str = field(default_factory=lambda: str(uuid.uuid4())) | |
| created_at: str = field(default_factory=_utc_now_iso) | |
| phase_id: str = "" | |
| case_id: str = "" | |
| benchmark_version: str = "" | |
| profile_id: str = "" | |
| attempt_index: int = 0 | |
| streaming: bool = False | |
| thinking_mode: str = "" | |
| requested_thinking_budget: int | None = None | |
| prompt_hash: str = "" | |
| request_started_at: str = "" | |
| first_token_at: str = "" | |
| completed_at: str = "" | |
| latency_ms: int = 0 | |
| prompt_input_tokens: int = 0 | |
| cache_hit_tokens: int = 0 | |
| cache_miss_tokens: int = 0 | |
| output_tokens: int = 0 | |
| reasoning_or_thought_tokens: int = 0 | |
| total_tokens: int = 0 | |
| finish_reason: str = "" | |
| http_status: int | None = None | |
| error_class: str = "" | |
| retry_reason: str = "" | |
| escalation_reason: str = "" | |
| provider_service_tier: str = "" | |
| provider_model_version: str = "" | |
| provider_response_id: str = "" | |
| price_snapshot_id: str = "" | |
| estimated_cost_microunits: int = 0 | |
| currency: str = "" | |
| metadata: dict[str, Any] = field(default_factory=dict) | |
| def to_dict(self) -> dict[str, Any]: | |
| payload = asdict(self) | |
| payload["metadata"] = sanitize_metadata(payload.get("metadata") or {}) | |
| return payload | |
| class NoOpModelUsageSink: | |
| def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None: | |
| return None | |
| class JsonlModelUsageSink: | |
| """Thread-safe append-only JSONL sink for benchmark artifacts.""" | |
| def __init__(self, path: str | Path): | |
| self.path = Path(path) | |
| self.path.parent.mkdir(parents=True, exist_ok=True) | |
| self._lock = threading.Lock() | |
| def emit(self, event: ModelUsageEvent | Mapping[str, Any]) -> None: | |
| if isinstance(event, ModelUsageEvent): | |
| row = event.to_dict() | |
| else: | |
| row = sanitize_metadata(dict(event or {})) | |
| line = json.dumps(row, ensure_ascii=False, sort_keys=True) | |
| with self._lock: | |
| with self.path.open("a", encoding="utf-8") as handle: | |
| handle.write(line + "\n") | |
| def emit_usage_event(sink: Any, event: ModelUsageEvent) -> None: | |
| """Best-effort telemetry emission that can never fail model generation.""" | |
| if sink is None: | |
| return | |
| try: | |
| if hasattr(sink, "emit") and callable(sink.emit): | |
| sink.emit(event) | |
| elif callable(sink): | |
| sink(event) | |
| except Exception: | |
| # Telemetry is strictly observational in A3-R5. | |
| return | |
| def normalize_deepseek_usage(raw: Mapping[str, Any] | None) -> dict[str, int]: | |
| usage = dict(raw or {}) | |
| details = usage.get("completion_tokens_details") | |
| if not isinstance(details, Mapping): | |
| details = {} | |
| return { | |
| "prompt_input_tokens": _safe_int(usage.get("prompt_tokens")), | |
| "cache_hit_tokens": _safe_int(usage.get("prompt_cache_hit_tokens")), | |
| "cache_miss_tokens": _safe_int(usage.get("prompt_cache_miss_tokens")), | |
| "output_tokens": _safe_int(usage.get("completion_tokens")), | |
| "reasoning_or_thought_tokens": _safe_int(details.get("reasoning_tokens")), | |
| "total_tokens": _safe_int(usage.get("total_tokens")), | |
| } | |
| def normalize_gemini_usage(raw: Mapping[str, Any] | None) -> dict[str, int | str]: | |
| usage = dict(raw or {}) | |
| return { | |
| "prompt_input_tokens": _safe_int(usage.get("promptTokenCount")), | |
| "cache_hit_tokens": _safe_int(usage.get("cachedContentTokenCount")), | |
| "cache_miss_tokens": max( | |
| 0, | |
| _safe_int(usage.get("promptTokenCount")) - _safe_int(usage.get("cachedContentTokenCount")), | |
| ), | |
| "output_tokens": _safe_int(usage.get("candidatesTokenCount")), | |
| "reasoning_or_thought_tokens": _safe_int(usage.get("thoughtsTokenCount")), | |
| "total_tokens": _safe_int(usage.get("totalTokenCount")), | |
| "provider_service_tier": _safe_text(usage.get("trafficType") or usage.get("serviceTier"), 80), | |
| } | |
| def normalize_zhipu_usage(raw: Mapping[str, Any] | None) -> dict[str, int]: | |
| """Normalize Zhipu/GLM Chat Completions usage fields. | |
| Official GLM responses expose prompt/completion/total token counts and | |
| prompt_tokens_details.cached_tokens. reasoning_tokens is consumed only | |
| when the provider actually returns it; it is never inferred from | |
| reasoning_content text. | |
| """ | |
| usage = dict(raw or {}) | |
| prompt_details = usage.get("prompt_tokens_details") | |
| if not isinstance(prompt_details, Mapping): | |
| prompt_details = {} | |
| completion_details = usage.get("completion_tokens_details") | |
| if not isinstance(completion_details, Mapping): | |
| completion_details = {} | |
| prompt_tokens = _safe_int(usage.get("prompt_tokens")) | |
| cached_tokens = _safe_int(prompt_details.get("cached_tokens")) | |
| return { | |
| "prompt_input_tokens": prompt_tokens, | |
| "cache_hit_tokens": cached_tokens, | |
| "cache_miss_tokens": max(0, prompt_tokens - cached_tokens), | |
| "output_tokens": _safe_int(usage.get("completion_tokens")), | |
| "reasoning_or_thought_tokens": _safe_int(completion_details.get("reasoning_tokens")), | |
| "total_tokens": _safe_int(usage.get("total_tokens")), | |
| } | |