PricingLoginStart free
AI-agenter · leksjon

Asynkrone arbeidsflyter og bakgrunnsjobber

Langvarige agenter trenger køer (Celery, RQ, Temporal) – returner en jobb-ID og hent status med jevne mellomrom.

Leksjon 2 av 415 trinn

Asynkrone arbeidsflyter og bakgrunnsjobber er en gratis leksjon i AI-agenter på CoddyKit. Dette er leksjon 2 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i AI-agenter, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i AI-agenter inneholder totalt 4 leksjoner.

Deler av denne leksjonen er ennå ikke oversatt og vises på engelsk.

Når forespørselssyklusen er for treg

Noen agentoppgaver tar 30 sekunder, 5 minutter eller flere timer. HTTP-tilkoblingen kan ikke holdes åpen så lenge. Flytt dem til asynkrone bakgrunnsjobber.

Mønsteret

  1. Klienten sender en POST-forespørsel for oppgaven -> serveren oppretter en jobb og returnerer job_id
  2. En worker henter jobben fra en kø
  3. Klienten spør jevnlig med GET /jobs/{id} om status
  4. Når jobben er ferdig, returnerer serveren resultatet

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

En separat prosess konsumerer køen:

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

Valg av køsystem

  • RQ (Redis Queue) — minimalistisk Python
  • Celery — klassisk Python-verktøy med mange funksjoner
  • Temporal — robuste arbeidsflyter, nye forsøk og observability
  • Dramatiq — alternativ til Celery
  • Cloud-native — Cloud Tasks, SQS, Pub/Sub

Temporal for robuste arbeidsflyter

Agentarbeidsflyter har tilstand. Tem­porals modell for robust kjøring passer svært godt:

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)

Strømming av delvise oppdateringer

Langvarige jobber har nytte av fremdriftsoppdateringer. Bruk Server-Sent Events eller 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-er for jobber

Ikke behold jobboppføringer for alltid:

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

Nye forsøk

Midlertidige feil prøves automatisk på nytt, mens permanente feil sendes til en DLQ for manuell gjennomgang:

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

Grenser for samtidighet

Begrens antallet samtidige workere for å kontrollere GPU- og API-kostnader:

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

# Or per-queue concurrency in Temporal worker config.

Kvoter per bruker

Følg med på antallet jobber per bruker som er under behandling:

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

Observability

Spor hver jobb fra start til slutt. Merk spans med job_id og user_id. Mislykkede jobber bør automatisk opprette en sak eller et varsel.

Mønster for statussjekking

Hvorfor bruke mønsteret med innsending og statussjekking for langvarige agenter?

Besvar den i den interaktive leksjonen

Oppsummering

Send en POST-forespørsel for jobben, returner ID-en, la en worker behandle den fra køen, og la klienten spørre jevnlig. RQ/Celery for enkle arbeidsflyter, Temporal for arbeidsflyter med tilstand. TTL-er, nye forsøk, kvoter og sporing.

Gratis å komme i gang

Lær deg AI-agenter med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs
60
Leksjoner
239
Start læringen gratisSe hele kurset

Ofte stilte spørsmål

Er leksjonen «Asynkrone arbeidsflyter og bakgrunnsjobber» gratis?

Ja – hele teksten i «Asynkrone arbeidsflyter og bakgrunnsjobber» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av AI-agenter-kurset, kan du oppgradere til CoddyKit PRO. Kurset i AI-agenter inneholder totalt 4 leksjoner.

Hva lærer jeg i «Asynkrone arbeidsflyter og bakgrunnsjobber»?

Langvarige agenter trenger køer (Celery, RQ, Temporal) – returner en jobb-ID og hent status med jevne mellomrom. Du øver på AI-agenter med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med AI-agenter?

Ingen tidligere erfaring er nødvendig. AI-agenter på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 2 av 4.

Hvor lang tid tar leksjonen «Asynkrone arbeidsflyter og bakgrunnsjobber»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne AI-agenter-leksjonen?

Ja. Alle AI-agenter-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

  1. Eksponere agenter bak et API
  2. Asynkrone arbeidsflyter og bakgrunnsjobber
  3. Begrensning av forespørselsfrekvens og kvotehåndtering
  4. Blue-green- og canary-distribusjoner for agenter
← Tilbake til AI-agenter