Inputs et timeouts

Ce qui entre dans un workflow

Chaque workflow reçoit des paramètres d’entrée qui définissent son contexte d’exécution. Comme ces paramètres traversent le réseau et sont conservés dans le journal d’exécution, ils obéissent à une règle stricte : ils doivent être JSON-sérialisables. Les types primitifs de Python passent donc sans difficulté, tout comme les listes, les dictionnaires et les valeurs nulles.

# 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

En revanche, datetime, bytes, set, tuple et les classes personnalisées non-Pydantic ne sont pas supportés directement. Si votre workflow doit recevoir une date de facturation, convertissez-la en chaîne ISO 8601 avant de la transmettre, et reconstruisez l’objet côté activité. Le principe vaut pour tous ces types : la conversion est votre responsabilité, pas celle de la plateforme.

Structurer et valider avec Pydantic

Passer un dictionnaire libre fonctionne, mais vous perdez toute garantie sur son contenu. Pour des paramètres structurés et validés, appuyez-vous sur des modèles Pydantic, qui documentent l’interface du workflow autant qu’ils la protègent.

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

Côté appelant, rien de particulier à faire : les champs du BaseModel deviennent simplement des clés de premier niveau dans le dictionnaire d’input.

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 valide automatiquement les données reçues. Si un champ est invalide, le workflow échoue immédiatement, avec un message d’erreur explicite, plutôt que de partir en exécution et de planter trois activités plus loin. Cette validation se resserre autant que nécessaire : contraintes de format sur une adresse e-mail, bornes numériques sur une priorité, et surtout extra = "forbid", qui rejette les champs inconnus.

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

Ce dernier point mérite l’attention. Sans extra="forbid", une faute de frappe dans le nom d’un champ passe inaperçue : le champ attendu prend sa valeur par défaut et le workflow s’exécute avec des paramètres qui ne sont pas ceux que vous croyez. L’échec est silencieux, donc coûteux.

Lorsqu’un même workflow doit accepter plusieurs formes d’entrée, les union types permettent de décrire ces variantes sans multiplier les workflows.

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"

La durée de vie d’une exécution

Chaque workflow a une durée maximale d’exécution, fixée par le paramètre execution_timeout. Par défaut, elle est d’une heure : passé ce délai, le workflow est automatiquement annulé, quel que soit son état d’avancement. Pour un traitement censé durer trente secondes, cette valeur est confortable ; pour un pipeline nocturne, elle est un piège.

Vous ajustez donc ce timeout à la nature du traitement, à la baisse comme à la hausse. Un workflow de traitement rapide gagne à être serré, parce qu’un dépassement signale alors un vrai problème plutôt qu’une lenteur passagère.

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)

À l’autre extrémité, le plafond autorisé est de 7 jours. Cette durée n’a de sens que pour les processus qui attendent un événement extérieur, typiquement une validation humaine : un rapport collecté automatiquement mais signé par un responsable qui ne reviendra de congés que la semaine suivante.

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)

Pour calibrer, appuyez-vous sur cet ordre de grandeur par famille d’usage.

Deux niveaux de timeout à ne pas confondre

execution_timeout borne l’ensemble du workflow ; start_to_close_timeout borne une seule activité. Les deux cohabitent et se complètent : le premier vous protège d’un workflow qui n’aboutit jamais, le second d’une activité qui reste pendue sur un service muet. Notez au passage, dans l’exemple ci-dessous, que le timeout du client HTTP (240 secondes) est volontairement inférieur au start_to_close_timeout (5 minutes), pour que l’erreur remonte proprement avant que la plateforme ne coupe.

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 {"résultats": resultats}

Un dernier réflexe pour vos inputs : gardez-les légers. La limite est de 2 MB, et rien ne vous oblige à en approcher. Transmettez l’URL d’un document plutôt que son contenu, un identifiant de commande plutôt que la commande entière, et laissez les activités aller chercher la donnée. Définissez des valeurs par défaut pour tout ce qui est optionnel, documentez chaque champ avec Field(description="..."), et confiez la validation à Pydantic plutôt qu’à des if disséminés dans le workflow.

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