Spaces:
Running
Running
File size: 3,946 Bytes
ed28aa2 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Fail-open shadow telemetry for benchmark output validators.
Shadow mode records validator outcomes only. It never changes the answer, retry
budget, provider selection, or benchmark score. Raw model output is deliberately
not persisted; only length and normalized validator evidence are stored.
"""
from __future__ import annotations
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import threading
from typing import Any, Mapping, Optional
from .validators import ValidationResult, validate_coding_output, validate_mmlu_output
_ENABLED_VALUES = frozenset({"1", "true", "yes", "on"})
_WRITE_LOCK = threading.Lock()
def shadow_enabled() -> bool:
return os.getenv("BENCHMARK_SHADOW_MODE", "0").strip().lower() in _ENABLED_VALUES
def infer_benchmark_category(goal: Any) -> Optional[str]:
"""Infer only the two supported benchmark categories from explicit markers."""
text = str(goal or "")
lowered = text.lower()
if "mmlu" in lowered or "scelta multipla" in lowered or "a/b/c/d" in lowered:
return "mmlu"
if "code_correct" in lowered or "typescript" in lowered or "```typescript" in lowered:
return "coding"
return None
def _safe_metadata(metadata: Optional[Mapping[str, Any]]) -> dict[str, Any]:
allowed = {
"provider",
"model",
"profile",
"attempt",
"latency_ms",
"first_token_ms",
"task_id",
"source",
}
safe: dict[str, Any] = {}
for key in allowed:
value = (metadata or {}).get(key)
if value is None:
continue
if isinstance(value, (str, int, float, bool)):
safe[key] = value
else:
safe[key] = str(value)[:120]
return safe
def _evidence_for_log(result: ValidationResult) -> dict[str, Any]:
evidence: dict[str, Any] = {}
for key, value in result.evidence.items():
if key == "source_length":
evidence[key] = value
elif key in {"candidates", "distinct_candidates", "required_symbols", "missing_symbols", "declarations", "fence_count", "languages", "extraction", "significant_lines", "correct", "expected", "has_import_or_export", "has_syntax_tokens"}:
evidence[key] = value
return evidence
def _log_path() -> Path:
return Path(os.getenv("BENCHMARK_SHADOW_LOG_PATH", "/tmp/baida98-benchmark-shadow.jsonl"))
def _append_event(event: dict[str, Any]) -> None:
path = _log_path()
path.parent.mkdir(parents=True, exist_ok=True)
with _WRITE_LOCK:
with path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n")
def validate_and_record_shadow(
*,
goal: Any,
answer: Any,
metadata: Optional[Mapping[str, Any]] = None,
) -> Optional[ValidationResult]:
"""Validate and record a supported benchmark response in fail-open shadow mode."""
if not shadow_enabled():
return None
category = infer_benchmark_category(goal)
if category is None:
return None
if category == "mmlu":
result = validate_mmlu_output(answer)
validator = "mmlu_v1"
else:
result = validate_coding_output(answer)
validator = "coding_v1"
text = answer if isinstance(answer, str) else str(answer or "")
event = {
"schema_version": 1,
"event": "benchmark_shadow_validation",
"timestamp": datetime.now(timezone.utc).isoformat(),
"category": category,
"validator": validator,
"valid": result.valid,
"failure_code": result.failure_code,
"response_chars": len(text),
"evidence": _evidence_for_log(result),
"metadata": _safe_metadata(metadata),
}
try:
_append_event(event)
except Exception:
# Shadow telemetry must never break the agent loop.
return result
return result
|