feat(evaluation): Improve iteration loop efficiency

backend/app/alembic/versions/076_add_evaluation_iteration_run.py

@@ -0,0 +1,127 @@
+"""Add evaluation_iteration_run table
+
+Revision ID: 076
+Revises: 075
+Create Date: 2026-08-02 00:00:00.000000
+
+Thin tracking row for the eval-iterate-improve LangGraph loop (see
+docs/srd-ai-prompt-improvement.md follow-on: the Evaluation Iteration Loop). The
+round-by-round trajectory itself lives in the LangGraph checkpoint (owned by
+`langgraph-checkpoint-postgres`, set up separately via `checkpointer.setup()`,
+not this migration) — this table only tracks enough to create/look up a loop,
+scope it by org/project, and let the cron tick find loops still in flight.
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+revision = "076"
+down_revision = "075"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "evaluation_iteration_run",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("dataset_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "experiment_name",
+            sqlmodel.sql.sqltypes.AutoString(length=255),
+            nullable=False,
+        ),
+        sa.Column("config_id", sa.Uuid(), nullable=False),
+        sa.Column("initial_config_version", sa.Integer(), nullable=False),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "processing",
+                "completed",
+                "failed",
+                name="evaluationiterationstatusenum",
+            ),
+            nullable=False,
+            comment="Loop bookkeeping status: processing, completed, or failed",
+        ),
+        sa.Column(
+            "stop_reason",
+            sqlmodel.sql.sqltypes.AutoString(),
+            nullable=True,
+            comment="Copied from the final graph state once terminal: ceiling_reached, max_rounds_reached, or round_failed",
+        ),
+        sa.Column(
+            "callback_url",
+            sqlmodel.sql.sqltypes.AutoString(),
+            nullable=False,
+            comment="HTTPS webhook validated via validate_callback_url before create",
+        ),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column("organization_id", sa.Integer(), nullable=False),
+        sa.Column("project_id", sa.Integer(), nullable=False),
+        sa.Column("inserted_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["dataset_id"],
+            ["evaluation_dataset.id"],
+            name="fk_evaluation_iteration_run_dataset_id",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["config_id"],
+            ["config.id"],
+            name="fk_evaluation_iteration_run_config_id",
+            ondelete="RESTRICT",
+        ),
+        sa.ForeignKeyConstraint(
+            ["organization_id"], ["organization.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_evaluation_iteration_run_dataset_id"),
+        "evaluation_iteration_run",
+        ["dataset_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_evaluation_iteration_run_config_id"),
+        "evaluation_iteration_run",
+        ["config_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_evaluation_iteration_run_organization_id"),
+        "evaluation_iteration_run",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_evaluation_iteration_run_project_id"),
+        "evaluation_iteration_run",
+        ["project_id"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index(
+        op.f("ix_evaluation_iteration_run_project_id"),
+        table_name="evaluation_iteration_run",
+    )
+    op.drop_index(
+        op.f("ix_evaluation_iteration_run_organization_id"),
+        table_name="evaluation_iteration_run",
+    )
+    op.drop_index(
+        op.f("ix_evaluation_iteration_run_config_id"),
+        table_name="evaluation_iteration_run",
+    )
+    op.drop_index(
+        op.f("ix_evaluation_iteration_run_dataset_id"),
+        table_name="evaluation_iteration_run",
+    )
+    op.drop_table("evaluation_iteration_run")
+    sa.Enum(name="evaluationiterationstatusenum").drop(op.get_bind(), checkfirst=True)

backend/app/api/main.py

@@ -41,6 +41,9 @@
     router as evaluations_dataset_v2_router,
 )
 from app.api.routes.evaluations.evaluation_v2 import router as evaluations_v2_router
+from app.api.routes.evaluations.iteration_v2 import (
+    router as evaluations_iteration_v2_router,
+)
 from app.api.routes.evaluations.prompt_improvement_v2 import (
     router as evaluations_prompt_improvement_v2_router,
 )
@@ -89,3 +92,4 @@
 api_v2_router.include_router(evaluations_v2_router)
 api_v2_router.include_router(evaluations_dataset_v2_router)
 api_v2_router.include_router(evaluations_prompt_improvement_v2_router)
+api_v2_router.include_router(evaluations_iteration_v2_router)

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

@@ -0,0 +1,68 @@
+"""v2 evaluation iteration loop trigger — chains eval -> improve-prompt -> eval."""
+
+import logging
+
+from asgi_correlation_id import correlation_id
+from fastapi import APIRouter, Depends, HTTPException
+
+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_iteration import (
+    EvaluationIterationCreateRequest,
+    EvaluationIterationRunImmediatePublic,
+)
+from app.services.evaluations.iteration import validate_and_start_evaluation_iteration
+from app.utils import APIResponse, load_description, validate_callback_url
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/evaluations", tags=["Evaluation v2"])
+
+
+@router.post(
+    "/iterations",
+    description=load_description("evaluation/create_evaluation_iteration_v2.md"),
+    response_model=APIResponse[EvaluationIterationRunImmediatePublic],
+    status_code=202,
+    dependencies=[
+        Depends(require_permission(Permission.REQUIRE_PROJECT)),
+        Depends(monitor_rate("evaluations")),
+    ],
+)
+def create_evaluation_iteration_v2(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    request: EvaluationIterationCreateRequest,
+) -> APIResponse[EvaluationIterationRunImmediatePublic]:
+    """Kick off a self-driving eval -> improve-prompt -> eval loop."""
+    try:
+        validate_callback_url(str(request.callback_url))
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=f"invalid_callback_url: {exc}")
+
+    iteration_run = validate_and_start_evaluation_iteration(
+        session=session,
+        dataset_id=request.dataset_id,
+        experiment_name=request.experiment_name,
+        config_id=request.config_id,
+        config_version=request.config_version,
+        max_rounds=request.max_rounds,
+        callback_url=str(request.callback_url),
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+        trace_id=correlation_id.get() or "N/A",
+    )
+
+    return APIResponse.success_response(
+        data=EvaluationIterationRunImmediatePublic(
+            iteration_run_id=iteration_run.id,
+            status=iteration_run.status,
+            message=(
+                "Evaluation iteration loop is running; the round-by-round report "
+                "will be delivered to your callback_url."
+            ),
+            inserted_at=iteration_run.inserted_at,
+            updated_at=iteration_run.updated_at,
+        )
+    )

backend/app/celery/tasks/job_execution.py

@@ -4,7 +4,8 @@
 ordered by the per-task `priority`:
 
     9  LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job)
-    6  Fast evaluation (run_evaluation_fast_chunk, run_evaluation_fast_aggregate)
+    6  Fast evaluation (run_evaluation_fast_chunk, run_evaluation_fast_aggregate,
+       run_prompt_improvement, run_evaluation_iteration_graph_step)
     2  Everything else (doctransform, collections, STT/TTS evaluation, assessment)
     1  Notifications (send_eval_completion_notification)
 
@@ -425,6 +426,52 @@ def run_evaluation_fast_aggregate(
     )
 
 
+# Priority 6 (fast-eval tier): same band as run_prompt_improvement — one graph
+# step per invocation is either a cheap status re-check (re-interrupts) or the
+# same-cost work run_prompt_improvement/run_evaluation_fast_* already does.
+@celery_app.task(bind=True, queue="default", priority=6)
+@gevent_timeout(
+    settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_iteration_graph_step"
+)
+def run_evaluation_iteration_graph_step(
+    self,
+    iteration_run_id: int,
+    resume: bool,
+    organization_id: int,
+    project_id: int,
+    trace_id: str = DEFAULT_TRACE_ID,
+    **kwargs,
+):
+    """Advance one step of an evaluation-iteration LangGraph loop.
+
+    `resume=False` (kickoff) invokes the graph with a fresh initial state built
+    from `kwargs` (`max_rounds`, `config_version`) + the thin tracking row;
+    `resume=True` (cron tick) sends `Command(resume=True)` to the persisted
+    checkpoint. Either call returns once the graph re-interrupts or reaches
+    `finalize_node` — see `execute_evaluation_iteration_graph_step`.
+    """
+    from app.services.evaluations.iteration_graph import (
+        execute_evaluation_iteration_graph_step,
+    )
+
+    _set_trace(trace_id)
+    logger.info(
+        f"[run_evaluation_iteration_graph_step] Starting | "
+        f"iteration_run_id={iteration_run_id} | resume={resume} | "
+        f"task_id={current_task.request.id}"
+    )
+    return _run_with_otel_parent(
+        self,
+        lambda: execute_evaluation_iteration_graph_step(
+            iteration_run_id=iteration_run_id,
+            resume=resume,
+            organization_id=organization_id,
+            project_id=project_id,
+            **kwargs,
+        ),
+    )
+
+
 @celery_app.task(bind=True, queue="default", priority=1)
 @gevent_timeout(
     settings.CELERY_TASK_SOFT_TIME_LIMIT, "send_eval_completion_notification"

backend/app/celery/utils.py

@@ -305,6 +305,38 @@ def start_fast_evaluation_aggregate(eval_run_id: int, trace_id: str = "N/A") ->
     return task_id
 
 
+def start_evaluation_iteration_round(
+    iteration_run_id: int,
+    resume: bool,
+    organization_id: int,
+    project_id: int,
+    trace_id: str = "N/A",
+    **kwargs: Any,
+) -> str:
+    """Enqueue one graph step (kickoff or resume) for an evaluation iteration loop.
+
+    `**kwargs` (`max_rounds`, `config_version`) is only meaningful on the
+    `resume=False` kickoff call, to seed the graph's initial state — a resume
+    call ignores them since the checkpoint already carries that state.
+    """
+    from app.celery.tasks.job_execution import run_evaluation_iteration_graph_step
+
+    task_id = _enqueue_with_trace_context(
+        run_evaluation_iteration_graph_step,
+        iteration_run_id=iteration_run_id,
+        resume=resume,
+        organization_id=organization_id,
+        project_id=project_id,
+        trace_id=trace_id,
+        **kwargs,
+    )
+    logger.info(
+        f"[start_evaluation_iteration_round] Enqueued | "
+        f"iteration_run_id={iteration_run_id} | resume={resume} | task_id={task_id}"
+    )
+    return task_id
+
+
 def get_task_status(task_id: str) -> Dict[str, Any]:
     result = AsyncResult(task_id)
     return {

backend/app/core/config.py

@@ -211,6 +211,15 @@ def AWS_S3_BUCKET(self) -> str:
     # task well under CELERY_TASK_SOFT_TIME_LIMIT.
     EVAL_FAST_CHUNK_SIZE: int = 50
 
+    # Evaluation iteration loop (eval -> improve-prompt -> eval, LangGraph-orchestrated).
+    EVAL_ITERATION_MAX_ROUNDS_DEFAULT: int = 10
+    EVAL_ITERATION_MAX_ROUNDS_HARD_CAP: int = 25
+    # Absolute (0-1 scale) stop-score delta below which a round counts as "no
+    # meaningful improvement" toward the ceiling-reached stop condition.
+    EVAL_ITERATION_CEILING_DELTA_THRESHOLD: float = 0.05
+    # Consecutive low-delta rounds required before the loop stops as ceiling_reached.
+    EVAL_ITERATION_CEILING_CONSECUTIVE_ROUNDS: int = 3
+
     EVAL_JUDGE_MODEL: str = "gpt-5-mini"
 
     # One of: none | minimal | low | medium | high | xhigh.

backend/app/crud/evaluations/__init__.py

@@ -33,6 +33,12 @@
     run_fast_evaluation,
     run_response_chunk,
 )
+from app.crud.evaluations.iteration import (
+    create_evaluation_iteration_run,
+    get_evaluation_iteration_run_by_id,
+    list_processing_evaluation_iteration_runs,
+    update_evaluation_iteration_run,
+)
 from app.crud.evaluations.langfuse import (
     create_langfuse_dataset_run,
     fetch_trace_scores_from_langfuse,
@@ -84,6 +90,11 @@
     "list_response_chunk_jobs",
     "run_fast_evaluation",
     "run_response_chunk",
+    # Iteration loop
+    "create_evaluation_iteration_run",
+    "get_evaluation_iteration_run_by_id",
+    "list_processing_evaluation_iteration_runs",
+    "update_evaluation_iteration_run",
     # Processing
     "check_and_process_evaluation",
     "poll_all_pending_evaluations",