Wiederholungsstrategie

Warum Wiederholungen unerlässlich sind

In verteilten Systemen sind vorübergehende Fehler unvermeidbar: Netzwerk-Timeouts, überschrittene Rate-Limits oder vorübergehend nicht verfügbare Dienste. Keine dieser Fehler deutet auf einen Fehler in Ihrem Code hin, doch jeder einzelne kann einen Prozess zum Scheitern bringen, wenn nichts den Schock auffängt. Eine Wiederholungsstrategie ermöglicht es Ihren Aktivitäten, diese Fehler ohne manuelle Intervention zu überstehen, wobei die Plattform sie automatisch gemäß der für jede Aktivität definierten Konfiguration neu startet.

Konfigurieren einer Wiederholungsstrategie

Der Dekoratoren @workflows.activity() akzeptiert Parameter, die dieses Verhalten steuern. Hier ist eine Aktivität für einen externen API-Aufruf, die entsprechend für Resilienz gerüstet ist.

from datetime import timedelta
import mistralai.workflows as workflows

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def call_external_api(url: str, data: dict) -> dict:
    """Ruft eine externe API mit automatischer Wiederholung auf."""
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(url, json=data)
        response.raise_for_status()
        return response.json()

Drei Parameter, drei komplementäre Rollen.

start_to_close_timeout – Maximale Dauer für einen einzelnen Ausführungsversuch. Wenn die Aktivität innerhalb dieses Zeitraums nicht abgeschlossen ist, wird sie als fehlgeschlagen betrachtet und ein neuer Versuch wird gestartet.
retry_policy_max_attempts – Maximale Anzahl der Versuche. Der Standardwert hängt von der Plattform ab. Nach dieser Anzahl an Versuchen scheitert die Aktivität endgültig und die Workflow-Einheit erhält einen Fehler.
retry_policy_backoff_coefficient – Multiplikator für die Verzögerung zwischen den Versuchen. Bei einem Koeffizienten von 2,0 verdoppeln sich die Verzögerungen bei jedem Versuch (1s, 2s, 4s, 8s…). Dies ist ein exponentieller Backoff.

Dieser Backoff-Mechanismus verdient besondere Aufmerksamkeit, da sein Zweck nicht nur darin besteht, zu warten: Er gibt dem kämpfenden Dienst Zeit zur Erholung, anstatt ihn zu überlasten. Der Verlauf eines Aufrufs, der schließlich erfolgreich ist, illustriert dies deutlich.

Versuch 1: sofortige Ausführung
  → Fehler (Netzwerk-Timeout)
  → Warten 1 Sekunde

Versuch 2: Ausführung
  → Fehler (Dienst nicht verfügbar)
  → Warten 2 Sekunden (1 × 2,0)

Versuch 3: Ausführung
  → Fehler (Rate-Limit)
  → Warten 4 Sekunden (2 × 2,0)

Versuch 4: Ausführung
  → Erfolg!

Bei einer Antwort “429 Too Many Requests” ist dieser Mechanismus besonders effektiv: Eine Strategie, die sofort wiederholt, würde die Sättigung lediglich aufrechterhalten, wohingegen das Erhöhen der Abstände schließlich unter die Schwelle des Anbieters fällt.

`start_to_close_timeout` niemals weglassen

Dieser Timeout ist kritisch und sollte auf jeder Aktivität ohne Ausnahme erscheinen. Ohne ihn wartet eine Aktivität, die stecken bleibt – Deadlock, hängende Verbindung, Dienst, der die Anfrage akzeptiert, aber nie antwortet – unendlich lang und friert den gesamten Workflow ein. Schlimmer noch: Es löst keine Wiederholung aus, da aus seiner Perspektive noch kein Fehler vorliegt.

# SCHLECHTE PRAXIS: kein Timeout
@workflows.activity()
async def dangerous_activity(params: dict) -> dict:
    # Wenn der Dienst nie antwortet, wartet die Aktivität unendlich lang
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.lente.example.com/data")
    return response.json()

# GUTE PRAXIS: expliziter Timeout
@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2)
)
async def safe_activity(params: dict) -> dict:
    async with httpx.AsyncClient(timeout=90) as client:
        response = await client.get("https://api.lente.example.com/data")
    return response.json()

Um den Wert festzulegen, messen Sie zunächst die normale Ausführungszeit der Aktivität und fügen Sie einen Puffer von zwei bis drei Mal hinzu, um langsame Fälle abzudecken. Stellen Sie schließlich sicher, dass der interne HTTP-Client-Timeout niedriger als der start_to_close_timeout ist – im obigen Beispiel 90 Sekunden gegenüber 2 Minuten –, damit der Fehler sauber aus Ihrem Code propagiert wird und nicht von der Plattform abgeschnitten wird.

Idempotenz: eine nicht verhandelbare Bedingung

Da eine Aktivität mit denselben Parametern mehrfach wiedergegeben werden kann, muss sie unabhängig von der Anzahl der Ausführungen denselben Effekt erzeugen. Diese Eigenschaft wird Idempotenz genannt, und ihr Fehlen verwandelt eine Wiederholungsstrategie in einen Incident-Generator.

@workflows.activity(retry_policy_max_attempts=3)
async def add_credit(user_id: str, amount: float) -> dict:
    """GEFAHRLICH: Wenn wiederholt, wird Guthaben mehrfach hinzugefügt!"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://api.billing.com/credits/add",
            json={"user_id": user_id, "amount": amount}
        )
    return response.json()

Das gefürchtete Szenario ist häufig: Die Billing-API verarbeitet die Anfrage korrekt, aber die Antwort geht im Netzwerk verloren. Die Aktivität schließt mit einem Fehler ab, die Plattform wiederholt den Versuch, und der Kunde erhält das Guthaben doppelt. Ein Idempotenz-Schlüssel eliminiert das Problem, indem er dem entfernten Dienst eine Möglichkeit gibt, eine bereits verarbeitete Anfrage zu erkennen.

@workflows.activity(retry_policy_max_attempts=3)
async def add_credit(user_id: str, amount: float, idempotency_key: str) -> dict:
    """KORREKT: Idempotenz-Schlüssel verhindert Duplizierung."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"https://api.billing.com/credits/add",
            json={
                "user_id": user_id,
                "amount": amount,
                "idempotency_key": idempotency_key
            }
        )
    return response.json()

Generieren Sie den Schlüssel im Workflow mit workflow.uuid4(), was garantiert, dass er bei jeder Wiedergabe identisch bleibt – ein klassisches uuid.uuid4() würde bei jedem Versuch einen neuen Wert erzeugen und alle Schutzmaßnahmen zunichtemachen. Wenn der entfernte Dienst diesen Mechanismus nicht anbietet, bestehen zwei weitere Wege: Upsert in der Datenbank, indem INSERT ... ON CONFLICT UPDATE statt nur INSERT verwendet wird, oder eine Vorab-Prüfung, die eine Abfrage des Zustands vor der Aktion beinhaltet.

Drei Konfigurationsprofile

Werte werden entsprechend der Art der Operation kalibriert. Eine Zahlung oder Benachrichtigung muss schnell erfolgreich oder schnell fehlschlagen: kurzer Timeout, drei Versuche, moderater Backoff.

@workflows.activity(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy_max_attempts=3,
    retry_policy_backoff_coefficient=2.0,
)
async def process_payment(params: dict) -> dict:
    # ...
    pass

Läufige Verarbeitung, wie Berichterstellung oder schwere Inferenz, invertiert die Logik: Jeder Versuch kostet erhebliche Rechenzeit, daher autorisieren Sie weniger Versuche mit größeren Abständen.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=15),
    retry_policy_max_attempts=2,
    retry_policy_backoff_coefficient=3.0,
)
async def generate_ml_report(params: dict) -> dict:
    # ...
    pass

Bei einer instabilen Drittanbieter-API, die Rate-Limits unterliegt, nehmen Sie schließlich Persistenz an: fünf Versuche, auch wenn dies mehrere Minuten dauert.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2),
    retry_policy_max_attempts=5,
    retry_policy_backoff_coefficient=2.0,
)
async def call_third_party_api(params: dict) -> dict:
    # ...
    pass

Kernaussagen