feat(evaluation): v2 native LLM-as-judge (ground truth) + Langfuse-free

backend/app/alembic/versions/075_add_judge_column_to_evaluation_run.py

@@ -0,0 +1,36 @@
+"""Add is_judge_run to evaluation_run
+
+Revision ID: 075
+Revises: 074
+Create Date: 2026-07-16 00:00:00.000000
+
+The chunked aggregate task only knows an eval_run_id, so the judge intent has to be
+durable on the run row for it to read at judge time.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "075"
+down_revision = "074"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "is_judge_run",
+            sa.Boolean(),
+            nullable=True,
+            comment=(
+                "True for v2 runs that run the native LLM-as-judge (and skip the "
+                "Langfuse score sync). NULL/False = v1 run, cosine-only, Langfuse-synced"
+            ),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("evaluation_run", "is_judge_run")

backend/app/api/docs/evaluation/create_evaluation_dataset_v2.md

@@ -0,0 +1,18 @@
+Upload a CSV of golden Q&A pairs for evaluation, stored natively by Kaapi with no
+Langfuse dataset (v2).
+
+Same multipart shape as the v1 dataset upload (`file`, `dataset_name`,
+`description`, `duplication_factor`), but the dataset is stored in object storage
+(S3) only: the row is created with `langfuse_dataset_id` null and the CSV holds
+exactly the **original** rows — no physical duplication. The `duplication_factor`
+is recorded in `dataset_metadata` (with a run-time-duplication marker) and applied
+when the run executes, so an 8-row CSV with factor 5 evaluates 40 rows.
+
+**Response:** `APIResponse[DatasetUploadResponse]`, same shape as v1 with
+`langfuse_dataset_id` null. `original_items` is the stored row count and
+`total_items` is `original_items × duplication_factor` (the count the run will
+produce, not stored rows). `eligible_for_fast` reflects only the unique-row size
+cap (`EVAL_FAST_MAX_UNIQUE_ROWS`).
+
+**CSV format:** required columns `question`, `answer`; optional `category`. Rows
+with a missing question or answer are skipped. Maximum upload size is 1 MB.

backend/app/api/docs/evaluation/create_evaluation_v2.md

@@ -0,0 +1,36 @@
+Start a v2 evaluation run. Replicates the v1 `POST /api/v1/evaluations` request
+body with Kaapi's native LLM-as-Judge built in — v1 is left unchanged.
+
+v2 runs are **always fast** and always judged (there is no `run_mode`; batch is
+deferred to a later phase). Every scoreable row is automatically judged (no opt-in
+flag) by one combined LLM-judge call scoring each applicable metric in [0, 1] with
+reasoning: **Adherence to Ground Truth** (answer conveys the same correct
+information as the golden answer), **Adherence to Prompt** (answer obeys the
+assistant's configured instructions; applies only when the run resolves a config
+prompt), and **Adherence to Knowledge Base** (groundedness of the answer against
+the retrieved chunks; applies only to rows that retrieved chunks). Per-row scores +
+reasoning are stored natively by Kaapi in the `score_trace_url` trace unit; the
+single `judge` cost stage covers the one combined call. v2 runs compute **no cosine
+similarity** and do **not** touch Langfuse.
+
+Judging is system-config only: the judge always uses the configured model
+(`EVAL_JUDGE_MODEL`, default `gpt-5-mini`) and the built-in per-metric prompts.
+There is no per-run or ad-hoc judge configuration.
+
+## Example
+
+```json
+{
+  "dataset_id": 123,
+  "experiment_name": "judge-smoke-1",
+  "config_id": "f54f0d67-4817-4103-9fdf-b74b3d46733e",
+  "config_version": 1
+}
+```
+
+## Error responses
+
+| Status | Code | When |
+| --- | --- | --- |
+| 409 | `run_name_already_exists` | A run with the same `experiment_name` already exists for this (organization, project) |
+| 422 | — | An unsupported config type / oversized dataset |

backend/app/api/main.py

@@ -3,7 +3,6 @@
 from app.api.routes import (
     analytics,
     api_keys,
-    assessment as assessment_routes,
     assistants,
     auth,
     collection_job,
@@ -35,7 +34,13 @@
     users,
     utils,
 )
-from app.core.config import settings
+from app.api.routes import (
+    assessment as assessment_routes,
+)
+from app.api.routes.evaluations.dataset_v2 import (
+    router as evaluations_dataset_v2_router,
+)
+from app.api.routes.evaluations.evaluation_v2 import router as evaluations_v2_router
 
 api_router = APIRouter()
 api_router.include_router(analytics.router)
@@ -73,3 +78,10 @@
 api_router.include_router(private.router)
 # if settings.ENVIRONMENT in ["development", "testing"]:
 #     api_router.include_router(private.router)
+
+
+# v2 API surface (mounted at settings.API_V2_STR). Only the endpoints that differ
+# from v1 live here — currently the judged run trigger. Everything else stays v1.
+api_v2_router = APIRouter()
+api_v2_router.include_router(evaluations_v2_router)
+api_v2_router.include_router(evaluations_dataset_v2_router)

backend/app/api/routes/evaluations/dataset_v2.py

@@ -0,0 +1,61 @@
+"""v2 Langfuse-free evaluation dataset upload route.
+
+Same multipart shape and response, but no Langfuse dataset is created and only
+the original items are stored — duplication is metadata, applied at run time.
+"""
+
+import logging
+
+from fastapi import APIRouter, Depends, File, Form, UploadFile
+
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.api.routes.evaluations.dataset import _dataset_to_response
+from app.core.rate_monitor import monitor_rate
+from app.models.evaluation import DatasetUploadResponse
+from app.services.evaluations import upload_dataset_v2, validate_csv_file
+from app.utils import APIResponse, load_description
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/evaluations/datasets", tags=["Evaluation v2"])
+
+
+@router.post(
+    "",
+    description=load_description("evaluation/create_evaluation_dataset_v2.md"),
+    response_model=APIResponse[DatasetUploadResponse],
+    dependencies=[
+        Depends(require_permission(Permission.REQUIRE_PROJECT)),
+        Depends(monitor_rate("evaluations")),
+    ],
+)
+async def upload_dataset_v2_route(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    file: UploadFile = File(
+        ..., description="CSV file with 'question' and 'answer' columns"
+    ),
+    dataset_name: str = Form(..., description="Name for the dataset"),
+    description: str | None = Form(None, description="Optional dataset description"),
+    duplication_factor: int = Form(
+        default=1,
+        ge=1,
+        le=5,
+        description="Run-time duplication per item (min: 1, max: 5)",
+    ),
+) -> APIResponse[DatasetUploadResponse]:
+    """Upload a Langfuse-free evaluation dataset (v2)."""
+    csv_content = await validate_csv_file(file)
+
+    dataset = upload_dataset_v2(
+        session=session,
+        csv_content=csv_content,
+        dataset_name=dataset_name,
+        description=description,
+        duplication_factor=duplication_factor,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+
+    return APIResponse.success_response(data=_dataset_to_response(dataset))

backend/app/api/routes/evaluations/evaluation_v2.py

@@ -0,0 +1,57 @@
+"""v2 evaluation run trigger — replica of the v1 trigger plus native judging."""
+
+import logging
+from uuid import UUID
+
+from asgi_correlation_id import correlation_id
+from fastapi import APIRouter, Body, Depends
+
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.core.rate_monitor import monitor_rate
+from app.models.evaluation import EvaluationRunPublic
+from app.services.evaluations.fast import validate_and_start_fast_evaluation
+from app.utils import APIResponse, load_description
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/evaluations", tags=["Evaluation v2"])
+
+
+@router.post(
+    "",
+    description=load_description("evaluation/create_evaluation_v2.md"),
+    response_model=APIResponse[EvaluationRunPublic],
+    dependencies=[
+        Depends(require_permission(Permission.REQUIRE_PROJECT)),
+        Depends(monitor_rate("evaluations")),
+    ],
+)
+def evaluate_v2(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    dataset_id: int = Body(..., description="ID of the evaluation dataset"),
+    experiment_name: str = Body(
+        ..., description="Name for this evaluation experiment/run"
+    ),
+    config_id: UUID = Body(..., description="Stored config ID"),
+    config_version: int = Body(..., ge=1, description="Stored config version"),
+) -> APIResponse[EvaluationRunPublic]:
+    """Start a v2 evaluation run.
+
+    Always fast and judged; there is no `run_mode` — batch judging is deferred.
+    Judging always runs (`is_judge_run=True`); there is no per-run judge config.
+    """
+
+    eval_run = validate_and_start_fast_evaluation(
+        session=session,
+        dataset_id=dataset_id,
+        run_name=experiment_name,
+        config_id=config_id,
+        config_version=config_version,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+        trace_id=correlation_id.get() or "N/A",
+        is_judge_run=True,
+    )
+    return APIResponse.success_response(data=eval_run)

backend/app/core/config.py

@@ -33,6 +33,8 @@ class Settings(BaseSettings):
     )
 
     API_V1_STR: str = "/api/v1"
+    # v2 hosts
+    API_V2_STR: str = "/api/v2"
     SECRET_KEY: str = secrets.token_urlsafe(32)
     # 60 minutes * 24 hours * 1 days = 1 days
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1
@@ -209,6 +211,12 @@ def AWS_S3_BUCKET(self) -> str:
     # task well under CELERY_TASK_SOFT_TIME_LIMIT.
     EVAL_FAST_CHUNK_SIZE: int = 50
 
+    EVAL_JUDGE_MODEL: str = "gpt-5-mini"
+
+    # Reasoning effort for the judge model; "minimal" keeps per-row judging fast.
+    # One of: none | minimal | low | medium | high | xhigh.
+    EVAL_JUDGE_REASONING_EFFORT: str = "minimal"
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def COMPUTED_CELERY_WORKER_CONCURRENCY(self) -> int:

backend/app/crud/evaluations/cost.py

@@ -9,12 +9,15 @@
 Persisted shape on `eval_run.cost`:
 
     {
-        "response":  {model, input_tokens, output_tokens, total_tokens, cost_usd},
-        "embedding": {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "response":          {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "embedding":         {model, input_tokens, output_tokens, total_tokens, cost_usd},
+        "judge":             {model, input_tokens, output_tokens, total_tokens, cost_usd},
         "total_cost_usd": float,
     }
 
-Either stage entry is optional. Embedding entries use output_tokens=0.
+Any stage entry is optional. Embedding entries use output_tokens=0. One combined call
+grades every metric, so judge tokens can't be split per metric and form a single stage,
+priced from per-row usage like the response stage.
 """
 
 import logging
@@ -112,9 +115,29 @@ def _build_embedding_cost_entry(
     return _build_cost_entry(session=session, model=model, totals=totals)
 
 
+# Everything else at eval_run.cost's top level is a judge stage, preserved across
+# partial updates.
+_NON_JUDGE_COST_KEYS: frozenset[str] = frozenset(
+    {"response", "embedding", "total_cost_usd"}
+)
+
+
+def _build_judge_cost_entry(
+    session: Session, model: str, results: list[dict[str, Any]]
+) -> dict[str, Any]:
+    """Build a judge-stage cost entry from per-row judge usage results."""
+    totals = _sum_tokens(
+        items=results,
+        usage_extractor=lambda r: r.get("usage"),
+        input_key="input_tokens",
+    )
+    return _build_cost_entry(session=session, model=model, totals=totals)
+
+
 def _build_cost_dict(
     response_entry: dict[str, Any] | None,
     embedding_entry: dict[str, Any] | None,
+    judge_entries: dict[str, dict[str, Any]],
 ) -> dict[str, Any]:
     """Combine per-stage entries into the `eval_run.cost` payload with a grand total."""
     cost: dict[str, Any] = {}
@@ -128,6 +151,11 @@ def _build_cost_dict(
         cost["embedding"] = embedding_entry
         total += embedding_entry.get("cost_usd", 0.0)
 
+    for stage, entry in judge_entries.items():
+        if entry:
+            cost[stage] = entry
+            total += entry.get("cost_usd", 0.0)
+
     cost["total_cost_usd"] = round(total, COST_USD_DECIMALS)
     return cost
 
@@ -141,10 +169,13 @@ def attach_cost(
     response_results: list[dict[str, Any]] | None = None,
     embedding_model: str | None = None,
     embedding_raw_results: list[dict[str, Any]] | None = None,
+    judge_stage: str | None = None,
+    judge_model: str | None = None,
+    judge_results: list[dict[str, Any]] | None = None,
 ) -> None:
     """Compute cost for the given stage(s) and attach to `eval_run.cost`, never raising.
 
-    Caller is responsible for persisting `eval_run` afterwards. Either stage's
+    Caller is responsible for persisting `eval_run` afterwards. Any stage's
     previously-computed entry on `eval_run.cost` is preserved when that stage's
     inputs are not supplied, so partial updates never clobber prior data.
     """
@@ -167,9 +198,25 @@ def attach_cost(
         else:
             embedding_entry = existing_cost.get("embedding")
 
+        # Carry forward prior judge stages; recompute only the one supplied this call.
+        judge_entries: dict[str, dict[str, Any]] = {
+            k: v
+            for k, v in existing_cost.items()
+            if k not in _NON_JUDGE_COST_KEYS and isinstance(v, dict)
+        }
+        if (
+            judge_stage is not None
+            and judge_model is not None
+            and judge_results is not None
+        ):
+            judge_entries[judge_stage] = _build_judge_cost_entry(
+                session=session, model=judge_model, results=judge_results
+            )
+
         eval_run.cost = _build_cost_dict(
             response_entry=response_entry,
             embedding_entry=embedding_entry,
+            judge_entries=judge_entries,
         )
     except Exception as cost_err:
         logger.warning(

backend/app/crud/evaluations/dataset.py

@@ -29,6 +29,16 @@
 logger = logging.getLogger(__name__)
 
 
+# dataset_metadata keys, shared by the upload services and the run-time loader so
+DATASET_META_ORIGINAL_ITEMS = "original_items_count"
+DATASET_META_TOTAL_ITEMS = "total_items_count"
+DATASET_META_DUPLICATION_FACTOR = "duplication_factor"
+# v2 marker: the stored CSV holds only the original rows and duplication is applied
+# at run time. Absent/false means the S3 data is already physically duplicated (v1),
+# so the run reads it as-is and must not multiply again.
+DATASET_META_DUPLICATE_AT_RUNTIME = "duplicate_at_runtime"
+
+
 def create_evaluation_dataset(
     session: Session,
     name: str,