Jeux de données

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

from mistralai.evaluations import TaskContext, ScorerContext

async def task(ctx: TaskContext) -> str:
    return ctx.input_record["sentence"]  # access any key you defined

def scorer(ctx: ScorerContext) -> int:
    return 1 if ctx.input_record["groundtruth"].lower() == str(ctx.output).lower() else 0

from mistralai.evaluations import Dataset, Evaluation, Evaluator, Project

run = await client.evaluation.run(
    project=Project(name="Language Detection"),
    evaluation=Evaluation(name="Managed Dataset Eval"),
    dataset=Dataset(slug="language-detection-golden-set"),
    task=task,
    evaluators=[Evaluator(name="accuracy", scorer=scorer)],
)

payload = {"sentence": "Bonjour, comment ça va?", "groundtruth": "French"}

{"payload":{"sentence":"Bonjour, comment ça va?","groundtruth":"French"},"properties":{}}

from typing import Any, TypedDict

from mistralai.evaluations import Mistral

class LangItem(TypedDict):
    sentence: str
    groundtruth: str

def as_dict(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if hasattr(value, "model_dump"):
        return value.model_dump(mode="json")
    return dict(value)

def require_string(value: Any, *, record_id: str, field: str) -> str:
    if isinstance(value, str) and value:
        return value
    raise ValueError(f"Dataset record {record_id} is missing {field}")

def to_lang_item(record: Any) -> LangItem:
    payload = as_dict(record.payload)

    return {
        "sentence": require_string(payload.get("sentence"), record_id=record.id, field="payload.sentence"),
        "groundtruth": require_string(
            payload.get("groundtruth"),
            record_id=record.id,
            field="payload.groundtruth",
        ),
    }

async def fetch_eval_dataset(client: Mistral, dataset_id: str) -> list[LangItem]:
    dataset: list[LangItem] = []
    page = 1
    page_size = 100

    while True:
        result = await client.beta.observability.datasets.list_records_async(
            dataset_id=dataset_id,
            page_size=page_size,
            page=page,
        )
        records = result.records.results

        dataset.extend(to_lang_item(record) for record in records)

        if not result.records.next:
            break
        page += 1

    return dataset

import asyncio
import os

from mistralai.evaluations import (
    Evaluation, Evaluator, Mistral, Project, ScorerContext, TaskContext,
)

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

async def task(ctx: TaskContext) -> str:
    response = await client.chat.complete_async(
        model="mistral-small-latest",
        messages=[
            {
                "role": "user",
                "content": (
                    "What language is this sentence in? Reply with ONLY the language name. "
                    f"Sentence: {ctx.input_record['sentence']}"
                ),
            }
        ],
    )
    return str(response.choices[0].message.content)

def scorer(ctx: ScorerContext) -> int:
    groundtruth = ctx.input_record["groundtruth"].lower()
    predicted = str(ctx.output).lower()
    return 1 if groundtruth == predicted else 0

async def main():
    dataset_id = os.environ["MISTRAL_DATASET_ID"]
    dataset = await fetch_eval_dataset(client, dataset_id=dataset_id)

    run = await client.evaluation.run(
        project=Project(name="Language Detection"),
        evaluation=Evaluation(name="Managed Dataset Eval"),
        dataset=dataset,
        task=task,
        evaluators=[
            Evaluator(
                name="accuracy",
                description="1 if the detected language matches the groundtruth.",
                scorer=scorer,
            ),
        ],
        metadata={
            "studio_dataset_id": dataset_id,
            "record_count": len(dataset),
        },
    )
    run.show(level="records")

asyncio.run(main())

from typing import TypedDict

class LanguageRecord(TypedDict):
    sentence: str
    groundtruth: str

dataset: list[LanguageRecord] = [
    {"sentence": "Hello, how are you?", "groundtruth": "English"},
    {"sentence": "Bonjour, comment ça va?", "groundtruth": "French"},
]

dataset = [
    {
        "prompt": "What is the capital of France?",
        "expected": "Paris",
        "difficulty": "easy",
    },
    {
        "prompt": "Explain the difference between precision and recall.",
        "expected": "Precision measures true positives over predicted positives; recall measures true positives over actual positives.",
        "difficulty": "medium",
    },
]

def accuracy_scorer(ctx: ScorerContext) -> int:
    return 1 if ctx.input_record["expected"].lower() in str(ctx.output).lower() else 0

from mistralai.evaluations import Evaluation, Project

run = await client.evaluation.run(
    project=Project(name="Language Detection"),
    evaluation=Evaluation(name="Accuracy Eval"),
    dataset=dataset,  # your list of dicts, or a Dataset
    task=task,
    evaluators=[...],
)

run = await client.evaluation.run(
    ...
    tags=["dataset:qa_baseline_2025_06", "model:mistral-small"],
    metadata={"dataset_version": "2025-06", "record_count": len(dataset)},
)