Zum Inhalt
wannacode

Ein Modell über eine API aufrufen

Kurz gesagt: Ein Sprachmodell über eine API aufzurufen heißt: eine Anfrage mit Modell, System-Prompt und Nachrichten mit Rollen schicken; die Antwort enthält den Text, den Grund fürs Aufhören und die Zahl der Tokens.

Problem

Der Assistent tritt vor die Kunden

Bisher waren die Modelle klein und lebten im Programm. Der Support-Assistent muss echten Kunden antworten, und dafür braucht es ein großes Modell bei einem Anbieter. Das Programm fragt es über eine API: Es schickt eine Anfrage über das Netzwerk und wartet auf eine Antwort.

Am ersten Tag des Sales sehen Kunden „Fehler 429“ statt einer Antwort. Ein anderer Kunde wird vom Assistenten nach der Bestellnummer gefragt, die er gerade genannt hat.

Wie sieht eine Anfrage aus, was liefert das Modell zurück, und was tust du, wenn die Antwort ein Fehler ist?

Die Idee

Anfrage und Antwort

Ein Modell aufzurufen ist eine HTTP-Anfrage mit JSON. Die Anfrage enthält:

  • model: welches Modell;
  • max_tokens: aus höchstens wie vielen Tokens die Antwort bestehen darf;
  • system: den System-Prompt, die Anweisung für Rolle und Ton;
  • messages: die Nachrichten mit ihren Rollen, user (der Kunde) und assistant (frühere Antworten des Modells).

Die Antwort enthält:

  • content: eine Liste von Blöcken mit einem type; der Text steht in einem text-Block;
  • stop_reason: warum das Modell aufgehört hat, end_turn (es war fertig) oder max_tokens (es hat die Obergrenze erreicht);
  • usage: wie viele Tokens hineingingen (input_tokens) und wie viele herauskamen (output_tokens). Danach bezahlst du.

Die API hat kein Gedächtnis. Jede Anfrage steht für sich und trägt alles mit, was das Modell sehen muss, wie in der Lektion über das Kontextfenster. Die APIs verschiedener Anbieter haben dieselbe Form: Nachrichten mit Rollen, Tokens, Tools.

Sieh es dir an

Ein falscher Client mit aufgezeichneten Antworten

Die Programme der Lektion rufen kein echtes Modell auf: Ihre Prüfung hat kein Netzwerk, und die echte Antwort ist jedes Mal anders (die Lektion über die Grenzen). Statt des Clients des Anbieters gibt es deshalb einen Ersatz mit derselben Form: Er nimmt model, max_tokens, system und messages und liefert content, stop_reason und usage. Die Antworten sind vorab aufgezeichnet und werden nach einem Wort in der letzten Nachricht gewählt. Die Tokens werden grob gezählt, 4 auf 3 Wörter.

@dataclass
class Block:
type: str
text: str
@dataclass
class Usage:
input_tokens: int
output_tokens: int
@dataclass
class Response:
content: list
stop_reason: str
usage: Usage
class ApiError(Exception):
def __init__(self, status, message):
super().__init__(message)
self.status = status
# ein Ersatz für den echten Client: aufgezeichnete Antworten, kein Netzwerk
class FakeClient:
def __init__(self, failures=()):
self.failures = list(failures)
def create(self, model, max_tokens, system, messages):
if self.failures:
status = self.failures.pop(0)
raise ApiError(status, reason(status))
text = answer(messages)
stop_reason = "end_turn"
if tokens(text) > max_tokens:
# max_tokens: die Antwort bricht mittendrin ab
text = " ".join(text.split()[: max_tokens * 3 // 4])
stop_reason = "max_tokens"
input_tokens = tokens(system)
for message in messages:
input_tokens += tokens(message["content"])
usage = Usage(input_tokens, tokens(text))
return Response([Block("text", text)], stop_reason, usage)
# die aufgezeichnete Antwort: gewählt nach einem Wort in der letzten Nachricht
def answer(messages):
last = messages[-1]["content"]
has_order = any("1042" in m["content"] for m in messages)
if "zurück" in last:
return "Du kannst die Ware bis 14 Tage nach Erhalt zurückgeben."
if "kommt" in last:
return "Bestellung 1042 ist beim Kurier und kommt morgen bis 18 Uhr." if has_order else "Wie lautet deine Bestellnummer?"
return "Wie kann ich helfen: Bestellstatus, Lieferung oder Rückgabe?"
# der Fehlertext zu seinem Code
def reason(status):
reasons = {
400: "ungültige Anfrage",
401: "ungültiger Schlüssel",
429: "zu viele Anfragen",
}
return reasons.get(status, "Serverfehler")
# grob: 4 Tokens auf 3 Wörter (die echte Zahl liefert der Tokenizer)
def tokens(text):
words = len(text.split())
return (words * 4 + 2) // 3

Ein Aufruf, dann dasselbe mit max_tokens 8:

# die Textblöcke, dann warum es aufgehört hat und wie viele Tokens
def show(response):
for block in response.content:
if block.type == "text":
print(block.text)
u = response.usage
print(f"stop_reason: {response.stop_reason}, input_tokens: {u.input_tokens}, output_tokens: {u.output_tokens}")
client = FakeClient()
system = "Du bist ein Support-Assistent eines Online-Shops. Antworte kurz."
messages = [{"role": "user", "content": "Wann kommt Bestellung 1042 an?"}]
response = client.create(model=MODEL, max_tokens=1024, system=system, messages=messages)
show(response)
response = client.create(model=MODEL, max_tokens=8, system=system, messages=messages)
show(response)
Ausgabe
Bestellung 1042 ist beim Kurier und kommt morgen bis 18 Uhr.
stop_reason: end_turn, input_tokens: 18, output_tokens: 15
Bestellung 1042 ist beim Kurier und
stop_reason: max_tokens, input_tokens: 18, output_tokens: 8

Bei 8 Tokens bricht die Antwort mitten im Satz ab, und stop_reason ist max_tokens. Das Programm muss das prüfen: Eine abgeschnittene Antwort sieht beim Kunden wie ein Fehler des Shops aus.

Sieh es dir an

Bei dir ausführen: der echte Aufruf

Derselbe Aufruf an ein echtes Modell. Er gehört nicht zu den geprüften Programmen und läuft hier nicht: Du brauchst ein Konto bei einem Anbieter und einen Schlüssel in der Umgebung, und jeder Aufruf kostet Geld.

Über das Netzwerk geht JSON. Die Anfrage:

Anfrage
{
"model": "<Name des Modells>",
"max_tokens": 1024,
"system": "Du bist ein Support-Assistent eines Online-Shops. Antworte kurz.",
"messages": [
{
"role": "user",
"content": "Wann kommt Bestellung 1042 an?"
}
]
}

Und die Antwort (die Zahlen hier stammen vom falschen Client):

Antwort
{
"content": [
{
"type": "text",
"text": "Bestellung 1042 ist beim Kurier und kommt morgen bis 18 Uhr."
}
],
"stop_reason": "end_turn",
"usage": {
"input_tokens": 18,
"output_tokens": 15
}
}

Mit dem Standard-HTTP-Client der Sprache ist das eine einzige POST-Anfrage. Die Adresse, der Name des Modells und der Header mit dem Schlüssel sind Platzhalter: Die echten stehen in der Dokumentation des Anbieters. Der Schlüssel wird aus der Variable LLM_API_KEY gelesen.

import json
import os
import urllib.request
body = {
"model": "<Name des Modells>",
"max_tokens": 1024,
"system": "Du bist ein Support-Assistent eines Online-Shops. Antworte kurz.",
"messages": [{"role": "user", "content": "Wann kommt Bestellung 1042 an?"}],
}
request = urllib.request.Request(
"https://<anbieter>/messages",
data=json.dumps(body).encode("utf-8"),
headers={
"Authorization": "Bearer " + os.environ["LLM_API_KEY"],
"Content-Type": "application/json",
},
)
with urllib.request.urlopen(request) as response:
reply = json.load(response)
for block in reply["content"]:
if block["type"] == "text":
print(block["text"])

Anbieter liefern auch offizielle Bibliotheken (SDKs) für Java, C#, Python und JavaScript mit derselben Form: Nachrichten mit Rollen, Tokens, Tools. Die genauen Namen der Klassen und Methoden stehen in ihrer Dokumentation.

Die Idee

Der Schlüssel liegt in der Umgebung, nicht im Code

Der API-Schlüssel ist wie ein Passwort für das Konto: Wer ihn hat, gibt davon Geld aus. Im Code landet er in Git, bei jedem Kollegen und manchmal in einem öffentlichen Repository. Deshalb wird er aus einer Umgebungsvariable gelesen, hier LLM_API_KEY.

key = os.environ.get("LLM_API_KEY", "")
if not key.strip():
print("Kein Schlüssel: setze LLM_API_KEY in der Umgebung.")
else:
print("Der Schlüssel wird aus der Umgebung geladen. Gib ihn nicht aus und schreib ihn nicht in den Code.")
Ausgabe
Kein Schlüssel: setze LLM_API_KEY in der Umgebung.

Das Programm sagt, ob der Schlüssel da ist, gibt ihn aber nie aus. Die Datei mit lokalen Geheimnissen (zum Beispiel .env) steht in der .gitignore.

Wie Geheimnisse in einem echten Projekt aufbewahrt werden, steht in der Lektion über Geheimnisse.

Sieh es dir an

Eine API ohne Gedächtnis

ask hängt die Frage an den Verlauf an, schickt den ganzen Verlauf und behält die Antwort mit der Rolle assistant. Am Ende dieselbe Frage, allein geschickt:

# hängt die Frage an, schickt den ganzen Verlauf und behält die Antwort
def ask(client, system, history, text):
history.append({"role": "user", "content": text})
r = client.create(model=MODEL, max_tokens=1024, system=system, messages=history)
reply = r.content[0].text
history.append({"role": "assistant", "content": reply})
return reply
client = FakeClient()
system = "Du bist ein Support-Assistent eines Online-Shops. Antworte kurz."
history = []
for text in ["Hallo, ich habe Bestellung 1042.", "Wann kommt sie an?"]:
print(f"Kunde: {text}")
print(f"Assistent: {ask(client, system, history, text)}")
alone = client.create(model=MODEL, max_tokens=1024, system=system,
messages=[{"role": "user", "content": "Wann kommt sie an?"}])
print(f"nur die Frage, ohne den Verlauf: {alone.content[0].text}")
Ausgabe
Kunde: Hallo, ich habe Bestellung 1042.
Assistent: Wie kann ich helfen: Bestellstatus, Lieferung oder Rückgabe?
Kunde: Wann kommt sie an?
Assistent: Bestellung 1042 ist beim Kurier und kommt morgen bis 18 Uhr.
nur die Frage, ohne den Verlauf: Wie lautet deine Bestellnummer?

Mit dem Verlauf sieht das Modell die Nummer 1042 aus der ersten Nachricht. Ohne ihn fragt es danach. Das Gedächtnis des Gesprächs liegt im Programm, nicht im Modell, und der Verlauf wächst und wird in jeder Runde mitbezahlt.

Probier es aus

Finde den Fehler

Ein Kollege schreibt ask. Bei der zweiten Frage fragt der Assistent wieder nach der Bestellnummer. In welcher Zeile steckt der Fehler?

Wähle die Zeile mit dem Fehler.

Probier es aus

Ergänze

Mit welcher Rolle kommt die Antwort des Modells in den Verlauf?

def ask(client, system, history, text):
    history.append({"role": "user", "content": text})
    r = client.create(model=MODEL, max_tokens=1024, system=system, messages=history)
    reply = r.content[0].text
    history.append({"role": , "content": reply})
    return reply

Die Idee

Die Fehler der API

Das Netzwerk und der Server des Anbieters fallen aus wie jeder fremde Dienst. Die Fehler kommen mit einem HTTP-Code und fallen in zwei Gruppen:

  • Vorübergehend: 429 (zu viele Anfragen, ein Limit erreicht), 500 und die anderen 5xx (der Server hat ein Problem oder ist überlastet), 408 und 409. Einen Moment später kann dieselbe Anfrage durchgehen. Diese lohnt es sich zu wiederholen.
  • Dauerhaft: 400 (fehlerhafte Anfrage), 401 (ungültiger Schlüssel), 404. Eine Wiederholung liefert dasselbe. Diese meldet das Programm sofort.

Eine Wiederholung wartet zwischen den Versuchen, und die Pause verdoppelt sich: 1, 2, 4 Sekunden. Ein kurzes Problem geht schnell vorbei, und ein überlasteter Server bekommt nicht einen Strom neuer Anfragen. Das ist exponentielles Backoff. Nach einer festen Zahl von Versuchen gibt das Programm auf.

Sieh es dir an

Wiederholungen mit einer Wartezeit, die sich verdoppelt

# wiederholt die vorübergehenden Fehler, die Wartezeit verdoppelt sich
def create_with_retry(client, system, messages):
max_attempts = 4
delay = 1
for attempt in range(1, max_attempts + 1):
try:
return client.create(
model=MODEL, max_tokens=1024,
system=system, messages=messages)
except ApiError as e:
last = attempt == max_attempts
if last or not retryable(e.status):
raise
print(f" Versuch {attempt}: {e.status},",
f"warte {delay} s")
# ein echtes Programm wartet hier:
# time.sleep(delay)
delay *= 2
# vorübergehend: 408, 409, 429 und 5xx
def retryable(status):
return (status == 408 or status == 409
or status == 429 or status >= 500)
Schritt für Schritt, in Worten
  1. Die Anfrage: das Modell, der System-Prompt und die Nachrichten mit ihren Rollen. Hier ist es eine, vom Kunden (user).
  2. Der erste Versuch liefert 429: „zu viele Anfragen“. Ein vorübergehender Fehler.
  3. 429 steht bei den vorübergehenden Fehlern, also wartet das Programm 1 s und versucht es noch einmal.
  4. Der zweite Versuch: 500, „Serverfehler“. Auch vorübergehend.
  5. Die Wartezeit verdoppelt sich: 2 s. Der überlastete Server bekommt Luft.
  6. Der dritte Versuch klappt.
  7. Die Antwort: ein Textblock, stop_reason end_turn und usage: 18 Eingabe- und 15 Ausgabe-Tokens.
  8. 401 und 400 werden nicht wiederholt: Warten repariert keinen falschen Schlüssel. Nach 4 Versuchen gibt das Programm auf.
# wiederholt die vorübergehenden Fehler, die Wartezeit verdoppelt sich
def create_with_retry(client, system, messages):
max_attempts = 4
delay = 1
for attempt in range(1, max_attempts + 1):
try:
return client.create(
model=MODEL, max_tokens=1024,
system=system, messages=messages)
except ApiError as e:
last = attempt == max_attempts
if last or not retryable(e.status):
raise
print(f" Versuch {attempt}: {e.status},",
f"warte {delay} s")
# ein echtes Programm wartet hier:
# time.sleep(delay)
delay *= 2
# vorübergehend: 408, 409, 429 und 5xx
def retryable(status):
return (status == 408 or status == 409
or status == 429 or status >= 500)

Drei Szenarien: 429 und 500, dann eine Antwort; 401, keine Wiederholung; vier Mal 429, nach dem vierten Versuch wird aufgegeben. Die Pause wird nur ausgegeben, damit das Programm sofort läuft.

system = "Du bist ein Support-Assistent eines Online-Shops. Antworte kurz."
messages = [{"role": "user", "content": "Wann kommt Bestellung 1042 an?"}]
scenarios = [[429, 500], [401], [429, 429, 429, 429]]
for failures in scenarios:
print(f"Szenario {failures}:")
try:
r = create_with_retry(FakeClient(failures), system, messages)
print(f" Antwort: {r.content[0].text}")
except ApiError as e:
print(f" aufgegeben: {e.status} {e}")
Ausgabe
Szenario [429, 500]:
Versuch 1: 429, warte 1 s
Versuch 2: 500, warte 2 s
Antwort: Bestellung 1042 ist beim Kurier und kommt morgen bis 18 Uhr.
Szenario [401]:
aufgegeben: 401 ungültiger Schlüssel
Szenario [429, 429, 429, 429]:
Versuch 1: 429, warte 1 s
Versuch 2: 429, warte 2 s
Versuch 3: 429, warte 4 s
aufgegeben: 429 zu viele Anfragen

Die offiziellen SDKs der Anbieter machen das meist intern: Sie wiederholen die vorübergehenden Fehler selbst, mit wachsender Pause. Lies die Dokumentation und setze keine eigene Schleife obendrauf: Die Versuche vervielfachen sich. Eine eigene Schleife schreibst du, wenn du die API direkt über HTTP aufrufst, wie im Block „Bei dir ausführen“.

Timeouts, Wiederholungen mit Jitter und den Circuit Breaker findest du in der Lektion über Ausfallsicherheit.

Probier es aus

Sag die Ausgabe voraus

Der falsche Client liefert zuerst 429, dann 400. system und messages sind wie oben. Was gibt das Programm aus?

client = FakeClient([429, 400])
try:
r = create_with_retry(client, system, messages)
print(f"Antwort: {r.content[0].text}")
except ApiError as e:
print(f"aufgegeben: {e.status}")
Antwort

Die Idee

Was der Kunde sieht

Wiederholungen helfen nicht immer. Wenn auch der letzte Versuch scheitert, darf der Kunde nicht „ApiException: 503“ sehen. Das Programm fängt den Fehler ab und sagt etwas Höfliches und Wahres: „Ich kann gerade nicht antworten. Ich verbinde dich mit einem Kollegen.“ Der Fehler geht für das Team ins Log.

Genauer, für Neugierige

Jitter. Wenn Tausende Programme gleichzeitig scheitern und genau 1, 2, 4 Sekunden warten, versuchen sie es gleichzeitig wieder. Zur Pause kommt ein zufälliger Teil (Jitter), um sie zu verteilen.

Timeout. Die Anfrage kann auch hängen, deshalb hat jeder Aufruf über das Netzwerk eine Obergrenze für die Zeit bis zur Antwort. Eine lange Antwort ist langsam, also schützt max_tokens Zeit und Geld zugleich. Mehr zu Langsamkeit und Kosten steht in der Lektion über Kosten und Latenz.

Limits. 429 heißt, dass du das Limit der Anfragen oder Tokens pro Minute erreicht hast. Eine Wiederholung hilft bei einer Spitze. Wenn du es ständig erreichst, hilft nicht mehr Wiederholen, sondern weniger Anfragen, kürzere Anfragen oder ein höheres Limit.

Wie es weitergeht. Der System-Prompt ist der erste Schritt zu Prompts als Code. Eine Antwort, die das Programm liest und nicht nur anzeigt, ist strukturierte Ausgabe, und ein Modell, das das Programm selbst nach Daten fragt, ist Tool Use. Text von einem Nutzer oder aus einem Dokument, der die Anweisungen ändern will (Prompt Injection), kommt im Modul über Agenten weiter hinten im Kurs.

Übung

Eine Antwort oder eine höfliche Absage

Löse die Aufgabe in deinem Editor.

Schreibe askSupport(client, system, question): Sie ruft createWithRetry auf und liefert den Text der Antwort, und wenn auch der letzte Versuch scheitert (oder der Fehler dauerhaft ist), liefert sie „Ich kann gerade nicht antworten. Ich verbinde dich mit einem Kollegen.“. Starte sie mit der Frage „Kann ich die Jacke zurückgeben?“ und vier Szenarien des falschen Clients: ohne Fehler; vier Mal 503; 401; ein Mal 429.

Ausgabe
Szenario []:
Du kannst die Ware bis 14 Tage nach Erhalt zurückgeben.
Szenario [503, 503, 503, 503]:
Versuch 1: 503, warte 1 s
Versuch 2: 503, warte 2 s
Versuch 3: 503, warte 4 s
Ich kann gerade nicht antworten. Ich verbinde dich mit einem Kollegen.
Szenario [401]:
Ich kann gerade nicht antworten. Ich verbinde dich mit einem Kollegen.
Szenario [429]:
Versuch 1: 429, warte 1 s
Du kannst die Ware bis 14 Tage nach Erhalt zurückgeben.

Weitere Aufgaben

  • Ein Log für das Team. Gib bei einer Absage vor der höflichen Antwort eine eigene Zeile mit dem Code und dem Text des Fehlers aus. Der Kunde sieht nur die höfliche.
  • Die abgeschnittene Antwort. Füge createWithRetry einen Parameter für max_tokens hinzu. Wenn stop_reason gleich max_tokens ist, gib die Antwort mit „…“ am Ende zurück und logge eine Warnung. Probiere es mit 8.

Test

Teste dich

1. Der Kunde schreibt „Hallo, ich habe Bestellung 1042.“ und dann „Wann kommt sie an?“. Bei der zweiten Anfrage schickt das Programm nur die neue Frage. Was passiert?
2. Die Antwort kommt mit stop_reason: max_tokens. Was bedeutet das?
3. Welche Fehler lohnt es sich zu wiederholen?
4. Warum verdoppelt sich die Pause zwischen den Versuchen (1, 2, 4 s)?
5. Wohin gehört der API-Schlüssel?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·