0Pricing
AI Agents · Lezione

Workflow asincroni e processi in background

Gli agenti di lunga durata hanno bisogno di code, come Celery, RQ o Temporal: restituisca un job id e verifichi lo stato.

Workflow asincroni e processi in background è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.

Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.

Quando il ciclo della richiesta è troppo lento

Alcune attività degli agent richiedono 30 s, 5 min o ore. Non può mantenere aperta la connessione HTTP. Le sposti in job asincroni in background.

Il pattern

  1. Il client invia un POST per l'attività -> il server crea un job e restituisce job_id
  2. Un worker preleva il job da una coda
  3. Il client esegue il polling di GET /jobs/{id} per conoscere lo stato
  4. Al termine, il server restituisce il risultato

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

Worker

Un processo separato consuma la coda:

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

Scelte per la coda

  • RQ (Redis Queue) — Python minimalista
  • Celery — il classico di Python, ricco di funzionalità
  • Temporal — workflow durevoli, retry, osservabilità
  • Dramatiq — alternativa a Celery
  • Cloud-native — Cloud Tasks, SQS, Pub/Sub

Temporal per workflow durevoli

I workflow degli agent mantengono lo stato. Il modello di esecuzione durevole di Temporal è particolarmente adatto:

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)

Aggiornamenti parziali in streaming

I job di lunga durata traggono vantaggio dagli aggiornamenti sullo stato di avanzamento. Usi Server-Sent Events o WebSocket:

@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 dei job

Non conservi i record dei job per sempre:

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

Retry

Gli errori transitori vengono ritentati automaticamente; gli errori permanenti vengono inviati a una DLQ per la revisione umana:

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

Limiti di concorrenza

Per controllare i costi di GPU/API, limiti il numero di worker concorrenti:

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

# Or per-queue concurrency in Temporal worker config.

Quote per utente

Tenga traccia dei job in corso per ogni utente:

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

Osservabilità

Tracci ogni job dall'inizio alla fine. Aggiunga job_id e user_id agli span. I job falliti dovrebbero aprire automaticamente un ticket o generare un avviso.

Pattern di polling dello stato

Perché usare il pattern invio/polling dello stato per gli agent di lunga durata?

Riepilogo

Invii il job con POST, restituisca l'ID, faccia elaborare il job a un worker dalla coda e faccia eseguire il polling al client. RQ/Celery per i casi semplici; Temporal per i workflow con stato. TTL, retry, quote e tracing.

Domande Frequenti

La lezione «Workflow asincroni e processi in background» è gratuita?

Sì — il testo completo di «Workflow asincroni e processi in background» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.

Cosa imparerò in «Workflow asincroni e processi in background»?

Gli agenti di lunga durata hanno bisogno di code, come Celery, RQ o Temporal: restituisca un job id e verifichi lo stato. Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Agents?

Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.

Quanto tempo richiede la lezione «Workflow asincroni e processi in background»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Agents?

Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Servire gli agenti tramite un'API
  2. Workflow asincroni e processi in background
  3. Limitazione della frequenza e gestione delle quote
  4. Deploy blue-green e canary per gli agenti
← Torna a AI Agents