From 897e466e1bad28faed202cf8d1a0cc1a2061b2a5 Mon Sep 17 00:00:00 2001 From: AkhileshNegi Date: Wed, 12 Aug 2026 10:40:08 +0530 Subject: [PATCH] added new queue for this --- .claude/conventions/celery.md | 36 +++++++++++++++-------- .env.example | 3 ++ backend/app/celery/celery_app.py | 17 +++++++++-- backend/app/celery/tasks/job_execution.py | 28 +++++++++++------- docker-compose.dev.yml | 18 +++++++++++- docker-compose.staging.yml | 25 +++++++++++++++- docker-compose.yml | 18 +++++++++++- docs/wiki/services.md | 4 +-- 8 files changed, 118 insertions(+), 31 deletions(-) diff --git a/.claude/conventions/celery.md b/.claude/conventions/celery.md index f83413f0a..d7feaa6cb 100644 --- a/.claude/conventions/celery.md +++ b/.claude/conventions/celery.md @@ -1,10 +1,14 @@ # Celery task conventions (`app/celery/tasks/`) Authoritative conventions for Celery tasks in kaapi-backend. Tasks live in `app/celery/tasks/`. -Celery uses RabbitMQ as broker. There is a **single `default` queue** declared with -`x-max-priority=10`; tasks are ordered by a per-task `priority` (higher drains first, FIFO within -a band). Read `app/celery/tasks/job_execution.py` before writing — it shows the full pattern -(decorator + timeout + OTel propagation + delegation to a service). +Celery uses RabbitMQ as broker. There are **two queues**, each declared with `x-max-priority=10`: +`default` (everything except fast-eval burst tasks) and `evaluations` (only +`run_evaluation_fast_chunk` / `run_evaluation_fast_aggregate` — physically isolated on their own +worker pool so a fast-eval burst can't occupy `default`'s worker slots and delay higher-priority +LLM jobs; priority only reorders queued-not-yet-claimed messages, it can't preempt a task a +worker already started). Within each queue, tasks are ordered by a per-task `priority` (higher +drains first, FIFO within a band). Read `app/celery/tasks/job_execution.py` before writing — it +shows the full pattern (decorator + timeout + OTel propagation + delegation to a service). ## Canonical decorator stack @@ -30,17 +34,23 @@ def run_my_job(self, project_id: int, job_id: str, trace_id: str, **kwargs): `_set_trace`, `_run_with_otel_parent`, and `gevent_timeout` already exist in this module / `app/celery/utils.py` — reuse them, don't reinvent. -## Priority choice — be explicit +## Queue and priority choice — be explicit -All tasks share the one `default` queue; set `priority` to place the task in the right band. Match +Only `run_evaluation_fast_chunk` and `run_evaluation_fast_aggregate` use `queue="evaluations"` +(fast-eval burst fan-out — many chunk tasks fired at once). Everything else uses +`queue="default"`. Don't move other eval-adjacent tasks (e.g. `run_evaluation_batch_submission`, +`send_eval_completion_notification`) onto `evaluations` — they're one-shot/fire-and-forget, no +burstiness to isolate. + +Within whichever queue the task belongs to, set `priority` to place it in the right band. Match the bands already in use (see the `job_execution.py` module docstring): -| Priority | When | -|---|---| -| `9` | User-blocking, interactive — LLM call / chain / response jobs | -| `6` | Fast evaluation | -| `2` | Default — doctransform, collections, STT/TTS evaluation, assessment | -| `1` | Notifications and other fire-and-forget background work | +| Priority | Queue | When | +|---|---|---| +| `9` | `default` | User-blocking, interactive — LLM call / chain / response jobs | +| `6` | `evaluations` | Fast evaluation chunk/aggregate | +| `2` | `default` | Default — doctransform, collections, STT/TTS evaluation, assessment | +| `1` | `default` | Notifications and other fire-and-forget background work | Pick the band that matches the task's user-facing urgency; document the choice in a comment if it's not obvious. `task_inherit_parent_priority=True` is set, so a task enqueued from another task @@ -49,7 +59,7 @@ inherits its priority unless you override it. ## Hard rules - **`bind=True`** so you have `self` (the task instance) for retries, IDs, etc. -- **Always pass `queue="default"` and an explicit `priority`** — there is only one queue; the priority is what matters. +- **Always pass an explicit `queue` (`"default"` or `"evaluations"`) and `priority`.** Only fast-eval chunk/aggregate tasks use `"evaluations"`; everything else uses `"default"`. - **Pass `trace_id` explicitly** as a parameter and call `_set_trace(trace_id)` first thing. This wires `asgi_correlation_id` so logs from inside the task match the originating request. - **Wrap the work in `_run_with_otel_parent(self, lambda: ...)`** so OpenTelemetry parent context propagates from the enqueueing process. - **Delegate to a service.** The task body should be a thin shim over `app/services//`. No DB queries, no external HTTP, no business logic inside the task itself. diff --git a/.env.example b/.env.example index 5a041fe26..446e5476c 100644 --- a/.env.example +++ b/.env.example @@ -71,6 +71,9 @@ REDIS_PASSWORD= # Celery Configuration # Leave CELERY_WORKER_CONCURRENCY empty to auto-detect CPU cores, or set to specific number (e.g., number_of_cores * 2) CELERY_WORKER_CONCURRENCY= +# Dedicated pool for the "evaluations" queue (fast-eval chunk/aggregate tasks), +# isolated from CELERY_WORKER_CONCURRENCY so an eval burst can't starve LLM-call tasks. +CELERY_EVAL_WORKER_CONCURRENCY=2 CELERY_WORKER_MAX_TASKS_PER_CHILD=1000 CELERY_WORKER_MAX_MEMORY_PER_CHILD=200000 CELERY_TASK_SOFT_TIME_LIMIT=300 diff --git a/backend/app/celery/celery_app.py b/backend/app/celery/celery_app.py index e0ddf8dcd..ec5d505f6 100644 --- a/backend/app/celery/celery_app.py +++ b/backend/app/celery/celery_app.py @@ -167,6 +167,9 @@ def initialize_worker(**_) -> None: ) # Define exchanges and queues with priority +DEFAULT_QUEUE = "default" +EVALUATIONS_QUEUE = "evaluations" + default_exchange = Exchange("default", type="direct") # Celery configuration using environment variables @@ -174,13 +177,21 @@ def initialize_worker(**_) -> None: # Queue configuration with priority support task_queues=( Queue( - "default", + DEFAULT_QUEUE, + exchange=default_exchange, + routing_key=DEFAULT_QUEUE, + queue_arguments={"x-max-priority": 10}, + ), + # Routing key must differ per queue on the shared direct exchange, or + # messages fan out to both. + Queue( + EVALUATIONS_QUEUE, exchange=default_exchange, - routing_key="default", + routing_key=EVALUATIONS_QUEUE, queue_arguments={"x-max-priority": 10}, ), ), - task_default_queue="default", + task_default_queue=DEFAULT_QUEUE, # Enable priority support task_inherit_parent_priority=True, worker_prefetch_multiplier=settings.CELERY_WORKER_PREFETCH_MULTIPLIER, diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py index 000c6025a..b55703074 100644 --- a/backend/app/celery/tasks/job_execution.py +++ b/backend/app/celery/tasks/job_execution.py @@ -1,14 +1,22 @@ -"""Celery task definitions for the single priority `default` queue. +"""Celery task definitions, split across two priority queues. -All tasks share one queue (`default`, declared with `x-max-priority=10`) and are -ordered by the per-task `priority`: +`default` (every task below except the two fast-eval ones), 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) + 9 LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job, + run_guardrails_job) + 6 Prompt improvement (run_prompt_improvement) 2 Everything else (doctransform, collections, STT/TTS evaluation, assessment) 1 Notifications (send_eval_completion_notification) -Higher priority drains first; within the same priority, delivery is FIFO. +`evaluations` (priority 6): run_evaluation_fast_chunk and +run_evaluation_fast_aggregate only. A fast eval fans out into many chunk tasks +at once; priority only reorders messages still waiting in a queue, so on a +shared queue that burst occupies every worker slot and delays priority-9 LLM +jobs. A separate queue with its own worker pool makes the isolation physical. + +Both queues are declared with `x-max-priority=10`. Higher priority drains first; +within the same priority, delivery is FIFO. """ import logging @@ -19,7 +27,7 @@ from opentelemetry import trace from opentelemetry.propagate import extract -from app.celery.celery_app import celery_app +from app.celery.celery_app import EVALUATIONS_QUEUE, celery_app from app.celery.utils import gevent_timeout from app.core.config import settings @@ -246,7 +254,7 @@ def run_evaluation_batch_submission( ) -# Priority 6 (fast-eval tier): user-blocking interactive prompt iteration in the +# Priority 6: user-blocking interactive prompt iteration in the # evaluation domain, above default batch work but below core LLM call/chain jobs. @celery_app.task(bind=True, queue="default", priority=6) @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_prompt_improvement") @@ -370,7 +378,7 @@ def run_tts_result_processing( ) -@celery_app.task(bind=True, queue="default", priority=6) +@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=6) @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_chunk") def run_evaluation_fast_chunk( self: Task, @@ -400,7 +408,7 @@ def run_evaluation_fast_chunk( ) -@celery_app.task(bind=True, queue="default", priority=6) +@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=6) @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_aggregate") def run_evaluation_fast_aggregate( self: Task, eval_run_id: int, trace_id: str = DEFAULT_TRACE_ID diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index f9e3b3e1c..d87e75d4f 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -57,4 +57,20 @@ services: required: true - path: .env.secrets required: false - command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"] + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"] + + celery_worker_eval: + image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}" + container_name: celery-worker-eval + restart: always + build: + context: ./backend + depends_on: + backend: + condition: service_healthy + env_file: + - path: .env + required: true + - path: .env.secrets + required: false + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"] diff --git a/docker-compose.staging.yml b/docker-compose.staging.yml index bbcece6bb..edc7ca7de 100644 --- a/docker-compose.staging.yml +++ b/docker-compose.staging.yml @@ -66,7 +66,7 @@ services: required: true - path: .env.secrets required: false - command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"] + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"] logging: driver: awslogs options: @@ -74,3 +74,26 @@ services: awslogs-group: /ec2/kaapi-staging awslogs-stream: celery-worker awslogs-create-group: "true" + + celery_worker_eval: + image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}" + container_name: celery-worker-eval + restart: always + build: + context: ./backend + depends_on: + backend: + condition: service_healthy + env_file: + - path: .env + required: true + - path: .env.secrets + required: false + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"] + logging: + driver: awslogs + options: + awslogs-region: ap-south-1 + awslogs-group: /ec2/kaapi-staging + awslogs-stream: celery-worker-eval + awslogs-create-group: "true" diff --git a/docker-compose.yml b/docker-compose.yml index 5d50e205b..5cdaf63d6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -146,7 +146,23 @@ services: environment: REDIS_HOST: redis RABBITMQ_HOST: rabbitmq - command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"] + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"] + + celery_worker_eval: + image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}" + container_name: celery-worker-eval + restart: always + build: + context: ./backend + depends_on: + backend: + condition: service_healthy + env_file: + - .env + environment: + REDIS_HOST: redis + RABBITMQ_HOST: rabbitmq + command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"] celery_flower: image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}" diff --git a/docs/wiki/services.md b/docs/wiki/services.md index 09e9c936b..06bf45492 100644 --- a/docs/wiki/services.md +++ b/docs/wiki/services.md @@ -10,7 +10,7 @@ FastAPI (backend/app/main.py, api/routes/*) │ ├── Postgres (SQLModel, core/db.py, alembic migrations) ├── Celery workers ── RabbitMQ (broker) + Redis (results) - │ └── app/celery/tasks/job_execution.py, priority queues + │ └── app/celery/tasks/job_execution.py, 2 queues (default, evaluations) + priority bands ├── Object storage core/cloud/storage.py ├── Langfuse core/langfuse/langfuse.py (traces, scores) ├── Sentry core/sentry_filters.py @@ -25,7 +25,7 @@ FastAPI (backend/app/main.py, api/routes/*) | Process | Entry | Notes | |---|---|---| | API server | `fastapi run app/main.py` | routes in `app/api/routes/` | -| Celery worker | `app/celery/` config | async jobs, priority queues | +| Celery worker | `app/celery/` config | `default` queue (most tasks) + dedicated `evaluations` queue/pool (fast-eval chunk/aggregate, isolated so eval bursts can't delay LLM jobs), priority bands within each | | Celery beat / cron | `app/api/routes/cron.py` + `crud/evaluations/cron.py` | batch polling for eval/assessment runs | ## Environment