Asynchrone I/O und Best Practices

Asynchrone I/O: Eine technische Anforderung

Alle Aktivitäten in Mistral Workflows müssen asynchrone Eingabe- und Ausgabeoperationen verwenden. Dies ist keine stilistische Empfehlung, sondern eine technische Einschränkung. Ein blockierender synchroner Aufruf wie requests.post() friert den Worker ein und verhindert, dass andere Aktivitäten ausgeführt werden.

Diese Anforderung ergibt sich aus der Architektur des Workers, der Aktivitäten innerhalb eines asyncio-Event-Loops ausführt. Wenn eine Coroutine asynchron auf eine Netzwerkanwort wartet, gibt sie die Kontrolle an den Loop zurück, sodass andere Aktivitäten fortfahren können. Ein synchroner Aufruf hingegen gibt nichts zurück.

import requests

# FALSCH: Blockiert den gesamten Worker
@workflows.activity()
async def blocking_activity(url: str) -> dict:
    # requests.post() ist synchron — es blockiert den Event-Loop
    response = requests.post(url, json={"query": "test"})
    return response.json()

Das Symptom in der Produktion kann verwirrend sein: Sie starten zehn Aktivitäten parallel mit asyncio.gather, sie führen jedoch nacheinander aus. Wenn jede zwei Sekunden auf eine externe API wartet, messen Sie zwanzig Sekunden anstatt der erwarteten zwei, ohne dass Fehler geworfen werden.

Wählen Sie nativ asynchrone Bibliotheken

Die Lösung besteht darin, jede blockierende Bibliothek durch ihre asynchrone Entsprechung zu ersetzen. Für HTTP-Aufrufe ist httpx die Standardwahl.

import httpx

# RICHTIG: Nicht blockierend
@workflows.activity()
async def async_activity(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json={"query": "test"})
    return response.json()

aiohttp erfüllt denselben Zweck mit einer leicht unterschiedlichen API, die auf Sitzungen basiert.

import aiohttp

@workflows.activity()
async def aiohttp_activity(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json={"query": "test"}) as response:
            return await response.json()

Für Datenbanken ersetzt asyncpg psycopg2 für PostgreSQL. Beachten Sie den try/finally-Block: Die Verbindung muss geschlossen werden, auch wenn die Abfrage eine Ausnahme wirft. Andernfalls werden bei wiederholten Versuchen die Verbindungsressourcen erschöpft.

import asyncpg

# RICHTIG: Asynchroner PostgreSQL-Treiber
@workflows.activity()
async def async_db_query(user_id: str) -> dict:
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        row = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", user_id
        )
        return dict(row) if row else {"error": "user not found"}
    finally:
        await conn.close()

Der Dateizugriff folgt derselben Logik mit aiofiles, selbst für kleine Dateien, bei denen Sie geneigt sein könnten, open() weiterhin zu verwenden.

import aiofiles

# RICHTIG: Asynchrones Datei-Lesen
@workflows.activity()
async def read_file_async(path: str) -> dict:
    async with aiofiles.open(path, mode="r") as f:
        content = await f.read()
    return {"content": content, "size": len(content)}

Auch die Mistral SDK ist keine Ausnahme: Jede Methode hat eine Variante, die mit _async endet, und dies ist diejenige, die Sie von einer Aktivität aus aufrufen müssen.

from mistralai.client import Mistral

@workflows.activity()
async def call_mistral(prompt: str) -> dict:
    client = Mistral()
    # Verwenden Sie complete_async, NICHT complete
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[{"role": "user", "content": prompt}]
    )
    return {"response": response.choices[0].message.content}

Halten Sie diese Zuordnungstabelle während der Code-Reviews griffbereit: Jede Zeile aus der linken Spalte, die in einer Aktivität entdeckt wird, ist ein potenzielles Performance-Problem.

Wenn keine asynchrone Version existiert

Manche Geschäftsbibliotheken, wie ein proprietäres SDK oder ein Legacy-Client, bieten einfach keine asynchrone Variante an. In diesen Fällen ermöglicht asyncio.to_thread(), den blockierenden Code in einem separaten Thread auszuführen, ohne den Event-Loop zu frieren.

import asyncio
from some_library import blocking_function

@workflows.activity()
async def compatibility_activity(params: dict) -> dict:
    """Hüllt einen blockierenden Aufruf in einen Thread ein."""
    result = await asyncio.to_thread(
        blocking_function,
        params["arg1"],
        params["arg2"]
    )
    return {"result": result}

Dies ist ein Kompromiss, keine äquivalente Lösung: Jeder Aufruf verbraucht einen Thread aus dem Pool, der eine endliche Größe hat. Wenn eine Aktivität massiv parallel aufgerufen wird, verschieben Sie lediglich die Engstelle, anstatt sie zu beseitigen. Bevorzugen Sie immer eine nativ asynchrone Bibliothek, wenn eine verfügbar ist.

Entwurf sauberer Aktivitäten

Neben der Asynchronität unterscheiden vier Gewohnheiten eine robuste Aktivität von einer, die Ihnen Stunden an Debugging-Kosten verursachen könnte.

Die erste ist der Umfang. Eine Aktivität sollte genau eine Verantwortung haben, da sie auch die Einheit für Wiederholungsversuche und Persistenz ist: Eine Aktivität, die alles versucht und beim Senden einer E-Mail fehlschlägt, wird die Extraktion und Analyse von vorne wiederholen.

# GUT: Jede Aktivität hat eine klare Verantwortung
@workflows.activity()
async def extract_text(url: str) -> dict:
    # ...
    pass

@workflows.activity()
async def analyze_sentiment(text: str) -> dict:
    # ...
    pass

# SCHLECHT: Alles-in-einer-Aktivität
@workflows.activity()
async def do_everything(url: str) -> dict:
    text = ...  # Extraktion
    sentiment = ...  # Analyse
    # E-Mail senden
    # In DB speichern
    pass

Die zweite betrifft Ressourcen. Der async with-Context-Manager garantiert, dass der Client auch bei einer Ausnahme geschlossen wird, etwas, das manuelle Aufrufe früher oder später unvermeidlich vergessen.

@workflows.activity()
async def clean_request(params: dict) -> dict:
    # Verwenden Sie async with, um das Schließen zu garantieren
    async with httpx.AsyncClient() as client:
        response = await client.get(params["url"])
    # Der Client wird hier automatisch geschlossen
    return response.json()

Die dritte ist subtiler: Nicht alle Fehler verdienen einen Wiederholungsversuch. Ein Timeout, ein 429 oder ein 500 sind vorübergehend und sollten aufsteigen, um einen Wiederholungsversuch auszulösen. Ein 4xx-Fehler hingegen zeigt an, dass die Anfrage ungültig ist, und dreimaliges Wiederholen wird das Problem nicht lösen. Geben Sie stattdessen ein explizites Ergebnis zurück und lassen Sie das Workflow entscheiden.

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=2),
    retry_policy_max_attempts=3,
)
async def robust_activity(params: dict) -> dict:
    async with httpx.AsyncClient(timeout=60) as client:
        try:
            response = await client.post(params["url"], json=params["data"])
            response.raise_for_status()
            return {"status": "success", "data": response.json()}
        except httpx.TimeoutException:
            raise  # Wird von der Plattform wiederholt
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                raise  # Rate limitiert — wird wiederholt
            elif e.response.status_code >= 500:
                raise  # Serverfehler — wird wiederholt
            else:
                # Clientfehler (4xx) — nicht wiederholen
                return {"status": "error", "code": e.response.status_code}

Schließlich stammt die vierte Gewohnheit aus der 2-MB-Grenze pro Aufruf: Geben Sie nur zurück, was das Workflow tatsächlich benötigt, um den nächsten Schritt zu entscheiden. Eine Anzahl, eine Stichprobe und die URL zum vollständigen Ergebnis sind fast immer ausreichend.

@workflows.activity()
async def extract_data(source: str) -> dict:
    data = await retrieve_large_data(source)

    # Geben Sie nur zurück, was das Workflow benötigt
    return {
        "item_count": len(data),
        "summary": data[:10],  # Nur die ersten 10
        "full_url": await upload_results(data)
    }

Kernaussagen

• Asynchrone I/O ist obligatorisch: Verwenden Sie httpx, asyncpg, aiofiles statt ihrer synchronen Entsprechungen
• Blockierende Aufrufe (requests, psycopg2, open) frieren den gesamten Worker ein
• Verwenden Sie asyncio.to_thread() als letzte Instanz für Bibliotheken ohne asynchrone Version
• Jede Aktivität muss eine klare Verantwortung haben
• Schließen Sie Verbindungen immer mit async with
• Behandeln Sie Fehler explizit: Manche sollten wiederholt werden, andere nicht
• Begrenzen Sie die Rückgabegrößen, um die 2-MB-Grenze einzuhalten