langfuse.experiment

   1"""Langfuse experiment functionality for running and evaluating tasks on datasets.
   2
   3This module provides the core experiment functionality for the Langfuse Python SDK,
   4allowing users to run experiments on datasets with automatic tracing, evaluation,
   5and result formatting.
   6"""
   7
   8import asyncio
   9import logging
  10from typing import (
  11    TYPE_CHECKING,
  12    Any,
  13    Awaitable,
  14    Dict,
  15    List,
  16    Optional,
  17    Protocol,
  18    TypedDict,
  19    Union,
  20)
  21
  22from langfuse.api import ScoreDataType
  23
  24if TYPE_CHECKING:
  25    from langfuse._client.datasets import DatasetItemClient
  26
  27
  28class LocalExperimentItem(TypedDict, total=False):
  29    """Structure for local experiment data items (not from Langfuse datasets).
  30
  31    This TypedDict defines the structure for experiment items when using local data
  32    rather than Langfuse-hosted datasets. All fields are optional to provide
  33    flexibility in data structure.
  34
  35    Attributes:
  36        input: The input data to pass to the task function. Can be any type that
  37            your task function can process (string, dict, list, etc.). This is
  38            typically the prompt, question, or data that your task will operate on.
  39        expected_output: Optional expected/ground truth output for evaluation purposes.
  40            Used by evaluators to assess correctness or quality. Can be None if
  41            no ground truth is available.
  42        metadata: Optional metadata dictionary containing additional context about
  43            this specific item. Can include information like difficulty level,
  44            category, source, or any other relevant attributes that evaluators
  45            might use for context-aware evaluation.
  46
  47    Examples:
  48        Simple text processing item:
  49        ```python
  50        item: LocalExperimentItem = {
  51            "input": "Summarize this article: ...",
  52            "expected_output": "Expected summary...",
  53            "metadata": {"difficulty": "medium", "category": "news"}
  54        }
  55        ```
  56
  57        Classification item:
  58        ```python
  59        item: LocalExperimentItem = {
  60            "input": {"text": "This movie is great!", "context": "movie review"},
  61            "expected_output": "positive",
  62            "metadata": {"dataset_source": "imdb", "confidence": 0.95}
  63        }
  64        ```
  65
  66        Minimal item with only input:
  67        ```python
  68        item: LocalExperimentItem = {
  69            "input": "What is the capital of France?"
  70        }
  71        ```
  72    """
  73
  74    input: Any
  75    expected_output: Any
  76    metadata: Optional[Dict[str, Any]]
  77
  78
  79ExperimentItem = Union[LocalExperimentItem, "DatasetItemClient"]
  80"""Type alias for items that can be processed in experiments.
  81
  82Can be either:
  83- LocalExperimentItem: Dict-like items with 'input', 'expected_output', 'metadata' keys
  84- DatasetItemClient: Items from Langfuse datasets with .input, .expected_output, .metadata attributes
  85"""
  86
  87ExperimentData = Union[List[LocalExperimentItem], List["DatasetItemClient"]]
  88"""Type alias for experiment datasets.
  89
  90Represents the collection of items to process in an experiment. Can be either:
  91- List[LocalExperimentItem]: Local data items as dictionaries
  92- List[DatasetItemClient]: Items from a Langfuse dataset (typically from dataset.items)
  93"""
  94
  95
  96class Evaluation:
  97    """Represents an evaluation result for an experiment item or an entire experiment run.
  98
  99    This class provides a strongly-typed way to create evaluation results in evaluator functions.
 100    Users must use keyword arguments when instantiating this class.
 101
 102    Attributes:
 103        name: Unique identifier for the evaluation metric. Should be descriptive
 104            and consistent across runs (e.g., "accuracy", "bleu_score", "toxicity").
 105            Used for aggregation and comparison across experiment runs.
 106        value: The evaluation score or result. Can be:
 107            - Numeric (int/float): For quantitative metrics like accuracy (0.85), BLEU (0.42)
 108            - String: For categorical results like "positive", "negative", "neutral"
 109            - Boolean: For binary assessments like "passes_safety_check"
 110        comment: Optional human-readable explanation of the evaluation result.
 111            Useful for providing context, explaining scoring rationale, or noting
 112            special conditions. Displayed in Langfuse UI for interpretability.
 113        metadata: Optional structured metadata about the evaluation process.
 114            Can include confidence scores, intermediate calculations, model versions,
 115            or any other relevant technical details.
 116        data_type: Optional score data type. Required if value is not NUMERIC.
 117            One of NUMERIC, CATEGORICAL, or BOOLEAN. Defaults to NUMERIC.
 118        config_id: Optional Langfuse score config ID.
 119
 120    Examples:
 121        Basic accuracy evaluation:
 122        ```python
 123        from langfuse import Evaluation
 124
 125        def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
 126            if not expected_output:
 127                return Evaluation(name="accuracy", value=None, comment="No expected output")
 128
 129            is_correct = output.strip().lower() == expected_output.strip().lower()
 130            return Evaluation(
 131                name="accuracy",
 132                value=1.0 if is_correct else 0.0,
 133                comment="Correct answer" if is_correct else "Incorrect answer"
 134            )
 135        ```
 136
 137        Multi-metric evaluator:
 138        ```python
 139        def comprehensive_evaluator(*, input, output, expected_output=None, **kwargs):
 140            return [
 141                Evaluation(name="length", value=len(output), comment=f"Output length: {len(output)} chars"),
 142                Evaluation(name="has_greeting", value="hello" in output.lower(), comment="Contains greeting"),
 143                Evaluation(
 144                    name="quality",
 145                    value=0.85,
 146                    comment="High quality response",
 147                    metadata={"confidence": 0.92, "model": "gpt-4"}
 148                )
 149            ]
 150        ```
 151
 152        Categorical evaluation:
 153        ```python
 154        def sentiment_evaluator(*, input, output, **kwargs):
 155            sentiment = analyze_sentiment(output)  # Returns "positive", "negative", or "neutral"
 156            return Evaluation(
 157                name="sentiment",
 158                value=sentiment,
 159                comment=f"Response expresses {sentiment} sentiment",
 160                data_type="CATEGORICAL"
 161            )
 162        ```
 163
 164        Failed evaluation with error handling:
 165        ```python
 166        def external_api_evaluator(*, input, output, **kwargs):
 167            try:
 168                score = external_api.evaluate(output)
 169                return Evaluation(name="external_score", value=score)
 170            except Exception as e:
 171                return Evaluation(
 172                    name="external_score",
 173                    value=None,
 174                    comment=f"API unavailable: {e}",
 175                    metadata={"error": str(e), "retry_count": 3}
 176                )
 177        ```
 178
 179    Note:
 180        All arguments must be passed as keywords. Positional arguments are not allowed
 181        to ensure code clarity and prevent errors from argument reordering.
 182    """
 183
 184    def __init__(
 185        self,
 186        *,
 187        name: str,
 188        value: Union[int, float, str, bool],
 189        comment: Optional[str] = None,
 190        metadata: Optional[Dict[str, Any]] = None,
 191        data_type: Optional[ScoreDataType] = None,
 192        config_id: Optional[str] = None,
 193    ):
 194        """Initialize an Evaluation with the provided data.
 195
 196        Args:
 197            name: Unique identifier for the evaluation metric.
 198            value: The evaluation score or result.
 199            comment: Optional human-readable explanation of the result.
 200            metadata: Optional structured metadata about the evaluation process.
 201            data_type: Optional score data type (NUMERIC, CATEGORICAL, or BOOLEAN).
 202            config_id: Optional Langfuse score config ID.
 203
 204        Note:
 205            All arguments must be provided as keywords. Positional arguments will raise a TypeError.
 206        """
 207        self.name = name
 208        self.value = value
 209        self.comment = comment
 210        self.metadata = metadata
 211        self.data_type = data_type
 212        self.config_id = config_id
 213
 214
 215class ExperimentItemResult:
 216    """Result structure for individual experiment items.
 217
 218    This class represents the complete result of processing a single item
 219    during an experiment run, including the original input, task output,
 220    evaluations, and tracing information. Users must use keyword arguments when instantiating this class.
 221
 222    Attributes:
 223        item: The original experiment item that was processed. Can be either
 224            a dictionary with 'input', 'expected_output', and 'metadata' keys,
 225            or a DatasetItemClient from Langfuse datasets.
 226        output: The actual output produced by the task function for this item.
 227            Can be any type depending on what your task function returns.
 228        evaluations: List of evaluation results for this item. Each evaluation
 229            contains a name, value, optional comment, and optional metadata.
 230        trace_id: Optional Langfuse trace ID for this item's execution. Used
 231            to link the experiment result with the detailed trace in Langfuse UI.
 232        dataset_run_id: Optional dataset run ID if this item was part of a
 233            Langfuse dataset. None for local experiments.
 234
 235    Examples:
 236        Accessing item result data:
 237        ```python
 238        result = langfuse.run_experiment(...)
 239        for item_result in result.item_results:
 240            print(f"Input: {item_result.item}")
 241            print(f"Output: {item_result.output}")
 242            print(f"Trace: {item_result.trace_id}")
 243
 244            # Access evaluations
 245            for evaluation in item_result.evaluations:
 246                print(f"{evaluation.name}: {evaluation.value}")
 247        ```
 248
 249        Working with different item types:
 250        ```python
 251        # Local experiment item (dict)
 252        if isinstance(item_result.item, dict):
 253            input_data = item_result.item["input"]
 254            expected = item_result.item.get("expected_output")
 255
 256        # Langfuse dataset item (object with attributes)
 257        else:
 258            input_data = item_result.item.input
 259            expected = item_result.item.expected_output
 260        ```
 261
 262    Note:
 263        All arguments must be passed as keywords. Positional arguments are not allowed
 264        to ensure code clarity and prevent errors from argument reordering.
 265    """
 266
 267    def __init__(
 268        self,
 269        *,
 270        item: ExperimentItem,
 271        output: Any,
 272        evaluations: List[Evaluation],
 273        trace_id: Optional[str],
 274        dataset_run_id: Optional[str],
 275    ):
 276        """Initialize an ExperimentItemResult with the provided data.
 277
 278        Args:
 279            item: The original experiment item that was processed.
 280            output: The actual output produced by the task function for this item.
 281            evaluations: List of evaluation results for this item.
 282            trace_id: Optional Langfuse trace ID for this item's execution.
 283            dataset_run_id: Optional dataset run ID if this item was part of a Langfuse dataset.
 284
 285        Note:
 286            All arguments must be provided as keywords. Positional arguments will raise a TypeError.
 287        """
 288        self.item = item
 289        self.output = output
 290        self.evaluations = evaluations
 291        self.trace_id = trace_id
 292        self.dataset_run_id = dataset_run_id
 293
 294
 295class ExperimentResult:
 296    """Complete result structure for experiment execution.
 297
 298    This class encapsulates the complete results of running an experiment on a dataset,
 299    including individual item results, aggregate run-level evaluations, and metadata
 300    about the experiment execution.
 301
 302    Attributes:
 303        name: The name of the experiment as specified during execution.
 304        run_name: The name of the current experiment run.
 305        description: Optional description of the experiment's purpose or methodology.
 306        item_results: List of results from processing each individual dataset item,
 307            containing the original item, task output, evaluations, and trace information.
 308        run_evaluations: List of aggregate evaluation results computed across all items,
 309            such as average scores, statistical summaries, or cross-item analyses.
 310        dataset_run_id: Optional ID of the dataset run in Langfuse (when using Langfuse datasets).
 311        dataset_run_url: Optional direct URL to view the experiment results in Langfuse UI.
 312
 313    Examples:
 314        Basic usage with local dataset:
 315        ```python
 316        result = langfuse.run_experiment(
 317            name="Capital Cities Test",
 318            data=local_data,
 319            task=generate_capital,
 320            evaluators=[accuracy_check]
 321        )
 322
 323        print(f"Processed {len(result.item_results)} items")
 324        print(result.format())  # Human-readable summary
 325
 326        # Access individual results
 327        for item_result in result.item_results:
 328            print(f"Input: {item_result.item}")
 329            print(f"Output: {item_result.output}")
 330            print(f"Scores: {item_result.evaluations}")
 331        ```
 332
 333        Usage with Langfuse datasets:
 334        ```python
 335        dataset = langfuse.get_dataset("qa-eval-set")
 336        result = dataset.run_experiment(
 337            name="GPT-4 QA Evaluation",
 338            task=answer_question,
 339            evaluators=[relevance_check, accuracy_check]
 340        )
 341
 342        # View in Langfuse UI
 343        if result.dataset_run_url:
 344            print(f"View detailed results: {result.dataset_run_url}")
 345        ```
 346
 347        Formatted output:
 348        ```python
 349        # Get summary view
 350        summary = result.format()
 351        print(summary)
 352
 353        # Get detailed view with individual items
 354        detailed = result.format(include_item_results=True)
 355        with open("experiment_report.txt", "w") as f:
 356            f.write(detailed)
 357        ```
 358    """
 359
 360    def __init__(
 361        self,
 362        *,
 363        name: str,
 364        run_name: str,
 365        description: Optional[str],
 366        item_results: List[ExperimentItemResult],
 367        run_evaluations: List[Evaluation],
 368        dataset_run_id: Optional[str] = None,
 369        dataset_run_url: Optional[str] = None,
 370    ):
 371        """Initialize an ExperimentResult with the provided data.
 372
 373        Args:
 374            name: The name of the experiment.
 375            run_name: The current experiment run name.
 376            description: Optional description of the experiment.
 377            item_results: List of results from processing individual dataset items.
 378            run_evaluations: List of aggregate evaluation results for the entire run.
 379            dataset_run_id: Optional ID of the dataset run (for Langfuse datasets).
 380            dataset_run_url: Optional URL to view results in Langfuse UI.
 381        """
 382        self.name = name
 383        self.run_name = run_name
 384        self.description = description
 385        self.item_results = item_results
 386        self.run_evaluations = run_evaluations
 387        self.dataset_run_id = dataset_run_id
 388        self.dataset_run_url = dataset_run_url
 389
 390    def format(self, *, include_item_results: bool = False) -> str:
 391        r"""Format the experiment result for human-readable display.
 392
 393        Converts the experiment result into a nicely formatted string suitable for
 394        console output, logging, or reporting. The output includes experiment overview,
 395        aggregate statistics, and optionally individual item details.
 396
 397        This method provides a comprehensive view of experiment performance including:
 398        - Experiment metadata (name, description, item count)
 399        - List of evaluation metrics used across items
 400        - Average scores computed across all processed items
 401        - Run-level evaluation results (aggregate metrics)
 402        - Links to view detailed results in Langfuse UI (when available)
 403        - Individual item details (when requested)
 404
 405        Args:
 406            include_item_results: Whether to include detailed results for each individual
 407                item in the formatted output. When False (default), only shows aggregate
 408                statistics and summary information. When True, includes input/output/scores
 409                for every processed item, making the output significantly longer but more
 410                detailed for debugging and analysis purposes.
 411
 412        Returns:
 413            A formatted multi-line string containing:
 414            - Experiment name and description (if provided)
 415            - Total number of items successfully processed
 416            - List of all evaluation metrics that were applied
 417            - Average scores across all items for each numeric metric
 418            - Run-level evaluation results with comments
 419            - Dataset run URL for viewing in Langfuse UI (if applicable)
 420            - Individual item details including inputs, outputs, and scores (if requested)
 421
 422        Examples:
 423            Basic usage showing aggregate results only:
 424            ```python
 425            result = langfuse.run_experiment(
 426                name="Capital Cities",
 427                data=dataset,
 428                task=generate_capital,
 429                evaluators=[accuracy_evaluator]
 430            )
 431
 432            print(result.format())
 433            # Output:
 434            # ──────────────────────────────────────────────────
 435            # 📊 Capital Cities
 436            # 100 items
 437            # Evaluations:
 438            #   • accuracy
 439            # Average Scores:
 440            #   • accuracy: 0.850
 441            ```
 442
 443            Detailed output including all individual item results:
 444            ```python
 445            detailed_report = result.format(include_item_results=True)
 446            print(detailed_report)
 447            # Output includes each item:
 448            # 1. Item 1:
 449            #    Input:    What is the capital of France?
 450            #    Expected: Paris
 451            #    Actual:   The capital of France is Paris.
 452            #    Scores:
 453            #      • accuracy: 1.000
 454            #        💭 Correct answer found
 455            # [... continues for all items ...]
 456            ```
 457
 458            Saving formatted results to file for reporting:
 459            ```python
 460            with open("experiment_report.txt", "w") as f:
 461                f.write(result.format(include_item_results=True))
 462
 463            # Or create summary report
 464            summary = result.format()  # Aggregate view only
 465            print(f"Experiment Summary:\\n{summary}")
 466            ```
 467
 468            Integration with logging systems:
 469            ```python
 470            import logging
 471            logger = logging.getLogger("experiments")
 472
 473            # Log summary after experiment
 474            logger.info(f"Experiment completed:\\n{result.format()}")
 475
 476            # Log detailed results for failed experiments
 477            if any(eval['value'] < threshold for eval in result.run_evaluations):
 478                logger.warning(f"Poor performance detected:\\n{result.format(include_item_results=True)}")
 479            ```
 480        """
 481        if not self.item_results:
 482            return "No experiment results to display."
 483
 484        output = ""
 485
 486        # Individual results section
 487        if include_item_results:
 488            for i, result in enumerate(self.item_results):
 489                output += f"\\n{i + 1}. Item {i + 1}:\\n"
 490
 491                # Extract and display input
 492                item_input = None
 493                if isinstance(result.item, dict):
 494                    item_input = result.item.get("input")
 495                elif hasattr(result.item, "input"):
 496                    item_input = result.item.input
 497
 498                if item_input is not None:
 499                    output += f"   Input:    {_format_value(item_input)}\\n"
 500
 501                # Extract and display expected output
 502                expected_output = None
 503                if isinstance(result.item, dict):
 504                    expected_output = result.item.get("expected_output")
 505                elif hasattr(result.item, "expected_output"):
 506                    expected_output = result.item.expected_output
 507
 508                if expected_output is not None:
 509                    output += f"   Expected: {_format_value(expected_output)}\\n"
 510                output += f"   Actual:   {_format_value(result.output)}\\n"
 511
 512                # Display evaluation scores
 513                if result.evaluations:
 514                    output += "   Scores:\\n"
 515                    for evaluation in result.evaluations:
 516                        score = evaluation.value
 517                        if isinstance(score, (int, float)):
 518                            score = f"{score:.3f}"
 519                        output += f"     • {evaluation.name}: {score}"
 520                        if evaluation.comment:
 521                            output += f"\\n       💭 {evaluation.comment}"
 522                        output += "\\n"
 523
 524                # Display trace link if available
 525                if result.trace_id:
 526                    output += f"\\n   Trace ID: {result.trace_id}\\n"
 527        else:
 528            output += f"Individual Results: Hidden ({len(self.item_results)} items)\\n"
 529            output += "💡 Set include_item_results=True to view them\\n"
 530
 531        # Experiment overview section
 532        output += f"\\n{'─' * 50}\\n"
 533        output += f"🧪 Experiment: {self.name}"
 534        output += f"\n📋 Run name: {self.run_name}"
 535        if self.description:
 536            output += f" - {self.description}"
 537
 538        output += f"\\n{len(self.item_results)} items"
 539
 540        # Collect unique evaluation names across all items
 541        evaluation_names = set()
 542        for result in self.item_results:
 543            for evaluation in result.evaluations:
 544                evaluation_names.add(evaluation.name)
 545
 546        if evaluation_names:
 547            output += "\\nEvaluations:"
 548            for eval_name in evaluation_names:
 549                output += f"\\n  • {eval_name}"
 550            output += "\\n"
 551
 552        # Calculate and display average scores
 553        if evaluation_names:
 554            output += "\\nAverage Scores:"
 555            for eval_name in evaluation_names:
 556                scores = []
 557                for result in self.item_results:
 558                    for evaluation in result.evaluations:
 559                        if evaluation.name == eval_name and isinstance(
 560                            evaluation.value, (int, float)
 561                        ):
 562                            scores.append(evaluation.value)
 563
 564                if scores:
 565                    avg = sum(scores) / len(scores)
 566                    output += f"\\n  • {eval_name}: {avg:.3f}"
 567            output += "\\n"
 568
 569        # Display run-level evaluations
 570        if self.run_evaluations:
 571            output += "\\nRun Evaluations:"
 572            for run_eval in self.run_evaluations:
 573                score = run_eval.value
 574                if isinstance(score, (int, float)):
 575                    score = f"{score:.3f}"
 576                output += f"\\n  • {run_eval.name}: {score}"
 577                if run_eval.comment:
 578                    output += f"\\n    💭 {run_eval.comment}"
 579            output += "\\n"
 580
 581        # Add dataset run URL if available
 582        if self.dataset_run_url:
 583            output += f"\\n🔗 Dataset Run:\\n   {self.dataset_run_url}"
 584
 585        return output
 586
 587
 588class TaskFunction(Protocol):
 589    """Protocol defining the interface for experiment task functions.
 590
 591    Task functions are the core processing functions that operate on each item
 592    in an experiment dataset. They receive an experiment item as input and
 593    produce some output that will be evaluated.
 594
 595    Task functions must:
 596    - Accept 'item' as a keyword argument
 597    - Return any type of output (will be passed to evaluators)
 598    - Can be either synchronous or asynchronous
 599    - Should handle their own errors gracefully (exceptions will be logged)
 600    """
 601
 602    def __call__(
 603        self,
 604        *,
 605        item: ExperimentItem,
 606        **kwargs: Dict[str, Any],
 607    ) -> Union[Any, Awaitable[Any]]:
 608        """Execute the task on an experiment item.
 609
 610        This method defines the core processing logic for each item in your experiment.
 611        The implementation should focus on the specific task you want to evaluate,
 612        such as text generation, classification, summarization, etc.
 613
 614        Args:
 615            item: The experiment item to process. Can be either:
 616                - Dict with keys like 'input', 'expected_output', 'metadata'
 617                - Langfuse DatasetItem object with .input, .expected_output attributes
 618            **kwargs: Additional keyword arguments that may be passed by the framework
 619
 620        Returns:
 621            Any: The output of processing the item. This output will be:
 622            - Stored in the experiment results
 623            - Passed to all item-level evaluators for assessment
 624            - Traced automatically in Langfuse for observability
 625
 626            Can return either a direct value or an awaitable (async) result.
 627
 628        Examples:
 629            Simple synchronous task:
 630            ```python
 631            def my_task(*, item, **kwargs):
 632                prompt = f"Summarize: {item['input']}"
 633                return my_llm_client.generate(prompt)
 634            ```
 635
 636            Async task with error handling:
 637            ```python
 638            async def my_async_task(*, item, **kwargs):
 639                try:
 640                    response = await openai_client.chat.completions.create(
 641                        model="gpt-4",
 642                        messages=[{"role": "user", "content": item["input"]}]
 643                    )
 644                    return response.choices[0].message.content
 645                except Exception as e:
 646                    # Log error and return fallback
 647                    print(f"Task failed for item {item}: {e}")
 648                    return "Error: Could not process item"
 649            ```
 650
 651            Task using dataset item attributes:
 652            ```python
 653            def classification_task(*, item, **kwargs):
 654                # Works with both dict items and DatasetItem objects
 655                text = item["input"] if isinstance(item, dict) else item.input
 656                return classify_text(text)
 657            ```
 658        """
 659        ...
 660
 661
 662class EvaluatorFunction(Protocol):
 663    """Protocol defining the interface for item-level evaluator functions.
 664
 665    Item-level evaluators assess the quality, correctness, or other properties
 666    of individual task outputs. They receive the input, output, expected output,
 667    and metadata for each item and return evaluation metrics.
 668
 669    Evaluators should:
 670    - Accept input, output, expected_output, and metadata as keyword arguments
 671    - Return Evaluation dict(s) with 'name', 'value', 'comment', 'metadata' fields
 672    - Be deterministic when possible for reproducible results
 673    - Handle edge cases gracefully (missing expected output, malformed data, etc.)
 674    - Can be either synchronous or asynchronous
 675    """
 676
 677    def __call__(
 678        self,
 679        *,
 680        input: Any,
 681        output: Any,
 682        expected_output: Any,
 683        metadata: Optional[Dict[str, Any]],
 684        **kwargs: Dict[str, Any],
 685    ) -> Union[
 686        Evaluation, List[Evaluation], Awaitable[Union[Evaluation, List[Evaluation]]]
 687    ]:
 688        r"""Evaluate a task output for quality, correctness, or other metrics.
 689
 690        This method should implement specific evaluation logic such as accuracy checking,
 691        similarity measurement, toxicity detection, fluency assessment, etc.
 692
 693        Args:
 694            input: The original input that was passed to the task function.
 695                This is typically the item['input'] or item.input value.
 696            output: The output produced by the task function for this input.
 697                This is the direct return value from your task function.
 698            expected_output: The expected/ground truth output for comparison.
 699                May be None if not available in the dataset. Evaluators should
 700                handle this case appropriately.
 701            metadata: Optional metadata from the experiment item that might
 702                contain additional context for evaluation (categories, difficulty, etc.)
 703            **kwargs: Additional keyword arguments that may be passed by the framework
 704
 705        Returns:
 706            Evaluation results in one of these formats:
 707            - Single Evaluation dict: {"name": "accuracy", "value": 0.85, "comment": "..."}
 708            - List of Evaluation dicts: [{"name": "precision", ...}, {"name": "recall", ...}]
 709            - Awaitable returning either of the above (for async evaluators)
 710
 711            Each Evaluation dict should contain:
 712            - name (str): Unique identifier for this evaluation metric
 713            - value (int|float|str|bool): The evaluation score or result
 714            - comment (str, optional): Human-readable explanation of the result
 715            - metadata (dict, optional): Additional structured data about the evaluation
 716
 717        Examples:
 718            Simple accuracy evaluator:
 719            ```python
 720            def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
 721                if expected_output is None:
 722                    return {"name": "accuracy", "value": None, "comment": "No expected output"}
 723
 724                is_correct = output.strip().lower() == expected_output.strip().lower()
 725                return {
 726                    "name": "accuracy",
 727                    "value": 1.0 if is_correct else 0.0,
 728                    "comment": "Exact match" if is_correct else "No match"
 729                }
 730            ```
 731
 732            Multi-metric evaluator:
 733            ```python
 734            def comprehensive_evaluator(*, input, output, expected_output=None, **kwargs):
 735                results = []
 736
 737                # Length check
 738                results.append({
 739                    "name": "output_length",
 740                    "value": len(output),
 741                    "comment": f"Output contains {len(output)} characters"
 742                })
 743
 744                # Sentiment analysis
 745                sentiment_score = analyze_sentiment(output)
 746                results.append({
 747                    "name": "sentiment",
 748                    "value": sentiment_score,
 749                    "comment": f"Sentiment score: {sentiment_score:.2f}"
 750                })
 751
 752                return results
 753            ```
 754
 755            Async evaluator using external API:
 756            ```python
 757            async def llm_judge_evaluator(*, input, output, expected_output=None, **kwargs):
 758                prompt = f"Rate the quality of this response on a scale of 1-10:\n"
 759                prompt += f"Question: {input}\nResponse: {output}"
 760
 761                response = await openai_client.chat.completions.create(
 762                    model="gpt-4",
 763                    messages=[{"role": "user", "content": prompt}]
 764                )
 765
 766                try:
 767                    score = float(response.choices[0].message.content.strip())
 768                    return {
 769                        "name": "llm_judge_quality",
 770                        "value": score,
 771                        "comment": f"LLM judge rated this {score}/10"
 772                    }
 773                except ValueError:
 774                    return {
 775                        "name": "llm_judge_quality",
 776                        "value": None,
 777                        "comment": "Could not parse LLM judge score"
 778                    }
 779            ```
 780
 781            Context-aware evaluator:
 782            ```python
 783            def context_evaluator(*, input, output, metadata=None, **kwargs):
 784                # Use metadata for context-specific evaluation
 785                difficulty = metadata.get("difficulty", "medium") if metadata else "medium"
 786
 787                # Adjust expectations based on difficulty
 788                min_length = {"easy": 50, "medium": 100, "hard": 150}[difficulty]
 789
 790                meets_requirement = len(output) >= min_length
 791                return {
 792                    "name": f"meets_{difficulty}_requirement",
 793                    "value": meets_requirement,
 794                    "comment": f"Output {'meets' if meets_requirement else 'fails'} {difficulty} length requirement"
 795                }
 796            ```
 797        """
 798        ...
 799
 800
 801class RunEvaluatorFunction(Protocol):
 802    """Protocol defining the interface for run-level evaluator functions.
 803
 804    Run-level evaluators assess aggregate properties of the entire experiment run,
 805    computing metrics that span across all items rather than individual outputs.
 806    They receive the complete results from all processed items and can compute
 807    statistics like averages, distributions, correlations, or other aggregate metrics.
 808
 809    Run evaluators should:
 810    - Accept item_results as a keyword argument containing all item results
 811    - Return Evaluation dict(s) with aggregate metrics
 812    - Handle cases where some items may have failed processing
 813    - Compute meaningful statistics across the dataset
 814    - Can be either synchronous or asynchronous
 815    """
 816
 817    def __call__(
 818        self,
 819        *,
 820        item_results: List[ExperimentItemResult],
 821        **kwargs: Dict[str, Any],
 822    ) -> Union[
 823        Evaluation, List[Evaluation], Awaitable[Union[Evaluation, List[Evaluation]]]
 824    ]:
 825        r"""Evaluate the entire experiment run with aggregate metrics.
 826
 827        This method should implement aggregate evaluation logic such as computing
 828        averages, calculating distributions, finding correlations, detecting patterns
 829        across items, or performing statistical analysis on the experiment results.
 830
 831        Args:
 832            item_results: List of results from all successfully processed experiment items.
 833                Each item result contains:
 834                - item: The original experiment item
 835                - output: The task function's output for this item
 836                - evaluations: List of item-level evaluation results
 837                - trace_id: Langfuse trace ID for this execution
 838                - dataset_run_id: Dataset run ID (if using Langfuse datasets)
 839
 840                Note: This list only includes items that were successfully processed.
 841                Failed items are excluded but logged separately.
 842            **kwargs: Additional keyword arguments that may be passed by the framework
 843
 844        Returns:
 845            Evaluation results in one of these formats:
 846            - Single Evaluation dict: {"name": "avg_accuracy", "value": 0.78, "comment": "..."}
 847            - List of Evaluation dicts: [{"name": "mean", ...}, {"name": "std_dev", ...}]
 848            - Awaitable returning either of the above (for async evaluators)
 849
 850            Each Evaluation dict should contain:
 851            - name (str): Unique identifier for this run-level metric
 852            - value (int|float|str|bool): The aggregate evaluation result
 853            - comment (str, optional): Human-readable explanation of the metric
 854            - metadata (dict, optional): Additional structured data about the evaluation
 855
 856        Examples:
 857            Average accuracy calculator:
 858            ```python
 859            def average_accuracy(*, item_results, **kwargs):
 860                if not item_results:
 861                    return {"name": "avg_accuracy", "value": 0.0, "comment": "No results"}
 862
 863                accuracy_values = []
 864                for result in item_results:
 865                    for evaluation in result.evaluations:
 866                        if evaluation.name == "accuracy":
 867                            accuracy_values.append(evaluation.value)
 868
 869                if not accuracy_values:
 870                    return {"name": "avg_accuracy", "value": None, "comment": "No accuracy evaluations found"}
 871
 872                avg = sum(accuracy_values) / len(accuracy_values)
 873                return {
 874                    "name": "avg_accuracy",
 875                    "value": avg,
 876                    "comment": f"Average accuracy across {len(accuracy_values)} items: {avg:.2%}"
 877                }
 878            ```
 879
 880            Multiple aggregate metrics:
 881            ```python
 882            def statistical_summary(*, item_results, **kwargs):
 883                if not item_results:
 884                    return []
 885
 886                results = []
 887
 888                # Calculate output length statistics
 889                lengths = [len(str(result.output)) for result in item_results]
 890                results.extend([
 891                    {"name": "avg_output_length", "value": sum(lengths) / len(lengths)},
 892                    {"name": "min_output_length", "value": min(lengths)},
 893                    {"name": "max_output_length", "value": max(lengths)}
 894                ])
 895
 896                # Success rate
 897                total_items = len(item_results)  # Only successful items are included
 898                results.append({
 899                    "name": "processing_success_rate",
 900                    "value": 1.0,  # All items in item_results succeeded
 901                    "comment": f"Successfully processed {total_items} items"
 902                })
 903
 904                return results
 905            ```
 906
 907            Async run evaluator with external analysis:
 908            ```python
 909            async def llm_batch_analysis(*, item_results, **kwargs):
 910                # Prepare batch analysis prompt
 911                outputs = [result.output for result in item_results]
 912                prompt = f"Analyze these {len(outputs)} outputs for common themes:\n"
 913                prompt += "\n".join(f"{i+1}. {output}" for i, output in enumerate(outputs))
 914
 915                response = await openai_client.chat.completions.create(
 916                    model="gpt-4",
 917                    messages=[{"role": "user", "content": prompt}]
 918                )
 919
 920                return {
 921                    "name": "thematic_analysis",
 922                    "value": response.choices[0].message.content,
 923                    "comment": f"LLM analysis of {len(outputs)} outputs"
 924                }
 925            ```
 926
 927            Performance distribution analysis:
 928            ```python
 929            def performance_distribution(*, item_results, **kwargs):
 930                # Extract all evaluation scores
 931                all_scores = []
 932                score_by_metric = {}
 933
 934                for result in item_results:
 935                    for evaluation in result.evaluations:
 936                        metric_name = evaluation.name
 937                        value = evaluation.value
 938
 939                        if isinstance(value, (int, float)):
 940                            all_scores.append(value)
 941                            if metric_name not in score_by_metric:
 942                                score_by_metric[metric_name] = []
 943                            score_by_metric[metric_name].append(value)
 944
 945                results = []
 946
 947                # Overall score distribution
 948                if all_scores:
 949                    import statistics
 950                    results.append({
 951                        "name": "score_std_dev",
 952                        "value": statistics.stdev(all_scores) if len(all_scores) > 1 else 0,
 953                        "comment": f"Standard deviation across all numeric scores"
 954                    })
 955
 956                # Per-metric statistics
 957                for metric, scores in score_by_metric.items():
 958                    if len(scores) > 1:
 959                        results.append({
 960                            "name": f"{metric}_variance",
 961                            "value": statistics.variance(scores),
 962                            "comment": f"Variance in {metric} across {len(scores)} items"
 963                        })
 964
 965                return results
 966            ```
 967        """
 968        ...
 969
 970
 971def _format_value(value: Any) -> str:
 972    """Format a value for display."""
 973    if isinstance(value, str):
 974        return value[:50] + "..." if len(value) > 50 else value
 975    return str(value)
 976
 977
 978async def _run_evaluator(
 979    evaluator: Union[EvaluatorFunction, RunEvaluatorFunction], **kwargs: Any
 980) -> List[Evaluation]:
 981    """Run an evaluator function and normalize the result."""
 982    try:
 983        result = evaluator(**kwargs)
 984
 985        # Handle async evaluators
 986        if asyncio.iscoroutine(result):
 987            result = await result
 988
 989        # Normalize to list
 990        if isinstance(result, (dict, Evaluation)):
 991            return [result]  # type: ignore
 992
 993        elif isinstance(result, list):
 994            return result
 995
 996        else:
 997            return []
 998
 999    except Exception as e:
1000        evaluator_name = getattr(evaluator, "__name__", "unknown_evaluator")
1001        logging.getLogger("langfuse").error(f"Evaluator {evaluator_name} failed: {e}")
1002        return []
1003
1004
1005async def _run_task(task: TaskFunction, item: ExperimentItem) -> Any:
1006    """Run a task function and handle sync/async."""
1007    result = task(item=item)
1008
1009    # Handle async tasks
1010    if asyncio.iscoroutine(result):
1011        result = await result
1012
1013    return result
1014
1015
1016def create_evaluator_from_autoevals(
1017    autoevals_evaluator: Any, **kwargs: Optional[Dict[str, Any]]
1018) -> EvaluatorFunction:
1019    """Create a Langfuse evaluator from an autoevals evaluator.
1020
1021    Args:
1022        autoevals_evaluator: An autoevals evaluator instance
1023        **kwargs: Additional arguments passed to the evaluator
1024
1025    Returns:
1026        A Langfuse-compatible evaluator function
1027    """
1028
1029    def langfuse_evaluator(
1030        *,
1031        input: Any,
1032        output: Any,
1033        expected_output: Any,
1034        metadata: Optional[Dict[str, Any]],
1035        **langfuse_kwargs: Dict[str, Any],
1036    ) -> Evaluation:
1037        evaluation = autoevals_evaluator(
1038            input=input, output=output, expected=expected_output, **kwargs
1039        )
1040
1041        return Evaluation(
1042            name=evaluation.name,
1043            value=evaluation.score,
1044            comment=(evaluation.metadata or {}).get("comment"),
1045            metadata=evaluation.metadata,
1046        )
1047
1048    return langfuse_evaluator