Validierung und Retry-Strategien
Auf dieser Seite
Validierung: Die letzte Verteidigungslinie vor dem Produktivbetrieb
Selbst bei Verwendung von benutzerdefinierten strukturierten Ausgaben bleibt die Validierung unerlässlich. Das Schema garantiert die Struktur, nicht jedoch die semantische Kohärenz. Ein Feld price könnte beispielsweise -999.0 enthalten: ein gültiger Float, der das Schema perfekt erfüllt, aber für einen Preis keinen Sinn ergibt.
Der Unterschied ist klar. Pydantic und Zod übernehmen die strukturelle Validierung automatisch: erforderliche Felder, Datentypen, Enum-Beschränkungen und verschachtelte Strukturen. Sie müssen hierfür keinen Code schreiben. Die semantische Validierung liegt jedoch vollständig in Ihrer Verantwortung. Dazu gehören numerische Grenzen (ein Preis größer als null, eine Bewertung zwischen 0 und 1), die Konsistenz zwischen Feldern (die Summe muss der Summe der Positionen entsprechen), String-Formate (eine wirklich gültige E-Mail-Adresse, ein korrekt formatiertes Datum) und Geschäftslogik (ein Fälligkeitsdatum, das nach dem Rechnungsdatum liegt).
Geschäftsregeln im Schema verankern
Pydantic ermöglicht es, diese Prüfungen direkt in das Modell zu integrieren, sodass sie untrennbar mit der Struktur verbunden sind. @field_validator übernimmt die Prüfung einzelner Felder – beispielsweise die Sicherstellung, dass eine Menge positiv ist oder der Einheitspreis nicht negativ. @model_validator(mode="after") wird ausgeführt, sobald alle Felder besetzt sind, und ermöglicht Prüfungen über mehrere Felder hinweg, wie die Überprüfung, ob die angegebene Summe der Menge mal Einheitspreis innerhalb von einem Cent entspricht.
from pydantic import BaseModel, Field, field_validator, model_validator
class InvoiceLine(BaseModel):
description: str
quantity: float
unit_price: float
total: float
@field_validator("quantity")
@classmethod
def quantity_positive(cls, v):
if v <= 0:
raise ValueError(f"Menge muss positiv sein, erhalten: {v}")
return v
@field_validator("unit_price")
@classmethod
def price_non_negative(cls, v):
if v < 0:
raise ValueError(f"Einheitspreis kann nicht negativ sein: {v}")
return v
@model_validator(mode="after")
def check_total(self):
expected = round(self.quantity * self.unit_price, 2)
if abs(self.total - expected) > 0.01:
raise ValueError(
f"Inkonsistente Summe: {self.total} != {self.quantity} x {self.unit_price} = {expected}"
)
return selfZod drückt dieselbe Logik aus, indem es eingebaute Einschränkungen wie .positive() und .nonnegative() für Felder sowie .refine() für Regeln über das gesamte Objekt hinweg verwendet.
import { z } from "zod";
const InvoiceLineSchema = z
.object({
description: z.string(),
quantity: z.number().positive("Menge muss positiv sein"),
unitPrice: z.number().nonnegative("Preis kann nicht negativ sein"),
total: z.number(),
})
.refine(
(data) => Math.abs(data.total - data.quantity * data.unitPrice) < 0.01,
{ message: "Summe entspricht nicht Menge x Einheitspreis" }
);Retry mit Feedback
Wenn die Validierung fehlschlägt, ist das effektivste Muster, den Fehler an das Modell zurückzusenden, damit es sich selbst korrigieren kann. Die folgende Funktion verknüpft drei Prüfungen in Reihenfolge ihrer Wichtigkeit: vollständige Antwort, nicht-null-Objekt und Einhaltung der Geschäftsregeln. Bei einem Fehler fügt sie die fehlerhafte Antwort und die Fehlermeldung zum Gespräch hinzu, wodurch der nächsten Versuchsversuch präziser diagnostischer Kontext zur Verfügung steht. Beachten Sie die Kopie der Nachrichtenliste am Anfang der Funktion – ohne diese würden Sie die Gesprächshistorie des Aufrufers verseuchen.
from mistralai.client import Mistral
from pydantic import BaseModel, ValidationError
import json
def extract_with_retry(
client: Mistral,
messages: list,
schema,
max_retries: int = 3,
model: str = "mistral-large-latest"
):
"""Extraktion mit Retry und strukturiertem Fehlerfeedback."""
conversation = list(messages) # Kopie erstellen, um das Original nicht zu verändern
for attempt in range(max_retries):
try:
response = client.chat.parse(
model=model,
messages=conversation,
response_format=schema,
temperature=0,
max_tokens=2048
)
# Prüfen, ob die Antwort vollständig ist
if response.choices[0].finish_reason != "stop":
raise ValueError(
f"Unvollständige Antwort (finish_reason={response.choices[0].finish_reason})"
)
parsed = response.choices[0].message.parsed
if parsed is None:
raise ValueError("Geparsedte Antwort ist None")
# Zusätzliche semantische Validierung
validate_business_rules(parsed)
return parsed
except (ValidationError, ValueError) as e:
error_msg = str(e)
print(f"Versuch {attempt + 1}/{max_retries} fehlgeschlagen: {error_msg}")
if attempt < max_retries - 1:
# Feedback zum Kontext hinzufügen
raw = response.choices[0].message.content if response else ""
conversation.append({"role": "assistant", "content": raw})
conversation.append({
"role": "user",
"content": f"Ihre Antwort enthält einen Fehler: {error_msg}. Bitte korrigieren Sie diese."
})
else:
raise RuntimeError(
f"Fehlschlag nach {max_retries} Versuchen. Letzter Fehler: {error_msg}"
)Eine Kaskade aus Fallback-Plänen
Retry löst nicht alles: manchmal ist ein Dokument für das Schema zu untypisch. Anstatt abrupt zu versagen, degradieren Sie sanft. Der erste Versuch nutzt benutzerdefinierte strukturierte Ausgaben, die zuverlässigste Option. Scheitert dieser, wechseln Sie auf JSON Mode, der toleranter ist, und leiten Sie die Ausgabe dann durch model_validate() weiter – dies behält die Schema-Einschränkungen bei, ohne darauf angewiesen zu sein, dass das Modell sie perfekt generiert. Als letzter Rückgriff gibt die Funktion None zurück und überlässt es dem Aufrufer, zwischen manueller Warteschlangeneintragung oder Aufgabe zu entscheiden.
def extract_with_fallback(client, text, schema, system_prompt):
"""Extraktion mit Fallback-Kaskade."""
# Versuch 1: Benutzerdefinierte strukturierte Ausgabe (bestes Ergebnis)
try:
return extract_with_retry(
client,
[
{"role": "system", "content": system_prompt},
{"role": "user", "content": text}
],
schema,
max_retries=2,
model="mistral-large-latest"
)
except RuntimeError:
print("Benutzerdefiniertes Parsing fehlgeschlagen, Wechsel auf JSON Mode")
# Versuch 2: JSON Mode (flexibler)
try:
response = client.chat.complete(
model="mistral-large-latest",
messages=[
{"role": "system", "content": system_prompt + "\nGeben Sie gültiges JSON zurück."},
{"role": "user", "content": text}
],
response_format={"type": "json_object"},
temperature=0
)
raw = json.loads(response.choices[0].message.content)
return schema.model_validate(raw) # Pydantic-Validierung versuchen
except Exception:
print("JSON Mode fehlgeschlagen, Wechsel auf manuelle Extraktion")
# Versuch 3: Teilobjekt oder Fehler zurückgeben
return NoneDen Wasserhahn rechtzeitig abdrehen
In einem Hochfrequenz-Pipeline verwandelt eine API-Ausfall Retry in einen Verstärker: jedes Dokument schlägt dreimal fehl, die Rechnung verdreifacht sich und die Latenz steigt. Ein Circuit Breaker überwacht Fehler in einem gleitenden Fenster und blockiert Aufrufe, sobald ein Schwellenwert überschritten wird – fünf Fehler in 60 Sekunden in der hier gezeigten Konfiguration. Ein Erfolg schließt den Circuit und setzt den Zähler zurück.
from collections import deque
from time import time
class CircuitBreaker:
def __init__(self, failure_threshold=5, window_seconds=60):
self.failures = deque()
self.threshold = failure_threshold
self.window = window_seconds
self.is_open = False
def record_failure(self):
now = time()
self.failures.append(now)
# Fehler außerhalb des Fensters bereinigen
while self.failures and self.failures[0] < now - self.window:
self.failures.popleft()
if len(self.failures) >= self.threshold:
self.is_open = True
def record_success(self):
self.is_open = False
self.failures.clear()
def can_proceed(self) -> bool:
if not self.is_open:
return True
# Prüfen, ob das Fenster abgelaufen ist
now = time()
while self.failures and self.failures[0] < now - self.window:
self.failures.popleft()
if len(self.failures) < self.threshold:
self.is_open = False
return True
return False
# Verwendung
breaker = CircuitBreaker(failure_threshold=5, window_seconds=60)
def safe_extract(client, text, schema):
if not breaker.can_proceed():
raise RuntimeError("Circuit offen – zu viele kürzliche Fehler")
try:
result = extract_with_retry(client, [...], schema)
breaker.record_success()
return result
except RuntimeError:
breaker.record_failure()
raiseSichtbar machen
Alle diese Mechanismen bleiben blind ohne Spuren. Protokollieren Sie jeden Aufruf mit seiner Dauer und dem beteiligten Schema, unabhängig davon, ob er erfolgreich war oder nicht. Diese Daten ermöglichen es Ihnen später zu bestimmen, ob die Degradation auf ein Modellproblem, eine Dokumentart oder eine Prompt-Änderung zurückzuführen ist.
import logging
import time
logger = logging.getLogger("structured_output")
def monitored_extract(client, text, schema, **kwargs):
start = time.time()
try:
result = extract_with_retry(client, [...], schema, **kwargs)
duration = time.time() - start
logger.info(f"Extraktion erfolgreich in {duration:.2f}s — schema={schema.__name__}")
return result
except Exception as e:
duration = time.time() - start
logger.error(f"Extraktion fehlgeschlagen in {duration:.2f}s — schema={schema.__name__} — {e}")
raiseKernaussagen
- Strukturelle Validierung (Typen) ist automatisch; semantische Validierung (Logik) liegt in Ihrer Verantwortung
- Retry mit Feedback ist das effektivste Muster: Senden Sie den Fehler an das Modell zurück
- Implementieren Sie eine Fallback-Kaskade: Benutzerdefiniert > JSON Mode > Standardwert
- Verwenden Sie in hochfrequenten Produktivsystemen einen Circuit Breaker, um Fehlerkaskaden zu verhindern
- Protokollieren Sie jeden Aufruf zur Überwachung und Diagnose