"""Langfuse experiment functionality for running and evaluating tasks on datasets.

This module provides the core experiment functionality for the Langfuse Python SDK,

allowing users to run experiments on datasets with automatic tracing, evaluation,

and result formatting.

"""

import asyncio

from datetime import datetime

from typing import (

TYPE_CHECKING,

Any,

Awaitable,

Dict,

List,

Optional,

Protocol,

TypedDict,

Union,

overload,

)

from langfuse.api import DatasetItem

from langfuse.logger import langfuse_logger as logger

from langfuse.types import ExperimentScoreType

if TYPE_CHECKING:

from langfuse._client.client import Langfuse

from langfuse.batch_evaluation import CompositeEvaluatorFunction

class LocalExperimentItem(TypedDict, total=False):

"""Structure for local experiment data items (not from Langfuse datasets).

This TypedDict defines the structure for experiment items when using local data

rather than Langfuse-hosted datasets. All fields are optional to provide

flexibility in data structure.

Attributes:

input: The input data to pass to the task function. Can be any type that

your task function can process (string, dict, list, etc.). This is

typically the prompt, question, or data that your task will operate on.

expected_output: Optional expected/ground truth output for evaluation purposes.

Used by evaluators to assess correctness or quality. Can be None if

no ground truth is available.

metadata: Optional metadata dictionary containing additional context about

this specific item. Can include information like difficulty level,

category, source, or any other relevant attributes that evaluators

might use for context-aware evaluation.

Examples:

Simple text processing item:

```python

item: LocalExperimentItem = {

"input": "Summarize this article: ...",

"expected_output": "Expected summary...",

"metadata": {"difficulty": "medium", "category": "news"}

}

```

Classification item:

```python

item: LocalExperimentItem = {

"input": {"text": "This movie is great!", "context": "movie review"},

"expected_output": "positive",

"metadata": {"dataset_source": "imdb", "confidence": 0.95}

}

```

Minimal item with only input:

```python

item: LocalExperimentItem = {

"input": "What is the capital of France?"

}

```

"""

input: Any

expected_output: Any

metadata: Optional[Dict[str, Any]]

ExperimentItem = Union[LocalExperimentItem, DatasetItem]

"""Type alias for items that can be processed in experiments.

Can be either:

- LocalExperimentItem: Dict-like items with 'input', 'expected_output', 'metadata' keys

- DatasetItem: Items from Langfuse datasets with .input, .expected_output, .metadata attributes

"""

ExperimentData = Union[List[LocalExperimentItem], List[DatasetItem]]

"""Type alias for experiment datasets.

Represents the collection of items to process in an experiment. Can be either:

- List[LocalExperimentItem]: Local data items as dictionaries

- List[DatasetItem]: Items from a Langfuse dataset (typically from dataset.items)

"""

class Evaluation:

"""Represents an evaluation result for an experiment item or an entire experiment run.

This class provides a strongly-typed way to create evaluation results in evaluator functions.

Users must use keyword arguments when instantiating this class.

Attributes:

name: Unique identifier for the evaluation metric. Should be descriptive

and consistent across runs (e.g., "accuracy", "bleu_score", "toxicity").

Used for aggregation and comparison across experiment runs.

value: The evaluation score or result. Can be:

- Numeric (int/float): For quantitative metrics like accuracy (0.85), BLEU (0.42)

- String: For categorical results like "positive", "negative", "neutral"

- Boolean: For binary assessments like "passes_safety_check"

comment: Optional human-readable explanation of the evaluation result.

Useful for providing context, explaining scoring rationale, or noting

special conditions. Displayed in Langfuse UI for interpretability.

metadata: Optional structured metadata about the evaluation process.

Can include confidence scores, intermediate calculations, model versions,

or any other relevant technical details.

data_type: Optional score data type. Required if value is not NUMERIC.

One of NUMERIC, CATEGORICAL, or BOOLEAN. Defaults to NUMERIC.

config_id: Optional Langfuse score config ID.

Examples:

Basic accuracy evaluation:

```python

from langfuse import Evaluation

def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):

if not expected_output:

return Evaluation(name="accuracy", value=0, comment="No expected output")

is_correct = output.strip().lower() == expected_output.strip().lower()

return Evaluation(

name="accuracy",

value=1.0 if is_correct else 0.0,

comment="Correct answer" if is_correct else "Incorrect answer"

)

```

Multi-metric evaluator:

```python

def comprehensive_evaluator(*, input, output, expected_output=None, **kwargs):

return [

Evaluation(name="length", value=len(output), comment=f"Output length: {len(output)} chars"),

Evaluation(name="has_greeting", value="hello" in output.lower(), comment="Contains greeting"),

Evaluation(

name="quality",

value=0.85,

comment="High quality response",

metadata={"confidence": 0.92, "model": "gpt-4"}

)

]

```

Categorical evaluation:

```python

def sentiment_evaluator(*, input, output, **kwargs):

sentiment = analyze_sentiment(output) # Returns "positive", "negative", or "neutral"

return Evaluation(

name="sentiment",

value=sentiment,

comment=f"Response expresses {sentiment} sentiment",

data_type="CATEGORICAL"

)

```

Failed evaluation with error handling:

```python

def external_api_evaluator(*, input, output, **kwargs):

try:

score = external_api.evaluate(output)

return Evaluation(name="external_score", value=score)

except Exception as e:

return Evaluation(

name="external_score",

value=0,

comment=f"API unavailable: {e}",

metadata={"error": str(e), "retry_count": 3}

)

```

Note:

All arguments must be passed as keywords. Positional arguments are not allowed

to ensure code clarity and prevent errors from argument reordering.

"""

def __init__(

self,

*,

name: str,

value: Union[int, float, str, bool],

comment: Optional[str] = None,

metadata: Optional[Dict[str, Any]] = None,

data_type: Optional[ExperimentScoreType] = None,

config_id: Optional[str] = None,

):

"""Initialize an Evaluation with the provided data.

Args:

name: Unique identifier for the evaluation metric.

value: The evaluation score or result.

comment: Optional human-readable explanation of the result.

metadata: Optional structured metadata about the evaluation process.

data_type: Optional score data type (NUMERIC, CATEGORICAL, or BOOLEAN).

config_id: Optional Langfuse score config ID.

Note:

All arguments must be provided as keywords. Positional arguments will raise a TypeError.

"""

self.name = name

self.value = value

self.comment = comment

self.metadata = metadata

self.data_type = data_type

self.config_id = config_id

class ExperimentItemResult:

"""Result structure for individual experiment items.

This class represents the complete result of processing a single item

during an experiment run, including the original input, task output,

evaluations, and tracing information. Users must use keyword arguments when instantiating this class.

Attributes:

item: The original experiment item that was processed. Can be either

a dictionary with 'input', 'expected_output', and 'metadata' keys,

or a DatasetItem from Langfuse datasets.

output: The actual output produced by the task function for this item.

Can be any type depending on what your task function returns.

evaluations: List of evaluation results for this item. Each evaluation

contains a name, value, optional comment, and optional metadata.

trace_id: Optional Langfuse trace ID for this item's execution. Used

to link the experiment result with the detailed trace in Langfuse UI.

dataset_run_id: Optional dataset run ID if this item was part of a

Langfuse dataset. None for local experiments.

Examples:

Accessing item result data:

```python

result = langfuse.run_experiment(...)

for item_result in result.item_results:

print(f"Input: {item_result.item}")

print(f"Output: {item_result.output}")

print(f"Trace: {item_result.trace_id}")

# Access evaluations

for evaluation in item_result.evaluations:

print(f"{evaluation.name}: {evaluation.value}")

```

Working with different item types:

```python

# Local experiment item (dict)

if isinstance(item_result.item, dict):

input_data = item_result.item["input"]

expected = item_result.item.get("expected_output")

# Langfuse dataset item (object with attributes)

else:

input_data = item_result.item.input

expected = item_result.item.expected_output

```

Note:

All arguments must be passed as keywords. Positional arguments are not allowed

to ensure code clarity and prevent errors from argument reordering.

"""

def __init__(

self,

*,

item: ExperimentItem,

output: Any,

evaluations: List[Evaluation],

trace_id: Optional[str],

dataset_run_id: Optional[str],

):

"""Initialize an ExperimentItemResult with the provided data.

Args:

item: The original experiment item that was processed.

output: The actual output produced by the task function for this item.

evaluations: List of evaluation results for this item.

trace_id: Optional Langfuse trace ID for this item's execution.

dataset_run_id: Optional dataset run ID if this item was part of a Langfuse dataset.

Note:

All arguments must be provided as keywords. Positional arguments will raise a TypeError.

"""

self.item = item

self.output = output

self.evaluations = evaluations

self.trace_id = trace_id

self.dataset_run_id = dataset_run_id

class ExperimentResult:

"""Complete result structure for experiment execution.

This class encapsulates the complete results of running an experiment on a dataset,

including individual item results, aggregate run-level evaluations, and metadata

about the experiment execution.

Attributes:

name: The name of the experiment as specified during execution.

run_name: The name of the current experiment run.

description: Optional description of the experiment's purpose or methodology.

item_results: List of results from processing each individual dataset item,

containing the original item, task output, evaluations, and trace information.

run_evaluations: List of aggregate evaluation results computed across all items,

such as average scores, statistical summaries, or cross-item analyses.

experiment_id: ID of the experiment run propagated across all items. For

Langfuse datasets, this matches the dataset run ID. For local experiments,

this is a stable SDK-generated identifier for the run.

dataset_run_id: Optional ID of the dataset run in Langfuse (when using Langfuse datasets).

dataset_run_url: Optional direct URL to view the experiment results in Langfuse UI.

Examples:

Basic usage with local dataset:

```python

result = langfuse.run_experiment(

name="Capital Cities Test",

data=local_data,

task=generate_capital,

evaluators=[accuracy_check]

)

print(f"Processed {len(result.item_results)} items")

print(result.format()) # Human-readable summary

# Access individual results

for item_result in result.item_results:

print(f"Input: {item_result.item}")

print(f"Output: {item_result.output}")

print(f"Scores: {item_result.evaluations}")

```

Usage with Langfuse datasets:

```python

dataset = langfuse.get_dataset("qa-eval-set")

result = dataset.run_experiment(

name="GPT-4 QA Evaluation",

task=answer_question,

evaluators=[relevance_check, accuracy_check]

)

# View in Langfuse UI

if result.dataset_run_url:

print(f"View detailed results: {result.dataset_run_url}")

```

Formatted output:

```python

# Get summary view

summary = result.format()

print(summary)

# Get detailed view with individual items

detailed = result.format(include_item_results=True)

with open("experiment_report.txt", "w") as f:

f.write(detailed)

```

"""

def __init__(

self,

*,

name: str,

run_name: str,

description: Optional[str],

item_results: List[ExperimentItemResult],

run_evaluations: List[Evaluation],

experiment_id: str,

dataset_run_id: Optional[str] = None,

dataset_run_url: Optional[str] = None,

):

"""Initialize an ExperimentResult with the provided data.

Args:

name: The name of the experiment.

run_name: The current experiment run name.

description: Optional description of the experiment.

item_results: List of results from processing individual dataset items.

run_evaluations: List of aggregate evaluation results for the entire run.

experiment_id: ID of the experiment run.

dataset_run_id: Optional ID of the dataset run (for Langfuse datasets).

dataset_run_url: Optional URL to view results in Langfuse UI.

"""

self.name = name

self.run_name = run_name

self.description = description

self.item_results = item_results

self.run_evaluations = run_evaluations

self.experiment_id = experiment_id

self.dataset_run_id = dataset_run_id

self.dataset_run_url = dataset_run_url

def format(self, *, include_item_results: bool = False) -> str:

r"""Format the experiment result for human-readable display.

Converts the experiment result into a nicely formatted string suitable for

console output, logging, or reporting. The output includes experiment overview,

aggregate statistics, and optionally individual item details.

This method provides a comprehensive view of experiment performance including:

- Experiment metadata (name, description, item count)

- List of evaluation metrics used across items

- Average scores computed across all processed items

- Run-level evaluation results (aggregate metrics)

- Links to view detailed results in Langfuse UI (when available)

- Individual item details (when requested)

Args:

include_item_results: Whether to include detailed results for each individual

item in the formatted output. When False (default), only shows aggregate

statistics and summary information. When True, includes input/output/scores

for every processed item, making the output significantly longer but more

detailed for debugging and analysis purposes.

Returns:

A formatted multi-line string containing:

- Experiment name and description (if provided)

- Total number of items successfully processed

- List of all evaluation metrics that were applied

- Average scores across all items for each numeric metric

- Run-level evaluation results with comments

- Dataset run URL for viewing in Langfuse UI (if applicable)

- Individual item details including inputs, outputs, and scores (if requested)

Examples:

Basic usage showing aggregate results only:

```python

result = langfuse.run_experiment(

name="Capital Cities",

data=dataset,

task=generate_capital,

evaluators=[accuracy_evaluator]

)

print(result.format())

# Output:

# ──────────────────────────────────────────────────

# 📊 Capital Cities

# 100 items

# Evaluations:

# • accuracy

# Average Scores:

# • accuracy: 0.850

```

Detailed output including all individual item results:

```python

detailed_report = result.format(include_item_results=True)

print(detailed_report)

# Output includes each item:

# 1. Item 1:

# Input: What is the capital of France?

# Expected: Paris

# Actual: The capital of France is Paris.

# Scores:

# • accuracy: 1.000

# 💭 Correct answer found

# [... continues for all items ...]

```

Saving formatted results to file for reporting:

```python

with open("experiment_report.txt", "w") as f:

f.write(result.format(include_item_results=True))

# Or create summary report

summary = result.format() # Aggregate view only

print(f"Experiment Summary:\n{summary}")

```

Integration with logging systems:

```python

import logging

logger = logging.getLogger("experiments")

# Log summary after experiment

logger.info(f"Experiment completed:\n{result.format()}")

# Log detailed results for failed experiments

if any(eval['value'] < threshold for eval in result.run_evaluations):

logger.warning(f"Poor performance detected:\n{result.format(include_item_results=True)}")

```

"""

if not self.item_results:

return "No experiment results to display."

output = ""

# Individual results section

if include_item_results:

for i, result in enumerate(self.item_results):

output += f"\n{i + 1}. Item {i + 1}:\n"

# Extract and display input

item_input = None

if isinstance(result.item, dict):

item_input = result.item.get("input")

elif hasattr(result.item, "input"):

item_input = result.item.input

if item_input is not None:

output += f" Input: {_format_value(item_input)}\n"

# Extract and display expected output

expected_output = None

if isinstance(result.item, dict):

expected_output = result.item.get("expected_output")

elif hasattr(result.item, "expected_output"):

expected_output = result.item.expected_output

if expected_output is not None:

output += f" Expected: {_format_value(expected_output)}\n"

output += f" Actual: {_format_value(result.output)}\n"

# Display evaluation scores

if result.evaluations:

output += " Scores:\n"

for evaluation in result.evaluations:

score = evaluation.value

if isinstance(score, (int, float)):

score = f"{score:.3f}"

output += f" • {evaluation.name}: {score}"

if evaluation.comment:

output += f"\n 💭 {evaluation.comment}"

output += "\n"

# Display trace link if available

if result.trace_id:

output += f"\n Trace ID: {result.trace_id}\n"

else:

output += f"Individual Results: Hidden ({len(self.item_results)} items)\n"

output += "💡 Set include_item_results=True to view them\n"

# Experiment overview section

output += f"\n{'─' * 50}\n"

output += f"🧪 Experiment: {self.name}"

output += f"\n📋 Run name: {self.run_name}"

if self.description:

output += f" - {self.description}"

output += f"\n{len(self.item_results)} items"

# Collect unique evaluation names across all items

evaluation_names = set()

for result in self.item_results:

for evaluation in result.evaluations:

evaluation_names.add(evaluation.name)

if evaluation_names:

output += "\nEvaluations:"

for eval_name in evaluation_names:

output += f"\n • {eval_name}"

output += "\n"

# Calculate and display average scores

if evaluation_names:

output += "\nAverage Scores:"

for eval_name in evaluation_names:

scores = []

for result in self.item_results:

for evaluation in result.evaluations:

if evaluation.name == eval_name and isinstance(

evaluation.value, (int, float)

):

scores.append(evaluation.value)

if scores:

avg = sum(scores) / len(scores)

output += f"\n • {eval_name}: {avg:.3f}"

output += "\n"

# Display run-level evaluations

if self.run_evaluations:

output += "\nRun Evaluations:"

for run_eval in self.run_evaluations:

score = run_eval.value

if isinstance(score, (int, float)):

score = f"{score:.3f}"

output += f"\n • {run_eval.name}: {score}"

if run_eval.comment:

output += f"\n 💭 {run_eval.comment}"

output += "\n"

# Add dataset run URL if available

if self.dataset_run_url:

output += f"\n🔗 Dataset Run:\n {self.dataset_run_url}"

return output

class TaskFunction(Protocol):

"""Protocol defining the interface for experiment task functions.

Task functions are the core processing functions that operate on each item

in an experiment dataset. They receive an experiment item as input and

produce some output that will be evaluated.

Task functions must:

- Accept 'item' as a keyword argument

- Return any type of output (will be passed to evaluators)

- Can be either synchronous or asynchronous

- Should handle their own errors gracefully (exceptions will be logged)

"""

def __call__(

self,

*,

item: ExperimentItem,

**kwargs: Dict[str, Any],

) -> Union[Any, Awaitable[Any]]:

"""Execute the task on an experiment item.

This method defines the core processing logic for each item in your experiment.

The implementation should focus on the specific task you want to evaluate,

such as text generation, classification, summarization, etc.

Args:

item: The experiment item to process. Can be either:

- Dict with keys like 'input', 'expected_output', 'metadata'

- Langfuse DatasetItem object with .input, .expected_output attributes

**kwargs: Additional keyword arguments that may be passed by the framework

Returns:

Any: The output of processing the item. This output will be:

- Stored in the experiment results

- Passed to all item-level evaluators for assessment

- Traced automatically in Langfuse for observability

Can return either a direct value or an awaitable (async) result.

Examples:

Simple synchronous task:

```python

def my_task(*, item, **kwargs):

prompt = f"Summarize: {item['input']}"

return my_llm_client.generate(prompt)

```

Async task with error handling:

```python

async def my_async_task(*, item, **kwargs):

try:

response = await openai_client.chat.completions.create(

model="gpt-4",

messages=[{"role": "user", "content": item["input"]}]

)

return response.choices[0].message.content

except Exception as e:

# Log error and return fallback

print(f"Task failed for item {item}: {e}")

return "Error: Could not process item"

```

Task using dataset item attributes:

```python

def classification_task(*, item, **kwargs):

# Works with both dict items and DatasetItem objects

text = item["input"] if isinstance(item, dict) else item.input

return classify_text(text)

```

"""

...

class EvaluatorFunction(Protocol):

"""Protocol defining the interface for item-level evaluator functions.

Item-level evaluators assess the quality, correctness, or other properties

of individual task outputs. They receive the input, output, expected output,

and metadata for each item and return evaluation metrics.

Evaluators should:

- Accept input, output, expected_output, and metadata as keyword arguments

- Return Evaluation dict(s) with 'name', 'value', 'comment', 'metadata' fields

- Be deterministic when possible for reproducible results

- Handle edge cases gracefully (missing expected output, malformed data, etc.)

- Can be either synchronous or asynchronous

"""

def __call__(

self,

*,

input: Any,

output: Any,

expected_output: Any,

metadata: Optional[Dict[str, Any]],

**kwargs: Dict[str, Any],

) -> Union[

Evaluation, List[Evaluation], Awaitable[Union[Evaluation, List[Evaluation]]]

]:

r"""Evaluate a task output for quality, correctness, or other metrics.

This method should implement specific evaluation logic such as accuracy checking,

similarity measurement, toxicity detection, fluency assessment, etc.

Args:

input: The original input that was passed to the task function.

This is typically the item['input'] or item.input value.

output: The output produced by the task function for this input.

This is the direct return value from your task function.

expected_output: The expected/ground truth output for comparison.

May be None if not available in the dataset. Evaluators should

handle this case appropriately.

metadata: Optional metadata from the experiment item that might

contain additional context for evaluation (categories, difficulty, etc.)

**kwargs: Additional keyword arguments that may be passed by the framework

Returns:

Evaluation results in one of these formats:

- Single Evaluation dict: {"name": "accuracy", "value": 0.85, "comment": "..."}

- List of Evaluation dicts: [{"name": "precision", ...}, {"name": "recall", ...}]

- Awaitable returning either of the above (for async evaluators)

Each Evaluation dict should contain:

- name (str): Unique identifier for this evaluation metric

- value (int|float|str|bool): The evaluation score or result

- comment (str, optional): Human-readable explanation of the result

- metadata (dict, optional): Additional structured data about the evaluation

Examples:

Simple accuracy evaluator:

```python

def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):

if expected_output is None:

return {"name": "accuracy", "value": 0, "comment": "No expected output"}

is_correct = output.strip().lower() == expected_output.strip().lower()

return {

"name": "accuracy",

"value": 1.0 if is_correct else 0.0,

"comment": "Exact match" if is_correct else "No match"

}

```

Multi-metric evaluator:

```python

def comprehensive_evaluator(*, input, output, expected_output=None, **kwargs):

results = []

# Length check

results.append({

"name": "output_length",

"value": len(output),

"comment": f"Output contains {len(output)} characters"

})

# Sentiment analysis

sentiment_score = analyze_sentiment(output)

results.append({

"name": "sentiment",

"value": sentiment_score,

"comment": f"Sentiment score: {sentiment_score:.2f}"

})

return results

```

Async evaluator using external API:

```python

async def llm_judge_evaluator(*, input, output, expected_output=None, **kwargs):

prompt = f"Rate the quality of this response on a scale of 1-10:\n"

prompt += f"Question: {input}\nResponse: {output}"

response = await openai_client.chat.completions.create(

model="gpt-4",

messages=[{"role": "user", "content": prompt}]

)

try:

score = float(response.choices[0].message.content.strip())

return {

"name": "llm_judge_quality",

"value": score,

"comment": f"LLM judge rated this {score}/10"

}

except ValueError:

return {

"name": "llm_judge_quality",

"value": 0,

"comment": "Could not parse LLM judge score"

}

```

Context-aware evaluator:

```python

def context_evaluator(*, input, output, metadata=None, **kwargs):

# Use metadata for context-specific evaluation

difficulty = metadata.get("difficulty", "medium") if metadata else "medium"

# Adjust expectations based on difficulty

min_length = {"easy": 50, "medium": 100, "hard": 150}[difficulty]

meets_requirement = len(output) >= min_length

return {

"name": f"meets_{difficulty}_requirement",

"value": meets_requirement,

"comment": f"Output {'meets' if meets_requirement else 'fails'} {difficulty} length requirement"

}

```

"""

...

class RunEvaluatorFunction(Protocol):

"""Protocol defining the interface for run-level evaluator functions.

Run-level evaluators assess aggregate properties of the entire experiment run,

computing metrics that span across all items rather than individual outputs.

They receive the complete results from all processed items and can compute

statistics like averages, distributions, correlations, or other aggregate metrics.

Run evaluators should:

- Accept item_results as a keyword argument containing all item results

- Return Evaluation dict(s) with aggregate metrics

- Handle cases where some items may have failed processing

- Compute meaningful statistics across the dataset

- Can be either synchronous or asynchronous

"""

def __call__(

self,

*,

item_results: List[ExperimentItemResult],

**kwargs: Dict[str, Any],

) -> Union[

Evaluation, List[Evaluation], Awaitable[Union[Evaluation, List[Evaluation]]]

]:

r"""Evaluate the entire experiment run with aggregate metrics.

This method should implement aggregate evaluation logic such as computing

averages, calculating distributions, finding correlations, detecting patterns

across items, or performing statistical analysis on the experiment results.

Args:

item_results: List of results from all successfully processed experiment items.

Each item result contains:

- item: The original experiment item

- output: The task function's output for this item

- evaluations: List of item-level evaluation results

- trace_id: Langfuse trace ID for this execution

- dataset_run_id: Dataset run ID (if using Langfuse datasets)

Note: This list only includes items that were successfully processed.

Failed items are excluded but logged separately.

**kwargs: Additional keyword arguments that may be passed by the framework

Returns:

Evaluation results in one of these formats:

- Single Evaluation dict: {"name": "avg_accuracy", "value": 0.78, "comment": "..."}

- List of Evaluation dicts: [{"name": "mean", ...}, {"name": "std_dev", ...}]

- Awaitable returning either of the above (for async evaluators)

Each Evaluation dict should contain:

- name (str): Unique identifier for this run-level metric

- value (int|float|str|bool): The aggregate evaluation result

- comment (str, optional): Human-readable explanation of the metric

- metadata (dict, optional): Additional structured data about the evaluation

Examples:

Average accuracy calculator:

```python

def average_accuracy(*, item_results, **kwargs):

if not item_results:

return {"name": "avg_accuracy", "value": 0.0, "comment": "No results"}

accuracy_values = []

for result in item_results:

for evaluation in result.evaluations:

if evaluation.name == "accuracy":

accuracy_values.append(evaluation.value)

if not accuracy_values:

return {"name": "avg_accuracy", "value": 0, "comment": "No accuracy evaluations found"}

avg = sum(accuracy_values) / len(accuracy_values)

return {

"name": "avg_accuracy",

"value": avg,

"comment": f"Average accuracy across {len(accuracy_values)} items: {avg:.2%}"

}

```

Multiple aggregate metrics:

```python

def statistical_summary(*, item_results, **kwargs):

if not item_results:

return []

results = []

# Calculate output length statistics

lengths = [len(str(result.output)) for result in item_results]

results.extend([

{"name": "avg_output_length", "value": sum(lengths) / len(lengths)},

{"name": "min_output_length", "value": min(lengths)},

{"name": "max_output_length", "value": max(lengths)}

])

# Success rate

total_items = len(item_results) # Only successful items are included

results.append({

"name": "processing_success_rate",

"value": 1.0, # All items in item_results succeeded

"comment": f"Successfully processed {total_items} items"

})

return results

```

Async run evaluator with external analysis:

```python

async def llm_batch_analysis(*, item_results, **kwargs):

# Prepare batch analysis prompt

outputs = [result.output for result in item_results]

prompt = f"Analyze these {len(outputs)} outputs for common themes:\n"

prompt += "\n".join(f"{i+1}. {output}" for i, output in enumerate(outputs))

response = await openai_client.chat.completions.create(

model="gpt-4",

messages=[{"role": "user", "content": prompt}]

)

return {

"name": "thematic_analysis",

"value": response.choices[0].message.content,

"comment": f"LLM analysis of {len(outputs)} outputs"

}

```

Performance distribution analysis:

```python

def performance_distribution(*, item_results, **kwargs):

# Extract all evaluation scores

all_scores = []

score_by_metric = {}

for result in item_results:

for evaluation in result.evaluations:

metric_name = evaluation.name

value = evaluation.value

if isinstance(value, (int, float)):

all_scores.append(value)

if metric_name not in score_by_metric:

score_by_metric[metric_name] = []

score_by_metric[metric_name].append(value)

results = []

# Overall score distribution

if all_scores:

import statistics

results.append({

"name": "score_std_dev",

"value": statistics.stdev(all_scores) if len(all_scores) > 1 else 0,

"comment": f"Standard deviation across all numeric scores"

})

# Per-metric statistics

for metric, scores in score_by_metric.items():

if len(scores) > 1:

results.append({

"name": f"{metric}_variance",

"value": statistics.variance(scores),

"comment": f"Variance in {metric} across {len(scores)} items"

})

return results

```

"""

...

def _format_value(value: Any) -> str:

"""Format a value for display."""

if isinstance(value, str):

return value[:50] + "..." if len(value) > 50 else value

return str(value)

def _normalize_evaluator_result(result: Any) -> List[Evaluation]:

"""Normalize documented evaluator return shapes to Evaluation objects."""

candidates = result if isinstance(result, list) else [result]

return [

Evaluation(**candidate) if isinstance(candidate, dict) else candidate

for candidate in candidates

if isinstance(candidate, (dict, Evaluation))

]

async def _run_evaluator(

evaluator: Union[EvaluatorFunction, RunEvaluatorFunction],