Ausführen und Testen eines Workflows

Lektion aktualisiert am 3. November 2026 basierend auf der mistralai SDK-Version 3.7.0: Das Verfolgen einer Ausführung erfolgt nun über client.workflows.executions.get_workflow_execution_async, und das Ergebnis wird aus result gelesen. Nachtrag am selben Tag: Die vorherigen Beispiele verwiesen noch auf execution.id, execution.output und execution.error, die im SDK nicht vorhanden sind, und verglichen den Status mit den Kleinbuchstaben-Werten "completed" oder "failed". Sie verwenden nun execution_id, result, wait_for_result=True und Großbuchstaben-Statuswerte. Quelle: client-python.

Workflow auslösen

Sie haben Ihre Aktivitäten geschrieben, Ihren Workflow definiert und Ihren Worker gestartet. Nun müssen Sie eine Ausführung starten. Sie haben zwei Optionen: das Python SDK oder einen cURL-Aufruf an die REST-API. Die erste Option passt zu Ihren Python-Skripten und Diensten; die zweite eignet sich für schnelle Befehlszeilentests oder Auslöser aus nicht-Python-Systemen – einem Webhook, einem Scheduler oder einem Backend, das in einer anderen Programmiersprache geschrieben ist.

Bevor Sie etwas auslösen können, müssen drei Bedingungen erfüllt sein. Ihr Worker muss laufen (uv run python worker.py), Ihr API-Schlüssel muss in MISTRAL_API_KEY gesetzt sein, und der Workflow muss bei diesem Worker registriert sein, was bedeutet, dass er in der Liste erscheint, die an run_worker() übergeben wird.

Auslösen über das Python SDK

Schreiben Sie den Auslöser in einem Skript, das vom Worker getrennt ist, da der Worker-Prozess die Kontrolle niemals zurückgibt:

import asyncio
from mistralai.client import Mistral

async def main():
    client = Mistral()

    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="salutation_workflow",
        input={"nom": "Marie"},
        wait_for_result=True,  # warten auf Abschluss und Ergebnis empfangen
    )

    print(f"Ausführungs-ID: {execution.execution_id}")
    print(f"Ergebnis: {execution.result}")

if __name__ == "__main__":
    asyncio.run(main())

Der workflow_identifier stimmt exakt mit dem name überein, das in @workflows.workflow.define() deklariert wurde, und die Schlüssel im input-Dictionary entsprechen den Parametern Ihrer run-Methode. Führen Sie dieses Skript in einem separaten Terminal aus, während das erste Terminal vom Worker belegt wird:

uv run python trigger.py

Mit wait_for_result=True wartet das SDK auf den Abschluss der Ausführung und gibt das Ergebnis direkt zurück, was das Testen für kurze Workflows sehr lesbar macht. Ohne diesen Parameter gibt der Aufruf sofort die Ausführungs-ID zurück, die Sie dann verfolgen müssen (siehe unten).

Auslösen über cURL

Für schnelle Tests oder die Integration mit anderen Systemen stellt die REST-API denselben Auslöser bereit:

curl -X POST https://api.mistral.ai/v1/workflows/salutation_workflow/execute \
  -H "Authorization: Bearer $MISTRAL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"input": {"nom": "Marie"}, "wait_for_result": true}'

Mit wait_for_result enthält die Antwort den Workflow-Namen, die Ausführungs-ID und das Ergebnis; dies sind die Felder des SDK-Modells WorkflowExecutionSyncResponse:

{
  "workflow_name": "salutation_workflow",
  "execution_id": "…",
  "result": {
    "message": "Bonjour, Marie !"
  }
}

Behalten Sie diese execution_id bei: Sie ist der Schlüssel, den Sie später verwenden, um die Ausführung abzufragen oder ihre Spuren (traces) abzurufen. Ohne wait_for_result beschreibt die Antwort die laufende Ausführung, einschließlich ihres status (RUNNING zunächst).

End-to-End-Beispiel

Schauen wir uns das Beispiel für die Dokumentenanalyse-Pipeline erneut an, diesmal vollständig konfiguriert. Die erste Datei kombiniert die Definitionen und den Worker-Start; beachten Sie die Trunkierung auf 5000 Zeichen, die sowohl die Aktivitätsgrößenbegrenzung als auch Ihr Token-Budget schützt.

Datei workflow.py (Worker + Definitionen)

import asyncio
import httpx
import mistralai.workflows as workflows
from mistralai.client import Mistral

@workflows.activity()
async def extraire_texte(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    return {"texte": response.text[:5000]}  # Begrenzung auf 5000 Zeichen

@workflows.activity()
async def resumer_texte(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Fassen Sie in 3 Sätzen zusammen."},
            {"role": "user", "content": texte}
        ]
    )
    return {"resume": response.choices[0].message.content}

@workflows.workflow.define(name="resume_page_web")
class ResumePageWeb:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        resume = await resumer_texte(extraction["texte"])
        return {
            "url_source": url,
            "resume": resume["resume"]
        }

async def main() -> None:
    await workflows.run_worker([ResumePageWeb])

if __name__ == "__main__":
    asyncio.run(main())

Die zweite Datei löst den Workflow aus und zeigt das Ergebnis an, wobei der Fehlerfall explizit behandelt wird – eine Reflex, die Sie aus Ihrem ersten Skript übernommen haben sollten:

Datei trigger.py (Auslöser)

import asyncio
from mistralai.client import Mistral

async def main():
    client = Mistral()

    print("Workflow starten...")
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="resume_page_web",
        input={"url": "https://example.com/article"},
    )
    print(f"Ausführung gestartet: {execution.execution_id}")

    # Ausführung bis zum Abschluss verfolgen
    while True:
        suivi = await client.workflows.executions.get_workflow_execution_async(
            execution_id=execution.execution_id
        )
        if suivi.status not in ("RUNNING", "RETRYING_AFTER_ERROR"):
            break
        await asyncio.sleep(2)

    if suivi.status == "COMPLETED":
        print(f"Zusammenfassung: {suivi.result['resume']}")
    else:
        print(f"Ausführung beendet mit Status {suivi.status}")

if __name__ == "__main__":
    asyncio.run(main())

Ausführung

Terminal 1 — Worker starten:

uv run python workflow.py

Terminal 2 — Ausführung auslösen:

uv run python trigger.py

Ergebnisse prüfen

Das Worker-Terminal ist Ihr erster Beobachtungspunkt: Es zeigt ausgeführte Aktivitäten in Echtzeit an, zusammen mit ihrer Dauer, sodass Sie sofort erkennen können, welcher Schritt die meiste Zeit verbraucht.

[INFO] Aktivität ausführen: extraire_texte
[INFO] Aktivität abgeschlossen: extraire_texte (1.2s)
[INFO] Aktivität ausführen: resumer_texte
[INFO] Aktivität abgeschlossen: resumer_texte (3.4s)
[INFO] Workflow abgeschlossen: resume_page_web

Bei langen Workflows werden Sie natürlich nicht am Terminal kleben bleiben. Sobald Sie die Ausführungs-ID haben, können Sie den Status jederzeit von einem anderen Prozess aus abfragen:

execution = await client.workflows.executions.get_workflow_execution_async(
    execution_id="exec_abc123"
)
print(f"Status: {execution.status}")
print(f"Ergebnis: {execution.result}")

Dieser Vorgang ist auch über cURL verfügbar, was in einem Überwachungsskript nützlich ist:

curl https://api.mistral.ai/v1/workflows/executions/exec_abc123 \
  -H "Authorization: Bearer $MISTRAL_API_KEY"

Fehlerbehandlung

Ein Fehler tritt nicht beim ersten Ausnahmefall auf: Die Plattform versucht zunächst, die Aktivität gemäß ihrer Retry-Richtlinie erneut auszuführen. Erst wenn alle Wiederholungsversuche erschöpft sind, wechselt die Ausführung in den Status FAILED (SDK-Statuswerte sind in Großbuchstaben: RUNNING, COMPLETED, FAILED, CANCELED, TERMINATED, TIMED_OUT usw.), und Ihr aufrufender Code muss diesen Fall behandeln, anstatt davon auszugehen, dass ein result immer vorhanden ist:

execution = await client.workflows.executions.get_workflow_execution_async(
    execution_id="exec_abc123"
)

if execution.status == "FAILED":
    print(f"Ausführung {execution.execution_id} fehlgeschlagen")

Die Antwort enthält keine Fehlermeldung: Die Ursache erscheint in der Ausführungsverlaufsgeschichte und den Protokollen (get_workflow_execution_history_async, get_workflow_execution_logs_async). Um die fehlerhafte Aktivität und die dazugehörige Sequenz zu identifizieren, analysieren Sie OpenTelemetry-Spuren, die in Lektion 14 behandelt werden.

Lokales Testen

Das Hin- und Herschalten zwischen zwei Terminals wird während der Entwicklung schnell mühsam. Für die Entwicklung kann ein einzelnes Skript den Worker im Hintergrund starten, den Workflow auslösen und dann alles stoppen:

import asyncio
import mistralai.workflows as workflows

# ... Aktivitäts- und Workflow-Definitionen ...

async def main():
    # Worker im Hintergrund starten
    worker_task = asyncio.create_task(
        workflows.run_worker([MonWorkflow])
    )

    # Warten, bis der Worker bereit ist
    await asyncio.sleep(2)

    # Workflow auslösen
    from mistralai.client import Mistral
    client = Mistral()
    execution = await client.workflows.execute_workflow_async(
        workflow_identifier="mon_workflow",
        input={"test": True},
        wait_for_result=True,
    )
    print(f"Ergebnis: {execution.result}")

    # Worker stoppen
    worker_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())

Das asyncio.sleep(2) gibt dem Worker Zeit, seine Verbindung herzustellen, bevor der Auslöser aktiviert wird, und das finale worker_task.cancel() verhindert, dass das Skript unendlich lange hängen bleibt. Reservieren Sie diese Einrichtung für die Entwicklung: In der Produktion bleiben Worker und Auslöser als separate Prozesse mit unabhängigen Lebenszyklen bestehen.

Kernaussagen

• Zwei Auslösemethoden: Python SDK (execute_workflow_async) und REST-API (cURL)
• Der Worker muss laufen, damit der Workflow ausgeführt werden kann
• Mit wait_for_result=True wartet das SDK auf den Abschluss und gibt das Ergebnis (result) zurück; andernfalls wird die execution_id zurückgegeben, um die Ausführung zu verfolgen
• Verwenden Sie client.workflows.executions.get_workflow_execution_async(), um den Status einer Ausführung zu prüfen
• Statuswerte sind in Großbuchstaben (RUNNING, COMPLETED, FAILED usw.); Fehlerursachen erscheinen im Ausführungsverlauf, in den Protokollen und in den Spuren
• Für die Entwicklung können Sie Worker und Auslöser in einem einzigen Skript kombinieren, indem Sie asyncio verwenden