0Pricing
AI Agents · Lektion

Asynchrone Workflows und Hintergrundjobs

Langlebige Agents benötigen Queues (Celery, RQ, Temporal) – geben Sie eine Job-ID zurück und fragen Sie den Status ab.

Asynchrone Workflows und Hintergrundjobs ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

Wenn der Request-Zyklus zu lange dauert

Manche Agentenaufgaben dauern 30 Sekunden, 5 Minuten oder mehrere Stunden. Sie können die HTTP-Verbindung nicht so lange offen halten. Verschieben Sie sie in asynchrone Hintergrundjobs.

Das Muster

  1. Der Client sendet einen POST für die Aufgabe -> Der Server erstellt einen Job und gibt job_id zurück
  2. Ein Worker übernimmt den Job aus einer Queue
  3. Der Client fragt mit GET /jobs/{id} den Status ab
  4. Nach Abschluss gibt der Server das Ergebnis zurück

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

Ein separater Prozess verarbeitet die Queue:

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

Auswahl der Queues

  • RQ (Redis Queue) — minimalistisches Python-Framework
  • Celery — Python-Klassiker mit vielen Funktionen
  • Temporal — persistente Workflows, Wiederholungsversuche, Observability
  • Dramatiq — Alternative zu Celery
  • Cloud-native — Cloud Tasks, SQS, Pub/Sub

Temporal für persistente Workflows

Agenten-Workflows sind zustandsbehaftet. Das Modell der persistenten Ausführung von Temporal passt hervorragend dazu:

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)

Streaming partieller Updates

Lang laufende Jobs profitieren von Fortschrittsmeldungen. Verwenden Sie Server-Sent Events oder 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')

Job-TTLs

Behalten Sie Job-Einträge nicht für immer:

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

Wiederholungsversuche

Vorübergehende Fehler werden automatisch erneut versucht; dauerhafte Fehler gelangen zur manuellen Prüfung in eine DLQ:

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

Limits für Nebenläufigkeit

Um GPU- und API-Kosten zu kontrollieren, begrenzen Sie die Anzahl nebenläufiger Worker:

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

# Or per-queue concurrency in Temporal worker config.

Kontingente pro Benutzer

Verfolgen Sie die laufenden Jobs pro Benutzer:

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

Observability

Verfolgen Sie jeden Job von Anfang bis Ende. Kennzeichnen Sie Spans mit job_id und user_id. Für fehlgeschlagene Jobs sollte automatisch ein Ticket oder Alarm erstellt werden.

Muster für Status-Polling

Warum sollten Sie bei langlebigen Agenten das Muster aus Einreichung und Status-Polling verwenden?

Zusammenfassung

Job per POST einreichen, ID zurückgeben, Verarbeitung des Jobs durch einen Worker aus der Queue, Statusabfrage durch den Client. RQ/Celery für einfache Fälle; Temporal für zustandsbehaftete Workflows. TTLs, Wiederholungsversuche, Kontingente, Traces.

Häufig gestellte Fragen

Ist die Lektion „Asynchrone Workflows und Hintergrundjobs“ kostenlos?

Ja — der vollständige Text von „Asynchrone Workflows und Hintergrundjobs“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Asynchrone Workflows und Hintergrundjobs“?

Langlebige Agents benötigen Queues (Celery, RQ, Temporal) – geben Sie eine Job-ID zurück und fragen Sie den Status ab. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.

Wie lange dauert die Lektion „Asynchrone Workflows und Hintergrundjobs“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Agents hinter einer API bereitstellen
  2. Asynchrone Workflows und Hintergrundjobs
  3. Rate-Limiting und Quotenverwaltung
  4. Blue-Green- und Canary-Deployments für Agents
← Zurück zu AI Agents