feat(evaluation): Integrate native LLM judge

backend/app/alembic/versions/073_add_per_item_correctness_to_evaluation_run.py

@@ -0,0 +1,38 @@
+"""add per_item_correctness column to evaluation_run
+
+Revision ID: 073
+Revises: 072
+Create Date: 2026-07-06 00:00:00.000000
+
+Native LLM-as-a-judge correctness rides the existing score/cost columns; this
+adds the one durable map it needs. Nullable with no backfill: pre-feature runs
+carry no correctness data, so NULL is the correct "never judged" sentinel and
+distinguishes them from a run judged into an empty map.
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "073"
+down_revision = "072"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "per_item_correctness",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment=("Durable {trace_id: correctness} map of LLM-as-a-judge scores; "),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("evaluation_run", "per_item_correctness")

backend/app/alembic/versions/074_add_judge_config_to_evaluation_run.py

@@ -0,0 +1,43 @@
+"""add judge_config column to evaluation_run
+
+Revision ID: 074
+Revises: 073
+Create Date: 2026-07-09 00:00:00.000000
+
+The chunked fast-eval pipeline enqueues the aggregate task from the cron barrier
+with only eval_run_id, so a per-run judge LLMCallConfig can't survive as a Celery
+arg. Persist it on the run row instead. Nullable with no backfill: NULL is the
+"zero-config default judge" sentinel, which is exactly how pre-feature runs behave.
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "074"
+down_revision = "073"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "judge_config",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment=(
+                "Per-run LLMCallConfig payload tailoring the correctness judge "
+                "(saved id+version ref or ad-hoc blob); NULL = zero-config default "
+                "judge. Persisted here because the cron barrier enqueues the "
+                "aggregate with only eval_run_id, so it can't ride a Celery arg"
+            ),
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("evaluation_run", "judge_config")

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

@@ -12,14 +12,14 @@
     Query,
 )
 
-
 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.crud.evaluations import list_evaluation_runs as list_evaluation_runs_crud
 from app.crud.evaluations.core import group_traces_by_question_id
 from app.models.config.version import ConfigVersionPublic
 from app.models.evaluation import EvaluationRunPublic, RunModeEnum
+from app.models.llm.request import LLMCallConfig
 from app.services.evaluations import (
     get_evaluation_with_scores,
     improve_prompt,
@@ -58,6 +58,14 @@ def evaluate(
         default=RunModeEnum.FAST,
         description="Execution mode: 'batch' or 'fast'. Omit to default to 'fast'.",
     ),
+    judge_config: LLMCallConfig
+    | None = Body(
+        default=None,
+        description=(
+            "Optional per-run tailoring for the native correctness judge (fast "
+            "mode only). If omitted, the default judge prompt is used."
+        ),
+    ),
 ) -> APIResponse[EvaluationRunPublic]:
     """Start an evaluation run."""
     logger.info(
@@ -76,6 +84,7 @@ def evaluate(
             config_version=config_version,
             organization_id=auth_context.organization_.id,
             project_id=auth_context.project_.id,
+            judge_config=judge_config,
             trace_id=correlation_id.get() or "N/A",
         )
         return APIResponse.success_response(data=eval_run)

backend/app/celery/utils.py

@@ -2,10 +2,11 @@
 Utility functions for easy Celery integration across the application.
 Business logic modules can use these functions without knowing Celery internals.
 """
-import logging
+
 import functools
-from typing import Any, Dict, TypeVar
+import logging
 from collections.abc import Callable
+from typing import Any, Dict, TypeVar
 
 from celery.result import AsyncResult
 from gevent import Timeout
@@ -272,7 +273,7 @@ def start_fast_evaluation_chunk(
 
 
 def start_fast_evaluation_aggregate(eval_run_id: int, trace_id: str = "N/A") -> str:
-    """Enqueue the fan-in aggregate task once a fast run's chunks are all done."""
+    """Enqueue the final aggregation task for a fast EvaluationRun."""
     from app.celery.tasks.job_execution import run_evaluation_fast_aggregate
 
     task_id = _enqueue_with_trace_context(

backend/app/core/config.py

@@ -208,6 +208,9 @@ def AWS_S3_BUCKET(self) -> str:
     # task well under CELERY_TASK_SOFT_TIME_LIMIT.
     EVAL_FAST_CHUNK_SIZE: int = 50
 
+    EVAL_JUDGE_FALLBACK_MODEL: str = "gpt-4o-mini"
+    EVAL_JUDGE_DEFAULT_TEMPERATURE: float = 0.7
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def COMPUTED_CELERY_WORKER_CONCURRENCY(self) -> int:

backend/app/crud/credentials.py

@@ -9,6 +9,7 @@
 from app.core.providers import validate_provider, validate_provider_credentials
 from app.core.security import decrypt_credentials, encrypt_credentials
 from app.core.util import now
+from app.crud.project import get_project_by_id
 from app.models import Credential, CredsCreate, CredsUpdate
 
 
@@ -167,6 +168,41 @@ def get_provider_credential(
     return None
 
 
+def get_tracing_credential(
+    *,
+    session: Session,
+    org_id: int,
+    project_id: int | str,
+) -> dict[str, Any] | None:
+    """Return langfuse credentials only when the project opted into tracing.
+
+    Tracing is gated by the project's `settings["tracing"]` flag (off by
+    default). When disabled, returns None so LangfuseTracer /
+    observe_llm_execution degrade to a no-op and evaluations (via
+    get_tracing_client) fall back to cosine-only scoring.
+    """
+    try:
+        pid = int(project_id)
+    except (TypeError, ValueError):
+        logger.info(
+            f"[get_tracing_credential] Invalid project_id; tracing off | "
+            f"project_id={project_id}"
+        )
+        return None
+
+    project = get_project_by_id(session=session, project_id=pid)
+    if not project or not (project.settings or {}).get("tracing", False):
+        logger.info(f"[get_tracing_credential] Tracing disabled | project_id={pid}")
+        return None
+
+    return get_provider_credential(
+        session=session,
+        org_id=org_id,
+        project_id=pid,
+        provider="langfuse",
+    )
+
+
 def get_providers(*, session: Session, org_id: int, project_id: int) -> list[str]:
     """Returns a list of all active providers for which credentials are stored."""
     creds = get_creds_by_org(session=session, org_id=org_id, project_id=project_id)

backend/app/crud/evaluations/batch.py

@@ -20,6 +20,12 @@
     start_batch_job,
 )
 from app.core.batch.client import GeminiClient
+from app.core.cloud import get_cloud_storage
+from app.crud.evaluations.dataset import (
+    download_csv_from_object_store,
+    get_dataset_by_id,
+)
+from app.crud.evaluations.score import DEFAULT_CATEGORY
 from app.models import EvaluationRun
 from app.models.batch_job import BatchJobType
 from app.services.llm.mappers import (
@@ -70,6 +76,87 @@ def fetch_dataset_items(langfuse: Langfuse, dataset_name: str) -> list[dict[str,
     return items
 
 
+def use_langfuse_client(
+    session: Session,
+    eval_run: EvaluationRun,
+    langfuse: Langfuse | None,
+) -> Langfuse | None:
+    """Return the live client only if the run's dataset is Langfuse-backed, else
+    None, so an opt-out dataset is never traced even if the flag is later on."""
+    if langfuse is None:
+        return None
+
+    dataset = get_dataset_by_id(
+        session=session,
+        dataset_id=eval_run.dataset_id,
+        organization_id=eval_run.organization_id,
+        project_id=eval_run.project_id,
+    )
+    return langfuse if (dataset and dataset.langfuse_dataset_id) else None
+
+
+def load_evaluation_dataset_items(
+    session: Session,
+    eval_run: EvaluationRun,
+    langfuse: Langfuse | None,
+) -> list[dict[str, Any]]:
+    """Load dataset items from Langfuse when a (reconciled) client is present,
+    else from the dataset's object-store CSV."""
+    dataset = get_dataset_by_id(
+        session=session,
+        dataset_id=eval_run.dataset_id,
+        organization_id=eval_run.organization_id,
+        project_id=eval_run.project_id,
+    )
+    if not dataset:
+        raise ValueError(f"Dataset {eval_run.dataset_id} not found")
+
+    if langfuse is not None:
+        return fetch_dataset_items(
+            langfuse=langfuse, dataset_name=eval_run.dataset_name
+        )
+
+    return _load_items_from_object_store(session=session, dataset=dataset)
+
+
+def _load_items_from_object_store(
+    session: Session, dataset: Any
+) -> list[dict[str, Any]]:
+    """Load items from the dataset's object-store CSV with deterministic ids."""
+    from app.services.evaluations.validators import parse_csv_items
+
+    if not dataset.object_store_url:
+        raise ValueError(f"Dataset {dataset.id} has no object-store backing")
+
+    storage = get_cloud_storage(session=session, project_id=dataset.project_id)
+    csv_content = download_csv_from_object_store(
+        storage=storage, object_store_url=dataset.object_store_url
+    )
+    original_items = parse_csv_items(csv_content)
+    duplication_factor = max(
+        1, int((dataset.dataset_metadata or {}).get("duplication_factor", 1))
+    )
+
+    items: list[dict[str, Any]] = []
+    for row_idx, item in enumerate(original_items):
+        for dup_idx in range(duplication_factor):
+            items.append(
+                {
+                    "id": f"item_{row_idx}_{dup_idx}",
+                    "input": {"question": item["question"]},
+                    "expected_output": {"answer": item["answer"]},
+                    "metadata": {
+                        "category": item.get("category") or DEFAULT_CATEGORY,
+                        # 1-based int, matching the Langfuse upload path
+                        # (langfuse.py upload_dataset_to_langfuse) so the Q.ID
+                        # column groups numerically, not by the item id string.
+                        "question_id": row_idx + 1,
+                    },
+                }
+            )
+    return items
+
+
 def build_openai_evaluation_jsonl(
     dataset_items: list[dict[str, Any]], openai_params: dict[str, Any]
 ) -> list[dict[str, Any]]:

backend/app/crud/evaluations/cost.py

@@ -11,10 +11,11 @@
     {
         "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.
 """
 
 import logging
@@ -112,9 +113,22 @@ def _build_embedding_cost_entry(
     return _build_cost_entry(session=session, model=model, totals=totals)
 
 
+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_entry: dict[str, Any] | None,
 ) -> dict[str, Any]:
     """Combine per-stage entries into the `eval_run.cost` payload with a grand total."""
     cost: dict[str, Any] = {}
@@ -128,6 +142,10 @@ def _build_cost_dict(
         cost["embedding"] = embedding_entry
         total += embedding_entry.get("cost_usd", 0.0)
 
+    if judge_entry:
+        cost["judge"] = judge_entry
+        total += judge_entry.get("cost_usd", 0.0)
+
     cost["total_cost_usd"] = round(total, COST_USD_DECIMALS)
     return cost
 
@@ -141,10 +159,12 @@ 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_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 +187,17 @@ def attach_cost(
         else:
             embedding_entry = existing_cost.get("embedding")
 
+        if judge_model is not None and judge_results is not None:
+            judge_entry = _build_judge_cost_entry(
+                session=session, model=judge_model, results=judge_results
+            )
+        else:
+            judge_entry = existing_cost.get("judge")
+
         eval_run.cost = _build_cost_dict(
             response_entry=response_entry,
             embedding_entry=embedding_entry,
+            judge_entry=judge_entry,
         )
     except Exception as cost_err:
         logger.warning(