async/await für LLM-Aufrufe

Track KI · M0 Baustein 05 · ca. 50 Min. (plus optional 10 Min. lokal)

Worum es geht

Ein LLM-Aufruf dauert Sekunden. Wenn deine Anwendung 50 davon braucht (50 Dokumente zusammenfassen, 50 Antworten bewerten), entscheidet async/await, ob der Nutzer 100 Sekunden oder 10 Sekunden wartet. In dieser Lektion lernst du, viele Modellaufrufe gleichzeitig zu starten, ohne das Rate Limit zu sprengen (Semaphore), einzelne hängende Aufrufe abzubrechen (Timeout) und mit Fehlern in einem Stapel umzugehen. Der Teil über Tests kommt in Teil 2: pytest für LLM-Code.

Drei Themen sind hier kein neuer Stoff, weil sie in den Konzept-Lektionen stehen: Threads und den GIL erklärt Konzepte 19a, den Event Loop und Backpressure Konzepte 19b, Retry und Timeouts als Resilienz-Muster Konzepte 07a, Testpyramide und Testen von KI-Code Konzepte 16. Hier wendest du es auf den KI-Alltag an.

Kein echtes Modell und kein Netzwerk im Browser. Darum gibt es ein Hilfsmittel, das du gleich kennenlernst: einen Event Loop mit virtueller Uhr (sleep(2) kostet keine echte Sekunde, die Ausgaben sind bei jedem Lauf gleich). Echte Zeitmessung übst du in der lokalen Zusatzübung am Ende.

Zeitplan: etwa 25 Minuten Lesen, 20 Minuten für die drei Übungen, dazu optional 10 Minuten lokal.

Von JS/TS her gedacht

Idee JS/TS Python
Aufgabe, die später ein Ergebnis liefert Promise Coroutine (async def gibt beim Aufruf nur ein Objekt zurück, es läuft noch nichts)
Warten await p await c (nur in async def)
Alle gleichzeitig starten, auf alle warten Promise.all([...]) asyncio.gather(...)
Fehler einsammeln statt abbrechen Promise.allSettled asyncio.gather(..., return_exceptions=True)
Höchstens n gleichzeitig Bibliothek (z. B. p-limit) asyncio.Semaphore(n) (eingebaut)
Timeout AbortSignal.timeout(ms) asyncio.timeout(sekunden)

Die wichtigste Gemeinsamkeit: Das Zeitverhalten ist in Node und Python gleich. Fünf Aufrufe von je 200 ms, einmal nacheinander, einmal mit Promise.all (ausgeführt mit Node 20, die Zahlen schwanken leicht):

const llm = (p, ms) => new Promise(r => setTimeout(() => r(`Antwort auf ${p}`), ms));
const prompts = ["a", "b", "c", "d", "e"];
let t0 = performance.now();
for (const p of prompts) await llm(p, 200);
console.log("nacheinander", ((performance.now() - t0) / 1000).toFixed(1), "s");
t0 = performance.now();
await Promise.all(prompts.map(p => llm(p, 200)));
console.log("Promise.all ", ((performance.now() - t0) / 1000).toFixed(1), "s");
nacheinander 1.0 s
Promise.all  0.2 s

Zwei Unterschiede zu JS: Eine Python-Coroutine startet erst, wenn jemand sie mit await oder gather anfasst (ein JS-Promise läuft sofort los). Und ein async def-Aufruf ohne await ergibt nur eine Warnung “coroutine was never awaited”, keinen Fehler, aber auch kein Ergebnis.

Konzept

Schritt 1: Zwei Hilfsmittel, damit alles im Browser läuft

Zuerst der Event Loop mit virtueller Uhr. Du musst seinen Code nicht verstehen. Die Idee: Wenn alle Aufgaben warten, springt die Uhr direkt zum nächsten Timer, statt wirklich zu schlafen. laufe(coroutine) führt eine Coroutine aus (wie asyncio.run) und gibt ihr Ergebnis zurück. asyncio.get_running_loop().time() liefert die virtuelle Zeit in Sekunden. Das ersetzt hier time.perf_counter() aus der Quellenübung, denn Zeitmessung im Browser ist unzuverlässig. Wie du echt mit perf_counter misst, steht in der lokalen Zusatzübung.

Dann das Fake-Modell. Es ist eine async-Funktion, die dauer virtuelle Sekunden “denkt” und eine feste Antwort liefert:

Ausgabe:

Antwort auf 'Hallo'

Schritt 2: Nacheinander oder gleichzeitig

Fünf Aufrufe zu je 2 Sekunden. Variante 1 wartet jeden Aufruf einzeln ab, Variante 2 startet alle mit asyncio.gather und wartet gemeinsam:

Ausgabe:

nacheinander 10.0 s virtuell, erste Antwort: Antwort auf 'Frage 0'
gather        2.0 s virtuell, erste Antwort: Antwort auf 'Frage 0'

So liest du das: await llm(p) in einer Schleife wartet jedes Mal, bis der Aufruf fertig ist, erst dann beginnt der nächste. Fünf mal 2 Sekunden sind 10. gather startet erst alle Aufrufe und wartet dann, bis der langsamste fertig ist: 2 Sekunden. Die Antworten kommen in der Reihenfolge der Eingabe zurück, nicht in der Reihenfolge des Fertigwerdens. Das ist Promise.all in Python.

Die Faustregel: Gesamtzeit nacheinander = Summe, gleichzeitig = Maximum. Mit einer Begrenzung (Schritt 4) liegt sie dazwischen.

Schritt 3: Wenn einer von vielen fehlschlägt

Bei 50 Aufrufen schlägt irgendwann einer fehl (Rate Limit, Timeout, Modellfehler). Was macht gather dann? Ein kaputter Aufruf unter drei:

Ausgabe:

Ausnahme beim Aufrufer: 429 rate limit
["Antwort auf 'a'", RuntimeError('429 rate limit'), "Antwort auf 'c'"]

Ohne return_exceptions=True fliegt die erste Ausnahme zum Aufrufer, und du bekommst keines der anderen Ergebnisse, obwohl zwei Aufrufe gut gingen (und bezahlt wurden). Mit return_exceptions=True steht die Ausnahme als Objekt in der Ergebnisliste an ihrer Stelle, und du entscheidest pro Eintrag (wiederholen, überspringen, melden). Das entspricht Promise.allSettled.

Noch eine Eigenheit, die oft überrascht: Fliegt die Ausnahme, laufen die anderen Aufrufe weiter (gather bricht sie nicht ab). Ein Beispiel mit einem langsamen Aufruf (5 s) neben einem kaputten (1 s):

Ausgabe:

gather warf: boom bei 1.0
  langsamer Aufruf: fertig
Ende bei 11.0

Der langsame Aufruf lief nach dem Fehler weiter und kostete trotzdem. Wer das nicht will, nimmt asyncio.TaskGroup (bricht beim ersten Fehler alle ab, siehe Konzepte 19), oder er sammelt mit return_exceptions=True ein. Für Batch-Jobs mit LLM-Aufrufen ist Einsammeln fast immer richtig.

Schritt 4: Rate Limit mit Semaphore und Timeout pro Aufruf

Ein Anbieter erlaubt nur wenige gleichzeitige Anfragen (Rate Limit, die genauen Zahlen stehen in der Dokumentation des Anbieters, bitte prüfen). 20 Aufrufe auf einmal zu starten ist dann gefährlich. Eine Semaphore (asyncio.Semaphore(n)) ist ein Zähler mit n Plätzen: async with sem: belegt einen Platz und gibt ihn am Ende des Blocks zurück (auch bei einer Ausnahme). Sind alle Plätze belegt, wartet der nächste.

Ausgabe:

höchstens  1 gleichzeitig:  40.0 s
höchstens  4 gleichzeitig:  10.0 s
höchstens 20 gleichzeitig:   2.0 s

Rechne nach: 20 Aufrufe, 4 gleichzeitig, das sind 5 Runden zu 2 Sekunden, also 10. Mit 1 Platz sind es 20 mal 2 gleich 40 Sekunden (das ist wieder “nacheinander”). Die Gesamtzeit ist ungefähr aufgerundet(Anzahl / n) * Dauer.

Dann der Timeout pro Aufruf. async with asyncio.timeout(sekunden): bricht den Block ab und löst TimeoutError aus, wenn er zu lange dauert:

Ausgabe:

Timeout bei 1.5 s

Wichtig ist die Reihenfolge der Verschachtelung, wenn beides zusammenkommt. Der Timeout soll die Zeit des Modells begrenzen, nicht die Zeit, die ein Aufruf in der Warteschlange vor der Semaphore verbringt. Steht der Timeout außerhalb von async with sem, läuft die Uhr schon während des Wartens auf einen freien Platz. Dann läuft bei 20 Aufrufen und 1 Platz fast jeder Aufruf in den Timeout, obwohl das Modell gar nicht langsam ist. Das übst du in Übung 2.

Eine echte Wiederholung bei Fehlern (Retry mit Backoff) steht in Konzepte 07. Du hängst sie später um den einzelnen Aufruf, nicht um die ganze Liste.

Schritt 5: Wann async nichts bringt

Aus der Quelle: async lohnt sich bei I/O-bound Wartezeiten (Netzwerk, Datenbank, Modellantwort), nicht bei CPU-bound Rechenarbeit. Wer 10000 Vektoren mit numpy vergleicht, wartet auf nichts. Dort gibt es kein await, an dem etwas anderes drankommen könnte. Das Ergebnis ist dasselbe, aber komplizierter.

Die Quelle sagt außerdem, ein async def-Endpunkt in FastAPI könne während der Wartezeit auf eine Modellantwort weitere Anfragen bedienen, ein normaler def-Endpunkt nicht. Präziser (allgemeines Fachwissen, bitte gegen die FastAPI-Doku prüfen): FastAPI führt def-Endpunkte in einem Thread-Pool mit begrenzter Größe aus. Sie blockieren den Event Loop nicht, aber bei vielen langen Modellwartezeiten gehen die Threads aus. Der Gewinn von async def hängt außerdem an allem darin: Ruft dein async def-Endpunkt eine blockierende Bibliothek auf (requests.post, time.sleep), hält er den ganzen Event Loop an. Dann bist du schlechter dran als mit def. Übung 3 spielt das durch.

Falle

  1. Alle gleichzeitig, ohne Grenze. gather über 500 Aufrufe läuft in das Rate Limit des Anbieters. Immer eine Semaphore.
  2. gather ohne return_exceptions=True im Batch. Ein Fehler wirft 49 gute (und bezahlte) Antworten weg, und die anderen Aufrufe laufen trotzdem weiter.
  3. Timeout vor der Semaphore. Er zählt die Wartezeit in der Schlange mit und löst bei langen Listen grundlos aus.
  4. Blockierender Aufruf in async def. requests.post, time.sleep oder eine große Rechenschleife halten alle anderen Aufgaben an.
  5. Async für CPU-Arbeit. Ohne await gewinnt nichts. Dafür sind Prozesse da (Konzepte 19).

Übungen

Übung 1: Gesamtzeit vorhersagen (leicht bis mittel, ca. 5 Min.)

Vier Varianten rufen ein Fake-Modell auf. anruf(d) wartet d virtuelle Sekunden. Wie viele Sekunden dauert jede Variante? Trage (a, b, c, d) als Zahlen ein. Die Dauern der Aufrufe stehen jeweils in den Klammern.

async def anruf(dauer):
    await asyncio.sleep(dauer)

async def variante_a():
    for d in (3, 1, 2):
        await anruf(d)

async def variante_b():
    await asyncio.gather(anruf(3), anruf(1), anruf(2))

async def variante_c():
    await asyncio.gather(anruf(3), anruf(1))
    await anruf(2)

async def variante_d():
    sem = asyncio.Semaphore(2)
    async def begrenzt(d):
        async with sem:
            await anruf(d)
    await asyncio.gather(begrenzt(2), begrenzt(2), begrenzt(2), begrenzt(1))

Pro Variante: Wer startet wann? Bei gather starten alle gleichzeitig, bei der Semaphore nur zwei, der nächste erst, wenn ein Platz frei wird. Zeichne dir einen Zeitstrahl.

antwort = (6, 3, 5, 4)
antwort

A: 3 + 1 + 2 = 6 (nacheinander, Summe). B: alle gleichzeitig, das Maximum 3. C: erst gather(3, 1) dauert 3, dann 2 dazu gleich 5. D: Zwei Plätze. Die ersten beiden Aufrufe (je 2 s) laufen von 0 bis 2, dann starten die letzten beiden (2 s und 1 s) und sind nach 4 Sekunden fertig.

Übung 2: Stapelverarbeitung mit Limit und Timeout (mittel, ca. 10 Min.)

Schreibe frage_alle(llm, prompts, max_parallel, timeout). Sie ruft das Modell für jeden Prompt auf und gibt pro Prompt ein Tupel zurück, in der Reihenfolge der Prompts:

  • ("ok", antwort), wenn die Antwort kam,
  • ("timeout", None), wenn der Aufruf länger als timeout Sekunden am Modell brauchte,
  • ("fehler", meldung), wenn das Modell eine Ausnahme warf (meldung ist str(e)).

Es dürfen nie mehr als max_parallel Aufrufe gleichzeitig laufen, und ein Fehler darf die anderen Ergebnisse nicht zerstören. Das Fake-Modell llm ist ein Objekt, das du mit await llm(prompt) aufrufst. Es braucht 2 s pro Aufruf, Prompts, die mit LANG beginnen, 8 s, und Prompts, die mit FEHLER beginnen, werfen nach 0,5 s RuntimeError("429 rate limit"). Der Rahmen steht schon, ergänze die Lücken.

Drei Fragen nacheinander: Wo muss der Timeout stehen, damit die Wartezeit auf die Semaphore nicht mitzählt? Welche Ausnahme löst asyncio.timeout aus, und wie fängst du sie von anderen Fehlern getrennt ab? Wie bekommst du aus gather mehrere Ergebnisse, ohne dass die Ausnahme alles wegwirft?

async def frage_alle(llm, prompts, max_parallel, timeout):
    sem = asyncio.Semaphore(max_parallel)

    async def einer(prompt):
        async with sem:
            try:
                async with asyncio.timeout(timeout):
                    return ("ok", await llm(prompt))
            except TimeoutError:
                return ("timeout", None)
            except Exception as e:
                return ("fehler", str(e))

    return list(await asyncio.gather(*(einer(p) for p in prompts)))

frage_alle

Der Timeout steht innerhalb der Semaphore, sonst zählt die Wartezeit in der Schlange mit. Die Fehler werden im einzelnen Aufruf zu Ergebnissen gemacht, darum braucht gather hier kein return_exceptions=True (das wäre die zweite gültige Lösung).

Übung 3: Multiple Choice mit Begründung (leicht, ca. 4 Min.)

Ein Entwickler schreibt einen FastAPI-Endpunkt async def zusammenfassen(...). Darin ruft er für den LLM-Aufruf requests.post(...) auf, die klassische blockierende HTTP-Bibliothek, kein await. Der Aufruf dauert 3 Sekunden. Zehn Nutzer schicken in derselben Sekunde eine Anfrage. Was passiert (ein Prozess, ein Event Loop, ohne weitere Konfiguration)?

  • a) Der Event Loop steht bei jedem Aufruf still, die Antworten kommen nach 3, 6, 9 bis 30 Sekunden.
  • b) Alle zehn laufen gleichzeitig und sind nach etwa 3 Sekunden fertig, weil der Endpunkt async def ist.
  • c) FastAPI startet pro Anfrage automatisch einen Thread, darum dauert alles insgesamt etwa 3 Sekunden.
  • d) Es kommt sofort ein Fehler, weil requests in einer async def-Funktion nicht benutzt werden darf.

Trage den Buchstaben als String ein.

async def allein macht nichts gleichzeitig. Wo genau gibt der Code im Endpunkt die Kontrolle an den Event Loop zurück, wenn er requests.post aufruft?

antwort = "a"
antwort

Zusatzübung (lokal): echt messen mit perf_counter (optional, 10 Min.)

Die Quellenübung verlangt echte Zeit statt virtueller. Das Skript lernlabor/uebung/ki/ki_03_async_zeit.py misst 5 Aufrufe zu je 0,3 s synchron (time.sleep) und mit asyncio.gather und variiert das Limit der Semaphore. Mit --echt und drei URLs misst es echte HTTP-Abrufe (nur mit Netz, die Zeiten hängen von deinem Netz ab). Ein Beispiellauf zeigt die Größenordnung, deine Zahlen weichen ab (ausgeführt mit Python 3.13.9, ohne --echt):

5 Aufrufe à 0.3 s
synchron (time.sleep)         1.52 s
asyncio.gather                0.30 s
gather, höchstens 1 parallel  1.51 s
gather, höchstens 2 parallel  0.90 s
gather, höchstens 5 parallel  0.30 s
gather, höchstens 10 parallel  0.32 s

So gehst du vor: Sage die Zahlen zuerst voraus (Summe gegen Maximum, dann Runden bei der Semaphore). Starte das Skript, vergleiche, und trage in aufgabe_3 ein, ab welcher Parallelität es nicht mehr schneller wird.

cd lernlabor && uv run python uebung/ki/ki_03_async_zeit.py

Wie viele Aufrufe gibt es überhaupt? Was bringt ein Platz mehr, wenn alle Aufrufe schon gleichzeitig laufen?

aufgabe_3 = 5

Bei 5 Aufrufen sind 5 Plätze das Maximum. Mehr Plätze (10) ändern nichts mehr, die Zeit bleibt bei etwa 0,3 s, die Unterschiede sind Messrauschen.

Merksatz

Viele Modellaufrufe gleichzeitig brauchen ein Limit (Semaphore) und einen Timeout direkt am Modell, innerhalb der Semaphore. Und wer Fehler pro Aufruf einsammelt, wirft bei einem Ausfall nicht alle guten Antworten weg.

Prüfstein

Du sollst 200 Dokumente mit einem LLM zusammenfassen lassen. Der Anbieter erlaubt laut Dokumentation nur wenige gleichzeitige Anfragen, und einzelne Aufrufe hängen manchmal. Beschreibe, wie du den Stapel mit asyncio aufbaust (Limit, Timeout, Fehler) und in welcher Reihenfolge die Teile ineinander stecken. Was passiert, wenn du die Reihenfolge vertauschst?

Eine Semaphore(n) mit dem erlaubten Limit, pro Dokument eine Funktion, die erst die Semaphore belegt und darin den Aufruf mit asyncio.timeout(...) umschließt. Fehler und Timeouts werden pro Dokument zu einem Ergebnis (ok, timeout, fehler), alles zusammen mit gather. Die Ergebnisliste behält die Reihenfolge der Dokumente, danach kannst du die Fehlgeschlagenen gezielt wiederholen (Retry mit Backoff, Konzepte 07). Steht der Timeout vor der Semaphore, zählt die Wartezeit in der Schlange mit, und bei langen Listen laufen grundlos viele Aufrufe in den Timeout.

Weiter: Teil 2: pytest für LLM-Code zeigt, wie du genau diese Funktion ohne einen einzigen echten Modellaufruf testest.

Quelle: quellen/kursbuch-lerninhalte.md, Modul M0, Baustein “05 async/await” (Zeilen 190 bis 237 der Datei): Warum, Kernidee mit async def/await, Stolperfalle I/O-bound gegen CPU-bound, die Übung (3 URLs synchron gegen gather messen).

Über die Quelle hinaus (allgemeines Fachwissen): Semaphore für Rate Limits, asyncio.timeout, das Verhalten von gather bei Fehlern (return_exceptions, weiterlaufende Aufgaben), die virtuelle Uhr (eigenes Hilfsmittel zum Üben, kein Teil von asyncio). Zahlen in den Ausgaben stammen aus ausgeführtem Code (Python 3.13.9, Node 20). Bitte prüfen: wie viele gleichzeitige Anfragen dein Anbieter erlaubt (nur in dessen Doku) und wie genau FastAPI def-Endpunkte ausführt (Thread-Pool).