Celery

# Install Celery with Redis support
pip install celery[redis] django-celery-results

# celery_app.py (project root)
import os
from celery import Celery

os.environ.setdefault(
    'DJANGO_SETTINGS_MODULE', 'agentapi.settings'
)

app = Celery('agentapi')
app.config_from_object('django.conf:settings',
                        namespace='CELERY')
app.autodiscover_tasks()  # auto-finds tasks.py in each app

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

# Start a worker (development)
celery -A agentapi worker --loglevel=info

# Start with concurrency control
celery -A agentapi worker \
    --concurrency=8 \         # 8 child processes
    --loglevel=info \
    --queues=default,agents   # listen to specific queues

# Start the beat scheduler (periodic tasks)
celery -A agentapi beat --loglevel=info

# Start Flower monitoring (web UI)
pip install flower
celery -A agentapi flower --port=5555

# Production: run as systemd service
# /etc/systemd/system/celery-worker.service
# ExecStart=/app/venv/bin/celery -A agentapi worker
# Restart=always

# agents/tasks.py
from celery import shared_task
import httpx

@shared_task(bind=True, max_retries=3)
def run_agent_inference(self, agent_id, prompt):
    """Call LLM API asynchronously."""
    try:
        response = httpx.post(
            "https://api.openai.com/v1/chat/completions",
            json={"model": "gpt-4", "messages": [
                {"role": "user", "content": prompt}
            ]},
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=60,
        )
        return response.json()
    except httpx.TimeoutException as exc:
        raise self.retry(exc=exc, countdown=10)

# Calling the task from a view
result = run_agent_inference.delay(agent_id=42,
                                   prompt="Explain K8s")
print(result.id)      # task UUID
print(result.status)  # PENDING → STARTED → SUCCESS

# Chain: sequential pipeline
from celery import chain

pipeline = chain(
    fetch_document.s(doc_id),      # step 1
    extract_embeddings.s(),         # step 2 (gets step 1 result)
    store_in_vectordb.s(),          # step 3
)
pipeline.apply_async()

# Group: parallel fan-out
from celery import group

batch = group(
    run_agent_inference.s(id, prompt)
    for id, prompt in agent_prompts
)
results = batch.apply_async()  # runs all in parallel

# Chord: parallel + callback when all done
from celery import chord

chord(
    [embed_chunk.s(chunk) for chunk in chunks],  # parallel
    merge_embeddings.s()  # callback with all results
).apply_async()

# Automatic retry with exponential backoff
@shared_task(
    bind=True,
    autoretry_for=(httpx.TimeoutException,),
    retry_backoff=True,       # exponential: 1s, 2s, 4s...
    retry_backoff_max=300,    # cap at 5 minutes
    retry_jitter=True,        # add randomness
    max_retries=5,
)
def call_llm(self, prompt):
    response = httpx.post(LLM_URL, json={"prompt": prompt},
                          timeout=30)
    response.raise_for_status()
    return response.json()

# Time limits (prevent stuck workers)
@shared_task(
    time_limit=120,       # hard kill after 120s
    soft_time_limit=90,   # SoftTimeLimitExceeded at 90s
)
def long_running_task():
    try:
        do_work()
    except SoftTimeLimitExceeded:
        save_progress()  # graceful cleanup

# settings.py — schedule periodic tasks
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-stale-agents': {
        'task': 'agents.tasks.cleanup_inactive',
        'schedule': crontab(hour=2, minute=0),  # 2 AM daily
    },
    'refresh-embeddings': {
        'task': 'agents.tasks.refresh_embeddings',
        'schedule': 3600.0,  # every hour (seconds)
    },
    'health-check': {
        'task': 'agents.tasks.health_ping',
        'schedule': 60.0,  # every minute
        'options': {'expires': 30},  # skip if > 30s late
    },
}

# Or use django-celery-beat for DB-backed
# schedules editable via Django Admin UI
pip install django-celery-beat
# Add 'django_celery_beat' to INSTALLED_APPS
# Run: celery -A agentapi beat -l info \
#      --scheduler django_celery_beat.schedulers:DatabaseScheduler