Setting the file. One moment. evals/evaluator.py
Python·91 lines·3 KB
DEFAULT_MODEL
, create_ai_gateway_client
11
12
13# Metrics scored 0–10 by the judge, normalized to 0.0–1.0
14_STRUCTURED_OUTPUT = {
15 "type": "object",
16 "properties": {
17 "diagnosis_accuracy": {"type": "number"},
18 "evidence_grounding": {"type": "number"},
19 "actionability": {"type": "number"},
20 "completeness": {"type": "number"},
21 "reasoning": {"type": "string"},
22 },
23 "required": ["diagnosis_accuracy", "evidence_grounding", "actionability", "completeness", "reasoning"],
24 "additionalProperties": False,
25}
26
27
28class _JudgeCache:
29 """Thread-safe cache: one LLM call per unique row, shared across all metric evaluators."""
30
31 def __init__(self):
32 self._cache: dict[str, dict[str, Any]] = {}
33 self._lock = threading.Lock()
34
35 def get(self, key: str) -> dict[str, Any] | None:
36 with self._lock:
37 return self._cache.get(key)
38
39 def set(self, key: str, scores: dict[str, Any]) -> None:
40 with self._lock:
41 self._cache[key] = scores
42
43
44class _LlmObsTraceRcaJudge(BaseEvaluator):
45 def __init__(self, metric_name: str, cache: _JudgeCache):
46 super().__init__(name=metric_name)
47 self.metric_name = metric_name
48 self._cache = cache
49 rubric_path = Path(__file__).parent / "prompts" / "judge_rubric.txt"
50 self._judge = LLMJudge(
51 user_prompt=rubric_path.read_text(),
52 client=create_ai_gateway_client(),
53 structured_output=_STRUCTURED_OUTPUT,
54 model=DEFAULT_MODEL,
55 )
56
57 def evaluate(self, context: EvaluatorContext) -> EvaluatorResult:
58 if not context.output_data or not context.expected_output:
59 return EvaluatorResult(value=None)
60
61 # Fields that uniquely identify a row — prevents redundant LLM calls.
62 cache_key = "|".join(
63 str(context.input_data.get(f, ""))
64 for f in ["ml_app", "eval_name", "timeframe", "mode", "failure_filter"]
65 )
66
67 scores = self._cache.get(cache_key)
68 if scores is None:
69 result = self._judge.evaluate(context)
70 scores = result.value
71 self._cache.set(cache_key, scores)
72
73 raw = scores.get(self.metric_name)
74 return EvaluatorResult(
75 value=float(raw) / 10.0 if raw is not None else None,
76 reasoning=scores.get("reasoning"),
77 )
78
79
80class LlmObsTraceRcaEvaluator(BaseProjectEvaluator):
81 def get_evaluators(self) -> list[Evaluator]:
82 cache = _JudgeCache()
83 return [
84 _LlmObsTraceRcaJudge("diagnosis_accuracy", cache),
85 _LlmObsTraceRcaJudge("evidence_grounding", cache),
86 _LlmObsTraceRcaJudge("actionability", cache),
87 _LlmObsTraceRcaJudge("completeness", cache),
88 ]
89
90 def get_summary_evaluators(self) -> list:
91 return []