Activités locales

Les activités locales s'exécutent directement dans le processus worker du workflow, en passant outre les files d'attente de tâches du déploiement. Elles sont utiles pour des opérations rapides (moins d'une seconde) lorsque la surcharge de planification est significative.

Utile pour :

• Calculs rapides, validations, transformations de données
• Scénarios à haut volume où la latence est critique
• Opérations durant moins d'une seconde

À éviter pour :

• Opérations longues (plusieurs secondes ou plus)
• Appels à des API externes ou tâches gourmandes en I/O
• Opérations nécessitant une isolation de relecture/retry ou du Scale indépendant

from mistralai.workflows import run_activities_locally

@workflows.activity()
async def validate_email(email: str) -> bool:
    return "@" in email

@workflows.workflow.define(name="user-workflow")
class UserWorkflow:
    @workflows.workflow.entrypoint
    async def execute(self, email: str) -> bool:
        with run_activities_locally():
            result = await validate_email(email)
        return result

from mistralai.workflows import run_activities_locally

@workflows.activity()
async def validate_email(email: str) -> bool:
    return "@" in email

@workflows.workflow.define(name="user-workflow")
class UserWorkflow:
    @workflows.workflow.entrypoint
    async def execute(self, email: str) -> bool:
        with run_activities_locally():
            result = await validate_email(email)
        return result

Combinez activités locales et distantes :

@workflows.workflow.define(name="mixed-workflow")
class MixedWorkflow:
    @workflows.workflow.entrypoint
    async def execute(self, email: str) -> dict:
        # Recherche rapide exécutée en local
        with run_activities_locally():
            domain = await quick_lookup(email)

        # Opération lente exécutée comme activité classique
        verified = await external_api_call(email)

        return {"domain": domain, "verified": verified}

@workflows.workflow.define(name="mixed-workflow")
class MixedWorkflow:
    @workflows.workflow.entrypoint
    async def execute(self, email: str) -> dict:
        # Recherche rapide exécutée en local
        with run_activities_locally():
            domain = await quick_lookup(email)

        # Opération lente exécutée comme activité classique
        verified = await external_api_call(email)

        return {"domain": domain, "verified": verified}

Les activités locales contournent les files d'attente et le passage par worker, elles sont donc en général bien plus rapides que les activités classiques. La latence réelle dépend de l'infrastructure et de la charge du workflow.

Contourne la planification standard :

• Pas d'isolation au niveau déploiement : les activités locales partagent les ressources avec les workflows.
• Blocage du process worker : une activité locale lente bloque l'ensemble du worker du workflow.
• Pas de Scale indépendant : les activités locales ne peuvent pas être scalées séparément du worker.

Paramètres d'activité (support partiel) :

Timeouts, politiques de retry et injection de dépendances fonctionnent de la même façon pour les activités locales. Les paramètres suivants ne s'appliquent pas car les activités locales s'exécutent déjà dans le processus worker du workflow : sticky_to_worker, rate_limit, heartbeat_timeout.

@workflows.activity(
    start_to_close_timeout=timedelta(seconds=5),
    retry_policy_max_attempts=3
)
async def timed_local_activity(data: str) -> str:
    # Les timeouts et politiques de retry s'appliquent même en local.
    # Le rate limiting n'est PAS supporté en activité locale.
    return data.upper()

@workflows.activity(
    start_to_close_timeout=timedelta(seconds=5),
    retry_policy_max_attempts=3
)
async def timed_local_activity(data: str) -> str:
    # Les timeouts et politiques de retry s'appliquent même en local.
    # Le rate limiting n'est PAS supporté en activité locale.
    return data.upper()