Inputs et timeouts

Les inputs de workflow

Chaque workflow reçoit des paramètres d’entrée qui définissent son contexte d’exécution. Ces paramètres doivent respecter des règles strictes de sérialisation.

Types JSON supportés

Les inputs de workflow doivent être JSON-sérialisables. Les types Python suivants sont directement supportés :

# Types primitifs
nom: str = "Marie"
age: int = 32
score: float = 0.95
actif: bool = True

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

# Nullable
commentaire: str | None = None

Les types non supportés directement incluent : datetime, bytes, set, tuple, classes personnalisées non-Pydantic. Si vous devez les utiliser, convertissez-les en types JSON (par exemple, une date en chaîne ISO 8601).

Inputs avec Pydantic BaseModel

Pour des paramètres structurés et validés, utilisez des modèles Pydantic :

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

Lorsque vous envoyez l’input au workflow, les champs du BaseModel deviennent des clés de premier niveau :

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
    }
)

Validation automatique

Pydantic valide automatiquement les inputs. Si un champ est invalide, le workflow échoue immédiatement avec un message d’erreur clair :

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

    class Config:
        extra = "forbid"  # Rejette les champs inconnus

Union types pour des inputs flexibles

Vous pouvez utiliser des union types pour accepter différentes structures :

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"

Timeout d’exécution

Chaque workflow a une durée maximale d’exécution configurée via le paramètre execution_timeout.

Timeout par défaut

Le timeout par défaut est de 1 heure. Si votre workflow n’a pas terminé après 1 heure, il est automatiquement annulé.

Configurer un timeout personnalisé

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)

Timeout maximum

Le timeout maximum autorisé est de 7 jours :

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:
        # Ce workflow peut durer jusqu'à 7 jours
        # Utile pour les processus avec attente humaine
        donnees = await collecter_donnees_multi_sources(params)
        validation = await attendre_validation_humaine(donnees)
        return await finaliser_rapport(validation)

Choisir le bon timeout

Timeout des activités vs timeout du workflow

Il est important de distinguer les deux niveaux de timeout :

• execution_timeout (workflow) — Durée maximale de l’ensemble du workflow
• start_to_close_timeout (activité) — Durée maximale d’une seule activité

from datetime import timedelta

@workflows.activity(
    start_to_close_timeout=timedelta(minutes=5)
)
async def appel_api_externe(params: dict) -> dict:
    # Cette activité a 5 minutes pour se terminer
    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:
        # Le workflow entier a 2 heures
        # Mais chaque activité individuelle a son propre timeout
        resultats = []
        for source in sources:
            resultat = await appel_api_externe(source)
            resultats.append(resultat)
        return {"resultats": resultats}

Bonnes pratiques pour les inputs

• Validez toujours vos inputs avec Pydantic plutôt qu’avec du code impératif
• Utilisez extra="forbid" pour rejeter les champs inconnus et éviter les erreurs silencieuses
• Gardez les inputs légers — Passez des identifiants ou des URLs plutôt que des données volumineuses (limite de 2 MB)
• Définissez des valeurs par défaut pour les paramètres optionnels
• Documentez chaque champ avec Field(description="...")

Points clés à retenir

• Les inputs de workflow doivent être JSON-sérialisables (str, int, float, bool, list, dict)
• Utilisez Pydantic BaseModel pour des inputs structurés avec validation automatique
• Le timeout par défaut est de 1 heure, configurable jusqu’à 7 jours maximum
• Distinguez le timeout du workflow (execution_timeout) de celui des activités (start_to_close_timeout)
• Gardez les inputs légers : identifiants et URLs plutôt que données brutes volumineuses