Eingaben und Zeitüberschreitungen

Was fließt in einen Workflow ein

Jeder Workflow empfängt Eingabeparameter, die seinen Ausführungskontext definieren. Da diese Parameter über das Netzwerk übertragen und im Ausführungsprotokoll gespeichert werden, müssen sie einer strengen Regel folgen: Sie müssen JSON-serialisierbar sein. Python-Primitive-Typen werden problemlos übernommen, ebenso wie Listen, Wörterbücher und Null-Werte.

# Primitive Typen
nom: str = "Marie"
age: int = 32
score: float = 0,95
actif: bool = True

# Sammlungen
tags: list = ["urgent", "client-vip"]
metadata: dict = {"source": "api", "version": 2}

# Nullable
commentaire: str | None = None

datetime, bytes, set, tuple und nicht-Pydantic-klassische benutzerdefinierte Klassen werden jedoch nicht direkt unterstützt. Wenn Ihr Workflow ein Rechnungsdatum empfangen muss, konvertieren Sie es vor der Übermittlung in einen ISO 8601-String und rekonstruieren Sie das Objekt innerhalb der Aktivität. Dieses Prinzip gilt für alle solchen Typen: Die Konvertierung ist Ihre Verantwortung, nicht die der Plattform.

Strukturieren und validieren mit Pydantic

Das Übergeben eines frei formatierten Wörterbuchs funktioniert, doch Sie verlieren alle Garantien bezüglich seines Inhalts. Für strukturierte, validierte Parameter verlassen Sie sich auf Pydantic-Modelle, die die Workflow-Schnittstelle nicht nur schützen, sondern auch dokumentieren.

from pydantic import BaseModel, Field

class ParametresAnalyse(BaseModel):
    url_document: str
    langue: str = "fr"
    max_tokens: int = Field(default=1000, ge=100, le=10000)
    inclure_metadata: bool = False

@workflows.workflow.define(name="analyse_documentaire")
class AnalyseDocumentaire:
    @workflows.workflow.entrypoint
    async def run(self, params: ParametresAnalyse) -> dict:
        extraction = await extraire_texte(params.url_document, params.langue)
        analyse = await analyser_contenu(extraction["texte"], params.max_tokens)
        
        if params.inclure_metadata:
            metadata = await extraire_metadata(params.url_document)
            return {**analyse, "metadata": metadata}
        
        return analyse

Auf der Seite des Aufrufers ist nichts Besonderes zu tun: Die Felder des BaseModel werden einfach zu Top-Level-Schlüsseln im Eingabewörterbuch.

execution = await client.workflows.execute_workflow_async(
    workflow_identifier="analyse_documentaire",
    input={
        "url_document": "https://example.com/rapport.pdf",
        "langue": "fr",
        "max_tokens": 2000,
        "inclure_metadata": True
    }
)

Pydantic validiert die empfangenen Daten automatisch. Ist ein Feld ungültig, schlägt der Workflow sofort mit einer expliziten Fehlermeldung fehl, statt die Ausführung zu starten und drei Aktivitäten später abstürzen zu lassen. Diese Validierung kann nach Bedarf verschärft werden: Formatbeschränkungen für eine E-Mail-Adresse, numerische Grenzen für eine Priorität und vor allem extra = "forbid", das unbekannte Felder ablehnt.

class ParametresStricts(BaseModel):
    email: str = Field(pattern=r"^[\w.+-]+@[\w-]+\.[\w.]+$")
    priorite: int = Field(ge=1, le=5)

    class Config:
        extra = "forbid"  # Lehnt unbekannte Felder ab

Dieser letzte Punkt verdient Aufmerksamkeit. Ohne extra="forbid" bleibt ein Tippfehler im Feldnamen unbemerkt: Das erwartete Feld erhält seinen Standardwert, und der Workflow läuft mit Parametern, die nicht beabsichtigt waren. Der Fehler ist stumm und damit kostspielig.

Wenn ein einzelner Workflow mehrere Eingabeformen akzeptieren muss, ermöglichen Union-Typen, diese Varianten zu beschreiben, ohne Workflows zu vervielfältigen.

from typing import Union

class TraitementTexte(BaseModel):
    texte: str
    type_traitement: str = "resume"

class TraitementURL(BaseModel):
    url: str
    type_traitement: str = "resume"

class ParametresFlexibles(BaseModel):
    source: Union[TraitementTexte, TraitementURL]

    class Config:
        extra = "forbid"

Ausführungslebensdauer

Jeder Workflow hat eine maximale Ausführungszeit, die durch den Parameter execution_timeout festgelegt wird. Standardmäßig beträgt diese eine Stunde: Ab diesem Fristablauf wird der Workflow automatisch abgebrochen, unabhängig von seinem Fortschritt. Für einen Prozess, der nur dreißig Sekunden dauern soll, ist dieser Wert komfortabel; für eine Pipeline, die über Nacht läuft, ist er eine Falle.

Passen Sie diese Zeitüberschreitungsgrenze der Art der Verarbeitung an, sowohl nach unten als auch nach oben. Ein Workflow mit schneller Verarbeitung profitiert von einer engen Grenze, da eine Überschreitung dann ein echtes Problem signalisiert, anstatt nur eine vorübergehende Verlangsamung.

from datetime import timedelta

@workflows.workflow.define(
    name="traitement_rapide",
    execution_timeout=timedelta(minutes=10)
)
class TraitementRapide:
    @workflows.workflow.entrypoint
    async def run(self, donnees: dict) -> dict:
        return await traiter(donnees)

An der anderen Seite liegt die erlaubte Obergrenze bei 7 Tagen. Diese Dauer macht nur Sinn für Prozesse, die auf ein externes Ereignis warten, typischerweise eine menschliche Validierung: ein automatisch gesammelter Bericht, der von einem Manager unterzeichnet wird, der bis zur folgenden Woche nicht vom Urlaub zurückkehrt.

from datetime import timedelta

@workflows.workflow.define(
    name="workflow_longue_duree",
    execution_timeout=timedelta(days=7)
)
class WorkflowLongueDuree:
    @workflows.workflow.entrypoint
    async def run(self, params: dict) -> dict:
        # Dieser Workflow kann bis zu 7 Tage dauern
        # Nützlich für Prozesse mit menschlichen Wartepausen
        donnees = await collecter_donnees_multi_sources(params)
        validation = await attendre_validation_humaine(donnees)
        return await finaliser_rapport(validation)

Um die Kalibrierung vorzunehmen, verwenden Sie diese Größenordnung nach Nutzungsfamilien.

Zwei Zeitüberschreitungsebenen, die nicht verwechselt werden dürfen

execution_timeout begrenzt den gesamten Workflow; start_to_close_timeout begrenzt eine einzelne Aktivität. Beide existieren nebeneinander und ergänzen sich: Der erste schützt Sie vor einem Workflow, der niemals abgeschlossen wird, der zweite vor einer Aktivität, die an einem stummen Dienst hängt. Beachten Sie im folgenden Beispiel, dass die HTTP-Client-Zeitüberschreitung (240 Sekunden) absichtlich niedriger ist als die start_to_close_timeout (5 Minuten), sodass der Fehler sauber auftritt, bevor die Plattform die Verbindung unterbricht.

from datetime import timedelta

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5)
)
async def appel_api_externe(params: dict) -> dict:
    # Diese Aktivität hat 5 Minuten Zeit zum Abschluss
    async with httpx.AsyncClient(timeout=240) as client:
        response = await client.post(params["url"], json=params["data"])
    return response.json()

@workflows.workflow.define(
    name="pipeline_multi_etapes",
    execution_timeout=timedelta(hours=2)
)
class PipelineMultiEtapes:
    @workflows.workflow.entrypoint
    async def run(self, sources: list) -> dict:
        # Der gesamte Workflow hat 2 Stunden
        # Aber jede einzelne Aktivität hat ihre eigene Zeitüberschreitung
        resultats = []
        for source in sources:
            resultat = await appel_api_externe(source)
            resultats.append(resultat)
        return {"résultats": resultats}

Ein letzter Reflex für Ihre Eingaben: Halten Sie sie leicht. Die Grenze liegt bei 2 MB, und nichts zwingt Sie, diese anzunähern. Übermitteln Sie die URL eines Dokuments statt dessen Inhalt, eine Bestell-ID statt der vollen Bestellung, und lassen Sie die Aktivitäten die Daten abrufen. Definieren Sie Standardwerte für alles, was optional ist, dokumentieren Sie jedes Feld mit Field(description="...") und überlassen Sie die Validierung Pydantic statt if-Anweisungen, die im Workflow verstreut sind.

Kernaussagen

• Workflow-Eingaben müssen JSON-serialisierbar sein (str, int, float, bool, list, dict)
• Verwenden Sie Pydantic BaseModel für strukturierte Eingaben mit automatischer Validierung
• Die Standard-Zeitüberschreitung beträgt 1 Stunde, konfigurierbar bis zu einem Maximum von 7 Tagen
• Unterscheiden Sie die Workflow-Zeitüberschreitung (execution_timeout) von der Aktivitäts-Zeitüberschreitung (start_to_close_timeout)
• Halten Sie Eingaben leicht: IDs und URLs statt großer Rohdaten