Class Evaluator

src/evaluation/evaluator.py:30–95

Run a batch of scenarios against their saved trajectories.

class Evaluator:
    """Run a batch of scenarios against their saved trajectories."""

    def __init__(
        self,
        default_scorer: str = "llm_judge",
        judge_model: str | None = None,
    ) -> None:
        self.default_scorer = default_scorer
        self.judge_model = judge_model

    def evaluate(
        self,
        *,
        trajectories_path: Path,
        scenarios_paths: list[Path],
    ) -> EvalReport:
        scenarios = load_scenarios(scenarios_paths)
        trajectories = load_trajectories(trajectories_path)

        results: list[ScenarioResult] = []
        for scenario, traj in join_records(scenarios, trajectories):
            results.append(self._score_one(scenario, traj))

        return build_report(results)

    def _score_one(
        self, scenario: Scenario, traj: PersistedTrajectory
    ) -> ScenarioResult:
        name = scenario.scoring_method or self.default_scorer
        scorer = self._resolve(name)
        self._validate_judge_model(name, traj)
        trajectory_text = _trajectory_to_text(traj)
        answer = _strip_think_blocks(traj.answer)
        score = scorer(scenario, answer, trajectory_text)

        return ScenarioResult(
            scenario_id=scenario.id,
            scenario_type=scenario.type,
            run_id=traj.run_id,
            runner=traj.runner,
            model=traj.model,
            question=traj.question,
            answer=answer,
            score=score,
            ops=metrics_from_trajectory(traj),
        )

    @staticmethod
    def _resolve(name: str) -> Scorer:
        return scorer_registry.get(name)

    def _validate_judge_model(self, scorer_name: str, traj: PersistedTrajectory) -> None:
        if scorer_name != "llm_judge" or not self.judge_model:
            return

        trajectory_model = _normalize_model_id(traj.model)
        judge_model = _normalize_model_id(self.judge_model)