feat(evaluation) 11/15: admin API routes, and promptfoo in both images

infra/docker/api.Dockerfile

@@ -18,7 +18,21 @@ RUN apt-get update && apt-get install -y \
 
 # install ffmpeg
 RUN apt update && \
-    apt install -y ffmpeg 
+    apt install -y ffmpeg
+
+# Node + promptfoo back the admin evaluation page, where EvalRunner shells out
+# to `promptfoo eval`. Also installed in ray.Dockerfile: Ray runs inside this
+# container unless the deployment uses a separate cluster. Pinned rather than
+# resolved at run time so a run never depends on npm reachability.
+ARG PROMPTFOO_VERSION=0.121.19
+# Node comes from NodeSource: the distro package predates promptfoo's floor.
+RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \
+    && apt-get install -y --no-install-recommends nodejs \
+    && npm install -g promptfoo@${PROMPTFOO_VERSION} \
+    && npm cache clean --force \
+    && rm -rf /var/lib/apt/lists/*
+ENV PROMPTFOO_DISABLE_TELEMETRY=1 \
+    PROMPTFOO_DISABLE_UPDATE=1
 
 # Set environment variables for Hugging Face cache location
 ENV XDG_CACHE_HOME=${XDG_CACHE_HOME:-/app/model_weights}

infra/docker/ray.Dockerfile

@@ -19,7 +19,20 @@ RUN apt-get update && apt-get install -y \
 
 # install ffmpeg
 RUN apt update && \
-    apt install -y ffmpeg 
+    apt install -y ffmpeg
+
+# Node + promptfoo back the admin evaluation page, where EvalRunner shells out
+# to `promptfoo eval`. Pinned rather than resolved at run time so a run never
+# depends on npm reachability.
+ARG PROMPTFOO_VERSION=0.121.19
+# Node comes from NodeSource: the distro package predates promptfoo's floor.
+RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \
+    && apt-get install -y --no-install-recommends nodejs \
+    && npm install -g promptfoo@${PROMPTFOO_VERSION} \
+    && npm cache clean --force \
+    && rm -rf /var/lib/apt/lists/*
+ENV PROMPTFOO_DISABLE_TELEMETRY=1 \
+    PROMPTFOO_DISABLE_UPDATE=1
 
 
 # Set environment variables for Hugging Face cache location

openrag/api/main.py

@@ -40,6 +40,7 @@
     SecurityHeadersMiddleware,
 )
 from api.routers.admin.cluster import router as actors_router
+from api.routers.admin.evaluation import router as evaluation_router
 from api.routers.admin.indexing import router as indexer_router
 from api.routers.admin.jobs import router as queue_router
 from api.routers.admin.model_endpoints import router as model_endpoints_router
@@ -117,6 +118,7 @@ class Tags(Enum):
     PARTITION = "Partitions & files"
     MODEL_ENDPOINTS = "Model Endpoints"
     PRESETS = "Presets"
+    EVALUATION = "Evaluation"
     QUEUE = "Queue management"
     ACTORS = "Ray Actors"
     USERS = "User management"
@@ -357,6 +359,7 @@ def get_config():
 app.include_router(partition_router, prefix="/partition", tags=[Tags.PARTITION])
 app.include_router(model_endpoints_router, prefix="/model-endpoints", tags=[Tags.MODEL_ENDPOINTS])
 app.include_router(presets_router, prefix="/presets", tags=[Tags.PRESETS])
+app.include_router(evaluation_router, prefix="/evaluation", tags=[Tags.EVALUATION])
 app.include_router(queue_router, prefix="/queue", tags=[Tags.QUEUE])
 app.include_router(actors_router, prefix="/actors", tags=[Tags.ACTORS])
 app.include_router(users_router, prefix="/users", tags=[Tags.USERS])

openrag/api/routers/admin/evaluation.py

@@ -0,0 +1,130 @@
+"""Admin routes for the evaluation page.
+
+Datasets are uploaded once and replayed by runs. Every route is admin-only:
+a run indexes a corpus, spends grader tokens, and occupies the single runner
+slot, so it is not something a partition editor should be able to trigger.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from typing import Any
+
+from api.dependencies.auth import current_user, require_admin
+from api.schemas.admin.evaluation_schemas import (
+    EvalDatasetResponse,
+    EvalRunResponse,
+    EvalRunSummaryResponse,
+    StartRunRequest,
+)
+from di.providers import get_evaluation_service
+from fastapi import APIRouter, Depends, File, Form, UploadFile, status
+
+router = APIRouter(dependencies=[Depends(require_admin)])
+
+
+def _run_summary(run: Any) -> EvalRunSummaryResponse:
+    return EvalRunSummaryResponse(
+        id=run.id,
+        dataset_id=run.dataset_id,
+        status=run.status.value,
+        started_at=run.started_at,
+        finished_at=run.finished_at,
+        hit_rate=run.retrieval.hit_rate if run.retrieval else None,
+        mrr=run.retrieval.mrr if run.retrieval else None,
+        answer_pass_rate=run.answer.pass_rate if run.answer else None,
+        files_per_minute=run.indexing.files_per_minute if run.indexing else None,
+        error=run.error,
+    )
+
+
+def _run_detail(run: Any) -> EvalRunResponse:
+    return EvalRunResponse(
+        id=run.id,
+        dataset_id=run.dataset_id,
+        status=run.status.value,
+        started_at=run.started_at,
+        finished_at=run.finished_at,
+        indexing=asdict(run.indexing) if run.indexing else None,
+        retrieval=asdict(run.retrieval) if run.retrieval else None,
+        answer=asdict(run.answer) if run.answer else None,
+        cases=[asdict(case) for case in run.cases],
+        error=run.error,
+        created_by=run.created_by,
+    )
+
+
+@router.get("/datasets", response_model=list[EvalDatasetResponse])
+async def list_datasets(service=Depends(get_evaluation_service)):
+    """List stored evaluation datasets, newest first."""
+    return [asdict(dataset) for dataset in await service.list_datasets()]
+
+
+@router.post(
+    "/datasets",
+    response_model=EvalDatasetResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def create_dataset(
+    name: str = Form(..., description="Human-readable dataset name"),
+    testset: UploadFile = File(..., description="CSV: question,expected_answer,expected_file_ids"),
+    corpus: list[UploadFile] = File(..., description="Documents to index for the run"),
+    user=Depends(current_user),
+    service=Depends(get_evaluation_service),
+):
+    """Upload a corpus and its test set.
+
+    The CSV is validated here, so a bad test set fails now rather than after a
+    run has already indexed the corpus.
+    """
+    # Pass the open streams, not the bytes: large uploads are already spooled
+    # to disk, and reading them here would pull them into memory unbounded.
+    dataset = await service.create_dataset(
+        name=name,
+        corpus=[(upload.filename or "unnamed", upload.file) for upload in corpus],
+        testset=testset.file,
+        user_id=user.get("id") if isinstance(user, dict) else None,
+    )
+    return asdict(dataset)
+
+
+@router.delete("/datasets/{dataset_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_dataset(dataset_id: str, service=Depends(get_evaluation_service)):
+    """Delete a dataset and its stored files."""
+    await service.delete_dataset(dataset_id)
+
+
+@router.get("/runs", response_model=list[EvalRunSummaryResponse])
+async def list_runs(limit: int = 50, service=Depends(get_evaluation_service)):
+    """Run history, newest first."""
+    return [_run_summary(run) for run in await service.list_runs(limit)]
+
+
+@router.post("/runs", response_model=EvalRunResponse, status_code=status.HTTP_202_ACCEPTED)
+async def start_run(
+    body: StartRunRequest,
+    user=Depends(current_user),
+    service=Depends(get_evaluation_service),
+):
+    """Queue a run against a dataset.
+
+    Returns ``409`` when a run is already in flight — runs execute one at a
+    time so that indexing timings stay comparable between them.
+    """
+    run = await service.start_run(
+        body.dataset_id,
+        user.get("id") if isinstance(user, dict) else None,
+    )
+    return _run_detail(run)
+
+
+@router.get("/runs/{run_id}", response_model=EvalRunResponse)
+async def get_run(run_id: str, service=Depends(get_evaluation_service)):
+    """One run with its metrics and per-question detail."""
+    return _run_detail(await service.get_run(run_id))
+
+
+@router.post("/runs/{run_id}/cancel", response_model=EvalRunResponse)
+async def cancel_run(run_id: str, service=Depends(get_evaluation_service)):
+    """Ask the runner to abandon an in-flight run."""
+    return _run_detail(await service.cancel_run(run_id))

openrag/api/schemas/admin/evaluation_schemas.py

@@ -0,0 +1,113 @@
+"""Response models for the admin evaluation endpoints."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class EvalDatasetResponse(BaseModel):
+    """A stored corpus + test set."""
+
+    id: str
+    name: str
+    corpus_file_count: int
+    testset_row_count: int
+    created_at: datetime | None = None
+    created_by: int | None = None
+
+
+class FileIndexingSampleResponse(BaseModel):
+    filename: str
+    size_bytes: int
+    duration_seconds: float
+    failed: bool = False
+
+
+class IndexingMetricsResponse(BaseModel):
+    files_total: int
+    files_failed: int
+    bytes_total: int
+    wall_seconds: float
+    files_per_minute: float
+    megabytes_per_second: float
+    p50_seconds: float
+    p95_seconds: float
+    by_extension: dict[str, dict[str, float]] = Field(default_factory=dict)
+    samples: list[FileIndexingSampleResponse] = Field(default_factory=list)
+
+
+class RetrievalMetricsResponse(BaseModel):
+    scored_cases: int
+    skipped_cases: int
+    hit_rate: float
+    mrr: float
+    recall: float
+    context_relevance: float | None = None
+
+
+class AnswerMetricsResponse(BaseModel):
+    scored_cases: int
+    pass_rate: float
+    factuality: float | None = None
+    rubric_score: float | None = None
+
+
+class EvalCaseResponse(BaseModel):
+    query: str
+    retrieved_file_ids: list[str] = Field(default_factory=list)
+    expected_file_ids: list[str] = Field(default_factory=list)
+    hit: bool | None = None
+    reciprocal_rank: float | None = None
+    answer: str | None = None
+    answer_passed: bool | None = None
+    grader_reason: str | None = None
+
+
+class EvalRunResponse(BaseModel):
+    """A run, with metrics once it has finished."""
+
+    id: str
+    dataset_id: str
+    status: str
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    indexing: IndexingMetricsResponse | None = None
+    retrieval: RetrievalMetricsResponse | None = None
+    answer: AnswerMetricsResponse | None = None
+    cases: list[EvalCaseResponse] = Field(default_factory=list)
+    error: str | None = None
+    created_by: int | None = None
+
+
+class EvalRunSummaryResponse(BaseModel):
+    """Run history row — metrics headline only, no per-case detail."""
+
+    id: str
+    dataset_id: str
+    status: str
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    hit_rate: float | None = None
+    mrr: float | None = None
+    answer_pass_rate: float | None = None
+    files_per_minute: float | None = None
+    error: str | None = None
+
+
+class StartRunRequest(BaseModel):
+    dataset_id: str
+
+
+__all__ = [
+    "AnswerMetricsResponse",
+    "EvalCaseResponse",
+    "EvalDatasetResponse",
+    "EvalRunResponse",
+    "EvalRunSummaryResponse",
+    "FileIndexingSampleResponse",
+    "IndexingMetricsResponse",
+    "RetrievalMetricsResponse",
+    "StartRunRequest",
+]