"""Evaluation runner: executes golden QA cases against a caller-supplied

``answer_fn``. The harness is provider-agnostic — wire any agent loop,

direct LLM call, or stub by passing a function ``str -> str`` to

``EvalRunner``.

The runner does NOT spin up tracing, databases, or LLM clients on its own.

That keeps it cheap to import in unit tests and lets the agent code own its

own startup story.

"""

from __future__ import annotations

import json

import logging

import re

import time

from pathlib import Path

from typing import TYPE_CHECKING

from src.eval.judge import evaluate_semantic_similarity

from src.eval.models import EvalCase, EvalResult

if TYPE_CHECKING:

from collections.abc import Callable

from src.eval.judge import LLMClient

logger = logging.getLogger(__name__)

_GOLDEN_QA_PATH = (

Path(__file__).resolve().parent.parent.parent / "eval" / "golden_qa.json"

)

def load_golden_dataset(path: Path | None = None) -> list[EvalCase]:

"""Load test cases from the golden QA JSON file.

Standalone so callers can introspect the dataset without paying the

cost of constructing an ``EvalRunner``.

"""

qa_path = path or _GOLDEN_QA_PATH

with qa_path.open() as fh:

data = json.load(fh)

return [EvalCase.model_validate(item) for item in data]

class EvalRunner:

"""Runs the golden QA dataset and collects results."""

def __init__(

self,

answer_fn: Callable[[str], str],

judge_client: LLMClient | None = None,

judge_model: str = "",

) -> None:

self._answer_fn = answer_fn

self._judge_client = judge_client

self._judge_model = judge_model

def evaluate(self, case: EvalCase) -> EvalResult:

"""Evaluate a single test case."""

start = time.monotonic()

try:

actual_answer = self._answer_fn(case.question)

except Exception as exc:

actual_answer = f"ERROR: {exc}"

latency_ms = int((time.monotonic() - start) * 1000)

passed, score, reason = self._compare(case, actual_answer)

return EvalResult(

case_id=case.id,

question=case.question,

category=case.category,

difficulty=case.difficulty,

expected_answer=case.expected_answer,

actual_answer=actual_answer,

tools_called=[],

reasoning_trace=[],

latency_ms=latency_ms,

pass_result=passed,

score=score,

failure_reason=reason,

)

def evaluate_all(self) -> list[EvalResult]:

"""Evaluate every case in the loaded golden dataset."""

cases = load_golden_dataset()

results: list[EvalResult] = []

for case in cases:

logger.info("Evaluating %s: %s", case.id, case.question[:50])

result = self.evaluate(case)

status = "PASS" if result.pass_result else "FAIL"

logger.info(" %s %s", status, result.failure_reason or "")

results.append(result)

return results

# ---------------------------------------------------------------------

def _compare(

self,

case: EvalCase,

actual: str,

) -> tuple[bool, float | None, str | None]:

if case.tolerance == "exact_match":

return self._exact_match(case.expected_answer, actual)

if case.tolerance == "numeric_close":

return self._numeric_close(case.expected_answer, actual)

if case.tolerance == "semantic_similar":

return self._semantic_similar(case, actual)

return (False, None, f"Unknown tolerance: {case.tolerance}")

@staticmethod

def _exact_match(

expected: str, actual: str

) -> tuple[bool, float | None, str | None]:

norm_expected = _normalise(expected)

norm_actual = _normalise(actual)

if norm_expected == norm_actual:

return (True, None, None)

return (False, None, f"Exact match failed: '{actual[:100]}'")

@staticmethod

def _numeric_close(

expected: str, actual: str

) -> tuple[bool, float | None, str | None]:

expected_nums = _extract_numbers(expected)

actual_nums = _extract_numbers(actual)

if not expected_nums:

return (False, None, "No numbers found in expected answer")

if not actual_nums:

return (False, None, "No numbers found in actual answer")

target = expected_nums[0]

for num in actual_nums:

if target == 0:

if num == 0:

return (True, None, None)

elif abs(num - target) / abs(target) <= 0.01:

return (True, None, None)

return (

False,

None,

f"Numeric mismatch: expected ~{target}, got {actual_nums}",

)

def _semantic_similar(

self,

case: EvalCase,

actual: str,

) -> tuple[bool, float | None, str | None]:

score, explanation = evaluate_semantic_similarity(

question=case.question,

expected=case.expected_answer,

actual=actual,

client=self._judge_client,

model=self._judge_model,

)

if score is None:

return (True, None, f"Judge inconclusive: {explanation}")

if score >= 0.8:

return (True, score, None)

return (False, score, f"Semantic score {score:.2f}: {explanation}")

def _normalise(text: str) -> str:

"""Lowercase, strip, collapse whitespace."""

return " ".join(text.lower().strip().split())

def _extract_numbers(text: str) -> list[float]:

"""Pull decimals + integers out of free-form text.

Filters out 4-digit values in the year range (2020-2029) so a question

referencing a year doesn't accidentally provide the comparison target.

"""

pattern = r"\d[\d,]*\.?\d*"

matches = re.findall(pattern, text)

numbers: list[float] = []

for m in matches:

try:

val = float(m.replace(",", ""))

except ValueError:

continue

if val == int(val) and 2020 <= val <= 2029:

continue

numbers.append(val)

return numbers