WorkflowsException

Workflows fournit une classe d’exception structurée, WorkflowsException, pour gérer les erreurs de manière cohérente dans l’ensemble des workflows et activités.

Levez une WorkflowsException pour des erreurs attendues ou structurées que les appelants (autres workflows, activités ou consommateurs de l’API) doivent pouvoir détecter et traiter : saisie invalide, violation de règles métier, modes d’échec connus d’un service externe. Les champs structurés code et status HTTP rendent ces erreurs reconnaissables dans tout le système et visibles dans les réponses API.

Pour les erreurs inattendues (un KeyError, un httpx.ConnectError, un panic dans une librairie tierce), laissez l’exception se propager. La plateforme enregistrera l’erreur dans l’historique d’exécution de l’activité ou du workflow, appliquera votre stratégie de reprise, et finira par marquer l’exécution comme échouée si toutes les tentatives sont dépassées. Envelopper toutes les exceptions dans une WorkflowsException ajoute du bruit sans apport d’information.

WorkflowsException fournit :

• Des codes d’erreur structurés grâce à l’énumération ErrorCode
• La correspondance avec le code de statut HTTP
• La sérialisation en réponses JSON
• Des méthodes factory pour les scénarios d’erreurs courants

import asyncio
import http
import mistralai.workflows as workflows
from mistralai.workflows.exceptions import WorkflowsException, ErrorCode
from pydantic import BaseModel


@workflows.activity()
async def validate_input(value: str) -> str:
    if not value:
        raise WorkflowsException(
            code=ErrorCode.INVALID_ARGUMENTS_ERROR,
            message="Input value cannot be empty",
            status=http.HTTPStatus.BAD_REQUEST,
        )
    return f"valid:{value}"


class Input(BaseModel):
    value: str


@workflows.workflow.define(name="exception_workflow")
class ExceptionWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, params: Input) -> str:
        return await validate_input(params.value)


async def main():
    result = await workflows.execute_workflow(
        ExceptionWorkflow,
        params=Input(value="hello"),
    )
    print(result)

    try:
        await workflows.execute_workflow(
            ExceptionWorkflow,
            params=Input(value=""),
        )
    except WorkflowsException as e:
        print(f"code:    {e.code}")
        print(f"message: {e.message}")
        print(f"status:  {e.status}")

import asyncio
import http
import mistralai.workflows as workflows
from mistralai.workflows.exceptions import WorkflowsException, ErrorCode
from pydantic import BaseModel


@workflows.activity()
async def validate_input(value: str) -> str:
    if not value:
        raise WorkflowsException(
            code=ErrorCode.INVALID_ARGUMENTS_ERROR,
            message="Input value cannot be empty",
            status=http.HTTPStatus.BAD_REQUEST,
        )
    return f"valid:{value}"


class Input(BaseModel):
    value: str


@workflows.workflow.define(name="exception_workflow")
class ExceptionWorkflow:
    @workflows.workflow.entrypoint
    async def run(self, params: Input) -> str:
        return await validate_input(params.value)


async def main():
    result = await workflows.execute_workflow(
        ExceptionWorkflow,
        params=Input(value="hello"),
    )
    print(result)

    try:
        await workflows.execute_workflow(
            ExceptionWorkflow,
            params=Input(value=""),
        )
    except WorkflowsException as e:
        print(f"code:    {e.code}")
        print(f"message: {e.message}")
        print(f"status:  {e.status}")

L’énumération ErrorCode définit des codes d’erreur structurés regroupés par catégorie :