PricingLoginStart free
AI-agenten · Les

Asynchrone workflows en achtergrondtaken

Langdurig draaiende agents hebben queues nodig (Celery, RQ, Temporal) — geef een job-id terug en vraag de status op.

Les 2 van 415 stappen

Asynchrone workflows en achtergrondtaken is een gratis AI-agenten-les op CoddyKit. Dit is les 2 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject AI-agenten. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus AI-agenten bevat in totaal 4 lessen.

Delen van deze les zijn nog niet vertaald en worden in het Engels weergegeven.

Wanneer de aanvraagcyclus te traag is

Sommige agenttaken duren 30 seconden, 5 minuten of uren. Je kunt de HTTP-verbinding niet openhouden. Verplaats ze naar asynchrone achtergrondtaken.

Het patroon

  1. De client verstuurt een POST voor de taak -> de server maakt een taak aan en retourneert job_id
  2. De worker haalt de taak uit een wachtrij
  3. De client vraagt met GET /jobs/{id} de status op
  4. Wanneer de taak klaar is, retourneert de server het resultaat

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

Een afzonderlijk proces verwerkt de wachtrij:

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

Keuze van wachtrij

  • RQ (Redis Queue) — minimalistisch voor Python
  • Celery — klassieke Python-oplossing met veel functies
  • Temporal — duurzame workflows, nieuwe pogingen en observeerbaarheid
  • Dramatiq — alternatief voor Celery
  • Cloud-native — Cloud Tasks, SQS, Pub/Sub

Temporal voor duurzame workflows

Agentworkflows hebben een toestand. Het model voor duurzame uitvoering van Temporal past hier uitstekend bij:

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)

Gedeeltelijke updates streamen

Langlopende taken hebben baat bij voortgangsupdates. Gebruik Server-Sent Events of 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's voor taken

Bewaar taakrecords niet voor altijd:

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

Nieuwe pogingen

Tijdelijke fouten worden automatisch opnieuw geprobeerd; permanente fouten gaan naar een DLQ voor beoordeling door een mens:

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

Limieten voor gelijktijdigheid

Beperk het aantal gelijktijdige workers om GPU- en API-kosten te beheersen:

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

# Or per-queue concurrency in Temporal worker config.

Quota per gebruiker

Houd het aantal taken dat per gebruiker in behandeling is bij:

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

Observeerbaarheid

Traceer elke taak van begin tot eind. Voorzie spans van de tags job_id en user_id. Mislukte taken moeten automatisch een ticket of waarschuwing openen.

Patroon voor status opvragen

Waarom gebruik je voor langlopende agents het patroon met indienen en daarna de status opvragen?

Beantwoord dit in de interactieve les

Samenvatting

POST een taak, retourneer de id, laat de worker de taak uit de wachtrij verwerken en laat de client de status opvragen. RQ/Celery voor eenvoudige taken; Temporal voor workflows met toestand. TTL's, nieuwe pogingen, quota en traces.

Gratis beginnen

Leer AI-agenten met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen
60
Lessen
239
Gratis beginnen met lerenBekijk de volledige cursus

Veelgestelde vragen

Is de les “Asynchrone workflows en achtergrondtaken” gratis?

Ja — de volledige tekst van “Asynchrone workflows en achtergrondtaken” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus AI-agenten wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus AI-agenten bevat in totaal 4 lessen.

Wat leer ik in “Asynchrone workflows en achtergrondtaken”?

Langdurig draaiende agents hebben queues nodig (Celery, RQ, Temporal) — geef een job-id terug en vraag de status op. Je oefent met AI-agenten door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met AI-agenten te beginnen?

Ervaring vooraf is niet nodig. AI-agenten op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 2 van 4.

Hoe lang duurt de les “Asynchrone workflows en achtergrondtaken”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over AI-agenten?

Ja. Elke les over AI-agenten bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

  1. Agents achter een API aanbieden
  2. Asynchrone workflows en achtergrondtaken
  3. Rate limiting en quotabeheer
  4. Blue-green- en canary-deployments voor agents
← Terug naar AI-agenten