diff --git a/infra/docker/api.Dockerfile b/infra/docker/api.Dockerfile index 0e3669f95..d6b934219 100644 --- a/infra/docker/api.Dockerfile +++ b/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} diff --git a/infra/docker/ray.Dockerfile b/infra/docker/ray.Dockerfile index 2029065cf..94c54f0ff 100644 --- a/infra/docker/ray.Dockerfile +++ b/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 diff --git a/openrag/api/main.py b/openrag/api/main.py index e3255a5d8..543ce22c0 100644 --- a/openrag/api/main.py +++ b/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]) diff --git a/openrag/api/routers/admin/evaluation.py b/openrag/api/routers/admin/evaluation.py new file mode 100644 index 000000000..c5ced0cb9 --- /dev/null +++ b/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)) diff --git a/openrag/api/schemas/admin/evaluation_schemas.py b/openrag/api/schemas/admin/evaluation_schemas.py new file mode 100644 index 000000000..f88dada89 --- /dev/null +++ b/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", +]