Build: Der Jev-Tausch

Die letzten beiden Lektionen sollten diese hier langweilig machen. Der Entscheidungs-Endpunkt und der Router teilen sich einen Vertrag: decide(schema, question, state) gibt einen Wert und eine Konfidenz zurück. Der Tausch gegen Jev ist deshalb eine Änderung in einer Datei. Den Router musst du dafür nicht neu schreiben. Du brauchst einen Schlüssel aus dem Early Access. Den Stand beschreibt Erste Schritte.

Die Naht

Gib dem Router eine einzige Importstelle, die keine Implementierung beim Namen nennt. Nur diese Datei ändert sich jemals:

# decider.py – die Naht. Wer die Implementierung tauscht, ändert nur diese Datei.
from decision import decide  # heute: der Stand-in mit Haiku

# from shadow import decide        # Probezeit: Stand-in antwortet, Jev läuft mit
# from jev_decider import decide   # danach: Jev

In router.py ersetzt du from decision import decide durch from decider import decide. Damit ist der Tausch vorbereitet.

Jev hinter dem Vertrag

Die Jev-Seite bildet die Schemas auf Jevs Fragetypen ab. Eine Enum-Liste wird zur Choice-Frage, ein Boolean zum Noul. Alles Offene lehnt die Funktion ab, denn Jev beantwortet nur begrenzte Fragen. Du brauchst pip install typesafe-sdk (getestet mit 0.7.5) und die Umgebungsvariable TYPESAFE_API_KEY.

# jev_decider.py – derselbe Vertrag, beantwortet von Jev
from typesafe_sdk import Choice, Noul, TypeSafeClient  # pip install typesafe-sdk

client = TypeSafeClient()  # liest TYPESAFE_API_KEY
MODEL = "jev-1.14.0"       # feste Version statt jev-latest, siehe Lektion 3


def decide(schema: dict, question: str, state: str) -> dict:
    if "enum" in schema:
        options = {str(o): o for o in schema["enum"]}
        q = Choice(instructions=question, criteria={k: None for k in options})
        r = client.system_one(model=MODEL, state=state, questions={"value": q})
        a = r.choices["value"]
        value, p = options[a.choice], a.probabilities[a.choice]
    elif schema.get("type") == "boolean":
        r = client.system_one(model=MODEL, state=state, questions={"value": Noul(instructions=question)})
        yes = r.nouls["value"].noul
        value, p = yes >= 0.5, max(yes, 1 - yes)
    else:
        raise ValueError("Jev beantwortet nur begrenzte Fragen. Das bleibt beim Sprachmodell.")
    return {
        "value": value,
        "confidence": p,  # Wahrscheinlichkeit der gewählten Option, nicht das Feld confidence
        "usage": {"input": r.usage.input_tokens, "output": r.usage.output_tokens},
    }

Erst im Schatten

Stell den Import nicht am ersten Tag um. Lass Jev erst im SchattenbetriebSchattenbetriebEin neues System beantwortet dieselben Anfragen wie das alte, aber nur fürs Log. Die Aufrufer sehen weiter die Antwort des alten Systems, bis die Messwerte den Wechsel rechtfertigen.Im Glossar laufen: Der Stand-in beantwortet weiterhin jede Anfrage. Jev beantwortet dieselbe Anfrage nebenher, und beide Antworten landen im Log. Fehler von Jev können der Produktion nicht schaden, denn noch liest niemand seine Antwort.

# shadow.py – Probezeit: Der Stand-in antwortet, Jev läuft daneben mit.
import json
import time
from concurrent.futures import ThreadPoolExecutor

import decision as standin
import jev_decider as jev

LOG_PATH = "shadow.jsonl"
pool = ThreadPoolExecutor(max_workers=4)


def _timed(fn, *args) -> tuple[dict, int]:
    t0 = time.perf_counter()
    result = fn(*args)
    return result, round((time.perf_counter() - t0) * 1000)


def decide(schema: dict, question: str, state: str) -> dict:
    shadow_future = pool.submit(_timed, jev.decide, schema, question, state)  # läuft parallel zum Stand-in
    live, live_ms = _timed(standin.decide, schema, question, state)
    try:
        shadow, shadow_ms = shadow_future.result(timeout=5)
    except Exception as exc:  # Jev-Fehler erreichen den Aufrufer nie
        shadow, shadow_ms = {"error": repr(exc)}, None
    with open(LOG_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": time.time(), "state": state,
                            "live": live, "live_ms": live_ms,
                            "shadow": shadow, "shadow_ms": shadow_ms},
                           ensure_ascii=False) + "\n")
    return live  # Aufrufer sehen nur den Stand-in

Der Code ruft beide Seiten gleichzeitig auf. Nacheinander würde jeder Aufrufer zusätzlich auf Jev warten, auch im Fehlerfall. Parallel kostet der Schattenbetrieb keine spürbare Zeit, solange Jev schneller ist als der Stand-in. Den Fehlerpfad haben wir mit einem falschen Schlüssel geprüft: Der Aufrufer bekam die Antwort des Stand-ins, im Log stand TypeSafeAuthenticationError mit Status 401.

Für einen Probelauf ohne echten Verkehr gibt es replay.py. Das Skript liest eine Datei mit gelabelten Tickets, schickt jedes durch shadow.decide und schreibt die Labels in eine eigene Datei. Für echten Verkehr setzt du in decider.py die Zeile from shadow import decide ein, und der Router schreibt shadow.jsonl von selbst.

Was du vor dem Umschalten misst

Lass den Schattenbetrieb eine Woche laufen und werte das Log dann aus. Vier Zahlen entscheiden:

• Latenz. p50 und p95 beider Seiten, gemessen aus deiner Infrastruktur. Die Herstellerangabe von 70 bis 500 Millisekunden ist ein Anhaltspunkt und ersetzt keine eigene Messung.
• Kosten. Fünf Haiku-Aufrufe gegen Jevs Eingabe-Tokens, gerechnet auf tausend Entscheidungen.
• Übereinstimmung. Wie oft beide denselben Wert liefern. Hohe Übereinstimmung bei besserer Latenz ist ein gutes Zeichen. Niedrige Übereinstimmung heißt, dass einer von beiden falsch liegt. Wer, zeigen nur Labels.
• Kalibrierung. Teil Jevs Wahrscheinlichkeiten in Bereiche ein und prüf die Genauigkeit pro Bereich gegen deine Labels. Das ist TypeSafes zentrale Behauptung, getestet auf deinem Verkehr. Dafür brauchst du einige hundert gelabelte Entscheidungen.

# report.py – Latenz, Kosten, Übereinstimmung und Genauigkeit aus shadow.jsonl
import json
from statistics import quantiles

PRICES = {"live": (0.10, 0.50), "shadow": (0.042, 0.0)}  # US-Dollar pro Million Tokens (ein, aus), Stand Nov. 2026

rows = [json.loads(l) for l in open("shadow.jsonl", encoding="utf-8")]
labels = {}
try:
    labels = {json.loads(l)["state"]: json.loads(l)["label"] for l in open("labels.jsonl", encoding="utf-8")}
except FileNotFoundError:
    pass
ok = [r for r in rows if "error" not in r["shadow"]]
print(f"{len(rows)} Entscheidungen, {len(rows) - len(ok)} Jev-Fehler")

for side in ("live", "shadow"):
    ms = [r[f"{side}_ms"] for r in ok]
    p50, p95 = quantiles(ms, n=100)[49], quantiles(ms, n=100)[94]
    cin, cout = PRICES[side]
    cost = sum(r[side]["usage"]["input"] * cin + r[side]["usage"]["output"] * cout for r in ok) / 1e6
    print(f"{side:6}  p50 {p50:6.0f} ms  p95 {p95:6.0f} ms  {cost / len(ok) * 1000:.4f} USD pro 1000")

agree = sum(r["live"]["value"] == r["shadow"]["value"] for r in ok)
print(f"Übereinstimmung: {agree} von {len(ok)}")

labelled = [r for r in ok if r["state"] in labels]
for side in ("live", "shadow"):
    hits = sum(r[side]["value"] == labels[r["state"]] for r in labelled)
    print(f"{side:6}  richtig {hits} von {len(labelled)}")

print("\nUneinig oder falsch:")
for r in labelled:
    lab = labels[r["state"]]
    if r["live"]["value"] != r["shadow"]["value"] or r["live"]["value"] != lab or r["shadow"]["value"] != lab:
        print(f'  Label {lab:14} live {r["live"]["value"]:14} {r["live"]["confidence"]:.2f}  '
              f'Jev {r["shadow"]["value"]:14} {r["shadow"]["confidence"]:.2f}  {r["state"][:60]}')

Was unser Schattenlauf zeigte

Wir haben die 20 gelabelten Tickets aus den letzten Lektionen durch replay.py geschickt (Stand: 13. Nov. 2026, von einem Laptop in Deutschland). report.py gab aus:

20 Entscheidungen, 0 Jev-Fehler
live    p50   1085 ms  p95   1462 ms  0.1686 USD pro 1000
shadow  p50    150 ms  p95    330 ms  0.0135 USD pro 1000
Übereinstimmung: 19 von 20
live    richtig 18 von 20
shadow  richtig 19 von 20

Uneinig oder falsch:
  Label password_reset live billing        1.00  Jev password_reset 0.62  Ich komme nicht mehr in mein Konto, und abgebucht wurde trot
  Label other          live billing        1.00  Jev billing        0.76  Wechselt ihr bald den Preis? Dann überlege ich, ob ich bleib

Jev war rund siebenmal so schnell und kostete pro Entscheidung etwa ein Zwölftel. Interessanter sind die beiden letzten Zeilen. Es sind genau die zwei Tickets, die zu zwei Routen passen. Haiku gab beiden fünf von fünf Stimmen. Jev gab ihnen 0,62 und 0,76.

Für den Router heißt das: Mit dem Schwellenwert 0,8 hätte Jev beide Tickets an Opus gegeben: das, bei dem Jev falsch lag, und auch das, bei dem es richtig lag. Von Jevs 20 Entscheidungen hätten 12 den schnellen Pfad genommen, ohne einen Fehler. Beim Stand-in waren es 15, davon 2 falsch.

Zwanzig Tickets sind eine Vorführung. Für eine Entscheidung über den Tausch reichen sie nicht. Die Labels stammen von uns. Ermessensfragen sind ausgerechnet die beiden mehrdeutigen Tickets. Wer Kontozugang mit Abbuchung und die Preisfrage der Abrechnung zuordnet, kommt auf 20 Richtige für Haiku und 19 für Jev. Auch eine einzelne Wiederholung kann anders ausgehen. In einem früheren Lauf gab Haiku dem Preis-Ticket 0,8 statt 1,0. Für die Kalibrierung brauchst du einige hundert Entscheidungen aus deinem eigenen Verkehr. Das Muster passt aber zu dem, was die beiden Werkzeuge versprechen: Der Stimmenanteil misst Stabilität. Jevs Wahrscheinlichkeit soll messen, wie wahrscheinlich die Antwort stimmt.

Stell den Import in decider.py um, wenn die Zahlen es rechtfertigen. Lass den Stand-in als Kommentar eine Zeile darüber stehen. Er ist dein Rückweg.

Prüf dich selbst