« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Evaluation harness for running evaluators on datasets."""
3from __future__ import annotations
5from lexigram.contracts.ai.evaluation import (
6 EvaluationDataset,
7 EvaluationHarnessProtocol,
8 EvaluationResult,
9 EvaluationScoreType,
10 EvaluatorProtocol,
11 RunReport,
12)
13from lexigram.logging import get_logger
14from lexigram.result import Err, Ok, Result
16logger = get_logger(__name__)
19class EvaluationHarness(EvaluationHarnessProtocol):
20 """Default evaluation harness implementation.
22 Runs an evaluator against a dataset and produces a run report.
23 """
25 def __init__(self, pass_threshold: float = 0.8) -> None:
26 self._pass_threshold = pass_threshold
28 @property
29 def name(self) -> str:
30 return "evaluation_harness"
32 async def run(
33 self,
34 dataset: EvaluationDataset,
35 evaluator: EvaluatorProtocol,
36 ) -> Result[RunReport, Exception]:
37 results: list[EvaluationResult] = []
38 passed = 0
40 logger.info(
41 "evaluation_run_started",
42 dataset=dataset.name,
43 samples=len(dataset.samples),
44 evaluator=evaluator.name,
45 )
47 try:
48 for sample in dataset.samples:
49 result = await evaluator.evaluate(
50 sample.input,
51 sample.output if hasattr(sample, "output") else "",
52 sample.reference,
53 )
55 if result.is_ok():
56 eval_result = result.unwrap()
57 results.append(eval_result)
58 if eval_result.score >= self._pass_threshold:
59 passed += 1
60 else:
61 error = result.unwrap_err()
62 logger.warning(
63 "sample_evaluation_failed",
64 sample_id=sample.id,
65 error=str(error),
66 )
67 results.append(
68 EvaluationResult(
69 score=0.0,
70 score_type=EvaluationScoreType.CUSTOM,
71 feedback=f"Error: {error}",
72 metrics={"error": str(error)},
73 )
74 )
76 total = len(results)
77 average = sum(r.score for r in results) / total if total > 0 else 0.0
79 report = RunReport(
80 dataset_name=dataset.name,
81 evaluator_name=evaluator.name,
82 total_samples=total,
83 passed_samples=passed,
84 average_score=average,
85 results=results,
86 metadata={
87 "pass_threshold": self._pass_threshold,
88 "pass_rate": passed / total if total > 0 else 0.0,
89 },
90 )
92 logger.info(
93 "evaluation_run_completed",
94 dataset=dataset.name,
95 total=total,
96 passed=passed,
97 average_score=average,
98 )
100 return Ok(report)
101 except Exception as e:
102 logger.error(
103 "evaluation_run_failed",
104 dataset=dataset.name,
105 error=str(e),
106 )
107 return Err(e)
110__all__ = ["EvaluationHarness"]