Kosten und Streaming: Tokens, Logging, Abbruch

Track KI · M1 Baustein 05 · Teil 1 von 2 · ca. 50 Min.

Worum es geht

Ein LLM-Aufruf kostet Geld pro Token, und er liefert seine Antwort bei langen Texten nur langsam. Die Quelle nennt eine Folge: Ohne Logging weiß niemand, welcher Anwendungsfall wie viel kostet, bis die Rechnung überrascht. Und ein Stream (Text kommt häppchenweise) lässt dich früh reagieren, verlangt aber, dass du ihn sauber schließt.

In diesem Teil baust du drei kleine Werkzeuge:

  1. eine Kostenrechnung pro Aufruf (tokens, cost),
  2. ein Kostenlogging pro Aufruf und eine Summe pro Anwendungsfall,
  3. das Einsammeln eines Streams mit Abbruch.

Alles läuft im Browser mit simulierten Antworten. Im zweiten Teil kommen Retry mit Backoff und die Projektaufgabe, der Abschluss von M1.

Von JS/TS her gedacht

Idee JS/TS Python
Stream lesen for await (const chunk of stream) for chunk in stream (Generator)
Stream abbrechen break ruft return() des Iterators auf, finally läuft gen.close() löst im Generator GeneratorExit aus, finally läuft

Ein Unterschied ist wichtig. Erstens: In Python endet eine for-Schleife mit break, aber der Generator wird dabei nicht von der Schleife geschlossen. Er bleibt angehalten, solange noch jemand eine Referenz auf ihn hat (eine Variable, der Aufrufer deiner Funktion). Erst close() beendet ihn zuverlässig. Dass ein unreferenzierter Generator in CPython sofort aufgeräumt wird, ist ein Implementierungsdetail, auf das du dich bei einem Netzwerk-Stream nicht verlassen solltest.

Konzept

Schritt 1: Tokens in Kosten umrechnen

Jede Antwort der API trägt Zählwerte: response.usage.input_tokens (was du geschickt hast) und response.usage.output_tokens (was das Modell geschrieben hat). Die Preise gelten je 1000 Token, Input und Output haben verschiedene Preise. Die Quelle nennt als Beispielwerte 0,003 für Input und 0,015 für Output je 1000 Token und sagt ausdrücklich: je Modell prüfen. Wir verwenden sie als Platzhalter, nicht als aktuelle Preise (bitte prüfen).

Zum Nachrechnen: 1200 Input-Token sind 1,2 mal 0,003 = 0,0036. 300 Output-Token sind 0,3 mal 0,015 = 0,0045. Zusammen 0,0081. Mit diesen Platzhalterpreisen kostet ein Output-Token das Fünffache eines Input-Tokens. Darum kann eine lange Antwort teurer sein als ein langer Prompt (die Quelle gibt kein festes Verhältnis an, das gilt hier nur für die Beispielwerte).

Hochrechnen ist dann einfache Multiplikation: 1000 solcher Aufrufe am Tag kosten 1000 mal 0,0081 = 8,10. Genau deshalb willst du pro Anwendungsfall wissen, wie viele Aufrufe mit wie vielen Token laufen.

Schritt 2: Kosten pro Aufruf loggen

Die Quelle zeigt eine Funktion logge_kosten(response), die die Zählwerte liest, die Kosten berechnet, eine Log-Zeile schreibt und die Kosten zurückgibt. Hier mit dem logging-Modul aus ki/04, auf der Konsole sichtbar gemacht:

Beachte %.4f: vier Nachkommastellen, weil einzelne Aufrufe nur Bruchteile einer Währungseinheit kosten. Mit zwei Stellen stünde bei fast jedem Aufruf 0.01 oder 0.00 im Log, und du könntest nichts summieren. Außerdem fehlt für eine Auswertung nach Anwendungsfall noch eine Angabe, nämlich wofür der Aufruf war (zum Beispiel ein Feld anwendung=rechnung_extrahieren in der Zeile). In Übung 2 ergänzt du diese Angabe und summierst die Kosten pro Anwendungsfall. (Die Währung “EUR” im Log-Text übernehmen wir aus der Quelle. Welche Währung dein Anbieter abrechnet, bitte prüfen.)

Schritt 3: Streaming ist ein Generator

Bei langen Antworten wartest du sonst bis zum letzten Token. Die Quelle nennt dafür client.messages.stream(...): Der Text kommt häppchenweise (Chunks). So sieht der Aufruf mit dem SDK aus (nicht ausführbar, braucht Schlüssel und Paket; der Name text_stream geht über die Quelle hinaus, bitte gegen die SDK-Doku prüfen):

with client.messages.stream(model=MODELL, max_tokens=1024, messages=nachrichten) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Für Python ist ein Stream einfach etwas, über das du mit for läufst. Im Browser bauen wir ihn mit einem Generator nach. Der finally-Block im Generator steht stellvertretend für das Aufräumen am Ende des with-Blocks (Verbindung schließen):

Ausgabe: erst alle vier Chunks mit | getrennt, dann die Zeile [Stream geschlossen], weil der Generator zu Ende gelaufen ist. Jetzt der Abbruch. Du liest zwei Chunks und hörst auf:

Ohne g.close() käme die Zeile [Stream geschlossen] hier nicht. Bei einem echten Stream hieße das: Die Verbindung bleibt offen, und das Modell schreibt womöglich weiter, obwohl niemand mehr zuhört. Wer abbrechen will (zum Beispiel, weil die Antwort lang genug ist oder der Nutzer auf “Stopp” klickt), muss den Stream also aktiv schließen. Ob der Anbieter dann auch nur die bis dahin erzeugten Output-Token berechnet, steht nicht in der Quelle (bitte prüfen).

Dasselbe in JavaScript. Hier schließt break den Iterator selbst, das ist der Unterschied zu Python. Mit Node 20 ausgeführt:

async function* fakeStream(chunks) {
  try { for (const c of chunks) yield c; }
  finally { console.log("[Stream geschlossen]"); }
}
let text = "";
for await (const c of fakeStream(["Die ", "Rechnung ", "ist ", "offen."])) {
  text += c;
  if (text.length >= 10) break;
}
console.log(text);

Ausgabe: [Stream geschlossen], danach Die Rechnung (zwei Chunks gelesen, dann break).

Falle

  1. Stream mit break oder return verlassen und für geschlossen halten. In Python bleibt der Generator offen, solange eine Referenz auf ihn existiert (zum Beispiel beim Aufrufer). Schließe ihn mit close(), am besten in einem try ... finally.
  2. Kosten nur schätzen, nie loggen. Ohne eine Zeile pro Aufruf weißt du nicht, welcher Anwendungsfall die Rechnung treibt. Ob Anbieter zum Beispiel zwischengespeicherte (gecachte) Prompt-Teile anders abrechnen, steht nicht in der Quelle (bitte prüfen), deshalb rechnet unsere Funktion nur mit zwei Preisen.
  3. Einheiten vermischen. Die Preise gelten je 1000 Token. Wer die Token-Zahl direkt mit dem Preis multipliziert, rechnet um den Faktor 1000 daneben, und das fällt nicht auf, weil das Ergebnis trotzdem eine plausibel aussehende Zahl ist (Übung 1).

Übungen

Übung 1: Kostenfunktion reparieren (leicht, ca. 6 Min.)

Die Funktion kosten soll die Kosten eines Aufrufs in Geldeinheiten liefern. Die Preise kommen als Dict je 1000 Token ({"input": ..., "output": ...}). Sie läuft ohne Fehler, rechnet aber falsch. Es stecken zwei Fehler drin. Repariere sie.

Zum Prüfen: 1000 Input-Token und 0 Output-Token kosten genau den Input-Preis. 0 Input-Token und 1000 Output-Token kosten genau den Output-Preis.

Rechne die beiden Probefälle aus dem Aufgabentext von Hand durch und vergleiche das Ergebnis mit dem, was die Funktion liefert. Welche der beiden Zeilen weicht jeweils ab?

def kosten(input_tokens, output_tokens, preise_je_1k):
    kosten_in = input_tokens / 1000 * preise_je_1k["input"]
    kosten_out = output_tokens / 1000 * preise_je_1k["output"]
    return kosten_in + kosten_out
kosten

Übung 2: Kosten pro Anwendungsfall loggen und summieren (mittel, ca. 12 Min.)

Die Quelle will wissen, welcher Anwendungsfall wie viel kostet. Dafür brauchst du zwei Funktionen. Die Funktion kosten aus Übung 1 steht (richtig) bereit.

  1. logge_kosten(antwort, preise, logger, anwendung): Lies antwort.usage.input_tokens und antwort.usage.output_tokens. Schreibe genau eine Log-Zeile auf Level INFO über den übergebenen logger (kein print). Das Format ist frei, aber die Zeile muss den Namen der anwendung, die Anzahl der Input-Token, die Anzahl der Output-Token und die Kosten mit 4 Nachkommastellen enthalten. Beispiel: [rechnung_extrahieren] 812 in / 156 out / 0.0048 EUR. Gib die Kosten als Zahl zurück.
  2. kosten_je_anwendung(aufrufe, preise): aufrufe ist eine Liste von Paaren (anwendung, antwort). Das Ergebnis ist ein dict, das jeder Anwendung die Summe der Kosten aller ihrer Aufrufe zuordnet. Eine Anwendung steht nur einmal im dict, auch wenn sie mehrfach vorkommt. Eine leere Liste ergibt ein leeres dict.

Trage beide Funktionen ein und gib am Ende beide als Tupel zurück: logge_kosten, kosten_je_anwendung.

Bei der ersten Funktion: Zählwerte aus usage holen, rechnen, loggen, zurückgeben. Bei der zweiten: Du brauchst pro Anwendung einen laufenden Wert, der bei jedem Aufruf wächst. Wie liest du aus einem dict einen Wert mit Standard, wenn der Schlüssel noch fehlt?

def logge_kosten(antwort, preise, logger, anwendung):
    u = antwort.usage
    k = kosten(u.input_tokens, u.output_tokens, preise)
    logger.info("[%s] %d in / %d out / %.4f EUR", anwendung, u.input_tokens, u.output_tokens, k)
    return k

def kosten_je_anwendung(aufrufe, preise):
    summen = {}
    for anwendung, antwort in aufrufe:
        u = antwort.usage
        summen[anwendung] = summen.get(anwendung, 0) + kosten(u.input_tokens, u.output_tokens, preise)
    return summen

logge_kosten, kosten_je_anwendung

Übung 3: Stream einsammeln und abbrechen (mittel, ca. 12 Min.)

Schreibe sammle_bis(stream, max_zeichen). Sie liest Chunks aus dem Stream und setzt sie zu einem Text zusammen. Sobald der Text mindestens max_zeichen Zeichen hat, hört sie auf. Sie gibt ein Tupel (text, abgebrochen) zurück:

  • text enthält alle gelesenen Chunks ganz (der letzte Chunk wird nicht abgeschnitten),
  • abgebrochen ist True, wenn der Text max_zeichen erreicht hat, sonst False,
  • bei Abbruch wird der Stream geschlossen und es werden keine weiteren Chunks mehr geholt.

Der Check benutzt auch einen Stream ohne Ende. Wer erst alles einliest und dann kürzt, kommt dort nie an.

Eine for-Schleife über den Stream, Text anhängen, nach jedem Chunk prüfen. Was passiert mit dem Generator, wenn du die Schleife per return oder break verlässt, und welche Methode kennst du aus Schritt 3?

def sammle_bis(stream, max_zeichen):
    text = ""
    for chunk in stream:
        text += chunk
        if len(text) >= max_zeichen:
            stream.close()
            return text, True
    return text, False
sammle_bis

Merksatz

Logge jeden LLM-Aufruf mit Anwendungsfall, Tokens und Kosten, damit du summieren kannst, und schließe einen Stream aktiv, wenn du früher aufhörst: ein break oder return schließt ihn in Python nicht.

Prüfstein

Dein Dienst hat drei Anwendungsfälle, und die Rechnung des Anbieters ist höher als erwartet. Welche Angaben stehen in der Log-Zeile jedes Aufrufs, wie findest du damit den teuersten Anwendungsfall, und was passiert mit der Verbindung, wenn jemand bei einer langen Antwort auf “Stopp” klickt?

Weiter mit Teil 2: Retry mit Backoff, M1-Abschluss.

Quelle: quellen/kursbuch-lerninhalte.md, Modul M1, Baustein “05 Kosten, Streaming & Retries” (Kostenlogging, response.usage, die Beispielpreise 0,003 und 0,015 je 1000 Token als Beispielwerte, je Modell prüfen, client.messages.stream(...)). Über die Quelle hinaus (allgemeines Fachwissen, mit Python 3.13 und Node 20 ausgeführt): die Summe pro Anwendungsfall, das Schließen eines Generators mit close() und das Verhalten von break in JavaScript. Bitte prüfen: aktuelle Preise und Währung des Anbieters, text_stream im SDK, ob ein Abbruch des Streams Output-Token spart, die Abrechnung von Prompt-Caching.