Einen Worker erstellen

Die Rolle des Workers

Der Worker ist der Python-Prozess, der Ihre Workflows und Aktivitäten tatsächlich ausführt. Er verbindet sich mit der Mistral-Plattform, wartet auf ausstehende Aufgaben und führt diese lokal auf Ihrem Gerät oder Server aus. Dieses architektonische Detail überrascht Neueinsteiger oft: Ihr Code verlässt Ihre Umgebung niemals; lediglich der Orchestrierungszustand verbleibt bei Mistral.

Die Implikation ist direkt und erklärt die meisten „nichts passiert”-Probleme, die zu Beginn eines Projekts auftreten: Ohne einen laufenden Worker können Ihre Workflows nicht ausgeführt werden. Sie können eine Ausführung auslösen, aber sie wird einfach warten, bis ein geeigneter Worker verfügbar ist. Der Worker stellt die Verbindung zwischen der Mistral-Orchestrierungsplattform und Ihrem lokalen Code her.

Erstellung eines minimalen Workers

Ein funktionierender Worker passt in nur wenige Zeilen. Das folgende Beispiel kombiniert absichtlich die Aktivität, den Workflow und die Startlogik in einer einzigen Datei, sodass Sie ihn sofort ausführen können:

import asyncio
import mistralai.workflows as workflows

# Importieren Sie Ihre Workflows und Aktivitäten
@workflows.activity()
async def dire_bonjour(nom: str) -> dict:
    return {"message": f"Bonjour, {nom} !"}

@workflows.workflow.define(name="salutation_workflow")
class SalutationWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, nom: str) -> dict:
        resultat = await dire_bonjour(nom)
        return resultat

# Starten Sie den Worker
async def main() -> None:
    await workflows.run_worker([SalutationWorkflow])

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

Die Schlüsselfunktion ist workflows.run_worker(). Sie nimmt eine Liste von Workflow-Klassen entgegen, die dieser Worker ausführen kann – und ausschließlich Workflows. Aktivitäten werden über diese entdeckt, sodass Sie Aktivitäten nicht separat auflisten müssen.

Was run_worker() tut

Wenn Sie workflows.run_worker([MyWorkflow]) aufrufen, geschehen fünf Dinge nacheinander. Erst authentifiziert sich der Worker mit der Mistral-Plattform unter Verwendung Ihres API-Schlüssels. Anschließend registriert er die deklarierten Workflows und Aktivitäten. Diese Registrierung ermöglicht es der Plattform, zu erkennen, dass ein Ausführer für die Identifizierung salutation_workflow existiert. Als nächstes öffnet er eine persistente Verbindung und wartet auf Aufgaben, bleibt dabei absichtlich inaktiv, bis eine Ausführung ausgelöst wird. Wenn eine Ausführung eintrifft, führt der Worker den Code lokal aus, und die Ergebnisse werden an die Plattform zurückgesendet, um dort persistiert zu werden.

Der Worker bleibt aktiv, solange der Python-Prozess läuft, weshalb er die Kontrolle niemals an Ihr Terminal zurückgibt. Verwenden Sie Ctrl+C, um ihn sauber zu stoppen.

Worker mit mehreren Workflows

Es ist nicht erforderlich, pro Workflow einen Worker zu verwenden. Ein einzelner Prozess kann mehrere Workflows beherbergen, auch wenn sie gemeinsame Aktivitäten teilen. Hier dienen extraire_texte und analyser_sentiment beiden Zwecken:

import asyncio
import mistralai.workflows as workflows

# --- Aktivitäten ---

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

@workflows.activity()
async def analyser_sentiment(texte: str) -> dict:
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Analysez le sentiment : positif, négatif ou neutre."},
            {"role": "user", "content": texte}
        ]
    )
    return {"sentiment": response.choices[0].message.content}

@workflows.activity()
async def envoyer_alerte(message: str) -> dict:
    # Envoi vers Slack, email, etc.
    return {"status": "envoyé"}

# --- Workflows ---

@workflows.workflow.define(name="analyse_document")
class DocumentAnalysis:
    @workflows.workflow.entrypoint
    async def run(self, url: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        return sentiment

@workflows.workflow.define(name="surveillance_mentions")
class MentionMonitoring:
    @workflows.workflow.entrypoint
    async def run(self, url: str, seuil: str) -> dict:
        extraction = await extraire_texte(url)
        sentiment = await analyser_sentiment(extraction["texte"])
        if sentiment["sentiment"] == "négatif":
            await envoyer_alerte(f"Mention négative détectée : {url}")
        return sentiment

# --- Worker ---

async def main() -> None:
    await workflows.run_worker([
        DocumentAnalysis,
        MentionMonitoring,
    ])

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

Empfohlene Dateistruktur

Eine einzelne Datei funktioniert zum Lernen, nicht jedoch für ein Live-Projekt. Für die Produktion sollten Sie Ihren Code nach Typ in Module aufteilen:

mon-projet/
├── activities/
│   ├── __init__.py
│   ├── extraction.py      # extraire_texte()
│   ├── analyse.py          # analyser_sentiment()
│   └── notifications.py    # envoyer_alerte()
├── workflows/
│   ├── __init__.py
│   ├── analyse_document.py
│   └── surveillance.py
└── worker.py               # Single entry point

Die Datei worker.py enthält dann nur noch Imports und Startlogik, wodurch sie eine lesbare Inventarliste dessen wird, was dieser Prozess ausführen kann:

import asyncio
import mistralai.workflows as workflows

from workflows.document_analysis import DocumentAnalysis
from workflows.monitoring import MentionMonitoring

async def main() -> None:
    await workflows.run_worker([
        DocumentAnalysis,
        MentionMonitoring,
    ])

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

Worker starten

uv run python worker.py

Der Worker zeigt eine Verbindungsmitteilung an und bleibt aktiv, während er auf Aufgaben wartet. Sie sollten etwas Ähnliches wie Folgendes sehen:

Worker connected. Listening for tasks...

Bis diese Zeile erscheint, sollten Sie nichts auslösen. Prüfen Sie zunächst Ihren API-Schlüssel und Ihre Netzwerkverbindung.

Worker in der Produktion

Ein Worker, der manuell in einem Terminal gestartet wurde, verschwindet nach dem ersten SSH-Abbruch. Übergeben Sie ihn einem Prozessmanager – systemd, supervisor oder pm2 –, der ihn nach einem Absturz oder einem Neustart automatisch wieder startet:

# Mit pm2 (falls Node.js verfügbar ist)
pm2 start "uv run python worker.py" --name workflow-worker

# Mit systemd (eine .service-Datei erstellen)

Dieser automatische Neustart passt gut zur Langlebigkeit: Laufende Workflows setzen sich von selbst fort, sobald der neue Prozess sich verbindet. Hinsichtlich Geheimnissen sollten Sie den API-Schlüssel niemals hartcodiert in die Worker-Datei schreiben, sonst landet er in Ihrem Git-Repository. Das SDK liest die Umgebungsvariable automatisch:

import os

# Das SDK liest MISTRAL_API_KEY automatisch
# Es ist nicht erforderlich, ihn explizit zu übergeben

Schließlich sollten Sie an Protokollierung denken. Der Worker sendet Nachrichten über das Standard-logging-Modul von Python. Eine einfache Konfiguration beim Start gibt Ihnen eine Spur der ausgeführten Aktivitäten, was wertvoll ist, um zu verstehen, was während eines Vorfalls lief:

import logging
logging.basicConfig(level=logging.INFO)

Kernaussagen

• Der Worker ist der Prozess, der Ihre Workflows und Aktivitäten lokal ausführt
• workflows.run_worker([...]) nimmt eine Liste von Workflow-Klassen entgegen, die ausgeführt werden sollen
• Ein einzelner Worker kann mehrere verschiedene Workflows ausführen
• Der Worker bleibt aktiv und wartet auf Aufgaben, solange der Prozess läuft
• In der Produktion verwenden Sie einen Prozessmanager, um den Worker am Laufen zu halten
• Der API-Schlüssel wird automatisch aus MISTRAL_API_KEY gelesen