Workflow evaluation plugin

The Workflow evaluation plugin (mistralai-workflows-plugins-evaluations) runs offline evaluations inside Mistral Workflows. It exposes the same evaluation.run() API as the Evaluation SDK, but splits task execution and scoring into parallel Temporal activities.

The plugin follows the same mental model as the SDK:

Dataset  →  Task  →  Evaluators  →  Results

Dataset  →  Task  →  Evaluators  →  Results

Instead of running everything in one process, the plugin fans out each record as a child workflow that runs the task, then the scorers. Results are uploaded to Studio as the records complete.

Both write to the same projects, evaluations, and runs in Studio, and share the same primitives: Evaluator, Goal, Statistic, System, and TunableSystem.

Prerequisites: Python 3.12 to 3.14, a Mistral API key, and a working Workflows project. If you don't have one yet, follow Workflows installation and Your first workflow.

Add the plugin to your Workflows project from PyPI:

pip install mistralai-workflows-plugins-evaluations

pip install mistralai-workflows-plugins-evaluations

The plugin pulls in the Evaluation SDK (mistralai-evaluations) as a dependency. Plugin 0.7.0 requires mistralai-evaluations 0.10.0 or later: upgrade both together. You don't need to register anything on the worker: it discovers installed plugins at startup and registers the evaluation activities and child workflows automatically.

The worker uses its own Mistral credentials to upload results to Studio. On a local or self-hosted worker, set MISTRAL_API_KEY in the worker's environment:

export MISTRAL_API_KEY=<your-api-key>

export MISTRAL_API_KEY=<your-api-key>

On a managed deployment, the worker has no API key and the plugin uses its service-account token instead. See Managed deployments.

Evaluations run on managed deployments with no API key. MISTRAL_API_KEY is a reserved name there: the platform gives the worker a rotating service-account token (MISTRAL_SA_TOKEN_PATH), and the plugin uses it automatically for every call to Studio: creating the run, uploading results, and updating its status. There is nothing to configure.

This needs mistralai-workflows 3.10 or later and mistralai-evaluations 0.9.2 or later in the worker image. Upgrading the plugin to 0.6.6 or later pulls in the right mistralai-evaluations. A worker with neither an API key nor a service-account token fails with No credentials found.

An API key never overrides the service-account token. If you copy a key into a managed worker's environment, it is not sent. Grant permissions to the service account, or use on_behalf_of=True, instead of adding a key.

Declare the workflow with on_behalf_of=True. The calls the plugin makes from your workflow then run as the user who executes it, with that user's permissions on their workspace:

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

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

On-behalf-of workflows require a hardened deployment. See On-behalf-of workflows.

The plugin uploads records, and runs your task and scorer code, in its own child workflows. With on_behalf_of=True, those child workflows run as the user too, so everything runs as that user and the service account needs no extra permissions.

Without on_behalf_of, the calls run as the deployment's service account, for example in a scheduled workflow, which has no user. The service account then needs a role that carries the offline evaluations, datasets, and judges permissions on the target workspace. The default workflow_executor role of a deployment does not carry them: add offline_evaluations_manager to the deployment's requested_roles when you create it (["offline_evaluations_manager"], or service_account_roles in the Mistral MCP). See also Service accounts and Roles and permissions. Runs, datasets, judges, and optimizations are then created as the service account, which Studio shows as their creator. Features tied to a person, such as favorites and stars, are not available to it. optimize() works in both modes.

:::note Your own model calls Your tasks and scorers choose their own credentials. On a managed deployment, call get_mistral_client() from mistralai.workflows.client, as in the quickstart: it carries the same service-account token. A Mistral(api_key=os.environ["MISTRAL_API_KEY"]) client is for local runs. :::

When the caller lacks a permission, the 403 raises an ObservabilityAPIError that names the caller (user or service account), the permissions involved (offline evaluations, datasets, or judges), and how to fix it on a managed deployment. This needs mistralai-evaluations 0.10.0 or later. Before, the error asked you to check the API key, which was misleading for a service account that has no key.

Create a src/workflows/language_detection_eval.py file in your Workflows project:

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}

Restart your worker so it registers the new workflow, then trigger it like any other workflow — from the Workflows page in Studio, or with the Mistral SDK:

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={},
)

The workflow returns the run ID and a link to the run. Results are available in Studio under Evaluate > Evaluations > Offline.

The plugin reuses the SDK types, so everything in the SDK guides applies — datasets, system params, evaluators, goals, statistics, and multiple generations. The differences come from running inside a workflow:

• Decorate every function. Tasks use @evaluation.task, scorers @evaluation.scorer, and run evaluators @evaluation.run_scorer. Each one becomes a Temporal activity with its own timeout, retries, and concurrency (workflow-class and workflow-name tasks are not retried; see Task modes). See Decorators.
• Call evaluation from the plugin, not client.evaluation. Import types from mistralai.workflows.plugins.evaluations.types: this module is safe to import inside the workflow sandbox.
• Tasks can be workflows. Besides an activity, the task can be a workflow class or the name of a deployed workflow. See Task modes.
• Scorers return a Score or a number. A plain number is wrapped in a Score automatically.
• run() returns an EvalResult with run_id, run_url, statistics, run_scores, and failed_record_count. It is serializable, so you can return it from the workflow. There is no show() method.
• Run evaluators read records back from Studio. To stay under Temporal's payload limits, the run evaluator activity fetches the persisted records instead of receiving them from the workflow. See Run evaluators.
• Prefer Studio datasets for large inputs. dataset takes a list of records or a Studio dataset (Dataset). A Studio dataset keeps the inputs out of the workflow history. See Datasets and limits.
• No classification statistics yet. Classification statistics aren't supported by the plugin yet.
• Failures are retried by Temporal. There is no retry_failed_records(): each activity task and scorer retries on its own, according to its decorator. Workflow tasks are not retried. A record that still fails is missing from statistics: check EvalResult.failed_record_count before reading the scores (0.7.0+).

Run-level evaluators work as in the SDK, with a decorator:

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)],
)

When results are uploaded to Studio, the run evaluator activity receives the run ID and reads the persisted records back before calling your function. The same applies to run_metadata callbacks. As a consequence:

• Records are listed in upload order, which can differ from the dataset order.
• A record whose output failed to upload appears with no generations.
• The function must be registered on the worker that runs the evaluation workflow.
• Reading the records adds up to 10 minutes to the callback's timeout.

In local mode, records are passed in memory.

local=True skips every call to Studio but keeps the same Temporal topology (child workflows and activities). You don't need a project or evaluation, result.run_id is "local", and result.run_url is None. Statistics and run scores are still computed and returned. Unlike the SDK, omitting evaluation doesn't keep the run local: the plugin saves it to an evaluation named after name, or "Workflow Evaluation", in the default project. See Iterate locally for the recommended workflow.

Workflows run on Temporal, which caps the size of a workflow history and of each payload. How you pass the dataset decides how much of it goes through that history.

For large datasets, pass a dataset stored in Studio instead of inline records:

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=[...],
)

The evaluation workflow only handles record IDs. Each record is fetched inside its own child workflow, and the task output is saved to Studio from there. Neither inputs nor outputs go through the evaluation workflow's history, so their size doesn't count against Temporal's limits. The run is linked to the dataset in Studio.

Inline records are loaded into the workflow first, so every record crosses the workflow history. Switch to Dataset(id=...) once the inputs get large.

A run accepts up to 10,000 records, inline or from a Studio dataset. Inline lists are also bounded by payload size, as described in the following list. Records are processed in batches of child workflows, and the evaluation workflow only keeps each record's score values. A full-size run stays within Temporal's history limits (50 MB and 51,200 events).

Each record is still processed in its own workflow, and score values still travel back to the evaluation workflow. Keep these limits in mind:

• Record size. A single record's input, output, and metadata must each stay under Temporal's 2 MB payload limit.
• Score values. Keep them short (numbers or short labels). Put long text in the score's rationale, which stays in Studio.
• System params and run metadata. Keep the System params and run metadata small: they are repeated in every batch, so a 10,000-record run fails before starting past about 200 KB. Split the dataset into several runs if you need more.
• Generations and scores per record. The evaluation workflow keeps about 300 bytes per record, plus, for each generation, about 50 bytes plus 100 bytes per scorer (at least 250 bytes per generation, to cover a failed one). The scorer count is per generation, so it is multiplied by num_generations. For example, with 10,000 small records a run fits one generation with up to 34 scorers, or up to 13 generations with one or two scorers.

A run that exceeds these limits fails before starting. To fix it, lower num_generations, drop scorers, or split the dataset into several runs.

The Studio API limits also apply: see Limits.

• Task modes: run the task as an activity, a workflow class, or a deployed workflow.
• Decorators: configure timeouts, retries, and concurrency, and attach metadata callbacks.
• Optimization: run the prompt and parameter optimizer as a workflow.
• Agents: evaluate an Agents SDK agent from a workflow.
• Rescoring: score persisted runs again without re-running the task.
• Building blocks: own the fan-out and call the setup, scoring, and upload steps yourself.
• API reference: evaluation.run(), the decorators, and the plugin types.

For release notes, see the release history on PyPI.