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 installe le SDK d'évaluation (mistralai-evaluations) comme dépendance. 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. Assurez-vous que MISTRAL_API_KEY est défini dans l'environnement du worker :

export MISTRAL_API_KEY=<your-api-key>

export MISTRAL_API_KEY=<your-api-key>

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 Observability > Evaluate > Evaluations.

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. 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 avec run_id, run_url, statistics et run_scores. Il est sérialisable, donc vous pouvez le renvoyer depuis le workflow. Il n'y a 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.
• Datasets en ligne uniquement, pour le moment. dataset prend une liste d'enregistrements : les ensembles de données de Studio (Dataset) et les statistiques de classification ne sont pas encore pris en charge par le plugin.
• Les échecs font l'objet de nouvelles tentatives par Temporal. Il n'y a pas de retry_failed_records() : chaque tâche et chaque scorer réessaie de son côté, selon son décorateur.

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é.

• 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.
• 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.