Plugin d'évaluation de workflow

Le plugin d'évaluation de workflow (mistralai-workflows-plugins-evaluations) exécute des évaluations hors ligne dans Mistral Workflows. Il expose la même API evaluation.run() que l'SDK d'évaluation, mais répartit l'exécution des tâches et le scoring en activités Temporal parallèles.

Le plugin suit le même modèle mental que le SDK :

Dataset  →  Task  →  Evaluators  →  Results

Dataset  →  Task  →  Evaluators  →  Results

Au lieu d'exécuter tout dans un seul processus, le plugin distribue chaque enregistrement sous forme de workflow enfant qui exécute la tâche, puis les scorers. Les résultats sont téléchargés dans Studio au fur et à mesure que les enregistrements sont terminés.

Les deux écrivent dans les mêmes projets, évaluations et exécutions dans Studio, et partagent les mêmes primitives : Evaluator, Goal, Statistic, System et TunableSystem.

Prérequis : Python 3.12+, une clé API Mistral et un projet Workflows fonctionnel. Si vous n'en avez pas encore un, suivez l'installation de Workflows et votre premier workflow.

Ajoutez le plugin à votre projet Workflows depuis PyPI :

pip install mistralai-workflows-plugins-evaluations

pip install mistralai-workflows-plugins-evaluations

Le plugin ajoute le SDK d'Évaluation (mistralai-evaluations) comme dépendance. Le plugin 0.7.0 nécessite mistralai-evaluations 0.10.0 ou une version ultérieure : mettez les deux à jour ensemble. Vous n'avez rien à enregistrer sur le worker : il découvre les plugins installés au démarrage et enregistre automatiquement les activités d'évaluation et les workflows enfants.

Le worker utilise ses propres identifiants Mistral pour télécharger les résultats dans Studio. Sur un worker local ou auto-hébergé, définissez MISTRAL_API_KEY dans l'environnement du worker :

export MISTRAL_API_KEY=<your-api-key>

export MISTRAL_API_KEY=<your-api-key>

Sur un déploiement managé, le worker ne dispose pas de clé API et le plugin utilise à la place le token de compte de service du déploiement. Voir Déploiements managés.

Les évaluations s'exécutent sur des déploiements managés sans clé API. MISTRAL_API_KEY y est un nom réservé : la plateforme fournit au worker un token de compte de service à rotation automatique (MISTRAL_SA_TOKEN_PATH), et le plugin l'utilise automatiquement pour chaque appel à Studio : création de l'exécution, téléchargement des résultats et mise à jour de son statut. Aucune configuration n'est nécessaire.

Cela nécessite mistralai-workflows 3.10 ou une version ultérieure et mistralai-evaluations 0.9.2 ou une version ultérieure dans l'image du worker. La mise à jour du plugin vers la version 0.6.6 ou ultérieure installe la bonne version de mistralai-evaluations. Un worker sans clé API ni token de compte de service échoue avec l'erreur No credentials found.

Une clé API ne remplace jamais le token de compte de service. Si vous copiez une clé dans l'environnement d'un worker managé, elle n'est pas envoyée. Accordez des autorisations au compte de service, ou utilisez on_behalf_of=True, plutôt que d'ajouter une clé.

Déclarez le workflow avec on_behalf_of=True. Les appels effectués par le plugin depuis votre workflow s'exécutent alors en tant que l'utilisateur qui le lance, avec les autorisations de cet utilisateur sur son espace de travail :

@workflow.define(name="my-eval-workflow", on_behalf_of=True)
class MyEvalWorkflow:
    ...

@workflow.define(name="my-eval-workflow", on_behalf_of=True)
class MyEvalWorkflow:
    ...

Les workflows on-behalf-of nécessitent un déploiement durci. Voir Workflows on-behalf-of.

Le plugin télécharge les enregistrements et exécute le code de vos tâches et scorers dans ses propres workflows enfants. Avec on_behalf_of=True, ces workflows enfants s'exécutent eux aussi en tant que l'utilisateur ; tout s'exécute donc en tant que cet utilisateur et le compte de service ne nécessite aucune autorisation supplémentaire.

Sans on_behalf_of, les appels s'exécutent sous l'identité du compte de service du déploiement, par exemple dans un workflow planifié, qui n'a pas d'utilisateur. Le compte de service doit alors disposer d'un rôle qui comporte les autorisations relatives aux évaluations hors ligne, aux ensembles de données et aux juges sur l'espace de travail cible. Le rôle par défaut workflow_executor d'un déploiement ne les comporte pas : ajoutez offline_evaluations_manager aux requested_roles du déploiement lors de sa création (["offline_evaluations_manager"], ou service_account_roles dans le MCP Mistral). Voir aussi Comptes de service et Rôles et autorisations. Les exécutions, les ensembles de données, les juges et les optimisations sont alors créés sous l'identité du compte de service, que Studio affiche comme leur créateur. Les fonctionnalités liées à une personne, comme les favoris et les étoiles, ne sont pas disponibles pour ce compte. optimize() fonctionne dans les deux modes.

:::note Vos propres appels au modèle Vos tâches et vos scorers choisissent leurs propres identifiants. Sur un déploiement managé, appelez get_mistral_client() depuis mistralai.workflows.client, comme dans le guide de démarrage : il embarque le même token de compte de service. Un client Mistral(api_key=os.environ["MISTRAL_API_KEY"]) est réservé aux exécutions locales. :::

Quand l'appelant ne dispose pas d'une autorisation requise, le code 403 lève une ObservabilityAPIError qui identifie l'appelant (utilisateur ou compte de service), les autorisations concernées (évaluations hors ligne, ensembles de données ou juges) et la marche à suivre pour corriger le problème sur un déploiement géré. Cela nécessite mistralai-evaluations 0.10.0 ou une version ultérieure. Avant, l'erreur vous demandait de vérifier la clé API, ce qui prêtait à confusion pour un compte de service, qui ne possède pas de clé.

Créez un fichier src/workflows/language_detection_eval.py dans votre projet Workflows :

from mistralai.workflows import workflow
from mistralai.workflows.client import get_mistral_client
from mistralai.workflows.plugins.evaluations import evaluation
from mistralai.workflows.plugins.evaluations.types import (
    Evaluation, Evaluator, Goal, Project, Score, ScorerContext, System, TaskContext,
)

dataset = [
    {"sentence": "Hello, how are you?", "groundtruth": "English"},
    {"sentence": "Bonjour, comment ça va?", "groundtruth": "French"},
    {"sentence": "Hola, ¿cómo estás?", "groundtruth": "Spanish"},
]

@evaluation.task
async def detect_language(ctx: TaskContext) -> str:
    client = get_mistral_client()
    response = await client.chat.complete_async(
        model=str(ctx.system.params["model"]),
        messages=[
            {"role": "system", "content": str(ctx.system.params["system_prompt"])},
            {"role": "user", "content": ctx.input_record["sentence"]},
        ],
    )
    return str(response.choices[0].message.content)

@evaluation.scorer
async def accuracy(ctx: ScorerContext) -> Score:
    match = ctx.input_record["groundtruth"].lower() == str(ctx.output).strip().lower()
    return Score(value=1 if match else 0)

@workflow.define(name="language-detection-eval")
class LanguageDetectionEval:
    @workflow.entrypoint
    async def run(self) -> dict:
        result = await evaluation.run(
            project=Project(name="Language Detection"),
            evaluation=Evaluation(name="Accuracy Eval"),
            system=System(name="mistral-small", params={
                "model": "mistral-small-latest",
                "system_prompt": "What language is this sentence in? Reply with ONLY the language name.",
            }),
            dataset=dataset,
            task=detect_language,
            evaluators=[
                Evaluator(
                    name="accuracy",
                    description="1 if the detected language matches the groundtruth, 0 otherwise.",
                    scorer=accuracy,
                    goal=Goal.gte(0.8),
                ),
            ],
        )
        return {"run_id": result.run_id, "run_url": result.run_url}

from mistralai.workflows import workflow
from mistralai.workflows.client import get_mistral_client
from mistralai.workflows.plugins.evaluations import evaluation
from mistralai.workflows.plugins.evaluations.types import (
    Evaluation, Evaluator, Goal, Project, Score, ScorerContext, System, TaskContext,
)

dataset = [
    {"sentence": "Hello, how are you?", "groundtruth": "English"},
    {"sentence": "Bonjour, comment ça va?", "groundtruth": "French"},
    {"sentence": "Hola, ¿cómo estás?", "groundtruth": "Spanish"},
]

@evaluation.task
async def detect_language(ctx: TaskContext) -> str:
    client = get_mistral_client()
    response = await client.chat.complete_async(
        model=str(ctx.system.params["model"]),
        messages=[
            {"role": "system", "content": str(ctx.system.params["system_prompt"])},
            {"role": "user", "content": ctx.input_record["sentence"]},
        ],
    )
    return str(response.choices[0].message.content)

@evaluation.scorer
async def accuracy(ctx: ScorerContext) -> Score:
    match = ctx.input_record["groundtruth"].lower() == str(ctx.output).strip().lower()
    return Score(value=1 if match else 0)

@workflow.define(name="language-detection-eval")
class LanguageDetectionEval:
    @workflow.entrypoint
    async def run(self) -> dict:
        result = await evaluation.run(
            project=Project(name="Language Detection"),
            evaluation=Evaluation(name="Accuracy Eval"),
            system=System(name="mistral-small", params={
                "model": "mistral-small-latest",
                "system_prompt": "What language is this sentence in? Reply with ONLY the language name.",
            }),
            dataset=dataset,
            task=detect_language,
            evaluators=[
                Evaluator(
                    name="accuracy",
                    description="1 if the detected language matches the groundtruth, 0 otherwise.",
                    scorer=accuracy,
                    goal=Goal.gte(0.8),
                ),
            ],
        )
        return {"run_id": result.run_id, "run_url": result.run_url}

Redémarrez votre worker pour qu'il enregistre le nouveau workflow, puis déclenchez-le comme n'importe quel autre workflow — depuis la page Workflows dans Studio, ou avec le SDK Mistral :

from mistralai.client import Mistral

client = Mistral(api_key="your_api_key")

execution = client.workflows.execute_workflow(
    workflow_identifier="language-detection-eval",
    input={},
)

from mistralai.client import Mistral

client = Mistral(api_key="your_api_key")

execution = client.workflows.execute_workflow(
    workflow_identifier="language-detection-eval",
    input={},
)

Le workflow renvoie l'identifiant de l'exécution et un lien vers celle-ci. Les résultats sont disponibles dans Studio sous Evaluate > Evaluations > Hors ligne.

Le plugin réutilise les types du SDK, donc tout ce qui est décrit dans les guides du SDK s'applique — ensembles de données, paramètres de système, évaluateurs, objectifs, statistiques et générations multiples. Les différences viennent de l'exécution dans un workflow :

• Décorez chaque fonction. Les tâches utilisent @evaluation.task, les scorers @evaluation.scorer et les évaluateurs d'exécution @evaluation.run_scorer. Chacune devient une activité Temporal avec son propre timeout, ses propres nouvelles tentatives et sa propre concurrence (les tâches workflow-class et workflow-name ne font pas l'objet de nouvelles tentatives ; voir Modes de tâche). Voir Decorators.
• Appelez evaluation depuis le plugin, pas client.evaluation. Importez les types depuis mistralai.workflows.plugins.evaluations.types : ce module peut être importé sans risque dans la sandbox du workflow.
• Les tâches peuvent être des workflows. En plus d'une activité, la tâche peut être une classe de workflow ou le nom d'un workflow déployé. Voir Task modes.
• Les scorers renvoient un Score ou un nombre. Un nombre simple est automatiquement encapsulé dans un Score.
• run() renvoie un EvalResult contenant run_id, run_url, statistics, run_scores et failed_record_count. Il est sérialisable, vous pouvez donc le renvoyer depuis le workflow. Il n'existe pas de méthode show().
• Les évaluateurs d'exécution relisent les enregistrements depuis Studio. Pour rester sous les limites de charge utile de Temporal, l'activité d'évaluation d'exécution récupère les enregistrements persistés au lieu de les recevoir du workflow. Voir Run evaluators.
• Privilégiez les ensembles de données Studio pour les entrées volumineuses. dataset prend une liste d'enregistrements ou un ensemble de données Studio (Dataset). Un ensemble de données Studio garde les entrées hors de l'historique du workflow. Voir Ensembles de données et limites.
• Pas encore de statistiques de classification. Statistiques de classification ne sont pas encore prises en charge par le plugin.
• Les échecs sont réessayés par Temporal. Il n'existe pas de retry_failed_records() : chaque tâche d'activité et chaque scorer est réessayé de manière autonome, selon son décorateur. Les tâches de workflow ne sont pas réessayées. Un enregistrement qui échoue malgré les nouvelles tentatives est absent de statistics : vérifiez EvalResult.failed_record_count avant de lire les scores (0.7.0+).

Les évaluateurs au niveau de l'exécution fonctionnent comme dans le SDK, avec un décorateur :

from mistralai.workflows.plugins.evaluations.types import RunEvaluator, RunEvaluatorContext, Score

@evaluation.run_scorer
async def mean_accuracy(ctx: RunEvaluatorContext) -> Score:
    return Score(value=ctx.statistics["accuracy"].avg)

result = await evaluation.run(
    ...,
    run_evaluators=[RunEvaluator(name="mean_accuracy", scorer=mean_accuracy)],
)

from mistralai.workflows.plugins.evaluations.types import RunEvaluator, RunEvaluatorContext, Score

@evaluation.run_scorer
async def mean_accuracy(ctx: RunEvaluatorContext) -> Score:
    return Score(value=ctx.statistics["accuracy"].avg)

result = await evaluation.run(
    ...,
    run_evaluators=[RunEvaluator(name="mean_accuracy", scorer=mean_accuracy)],
)

Quand les résultats sont téléchargés dans Studio, l'activité d'évaluation d'exécution reçoit l'identifiant de l'exécution et relit les enregistrements persistés avant d'appeler votre fonction. Il en va de même pour les callbacks run_metadata. Par conséquent :

• Les enregistrements sont listés dans l'ordre de téléchargement, qui peut différer de l'ordre de l'ensemble de données.
• Un enregistrement dont la sortie n'a pas pu être téléchargée apparaît sans générations.
• La fonction doit être enregistrée sur le worker qui exécute le workflow d'évaluation.
• La lecture des enregistrements ajoute jusqu'à 10 minutes au timeout du callback.

En mode local, les enregistrements sont passés en mémoire.

local=True ignore tout appel à Studio mais conserve la même topologie Temporal (workflows enfants et activités). Vous n'avez pas besoin de project ni d'evaluation, result.run_id vaut "local" et result.run_url vaut None. Les statistiques et les scores d'exécution sont toujours calculés et renvoyés. Voir Iterate locally pour le workflow recommandé.

Les workflows s'exécutent sur Temporal, qui plafonne la taille de l'historique d'un workflow et celle de chaque charge utile. La façon dont vous transmettez l'ensemble de données détermine la part de celui-ci qui passe par cet historique.

Pour les ensembles de données volumineux, transmettez un ensemble de données stocké dans Studio plutôt que des enregistrements définis directement :

from mistralai.workflows.plugins.evaluations.types import Dataset

result = await evaluation.run(
    dataset=Dataset(id="dataset-id"),  # or Dataset(slug="my-dataset")
    task=summarize,
    evaluators=[...],
)

from mistralai.workflows.plugins.evaluations.types import Dataset

result = await evaluation.run(
    dataset=Dataset(id="dataset-id"),  # or Dataset(slug="my-dataset")
    task=summarize,
    evaluators=[...],
)

Le workflow d'évaluation ne gère que les identifiants des enregistrements. Chaque enregistrement est récupéré dans son propre workflow enfant, où la sortie de la tâche est enregistrée dans Studio. Ni les entrées ni les sorties ne passent par l'historique du workflow d'évaluation : leur taille n'entre donc pas en compte dans les limites de Temporal. L'exécution est liée à l'ensemble de données dans Studio.

Les enregistrements définis directement sont d'abord chargés dans le workflow, si bien que chaque enregistrement passe par l'historique du workflow. Passez à Dataset(id=...) dès que les entrées deviennent volumineuses.

Une exécution sur un ensemble de données Studio accepte jusqu'à 10 000 enregistrements. Les listes fournies directement n'ont pas de limite de nombre d'enregistrements, mais restent limitées par la taille de la charge utile (voir ci-dessous). Les enregistrements sont traités par batchs de workflows enfants, et le workflow d'évaluation ne conserve que les valeurs de score de chaque enregistrement. Une exécution à taille maximale reste dans les limites d'historique de Temporal (50 Mo et 51 200 événements).

Chaque enregistrement est toujours traité dans son propre workflow, et les valeurs de score remontent toujours vers le workflow d'évaluation. Gardez ces limites à l'esprit :

• Taille des enregistrements. L'entrée, la sortie et les métadonnées d'un même enregistrement doivent chacune rester sous la limite de charge utile de 2 Mo de Temporal.
• Valeurs de score. Gardez-les courtes (des nombres ou des libellés courts). Placez les textes longs dans le rationale du score, qui reste dans Studio.
• Paramètres système et métadonnées d'exécution. Limitez la taille des paramètres System et des metadata d'exécution : ils sont répétés dans chaque batch, si bien qu'une exécution de 10 000 enregistrements échoue avant de démarrer au-delà d'environ 200 Ko. Divisez l'ensemble de données en plusieurs exécutions si vous avez besoin de plus.
• Générations et scores par enregistrement. Le workflow d'évaluation conserve environ 300 octets par enregistrement, auxquels s'ajoutent, pour chaque génération, environ 50 octets plus 100 octets par scorer (au moins 250 octets par génération, pour tenir compte d'une génération en échec). Le nombre de scorers s'applique à chaque génération ; il est donc multiplié par num_generations. Par exemple, avec 10 000 petits enregistrements, une exécution peut contenir une génération avec jusqu'à 34 scorers, ou jusqu'à 13 générations avec un ou deux scorers.

Une exécution qui dépasse ces limites échoue avant de démarrer. Pour corriger le problème, réduisez num_generations, supprimez des scorers ou divisez l'ensemble de données en plusieurs exécutions.

Les limites de l'API Studio s'appliquent également : voir Limites.

• Task modes : exécuter la tâche comme une activité, une classe de workflow ou un workflow déployé.
• Decorators : configurer les timeouts, les nouvelles tentatives et la concurrence, et attacher des callbacks de métadonnées.
• Optimization : exécuter l'optimiseur de prompt et de paramètres comme un workflow.
• Les agents : évaluez un agent de l'Agents SDK à partir d'un workflow.
• Rescoring : rescorer des exécutions persistées sans réexécuter la tâche.
• Building blocks : contrôler la distribution et appeler soi-même les étapes de configuration, de scoring et de téléchargement.
• Référence API : evaluation.run(), les décorateurs et les types du plugin.

Pour les notes de version, consultez l'historique des versions sur PyPI.