"""Evaluation endpoint over fixture Q/A pairs.""" from __future__ import annotations import json from pathlib import Path from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_session from app.schemas import EvalCaseResult, EvalReport from app.services.eval_score import ( citation_pages_from_hits, score_case, ) from app.services.rag import ask router = APIRouter(prefix="/api/evaluate", tags=["evaluate"]) def _fixture_path() -> Path: return Path("data/fixtures/eval_questions.json") @router.post("", response_model=EvalReport) async def run_evaluation( session: AsyncSession = Depends(get_session), ) -> EvalReport: """Run the built-in grounded Q/A evaluation set.""" path = _fixture_path() cases = json.loads(path.read_text(encoding="utf-8")) results: list[EvalCaseResult] = [] correct = 0 invented = 0 page_ok = 0 page_total = 0 for case in cases: result = await ask( session, question=case["question"], language=case.get("language"), folder=case.get("folder"), ) expected = case.get("expected_contains", []) ok, is_invented = score_case( answer=result.answer, grounded=result.grounded, expected_contains=expected, should_refuse=bool(case.get("should_refuse")), match_mode="all", ) expected_page = case.get("expected_page") citation_pages = citation_pages_from_hits( result.citations ) page_match: bool | None = None if expected_page is not None: page_total += 1 page_match = expected_page in citation_pages if page_match: page_ok += 1 if ok: correct += 1 if is_invented: invented += 1 results.append( EvalCaseResult( id=case["id"], question=case["question"], expected_contains=expected, answer=result.answer, grounded=result.grounded, correct=ok, invented=is_invented, citation_pages=citation_pages, expected_page=expected_page, page_ok=page_match, ) ) total = len(cases) or 1 return EvalReport( total=len(cases), correct_rate=correct / total, invent_rate=invented / total, citation_page_rate=( page_ok / page_total if page_total else 0.0 ), cases=results, )