0Pricing
AI Agents · Lección

Flujos de trabajo asíncronos y trabajos en segundo plano

Los agentes de larga duración necesitan colas —Celery, RQ o Temporal—: devuelva un ID de trabajo y consulte su estado.

Flujos de trabajo asíncronos y trabajos en segundo plano es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

Cuando el ciclo de solicitud es demasiado lento

Algunas tareas de los agentes tardan 30 s, 5 min o varias horas. No puede mantener abierta la conexión HTTP. Trasládelas a trabajos asíncronos en segundo plano.

El patrón

  1. El cliente envía un POST con la tarea -> el servidor crea el trabajo y devuelve job_id
  2. El trabajador recoge el trabajo de una cola
  3. El cliente consulta GET /jobs/{id} para obtener el estado
  4. Cuando termina, el servidor devuelve el resultado

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')
    }

Trabajador

Un proceso independiente consume la cola:

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)})

Opciones de colas

  • RQ (Redis Queue) — Python minimalista
  • Celery — clásico de Python, con muchas funciones
  • Temporal — flujos de trabajo duraderos, reintentos y observabilidad
  • Dramatiq — alternativa a Celery
  • Nativo de la nube — Cloud Tasks, SQS, Pub/Sub

Temporal para flujos de trabajo duraderos

Los flujos de trabajo de los agentes mantienen su estado. El modelo de ejecución duradera de Temporal encaja muy bien:

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)

Actualizaciones parciales en streaming

Los trabajos de larga duración se benefician de las actualizaciones de progreso. Use Server-Sent Events o 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')

TTL de los trabajos

No conserve los registros de los trabajos para siempre:

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

Reintentos

Los errores transitorios se reintentan automáticamente; los errores permanentes se envían a una DLQ para que una persona los revise:

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'))

Límites de concurrencia

Para controlar los costes de GPU y de las API, limite el número de trabajadores concurrentes:

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

# Or per-queue concurrency in Temporal worker config.

Cuotas por usuario

Realice un seguimiento de los trabajos en curso de cada usuario:

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

Observabilidad

Trace cada trabajo de principio a fin. Etiquete los spans con job_id y user_id. Los trabajos fallidos deberían abrir automáticamente un ticket o generar una alerta.

Patrón de consulta de estado

¿Por qué utilizar el patrón de envío y consulta de estado para agentes de larga duración?

Resumen

Envíe el trabajo mediante POST, devuelva el id, haga que el trabajador lo procese desde la cola y permita que el cliente consulte su estado. RQ/Celery para casos sencillos; Temporal para flujos de trabajo con estado. TTL, reintentos, cuotas y trazas.

Preguntas frecuentes

¿La lección «Flujos de trabajo asíncronos y trabajos en segundo plano» es gratis?

Sí — el texto completo de «Flujos de trabajo asíncronos y trabajos en segundo plano» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.

¿Qué aprenderé en «Flujos de trabajo asíncronos y trabajos en segundo plano»?

Los agentes de larga duración necesitan colas —Celery, RQ o Temporal—: devuelva un ID de trabajo y consulte su estado. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Agents?

No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.

¿Cuánto tiempo toma la lección «Flujos de trabajo asíncronos y trabajos en segundo plano»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Agents?

Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Servir agentes detrás de una API
  2. Flujos de trabajo asíncronos y trabajos en segundo plano
  3. Limitación de velocidad y gestión de cuotas
  4. Despliegues blue-green y canary para agentes
← Volver a AI Agents