Attente de conditions

Les workflows peuvent suspendre leur exécution jusqu’à ce qu’une condition spécifique soit remplie, permettant des schémas pilotés par événements comme la validation humaine ou les circuits d’approbation.

Le prédicat de condition est généralement activé par un signal envoyé depuis l’extérieur du workflow (appel API, autre service ou action utilisateur). Le workflow reste en pause sans consommer de ressources de calcul jusqu’à ce que le prédicat soit à True ou que le délai soit dépassé.

La même primitive wait_condition permet de couvrir plusieurs cas d’usage récurrents :

• Circuits d’approbation — attendre la validation ou le refus d’un humain (piloté par signal). Voir l’exemple ci-dessous.
• Rappel externe — attendre qu’un webhook ou job tiers signale que le workflow peut continuer.
• Polling progressif (rare) — attendre qu’un autre processus interne change un indicateur ; le plus souvent, il est préférable d’utiliser une activité dédiée au polling.

Si votre cas repose fortement sur l’intervention humaine, consultez aussi Signals pour définir et émettre le déclencheur.

Utilisez workflow.wait_condition() pour bloquer jusqu’à ce qu’un prédicat retourne True :

from datetime import timedelta
from mistralai.workflows import workflow

await workflow.wait_condition(
    lambda: self.ready,
    timeout=timedelta(minutes=5)
)

from datetime import timedelta
from mistralai.workflows import workflow

await workflow.wait_condition(
    lambda: self.ready,
    timeout=timedelta(minutes=5)
)

Combinez avec des signaux pour mettre en œuvre un circuit humain-dans-la-boucle :

import asyncio
import mistralai.workflows as workflows
from mistralai.workflows import workflow
from datetime import timedelta

@workflows.workflow.define(name="approval_workflow")
class ApprovalWorkflow:
    def __init__(self):
        self.approved = False

    @workflows.workflow.signal(name="approve")
    async def approve(self) -> None:
        self.approved = True

    @workflows.workflow.entrypoint
    async def run(self, request_id: str) -> str:
        try:
            # Attendre jusqu’à 24 heures que le signal `approve` fasse passer self.approved à True
            await workflow.wait_condition(
                lambda: self.approved,
                timeout=timedelta(hours=24),
            )
        except asyncio.TimeoutError:
            return f"Request {request_id} timed out"

        return f"Request {request_id} approved"

import asyncio
import mistralai.workflows as workflows
from mistralai.workflows import workflow
from datetime import timedelta

@workflows.workflow.define(name="approval_workflow")
class ApprovalWorkflow:
    def __init__(self):
        self.approved = False

    @workflows.workflow.signal(name="approve")
    async def approve(self) -> None:
        self.approved = True

    @workflows.workflow.entrypoint
    async def run(self, request_id: str) -> str:
        try:
            # Attendre jusqu’à 24 heures que le signal `approve` fasse passer self.approved à True
            await workflow.wait_condition(
                lambda: self.approved,
                timeout=timedelta(hours=24),
            )
        except asyncio.TimeoutError:
            return f"Request {request_id} timed out"

        return f"Request {request_id} approved"

Lorsqu’un délai est renseigné, wait_condition déclenche une exception asyncio.TimeoutError si la condition n’est pas satisfaite dans le temps imparti. Gérez-la pour définir une logique de traitement du délai :

import asyncio

try:
    await workflow.wait_condition(
        lambda: self.ready,
        timeout=timedelta(minutes=30)
    )
except asyncio.TimeoutError:
    # Gérer le dépassement de délai
    return "Operation timed out"

import asyncio

try:
    await workflow.wait_condition(
        lambda: self.ready,
        timeout=timedelta(minutes=30)
    )
except asyncio.TimeoutError:
    # Gérer le dépassement de délai
    return "Operation timed out"