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