Asynchrone Architektur und Warteschlangen

Warum Sie niemals auf einen LLM-Aufruf warten sollten

Ein API-Aufruf an ein Sprachmodell dauert zwischen einer und dreißig Sekunden, abhängig vom Modell und der Komplexität der Anfrage. Dreißig Sekunden sind für einen Serverprozess eine Ewigkeit: Während dieser Zeit tut der wartende Thread nichts, belegt aber dennoch einen Slot. Multiplizieren Sie dies mit hundert gleichzeitigen Anfragen, und Ihre Anwendung stürzt nicht ab, weil sie zu viel rechnet, sondern weil sie zu lange wartet.

Eine asynchrone Architektur, gestützt auf eine Warteschlange, löst drei Probleme gleichzeitig. Sie absorbiert Verkehrsspitzen, da die Warteschlange wächst, anstatt Anfragen abzulehnen. Sie zentralisiert die Wiederholungslogik, da eine fehlgeschlagene Aufgabe in die Warteschlange zurückkehrt, anstatt mit der HTTP-Anfrage, die sie ausgelöst hat, zu verschwinden. Und sie entkoppelt Ihre Dienste: Die Komponente, die Anfragen empfängt, muss nicht mehr wissen, wie lange die ausführende Komponente benötigt, um diese zu verarbeiten.

Das Producer-/Consumer-Muster

Die grundlegende Struktur basiert auf zwei Objekten. Eine Aufgabe beschreibt, was erledigt werden muss, und den aktuellen Verarbeitungsstatus; eine Warteschlange verteilt diese Aufgaben an einen Pool von Arbeitern, die sie parallel abarbeiten. Die Modellierung des Status als Aufzählungstyp (Enumeration) statt als frei formatierter String vermeidet die häufigste Klasse von Fehlern in diesem Code-Typ: dass zwei Teile des Programms "terminee" und "terminée" schreiben.

import asyncio
from dataclasses import dataclass, field
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    prompt: str
    model: str = "gpt-5.6-terra"
    status: TaskStatus = TaskStatus.PENDING
    result: str | None = None
    attempts: int = 0
    max_attempts: int = 3

class ProcessingQueue:
    """Asynchrone Warteschlange für LLM-Aufrufe."""

    def __init__(self, num_workers: int = 5):
        self.queue: asyncio.Queue[Task] = asyncio.Queue()
        self.results: dict[str, Task] = {}
        self.num_workers = num_workers

    async def add(self, task: Task) -> str:
        """Fügt eine Aufgabe zur Warteschlange hinzu."""
        self.results[task.id] = task
        await self.queue.put(task)
        return task.id

    async def worker(self, worker_id: int):
        """Worker, der Aufgaben aus der Warteschlange verarbeitet."""
        import openai
        client = openai.AsyncOpenAI()

        while True:
            task = await self.queue.get()
            task.status = TaskStatus.IN_PROGRESS
            task.attempts += 1

            try:
                response = await client.responses.create(
                    model=task.model,
                    input=task.prompt,
                )
                task.result = response.output_text
                task.status = TaskStatus.COMPLETED

            except Exception as e:
                if task.attempts < task.max_attempts:
                    task.status = TaskStatus.PENDING
                    await self.queue.put(task)
                else:
                    task.status = TaskStatus.FAILED
                    task.result = str(e)

            finally:
                self.queue.task_done()

    async def start(self):
        """Startet die Worker."""
        workers = [
            asyncio.create_task(self.worker(i))
            for i in range(self.num_workers)
        ]
        return workers

    async def wait_completion(self):
        """Wartet, bis alle Aufgaben verarbeitet sind."""
        await self.queue.join()

Zwei Details in diesem Worker verdienen Ihre Aufmerksamkeit, da sie in jeder realen Implementierung vorkommen. Der Zähler attempts befindet sich auf der Aufgabe selbst, nicht auf dem Worker, sodass jeder andere Worker eine neu eingereihte Aufgabe übernehmen kann, ohne die Historie zu verlieren. Der Aufruf task_done() befindet sich in einem finally-Block, um sicherzustellen, dass die Warteschlange niemals unendlich hängen bleibt, falls eine unerwartete Ausnahme den except-Block umgeht: Ohne ihn ist ein join(), der niemals zurückkehrt, das Symptom, und es ist schmerzhaft zu diagnostizieren.

Auf der Seite des Aufrufers ist die Verwendung einfach: Starten Sie die Worker, reihen Sie Aufgaben ein, warten Sie, sammeln Sie die Ergebnisse und stoppen Sie.

async def process_documents(documents: list[str]):
    queue = ProcessingQueue(num_workers=10)
    workers = await queue.start()

    # Aufgaben hinzufügen
    for i, doc in enumerate(documents):
        task = Task(
            id=f"doc-{i}",
            prompt=f"Zusammenfassen Sie dieses Dokument: {doc}",
        )
        await queue.add(task)

    # Warten auf Abschluss
    await queue.wait_completion()

    # Ergebnisse sammeln
    for task_id, task in queue.results.items():
        print(f"{task_id}: {task.status.value}")

    # Worker stoppen
    for w in workers:
        w.cancel()

Quoten respektieren statt sie zu überschreiten

OpenAI durchsetzt zwei gleichzeitige Grenzen: Anfragen pro Minute und Tokens pro Minute. Ein Pool von zehn Arbeitern, der alles sendet, was er kann, wird eine dieser Grenzen erreichen, Fehler für Rate Limits erhalten, wiederholen und die Verstopfung verschlimmern. Es ist besser, upstream freiwillig langsamer zu werden, als downstream abgelehnt zu werden.

Der untenstehende Begrenzer wendet einen Token-Bucket über ein gleitendes Fenster von einer Minute an. Er hält zwei Historien auf – Zeitstempel kürzlicher Anfragen und Zeitstempel/Token-Paare –, reinigt Einträge, die aus dem Fenster gefallen sind, und wartet, wenn eine der Grenzen überschritten werden würde. Der asyncio-Lock ist unerlässlich: Ohne ihn würden zehn Worker dieselbe Situation gleichzeitig bewerten, und alle zehn würden entscheiden, dass noch Platz ist.

import asyncio
import time

class RateLimiter:
    """Rate-Limiter basierend auf einem Token-Bucket."""

    def __init__(self, rpm: int = 500, tpm: int = 200_000):
        self.rpm = rpm
        self.tpm = tpm
        self.recent_requests: list[float] = []
        self.recent_tokens: list[tuple[float, int]] = []
        self.lock = asyncio.Lock()

    async def wait(self, estimated_tokens: int = 1000):
        """Wartet gegebenenfalls, um die Grenzen einzuhalten."""
        async with self.lock:
            now = time.time()
            window = 60.0

            # Einträge außerhalb des Fensters bereinigen
            self.recent_requests = [
                t for t in self.recent_requests
                if now - t < window
            ]
            self.recent_tokens = [
                (t, n) for t, n in self.recent_tokens
                if now - t < window
            ]

            # RPM prüfen
            if len(self.recent_requests) >= self.rpm:
                wait_time = self.recent_requests[0] + window - now
                await asyncio.sleep(wait_time)

            # TPM prüfen
            tokens_used = sum(n for _, n in self.recent_tokens)
            if tokens_used + estimated_tokens > self.tpm:
                wait_time = self.recent_tokens[0][0] + window - now
                await asyncio.sleep(wait_time)

            self.recent_requests.append(time.time())
            self.recent_tokens.append((time.time(), estimated_tokens))

Der Parameter estimated_tokens verdient ein Wort: Sie kennen die tatsächlichen Kosten erst, wenn die Antwort eingeht, daher arbeiten Sie mit einer Schätzung. Machen Sie diese großzügig statt präzise. Systematisches Unterschätzen entspricht dem Deaktivieren des Schutzes genau dann, wenn er nützlich wäre.

Kontrolle sofort zurückgeben

Für Verarbeitungen, die mehrere Minuten dauern, ist es für einen HTTP-Client nicht mehr lebensfähig, darauf zu warten: Proxy-, Browser- und Load-Balancer-Timeouts werden auslösen, bevor Sie fertig sind. Das Webhook-Muster kehrt die Verantwortung um. Ihr Endpunkt akzeptiert die Anfrage, gibt sofort eine Task-ID zurück und benachrichtigt den Aufrufer, wenn das Ergebnis vorhanden ist, ohne dass periodisches Abfragen erforderlich ist.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/api/analyze")
async def start_analysis(
    request: dict,
    background_tasks: BackgroundTasks,
):
    """Startet eine Hintergrundanalyse und benachrichtigt über ein Webhook."""
    task_id = generate_id()

    background_tasks.add_task(
        run_analysis,
        task_id=task_id,
        document=request["document"],
        webhook_url=request.get("webhook_url"),
    )

    return {"task_id": task_id, "status": "in_progress"}


async def run_analysis(
    task_id: str,
    document: str,
    webhook_url: str | None,
):
    """Führt die Analyse durch und sendet das Ergebnis über ein Webhook."""
    import httpx

    result = await call_llm(document)

    if webhook_url:
        async with httpx.AsyncClient() as http:
            await http.post(webhook_url, json={
                "task_id": task_id,
                "status": "completed",
                "result": result,
            })

Speichern Sie die zurückgegebene ID und den zugehörigen Zustand immerwo an einem dauerhaften Ort. Ein Webhook kann verloren gehen, und der Aufrufer muss die Fähigkeit behalten, das Ergebnis selbst abzuholen.

Kernaussagen

• Nutzen Sie asynchrone Warteschlangen, um die Ingestion von der Verarbeitung zu entkoppeln
• Implementieren Sie einen Rate-Limiter, um OpenAI-Quoten zu respektieren
• Behandeln Sie Wiederholungen mit exponentiellem Backoff in Ihren Workern
• Für langlaufende Aufgaben bevorzugen Sie ein Webhook-Muster gegenüber dem Abfragen