Die Regel der Determiniertheit — Mistral Workflows: Dauerhafte Orchestrierung

Warum Determiniertheit zwingend erforderlich ist

Determiniertheit ist die kritischste Einschränkung in Mistral-Workflows und ergibt sich direkt aus dem Replay-Mechanismus, den Sie bereits kennengelernt haben. Wenn ein Worker einen unterbrochenen Workflow wieder aufnimmt, spielt er den Workflow-Code von vorne ab und vergleicht Operation für Operation, was der Code erzeugt, mit dem, was das Historienprotokoll (History Log) aufgezeichnet hat. Solange die beiden Sequenzen übereinstimmen, ist die Wiederaufnahme unsichtbar. Sobald sie divergieren, weiß die Plattform nicht mehr, wo sie aufgehört hat: Sie erkennt die Inkonsistenz und versagt den Workflow mit einem Fehler der Nicht-Determiniertheit.

In der Praxis erzeugt ein deterministischer Workflow bei jedem Durchlauf mit denselben Eingaben exakt dieselbe Sequenz von Operationen. Beachten Sie die Nuance: Aktivitätsergebnisse können sich zwischen den Ausführungen unterscheiden, da sie aus dem Journal stammen. Es muss die Reihenfolge der Aufrufe identisch bleiben.

Wie dies im Code aussieht

Betrachten Sie einen Workflow, der basierend auf dem Abonnementstatus eines Benutzers einen Bericht an diesen sendet.

@workflows.workflow.define(name="deterministic_workflow")
class DeterministicWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # Immer dieselbe Aufrufsequenz für dieselbe Eingabe
        profile = await get_profile(user_id)
        
        if profile["premium"]:
            report = await generate_full_report(user_id)
        else:
            report = await generate_basic_report(user_id)
        
        await send_email(profile["email"], report)
        return {"status": "completed"}

Dieser Workflow ist deterministisch, obwohl er eine Verzweigung enthält. Der Grund liegt in der Quelle der Bedingung: profile["premium"] stammt aus einem Aktivitätsergebnis und somit aus dem Journal. Während des Replays gibt get_profile exakt denselben Wert zurück wie bei der ersten Ausführung, und derselbe Pfad wird gewählt.

Vergleichen Sie dies nun mit dieser Version, die auf den ersten Blick harmlos aussieht, aber alles zum Scheitern bringt.

import datetime
import random

@workflows.workflow.define(name="non_deterministic_workflow")
class NonDeterministicWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # VERBOTEN: datetime.now() ändert sich bei jeder Ausführung
        if datetime.datetime.now().hour < 12:
            await send_morning_email(user_id)
        else:
            await send_afternoon_email(user_id)
        
        # VERBOTEN: random.random() liefert jedes Mal ein anderes Ergebnis
        if random.random() > 0.5:
            await send_notification(user_id)
        
        return {"status": "completed"}

Stellen Sie sich vor, der Worker stürzt um 11:58 Uhr ab und startet um 12:03 Uhr neu. Beim Replay gibt datetime.now() eine Nachmittagszeit zurück, während die Historie einen Aufruf von send_morning_email enthält. Der Workflow versagt. random.random() erzeugt denselben Effekt, ist jedoch schwerer zu diagnostizieren.

SDK-Ersatzfunktionen

Das SDK verbietet diese Aufrufe nicht nur; es stellt für die drei häufigsten Operationen deterministische Äquivalente bereit.

from mistralai.workflows import workflow

# Aktuelle Zeit ermitteln (deterministisch während des Replays)
current_time = workflow.now()          # Anstelle von datetime.now()

# Einzigartige ID generieren (deterministisch während des Replays)
request_id = workflow.uuid4()          # Anstelle von uuid.uuid4()

# Zufallszahl ermitteln (deterministisch während des Replays)
rand_value = workflow.random()         # Anstelle von random.random()

Das Prinzip ist dasselbe wie bei Aktivitäten: Der Wert wird bei der ersten Ausführung berechnet und im Journal aufgezeichnet, danach wird er einfach wiedergegeben. Das vorherige Beispiel wird dann vollkommen sicher.

import mistralai.workflows as workflows
from mistralai.workflows import workflow

@workflows.workflow.define(name="fixed_workflow")
class FixedWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, user_id: str) -> dict:
        # KORREKT: workflow.now() ist während des Replays deterministisch
        current_hour = workflow.now()
        if current_hour.hour < 12:
            await send_morning_email(user_id)
        else:
            await send_afternoon_email(user_id)
        
        # KORREKT: workflow.random() ist während des Replays deterministisch
        if workflow.random() > 0.5:
            await send_notification(user_id)
        
        # KORREKT: workflow.uuid4() ist während des Replays deterministisch
        trace_id = workflow.uuid4()
        await log_trace(str(trace_id), user_id)
        
        return {"status": "completed", "trace_id": str(trace_id)}

Die Grenze, die zu beachten ist

Folgendes ist im Workflow-Code verboten: datetime.now(), datetime.utcnow() und time.time(); uuid.uuid4() und uuid.uuid1(); random.random(), random.randint() und random.choice(); jeglicher direkter Netzwerkaufruf (httpx, aiohttp, requests); jegliches Lesen oder Schreiben von Dateien (open(), aiofiles); jeglicher Datenbankzugriff; os.environ, da sich Umgebungsvariablen zwischen Replays ändern können; und schließlich threading und multiprocessing, da die Plattform die interne Parallelität selbst verwaltet.

Alles, was reproduzierbar ist, bleibt erlaubt: Aufrufen von Aktivitäten über await, Nutzung von workflow.now(), workflow.uuid4() und workflow.random(), Schreiben von Verzweigungslogik oder Schleifen, deren Bedingungen von Aktivitätsergebnissen abhängen, Manipulieren von Daten im Arbeitsspeicher (Berechnungen, Strings) und Protokollierung – Protokollierung gilt als akzeptabler Nebeneffekt.

Diagnose eines Nicht-Determinierheitsfehlers

Wenn der Fehler auftritt, reichen in der Regel vier Prüfungen, um ihn zu lokalisieren.

Vorhandensein nicht-deterministischer Aufrufe – Suchen Sie nach datetime.now, uuid.uuid4, random.random im Workflow-Code
Direkte I/O-Operationen – Jeglicher Netzwerkaufruf oder Dateizugriff muss in einer Aktivität stattfinden
Umgebungsvariablen – Wenn Sie os.environ für Verzweigungen nutzen, verschieben Sie das Lesen in eine Aktivität
Bibliotheken von Drittanbietern – Manche Bibliotheken führen Netzwerkaufrufe durch oder nutzen Zufälligkeit intern

Dieser letzte Punkt ist der tückischste: Eine Serialisierungslibrary, die nach einer zufällig generierten ID sortiert, oder ein HTTP-Client, der beim Laden des Moduls initialisiert wird, können einen Replay versagen, ohne dass in Ihrem Workflow eine einzige verdächtige Zeile erscheint.

Die effektivste Verteidigung besteht darin, den Workflow so schlank wie möglich zu halten und alle komplexe Logik in Aktivitäten zu verlagern. Ein Workflow, der lediglich drei Aufrufe verknüpft, hat mechanisch keine Möglichkeit, nicht-deterministisch zu sein.

# BESTE PRAKTIK: einfacher Workflow, Logik in Aktivitäten
@workflows.workflow.define(name="simple_pipeline")
class SimplePipeline:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        step1 = await prepare_data(params)
        step2 = await process_data(step1)
        step3 = await finalize(step2)
        return step3

Wichtige Erkenntnisse

Workflow-Code muss deterministisch sein: dieselben Eingaben = dieselbe Operationssequenz
Nutzen Sie workflow.now(), workflow.uuid4() und workflow.random() anstelle der Standard-Python-Äquivalente
Alle I/O-Operationen (Netzwerk, Datei, DB) müssen in einer Aktivität stattfinden, niemals im Workflow
Verzweigungen sind erlaubt, wenn sie von Aktivitätsergebnissen abhängen
Eine Verletzung der Determiniertheit führt dazu, dass der Workflow während des Replays versagt
Wenn Zweifel bestehen, verlagern Sie die Logik in eine Aktivität