0Pricing
AI Agents · Leçon

Flux de travail asynchrones et tâches en arrière-plan

Les agents de longue durée ont besoin de files d’attente (Celery, RQ, Temporal) : renvoyez un identifiant de tâche et interrogez son état.

Flux de travail asynchrones et tâches en arrière-plan est une leçon AI Agents gratuite sur CoddyKit. Ceci est la leçon 2 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage AI Agents, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours AI Agents comprend 4 leçons au total.

Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.

Quand le cycle de requête est trop lent

Certaines tâches d’agent prennent 30 s, 5 min ou des heures. Vous ne pouvez pas maintenir la connexion HTTP ouverte. Déplacez-les vers des tâches asynchrones en arrière-plan.

Le modèle

  1. Le client envoie la tâche via POST -> le serveur crée la tâche et renvoie job_id
  2. Le processus de traitement récupère la tâche dans une file d’attente
  3. Le client interroge GET /jobs/{id} pour connaître l’état
  4. Une fois la tâche terminée, le serveur renvoie le résultat

Submit Endpoint

from uuid import uuid4

@app.post('/jobs')
def submit(req: AgentRequest):
    job_id = str(uuid4())
    redis.hset(f'job:{job_id}', mapping={'status': 'queued', 'user_id': req.user_id})
    queue.enqueue('run_agent_job', job_id, req.query)
    return {'job_id': job_id, 'status': 'queued'}

Status Endpoint

@app.get('/jobs/{job_id}')
def status(job_id: str):
    data = redis.hgetall(f'job:{job_id}')
    if not data:
        raise HTTPException(404)
    return {
        'status': data['status'],
        'result': data.get('result'),
        'error': data.get('error')
    }

Processus de traitement

Un processus distinct consomme la file d’attente :

def run_agent_job(job_id, query):
    redis.hset(f'job:{job_id}', 'status', 'running')
    try:
        result = run_agent(query)
        redis.hset(f'job:{job_id}', mapping={'status': 'done', 'result': result})
    except Exception as e:
        redis.hset(f'job:{job_id}', mapping={'status': 'failed', 'error': str(e)})

Choix de files d’attente

  • RQ (Redis Queue) — Python minimaliste
  • Celery — solution Python classique, avec de nombreuses fonctionnalités
  • Temporal — flux de travail durables, nouvelles tentatives, observabilité
  • Dramatiq — alternative à Celery
  • Natif du cloud — Cloud Tasks, SQS, Pub/Sub

Temporal pour les flux de travail durables

Les flux de travail d’agent conservent un état. Le modèle d’exécution durable de Temporal convient parfaitement :

import temporalio

@temporalio.workflow.defn
class AgentWorkflow:
    @temporalio.workflow.run
    async def run(self, query: str) -> str:
        plan = await workflow.execute_activity(plan_step, query, schedule_to_close_timeout=timedelta(minutes=2))
        results = await workflow.execute_activity(execute_step, plan, schedule_to_close_timeout=timedelta(minutes=10))
        return await workflow.execute_activity(synthesise_step, results)

Diffusion des mises à jour partielles

Les tâches de longue durée bénéficient de mises à jour de progression. Utilisez les événements envoyés par le serveur ou WebSockets :

@app.get('/jobs/{job_id}/stream')
def stream_progress(job_id):
    def gen():
        while True:
            update = redis.brpop(f'updates:{job_id}', timeout=30)
            if not update:
                yield 'data: {"status": "timeout"}\n\n'
                break
            yield f'data: {update[1].decode()}\n\n'
            if 'done' in update[1].decode():
                break
    return StreamingResponse(gen(), media_type='text/event-stream')

Durées d’expiration des tâches

Ne conservez pas les enregistrements de tâches indéfiniment :

redis.expire(f'job:{job_id}', 86400)   # 1 day

Nouvelles tentatives

Les échecs transitoires font automatiquement l’objet de nouvelles tentatives ; les échecs permanents sont envoyés vers une DLQ pour examen humain :

import time

def retry(retries=3, retry_backoff=True):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(1, retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f'attempt {attempt} failed: {e}')
                    if attempt == retries:
                        raise
            return None
        return wrapper
    return decorator

attempts = {'n': 0}

@retry(retries=3, retry_backoff=True)
def run_agent_job(job_id):
    attempts['n'] += 1
    if attempts['n'] < 3:
        raise RuntimeError('transient error')
    return f'job {job_id} done'

print(run_agent_job('job-1'))

Limites de concurrence

Pour contrôler les coûts liés au GPU et aux API, limitez le nombre de processus de traitement simultanés :

rq worker --burst --max-jobs 1000 --queue agent

# Or per-queue concurrency in Temporal worker config.

Quotas par utilisateur

Suivez les tâches en cours par utilisateur :

key = f'inflight:{user_id}'
if redis.scard(key) >= 5:
    raise HTTPException(429, 'Too many in-flight jobs')
redis.sadd(key, job_id)

Observabilité

Suivez chaque tâche de bout en bout. Étiquetez les segments avec job_id et user_id. Les tâches en échec devraient automatiquement ouvrir un ticket ou déclencher une alerte.

Modèle d’interrogation de l’état

Pourquoi utiliser le modèle soumission/interrogation de l’état pour les agents de longue durée ?

Récapitulatif

Envoyez une tâche via POST, renvoyez son identifiant, le processus la traite depuis la file d’attente, puis le client interroge l’état. RQ/Celery pour les cas simples ; Temporal pour les flux de travail avec état. Durées d’expiration, nouvelles tentatives, quotas, traces.

Questions Fréquemment Posées

La leçon « Flux de travail asynchrones et tâches en arrière-plan » est-elle gratuite ?

Oui — le texte complet de « Flux de travail asynchrones et tâches en arrière-plan » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours AI Agents, passe à CoddyKit PRO. Le cours AI Agents comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Flux de travail asynchrones et tâches en arrière-plan » ?

Les agents de longue durée ont besoin de files d’attente (Celery, RQ, Temporal) : renvoyez un identifiant de tâche et interrogez son état. Tu pratiques AI Agents avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer AI Agents ?

Aucune expérience préalable n'est requise. AI Agents sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 2 sur 4.

Combien de temps prend la leçon « Flux de travail asynchrones et tâches en arrière-plan » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon AI Agents ?

Oui. Chaque leçon AI Agents inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Exposer des agents derrière une API
  2. Flux de travail asynchrones et tâches en arrière-plan
  3. Limitation du débit et gestion des quotas
  4. Déploiements blue-green et canary pour les agents
← Retour à AI Agents