Evaluations

import dreadnode as dn

@dn.task
async def summarize(text: str) -> str:
    # In a real-world scenario, this would call an LLM.
    # We'll use a simple truncation to keep the example focused.
    summary = " ".join(text.split()[:5]) + "..."
    return summary

# A dataset is just a list of dictionaries.
# The keys should match the parameter names of your task.
dataset = [
    {"text": "Dreadnode provides a modular framework for evaluating task outputs."},
    {"text": "By decoupling evaluation from execution, you can easily define and compose checks."},
]

# Create the evaluation by attaching the dataset to the task.
basic_eval = summarize.as_eval(
    dataset=dataset,
)

# Run the evaluation and inspect the result.
result = await basic_eval.run()
print(result)

import dreadnode as dn
from dreadnode import scorers

@dn.task
async def summarize(text: str) -> str:
    summary = " ".join(text.split()[:5]) + "..."
    return summary

dataset = [
    {"text": "Dreadnode provides a modular framework for evaluating task outputs."},
    {"text": "By decoupling evaluation from execution, you can easily define and compose checks."},
    {"text": "This text is too short."}, # This sample will produce a summary that fails the check.
]

# 1. Define a scorer to check the output length.
#    The `name` you provide here is how you'll reference it in assertions.
length_check = scorers.length_in_range(min_length=20, max_length=50, name="valid_length")

# 2. Create the Eval using `.as_eval()` and attach the scorer.
# 3. Use `assert_scores` to turn the scorer into a pass/fail condition.
eval_with_assertion = summarize.as_eval(
    dataset=dataset,
    scorers=[length_check],
    assert_scores=["valid_length"], # This string must match the scorer's `name`.
)

result = await eval_with_assertion.run()

print(f"Pass Rate: {result.pass_rate:.2%}")
print(f"Passed samples: {result.passed_count}")
print(f"Failed samples: {result.failed_count}")

import dreadnode as dn
from dreadnode import scorers

@dn.task
async def summarize(text: str, model: str) -> str:
    if model == "truncate_short":
        return " ".join(text.split()[:3]) + "..."
    elif model == "truncate_long":
        return " ".join(text.split()[:8]) + "..."
    return "Invalid model specified."

dataset = [
    {"text": "Dreadnode provides a modular framework for evaluating task outputs, solving the problem of embedding messy and repetitive validation logic."},
    {"text": "By decoupling evaluation from execution, you can easily define, compose, and reuse powerful checks for quality, safety, and correctness."},
]

# The `parameters` dict defines the axes of your experiment.
# An Eval will run a full scenario for `model='truncate_short'`
# and another for `model='truncate_long'`.
benchmark_eval = summarize.as_eval(
    dataset=dataset,
    parameters={
        "model": ["truncate_short", "truncate_long"]
    },
    scorers=[scorers.length_in_range(min_length=20, max_length=50, name="valid_length")],
    assert_scores=["valid_length"],
)

result = await benchmark_eval.run()

# The result object now contains results for each scenario.
for scenario in result.scenarios:
    model_name = scenario.params.get("model")
    print(f"--- Results for model: {model_name} ---")
    print(f"Pass Rate: {scenario.pass_rate:.2%}\n")

import dreadnode as dn
from dreadnode import scorers

@dn.task
async def summarize(text: str) -> str:
    # This task will now produce a more realistic, generative-style summary.
    return "Modular evaluation is possible by decoupling execution from validation."

# Add the "ground_truth_summary" column to your dataset.
dataset = [
    {
        "text": "By decoupling evaluation from execution, you can easily define, compose, and reuse powerful checks for quality, safety, and correctness across your entire application.",
        "ground_truth_summary": "Decoupling evaluation from execution enables reusable quality checks."
    },
]

# 1. The scorer references the `ground_truth_summary` column from the dataset.
# 2. We use a threshold to define "passing" as a similarity score greater than 0.7.
# 3. Finally, we give the composed scorer a clear name for use in assertions.
similarity_check = (
    scorers.similarity_with_sentence_transformers(
        reference=dn.DatasetField("ground_truth_summary")
    ) > 0.7
) >> "is_correct"


correctness_eval = summarize.as_eval(
    dataset=dataset,
    scorers=[similarity_check],
    assert_scores=["is_correct"],
)

result = await correctness_eval.run()

print(f"Pass Rate: {result.pass_rate:.2%}")

import dreadnode as dn
from dreadnode import scorers
import pandas as pd

# To display all columns in the DataFrame
pd.set_option('display.max_columns', None)

@dn.task
async def get_capital(country: str) -> str:
    capitals = {"France": "Paris", "Japan": "Tokyo"}
    return capitals.get(country, "I don't know.")

dataset = [
    {"country": "France", "expected_capital": "Paris"},
    {"country": "Japan", "expected_capital": "Tokyo"},
    {"country": "Germany", "expected_capital": "Berlin"}, # This will fail
]

# A scorer that checks for the correct answer from the dataset.
correctness_check = scorers.equals(dn.DatasetField("expected_capital")) >> "is_correct"

capital_eval = get_capital.as_eval(
    dataset=dataset,
    dataset_input_mapping=["country"], # Map 'country' column to task's 'country' param
    scorers=[correctness_check],
    assert_scores=["is_correct"],
)

# 1. Run the evaluation to get the result object.
result = await capital_eval.run()

# 2. Convert the result to a DataFrame for analysis.
results_df = result.to_dataframe()

# 3. Filter the DataFrame to find only the failed samples.
failed_samples_df = results_df[results_df["passed"] == False]

print("--- Failed Samples ---")
print(failed_samples_df[["input_country", "output", "metric_is_correct"]])

# This uses the same `capital_eval` object from the previous example.

# Instead of .run(), call .console() to see live progress.
# It still returns the same EvalResult object when finished.
result = await capital_eval.console()

import dreadnode as dn
from dreadnode import scorers

@dn.task
async def get_capital(country: str) -> str:
    capitals = {"France": "Paris", "Japan": "Tokyo"}
    return capitals.get(country, "I don't know.")

correctness_check = scorers.equals(dn.DatasetField("expected_capital")) >> "is_correct"

# Simply pass the file path to the `dataset` argument.
file_based_eval = get_capital.as_eval(
    dataset="dataset.jsonl",
    dataset_input_mapping=["country"],
    scorers=[correctness_check],
    assert_scores=["is_correct"],
)

result = await file_based_eval.run()
print(f"Pass Rate: {result.pass_rate:.2%}")

import dreadnode as dn
from dreadnode import scorers

@dn.task
async def get_capital(country: str) -> str: # Task expects `country`
    capitals = {"France": "Paris"}
    return capitals.get(country, "I don't know.")

# Dataset uses `location` instead of `country`.
dataset_with_mismatched_keys = [
    {"location": "France", "expected_capital": "Paris"}
]

# Use a dict to map `dataset_key: task_parameter_name`.
mapping_eval = get_capital.as_eval(
    dataset=dataset_with_mismatched_keys,
    dataset_input_mapping={"location": "country"},
    scorers=[scorers.equals(dn.DatasetField("expected_capital")) >> "is_correct"],
)

result = await mapping_eval.run()
print(result.samples[0].input)

@dn.task
async def flaky_task(value: int) -> int:
    if value == 2:
        raise ValueError("This value causes an error!")
    return value * 2

dataset = [{"value": 1}, {"value": 2}, {"value": 3}, {"value": 4}]

# This evaluation will stop after the first error.
# result = await flaky_task.as_eval(dataset=dataset).run() # This would raise an error and stop.

# This evaluation will tolerate up to 5 total errors and continue running.
resilient_eval = flaky_task.as_eval(
    dataset=dataset,
    max_errors=5,
)

result = await resilient_eval.run()
print(f"Total samples processed: {len(result.samples)}")
print(f"Samples with errors: {len([s for s in result.samples if s.error])}")

# Uses the `capital_eval` from a previous example.
async with capital_eval.stream() as stream:
    async for event in stream:
        if isinstance(event, dn.eval.SampleComplete):
            if event.sample.failed:
                print(f"Detected a failure on sample {event.sample.index}!")
        elif isinstance(event, dn.eval.EvalEnd):
            print(f"Evaluation finished with stop reason: {event.stop_reason}")