Skip to content

Commit 799130c

Browse files
authored
Merge branch 'main' into feat/kms-integration
2 parents de69c23 + 102b61c commit 799130c

20 files changed

Lines changed: 1992 additions & 740 deletions

File tree

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,8 +55,8 @@ def evaluate(
5555
config_id: UUID = Body(..., description="Stored config ID"),
5656
config_version: int = Body(..., ge=1, description="Stored config version"),
5757
run_mode: RunModeEnum = Body(
58-
default=RunModeEnum.BATCH,
59-
description="Execution mode: 'batch' (default) or 'fast'",
58+
default=RunModeEnum.FAST,
59+
description="Execution mode: 'batch' or 'fast'. Omit to default to 'fast'.",
6060
),
6161
) -> APIResponse[EvaluationRunPublic]:
6262
"""Start an evaluation run."""

‎backend/app/celery/tasks/job_execution.py‎

Lines changed: 40 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
ordered by the per-task `priority`:
55
66
9 LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job)
7-
6 Fast evaluation (run_evaluation_fast)
7+
6 Fast evaluation (run_evaluation_fast_chunk, run_evaluation_fast_aggregate)
88
2 Everything else (doctransform, collections, STT/TTS evaluation, assessment)
99
1 Notifications (send_eval_completion_notification)
1010
@@ -352,32 +352,57 @@ def run_tts_result_processing(
352352

353353

354354
@celery_app.task(bind=True, queue="default", priority=6)
355-
@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast")
356-
def run_evaluation_fast(
357-
self: Task, eval_run_id: int, trace_id: str = DEFAULT_TRACE_ID
355+
@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_chunk")
356+
def run_evaluation_fast_chunk(
357+
self: Task,
358+
eval_run_id: int,
359+
chunk_index: int,
360+
trace_id: str = DEFAULT_TRACE_ID,
358361
) -> None:
359-
"""Run the fast evaluation pipeline for one EvaluationRun.
362+
"""Run one responses chunk of a fast EvaluationRun.
363+
364+
Idempotent: the chunk is skipped when its (eval_run, chunk_index) batch_job
365+
already has a raw_output_url, so Celery redelivery never re-charges OpenAI.
366+
"""
367+
from app.services.evaluations.fast import execute_fast_evaluation_chunk
368+
369+
_set_trace(trace_id)
370+
logger.info(
371+
f"[run_evaluation_fast_chunk] Starting fast eval chunk | "
372+
f"eval_run_id={eval_run_id} | chunk_index={chunk_index} | "
373+
f"task_id={current_task.request.id}"
374+
)
375+
376+
return _run_with_otel_parent(
377+
self,
378+
lambda: execute_fast_evaluation_chunk(
379+
eval_run_id=eval_run_id, chunk_index=chunk_index
380+
),
381+
)
360382

361-
Idempotency: each stage is skipped on retry when its `batch_job` marker is
362-
already set on the EvaluationRun, so Celery redelivery never re-calls
363-
OpenAI for work that already succeeded.
364383

365-
Args:
366-
eval_run_id: ID of the EvaluationRun (run_mode="fast").
367-
trace_id: Correlation id from the enqueueing request, propagated into
368-
the worker for log correlation.
384+
@celery_app.task(bind=True, queue="default", priority=6)
385+
@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_aggregate")
386+
def run_evaluation_fast_aggregate(
387+
self: Task, eval_run_id: int, trace_id: str = DEFAULT_TRACE_ID
388+
) -> None:
389+
"""Fan-in: merge the response chunks then run embeddings + scoring.
390+
391+
Enqueued once by the cron barrier after every chunk has completed.
392+
Idempotent: each stage is skipped on retry when its `batch_job` marker is
393+
set, so redelivery never re-does completed work.
369394
"""
370-
from app.services.evaluations.fast import execute_fast_evaluation
395+
from app.services.evaluations.fast import execute_fast_evaluation_aggregate
371396

372397
_set_trace(trace_id)
373398
logger.info(
374-
f"[run_evaluation_fast] Starting fast evaluation task | "
399+
f"[run_evaluation_fast_aggregate] Starting fast eval aggregate | "
375400
f"eval_run_id={eval_run_id} | task_id={current_task.request.id}"
376401
)
377402

378403
return _run_with_otel_parent(
379404
self,
380-
lambda: execute_fast_evaluation(eval_run_id=eval_run_id),
405+
lambda: execute_fast_evaluation_aggregate(eval_run_id=eval_run_id),
381406
)
382407

383408

‎backend/app/celery/utils.py‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -252,17 +252,36 @@ def start_tts_result_processing(
252252
return task_id
253253

254254

255-
def start_fast_evaluation(eval_run_id: int, trace_id: str = "N/A") -> str:
256-
"""Enqueue the run_evaluation_fast orchestrator task for one EvaluationRun."""
257-
from app.celery.tasks.job_execution import run_evaluation_fast
255+
def start_fast_evaluation_chunk(
256+
eval_run_id: int, chunk_index: int, trace_id: str = "N/A"
257+
) -> str:
258+
"""Enqueue one parallel responses-chunk task for a fast EvaluationRun."""
259+
from app.celery.tasks.job_execution import run_evaluation_fast_chunk
260+
261+
task_id = _enqueue_with_trace_context(
262+
run_evaluation_fast_chunk,
263+
eval_run_id=eval_run_id,
264+
chunk_index=chunk_index,
265+
trace_id=trace_id,
266+
)
267+
logger.info(
268+
f"[start_fast_evaluation_chunk] Enqueued fast eval chunk | "
269+
f"eval_run_id={eval_run_id} | chunk_index={chunk_index} | task_id={task_id}"
270+
)
271+
return task_id
272+
273+
274+
def start_fast_evaluation_aggregate(eval_run_id: int, trace_id: str = "N/A") -> str:
275+
"""Enqueue the fan-in aggregate task once a fast run's chunks are all done."""
276+
from app.celery.tasks.job_execution import run_evaluation_fast_aggregate
258277

259278
task_id = _enqueue_with_trace_context(
260-
run_evaluation_fast,
279+
run_evaluation_fast_aggregate,
261280
eval_run_id=eval_run_id,
262281
trace_id=trace_id,
263282
)
264283
logger.info(
265-
f"[start_fast_evaluation] Enqueued fast eval | "
284+
f"[start_fast_evaluation_aggregate] Enqueued fast eval aggregate | "
266285
f"eval_run_id={eval_run_id} | task_id={task_id}"
267286
)
268287
return task_id

‎backend/app/core/config.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,9 @@ def AWS_S3_BUCKET(self) -> str:
186186
LLM_PENDING_THRESHOLD_MINUTES: int = 30
187187
COLLECTION_PENDING_THRESHOLD_MINUTES: int = 30
188188
DOC_TRANSFORMATION_PENDING_THRESHOLD_MINUTES: int = 30
189+
# A fast run stuck in `processing` past this with no chunk progress is stalled;
190+
# the cron healer re-enqueues its missing chunk tasks.
191+
EVAL_FAST_STALL_THRESHOLD_MINUTES: int = 15
189192
PENDING_JOB_QUERY_TIMEOUT_MS: int = 1000
190193

191194
# AI-assisted prompt improvement settings.
@@ -197,11 +200,14 @@ def AWS_S3_BUCKET(self) -> str:
197200

198201
# Fast evaluation (run_mode="fast") configuration.
199202
# See "Fast Evaluation SRD.md" for the full design rationale.
200-
EVAL_FAST_MAX_UNIQUE_ROWS: int = 10
203+
EVAL_FAST_MAX_UNIQUE_ROWS: int = 100
201204
EVAL_FAST_FAILURE_THRESHOLD: float = 0.5
202205
# Capped at 4 by default: higher values (8-10) across multiple Celery
203206
# workers can cause memory pressure on smaller EC2 instances.
204207
EVAL_FAST_API_CONCURRENCY: int = 4
208+
# Items per responses chunk task; smaller = more parallel workers and each
209+
# task well under CELERY_TASK_SOFT_TIME_LIMIT.
210+
EVAL_FAST_CHUNK_SIZE: int = 50
205211

206212
@computed_field # type: ignore[prop-decorator]
207213
@property

0 commit comments

Comments
 (0)