Eine Aktivität erstellen

Was ist eine Aktivität?

Eine Aktivität ist die grundlegende operative Einheit in einem Workflow. Hier findet die tatsächliche Arbeit statt: Aufrufe externer APIs, Datenbankabfragen, das Versenden von E-Mails, die Verarbeitung von Dateien und Anrufe bei KI-Modellen. Die Arbeitsteilung ist einfach und dient als Kompass für diesen gesamten Kurs: Jede Operation, die Eingabe/Ausgabe (I/O) oder Nebenwirkungen (Side Effects) beinhaltet, muss in einer Aktivität verkapselt werden.

Während der Workflow die Orchestrierung übernimmt, führt die Aktivität aus. Dies ist der fundamentale Unterschied, und er hat eine direkte Konsequenz: Da riskante Arbeiten innerhalb von Aktivitäten isoliert sind, kann der Engine-Retry-Mechanismus, Messungen und Wiedergaben durchführen, ohne Nebenwirkungen zu duplizieren.

Anatomie einer Aktivität

Eine Aktivität ist eine Python-async-Funktion, die mit @workflows.activity() dekoriert ist:

import mistralai.workflows as workflows

@workflows.activity()
async def get_user_profile(user_id: str) -> dict:
    """Ruft ein Benutzerprofil über die API ab."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
        return response.json()

Vier Elemente in diesen wenigen Zeilen verdienen Ihre Aufmerksamkeit. Der Decorator @workflows.activity() registriert die Funktion als Aktivität beim Engine, sodass sie von einem Workflow aus aufgerufen werden kann. async def ist zwingend erforderlich: Alle Aktivitäten müssen asynchron sein, ohne Ausnahme. Parameter müssen JSON-serialisierbar sein, das heißt einfache Typen—str, int, float, bool, list, dict—oder Pydantic-Modelle; der Versuch, ein Verbindungsobjekt oder einen Dateihandler als Argument zu übergeben, führt zu einem Serialisierungsfehler. Dieselbe Regel gilt für den Rückgabewert, der über das Netzwerk übertragen wird, bevor er in das Ereignisprotokoll geschrieben wird.

Konkrete Beispiele für Aktivitäten

Der erste Fall, der in diesem Kurs wahrscheinlich am häufigsten vorkommt, verkapselt einen Aufruf bei einem Mistral-Modell. Beachten Sie, dass die Aktivität auch den Tokenverbrauch zurückgibt: Sobald diese Metadaten im Protokoll stehen, können Sie nachträglich die Kosten einer Ausführung rekonstruieren.

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

@workflows.activity()
async def generate_summary(text: str) -> dict:
    """Generiert eine Zusammenfassung des bereitgestellten Textes über Mistral."""
    client = Mistral()
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "Sie sind ein Assistent, der sich auf die Textzusammenfassung spezialisiert hat."},
            {"role": "user", "content": f"Fassen Sie diesen Text in 3 Sätzen zusammen:\n\n{text}"}
        ]
    )
    return {
        "summary": response.choices[0].message.content,
        "tokens_used": response.usage.total_tokens
    }

Der zweite Fall veranschaulicht den Zugriff auf eine Datenbank. Der try/finally-Block ist kein Schmuck: Wenn die Aktivität fehlschlägt und erneut versucht wird, führt das offene Verlassen einer Verbindung bei jedem Versuch schnell zur Erschöpfung des PostgreSQL-Pools.

import mistralai.workflows as workflows
import asyncpg

@workflows.activity()
async def save_report(report: dict) -> dict:
    """Speichert einen Bericht in PostgreSQL."""
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        row = await conn.fetchrow(
            "INSERT INTO reports (title, content, created_at) VALUES ($1, $2, NOW()) RETURNING id",
            report["title"],
            report["content"]
        )
        return {"id": row["id"], "status": "saved"}
    finally:
        await conn.close()

Der dritte Fall, das Versenden einer E-Mail, ist der Ort, an dem Nebenwirkungen am deutlichsten sichtbar sind: Sobald die Nachricht gesendet wurde, kann kein Retry dies auffangen.

import mistralai.workflows as workflows
import aiosmtplib

@workflows.activity()
async def send_notification(recipient: str, subject: str, body: str) -> dict:
    """Sendet eine Benachrichtigungs-E-Mail."""
    message = f"Betreff: {subject}\nAn: {recipient}\n\n{body}"
    await aiosmtplib.send(
        message,
        hostname="smtp.example.com",
        port=587,
        username="[email protected]",
        password="secret",
        use_tls=True
    )
    return {"status": "sent", "recipient": recipient}

Kernmerkmale

Zunächst eine willkommene Freiheit: Im Gegensatz zu Workflows unterliegen Aktivitäten keinen Determinismus-Beschränkungen. Sie können frei datetime.now(), uuid.uuid4(), random.random() oder andere nicht-deterministische Aufrufe innerhalb einer Aktivität verwenden. Genau dafür sind Aktivitäten gedacht – sie absorbieren alles, was ein Workflow nicht darf.

Die Plattform behandelt jeden Aktivitätsaufruf als atomare Einheit: Entweder gelingt er vollständig oder er schlägt fehl; es gibt keinen Zwischenzustand, der für den Workflow sichtbar ist. Wenn Ihre Aktivität drei Dinge tut und das zweite davon abstürzt, sieht der Workflow nur einen Fehler, keinen teilweisen Erfolg. Deshalb sollten Sie vermeiden, fünf verschiedene Operationen in einer einzigen Funktion zu stapeln.

Sobald eine Aktivität erfolgreich ist, wird ihr Ergebnis dauerhaft im Ereignisprotokoll des Workflows aufgezeichnet. Während einer Wiedergabe (Replay) wird die Aktivität niemals erneut ausgeführt: Nur das aufgezeichnete Ergebnis wird verwendet. Im Gegenzug muss eine Aktivität bei einem Fehler, der erneut versucht wird (z. B. Netzwerkzeitüberschreitung, vorübergehender Fehler), idempotent sein: Wenn sie mit denselben Parametern mehrmals ausgeführt wird, muss sie dasselbe Ergebnis liefern. Der Wechsel von POST zu PUT bei einer stabilen Identifikationsnummer reicht oft aus, um diese Eigenschaft zu erreichen, wie hier, wo die E-Mail als Schlüssel dient:

@workflows.activity()
async def create_or_update_client(email: str, name: str) -> dict:
    """Idempotent: Verwendet ein Upsert statt eines Inserts."""
    async with httpx.AsyncClient() as client:
        response = await client.put(
            f"https://api.crm.com/clients/{email}",
            json={"name": name, "email": email}
        )
        return response.json()

Benennung einer Aktivität

Der Decorator akzeptiert einen expliziten Namen, der die Lesbarkeit der Traces erheblich verbessert, wenn eine technische Funktion einen englischen Namen trägt, der von niemandem im Team mit dem geschäftlichen Bereich verknüpft ist:

@workflows.activity(name="Process Customer Emails")
async def process_customer_emails(batch_id: str) -> dict:
    # ...
    pass

Dieser Name erscheint in OpenTelemetry-Traces und erleichtert das Debugging. Machen Sie es zur Gewohnheit, dies von Anfang an einzustellen: An dem Tag, an dem Sie herausfinden müssen, warum eine Ausführung vierzig Minuten dauerte, werden Sie aussagekräftige Labels statt einer Liste von Funktionsnamen lesen.

Größenbeschränkungen

Zwei Grenzen regeln den Austausch: Ein Maximum von 2 MB für Eingabe-Parameter und ein Maximum von 2 MB für den Rückgabewert. Diese ergeben sich direkt aus dem Event Sourcing, da diese Werte serialisiert und im Protokoll gespeichert werden.

Die praktische Konsequenz ist klar. Wenn Sie große Dateien verarbeiten müssen – ein 40 MB großes gescanntes PDF, ein Export-Archiv – übergeben Sie diese nicht von einer Aktivität zur anderen. Speichern Sie sie in einem externen Dienst (S3, GCS) und zirkulieren Sie nur die URL oder die Identifikationsnummer: Die Aktivität lädt die Datei herunter, verarbeitet sie, schreibt das Ergebnis zurück in den Speicher und gibt eine Referenz von wenigen hundert Bytes zurück.

Wichtige Kernaussagen

• Eine Aktivität ist eine async-Funktion, die mit @workflows.activity() dekoriert ist und die eigentliche Arbeit (I/O, API, DB) ausführt.
• Parameter und Rückgabewerte müssen JSON-serialisierbar sein.
• Aktivitäten unterliegen keinen Determinismus-Beschränkungen.
• Eine erfolgreiche Aktivität wird während einer Wiedergabe niemals erneut ausgeführt.
• Aktivitäten müssen idempotent sein (dasselbe Ergebnis, wenn sie mehrmals ausgeführt werden).
• 2 MB Limit für Eingabe und Ausgabe pro Aktivität.