feat(llmobs): add summary evaluator functionality for datasets & experiments (#14629)

- changes the ExperimentResult class to have summary eval results and
the per experiment row results
- adds the ability to define and run summary evaluators

example:
given this script
```
#!/usr/bin/env python
# coding: utf-8

import os
import math
import random

from dotenv import load_dotenv

# Load environment variables from the .env file.
load_dotenv(override=True)

from ddtrace.llmobs import LLMObs

LLMObs.enable(
    api_key=os.getenv("DD_API_KEY"),
    app_key=os.getenv("DD_APPLICATION_KEY"),
    project_name="Onboarding",
    ml_app="Onboarding-ML-App",
)


dataset = LLMObs.create_dataset_from_csv(
    csv_path="./data/taskmaster.csv",
    dataset_name="taskmaster-mini-314",
    input_data_columns=["prompt", "topics"],
    expected_output_columns=["labels"],
)

dataset.as_dataframe()


def return_hello(input_data, config):
    return "hello"


def return_2(input_data, output_data, expected_output):
    return 2


def sum_of_rows_times_2_and_hellos(
    inputs, outputs, expected_outputs, evaluators_results
):
    return sum(evaluators_results["return_2"]) + len(outputs)


experiment = LLMObs.experiment(
    name="taskmaster-experiment",
    dataset=dataset,
    task=return_hello,
    evaluators=[return_2],
    summary_evaluators=[sum_of_rows_times_2_and_hellos],
)

results = experiment.run(jobs=50, raise_errors=True)


print(experiment.url)
```

results in
https://dddev.datadoghq.com/llm/experiments/889df25f-95b8-4149-9ba0-756c1afcc5f6
<img width="327" height="298" alt="image"
src="https://github.com/user-attachments/assets/759393c2-109a-43f6-82fd-21bc1ffbe0fc"
/>
the summary evaluator is essentially 3*number of records -> correct
result


## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: gary-huang

ddtrace/llmobs/_experiment.py

@@ -82,7 +82,7 @@ class EvaluationResult(TypedDict):
     evaluations: Dict[str, Dict[str, JSONType]]
 
 
-class ExperimentResult(TypedDict):
+class ExperimentRowResult(TypedDict):
     idx: int
     record_id: Optional[str]
     span_id: str
@@ -96,6 +96,11 @@ class ExperimentResult(TypedDict):
     error: Dict[str, Optional[str]]
 
 
+class ExperimentResult(TypedDict):
+    summary_evaluations: Dict[str, Dict[str, JSONType]]
+    rows: List[ExperimentRowResult]
+
+
 class Dataset:
     name: str
     description: str
@@ -304,11 +309,19 @@ def __init__(
         tags: Optional[Dict[str, str]] = None,
         config: Optional[ExperimentConfigType] = None,
         _llmobs_instance: Optional["LLMObs"] = None,
+        summary_evaluators: Optional[
+            List[
+                Callable[
+                    [List[DatasetRecordInputType], List[JSONType], List[JSONType], Dict[str, List[JSONType]]], JSONType
+                ]
+            ]
+        ] = None,
     ) -> None:
         self.name = name
         self._task = task
         self._dataset = dataset
         self._evaluators = evaluators
+        self._summary_evaluators = summary_evaluators or []
         self._description = description
         self._tags: Dict[str, str] = tags or {}
         self._tags["ddtrace.version"] = str(ddtrace.__version__)
@@ -327,21 +340,12 @@ def __init__(
         self._id: Optional[str] = None
         self._run_name: Optional[str] = None
 
-    def run(
-        self, jobs: int = 1, raise_errors: bool = False, sample_size: Optional[int] = None
-    ) -> List[ExperimentResult]:
-        if not self._llmobs_instance:
+    def run(self, jobs: int = 1, raise_errors: bool = False, sample_size: Optional[int] = None) -> ExperimentResult:
+        if not self._llmobs_instance or not self._llmobs_instance.enabled:
             raise ValueError(
                 "LLMObs is not enabled. Ensure LLM Observability is enabled via `LLMObs.enable(...)` "
                 "and create the experiment via `LLMObs.experiment(...)` before running the experiment."
             )
-        if not self._llmobs_instance.enabled:
-            logger.warning(
-                "Skipping experiment as LLMObs is not enabled. "
-                "Ensure LLM Observability is enabled via `LLMObs.enable(...)` "
-                "or set `DD_LLMOBS_ENABLED=1` and use `ddtrace-run` to run your application."
-            )
-            return []
 
         project = self._llmobs_instance._dne_client.project_create_or_get(self._project_name)
         self._project_id = project.get("_id", "")
@@ -360,11 +364,13 @@ def run(
         self._run_name = experiment_run_name
         task_results = self._run_task(jobs, raise_errors, sample_size)
         evaluations = self._run_evaluators(task_results, raise_errors=raise_errors)
-        experiment_results = self._merge_results(task_results, evaluations)
+        summary_evals = self._run_summary_evaluators(task_results, evaluations, raise_errors)
+        experiment_results = self._merge_results(task_results, evaluations, summary_evals)
         experiment_evals = self._generate_metrics_from_exp_results(experiment_results)
         self._llmobs_instance._dne_client.experiment_eval_post(
             self._id, experiment_evals, convert_tags_dict_to_list(self._tags)
         )
+
         return experiment_results
 
     @property
@@ -476,17 +482,64 @@ def _run_evaluators(self, task_results: List[TaskResult], raise_errors: bool = F
             evaluations.append(evaluation)
         return evaluations
 
+    def _run_summary_evaluators(
+        self, task_results: List[TaskResult], eval_results: List[EvaluationResult], raise_errors: bool = False
+    ) -> List[EvaluationResult]:
+        evaluations: List[EvaluationResult] = []
+        inputs: List[DatasetRecordInputType] = []
+        outputs: List[JSONType] = []
+        expected_outputs: List[JSONType] = []
+        evals_dict = {}
+
+        # name of evaluator (not summary evaluator) -> list of eval results ordered by index of the list of task results
+        # this is being computed so that the user can use the evaluation results in its original form
+        eval_results_by_name: dict[str, List[JSONType]] = {}
+        for idx, task_result in enumerate(task_results):
+            outputs.append(task_result["output"])
+            record: DatasetRecord = self._dataset[idx]
+            inputs.append(record["input_data"])
+            expected_outputs.append(record["expected_output"])
+
+            eval_result_at_idx_by_name = eval_results[idx]["evaluations"]
+            for name, eval_value in eval_result_at_idx_by_name.items():
+                if name not in eval_results_by_name:
+                    eval_results_by_name[name] = []
+
+                eval_results_by_name[name].append(eval_value.get("value"))
+
+        for idx, summary_evaluator in enumerate(self._summary_evaluators):
+            eval_result: JSONType = None
+            eval_err: JSONType = None
+
+            try:
+                eval_result = summary_evaluator(inputs, outputs, expected_outputs, eval_results_by_name)
+            except Exception as e:
+                exc_type, exc_value, exc_tb = sys.exc_info()
+                exc_type_name = type(e).__name__ if exc_type is not None else "Unknown Exception"
+                exc_stack = "".join(traceback.format_exception(exc_type, exc_value, exc_tb))
+                eval_err = {"message": str(exc_value), "type": exc_type_name, "stack": exc_stack}
+                if raise_errors:
+                    raise RuntimeError(f"Summary evaluator {summary_evaluator.__name__} failed") from e
+            evals_dict[summary_evaluator.__name__] = {"value": eval_result, "error": eval_err}
+            evaluation: EvaluationResult = {"idx": idx, "evaluations": evals_dict}
+            evaluations.append(evaluation)
+
+        return evaluations
+
     def _merge_results(
-        self, task_results: List[TaskResult], evaluations: List[EvaluationResult]
-    ) -> List[ExperimentResult]:
+        self,
+        task_results: List[TaskResult],
+        evaluations: List[EvaluationResult],
+        summary_evaluations: Optional[List[EvaluationResult]],
+    ) -> ExperimentResult:
         experiment_results = []
         for idx, task_result in enumerate(task_results):
             output_data = task_result["output"]
             metadata: Dict[str, JSONType] = {"tags": cast(List[JSONType], convert_tags_dict_to_list(self._tags))}
             metadata.update(task_result.get("metadata") or {})
             record: DatasetRecord = self._dataset[idx]
             evals = evaluations[idx]["evaluations"]
-            exp_result: ExperimentResult = {
+            exp_result: ExperimentRowResult = {
                 "idx": idx,
                 "span_id": task_result.get("span_id", ""),
                 "trace_id": task_result.get("trace_id", ""),
@@ -500,10 +553,28 @@ def _merge_results(
                 "error": task_result["error"],
             }
             experiment_results.append(exp_result)
-        return experiment_results
+
+        summary_evals: Dict[str, Dict[str, JSONType]] = {}
+        if summary_evaluations:
+            for summary_evaluation in summary_evaluations:
+                for name, eval_data in summary_evaluation["evaluations"].items():
+                    summary_evals[name] = eval_data
+
+        result: ExperimentResult = {
+            "summary_evaluations": summary_evals,
+            "rows": experiment_results,
+        }
+        return result
 
     def _generate_metric_from_evaluation(
-        self, eval_name: str, eval_value: JSONType, err: JSONType, span_id: str, trace_id: str, timestamp_ns: int
+        self,
+        eval_name: str,
+        eval_value: JSONType,
+        err: JSONType,
+        span_id: str,
+        trace_id: str,
+        timestamp_ns: int,
+        source: str = "custom",
     ) -> "LLMObsExperimentEvalMetricEvent":
         metric_type = None
         if eval_value is None:
@@ -516,6 +587,7 @@ def _generate_metric_from_evaluation(
             metric_type = "categorical"
             eval_value = str(eval_value).lower()
         return {
+            "metric_source": source,
             "span_id": span_id,
             "trace_id": trace_id,
             "timestamp_ms": int(timestamp_ns / 1e6),
@@ -528,14 +600,18 @@ def _generate_metric_from_evaluation(
         }
 
     def _generate_metrics_from_exp_results(
-        self, experiment_results: List[ExperimentResult]
+        self, experiment_result: ExperimentResult
     ) -> List["LLMObsExperimentEvalMetricEvent"]:
         eval_metrics = []
-        for exp_result in experiment_results:
+        latest_timestamp: int = 0
+        for exp_result in experiment_result["rows"]:
             evaluations = exp_result.get("evaluations") or {}
             span_id = exp_result.get("span_id", "")
             trace_id = exp_result.get("trace_id", "")
             timestamp_ns = cast(int, exp_result.get("timestamp", 0))
+            if timestamp_ns > latest_timestamp:
+                latest_timestamp = timestamp_ns
+
             for eval_name, eval_data in evaluations.items():
                 if not eval_data:
                     continue
@@ -544,6 +620,20 @@ def _generate_metrics_from_exp_results(
                     eval_name, eval_value, eval_data.get("error"), span_id, trace_id, timestamp_ns
                 )
                 eval_metrics.append(eval_metric)
+
+        for name, summary_eval_data in experiment_result.get("summary_evaluations", {}).items():
+            if not summary_eval_data:
+                continue
+            eval_metric = self._generate_metric_from_evaluation(
+                name,
+                summary_eval_data.get("value"),
+                summary_eval_data.get("error"),
+                "",
+                "",
+                latest_timestamp,
+                source="summary",
+            )
+            eval_metrics.append(eval_metric)
         return eval_metrics
 
 

ddtrace/llmobs/_llmobs.py

@@ -752,6 +752,13 @@ def experiment(
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         config: Optional[ExperimentConfigType] = None,
+        summary_evaluators: Optional[
+            List[
+                Callable[
+                    [List[DatasetRecordInputType], List[JSONType], List[JSONType], Dict[str, List[JSONType]]], JSONType
+                ]
+            ]
+        ] = None,
     ) -> Experiment:
         """Initializes an Experiment to run a task on a Dataset and evaluators.
 
@@ -777,9 +784,21 @@ def experiment(
         for evaluator in evaluators:
             sig = inspect.signature(evaluator)
             params = sig.parameters
-            required_params = ("input_data", "output_data", "expected_output")
-            if not all(param in params for param in required_params):
-                raise TypeError("Evaluator function must have parameters {}.".format(required_params))
+            evaluator_required_params = ("input_data", "output_data", "expected_output")
+            if not all(param in params for param in evaluator_required_params):
+                raise TypeError("Evaluator function must have parameters {}.".format(evaluator_required_params))
+
+        if summary_evaluators and not all(callable(summary_evaluator) for summary_evaluator in summary_evaluators):
+            raise TypeError("Summary evaluators must be a list of callable functions.")
+        if summary_evaluators:
+            for summary_evaluator in summary_evaluators:
+                sig = inspect.signature(summary_evaluator)
+                params = sig.parameters
+                summary_evaluator_required_params = ("inputs", "outputs", "expected_outputs", "evaluators_results")
+                if not all(param in params for param in summary_evaluator_required_params):
+                    raise TypeError(
+                        "Summary evaluator function must have parameters {}.".format(summary_evaluator_required_params)
+                    )
         return Experiment(
             name,
             task,
@@ -790,6 +809,7 @@ def experiment(
             description=description,
             config=config,
             _llmobs_instance=cls._instance,
+            summary_evaluators=summary_evaluators,
         )
 
     @classmethod

ddtrace/llmobs/_writer.py

@@ -87,6 +87,7 @@ class LLMObsEvaluationMetricEvent(TypedDict, total=False):
 
 
 class LLMObsExperimentEvalMetricEvent(TypedDict, total=False):
+    metric_source: str
     span_id: str
     trace_id: str
     timestamp_ms: int

tests/llmobs/llmobs_cassettes/datadog/datadog_api_unstable_llm-obs_v1_experiments_9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8_events_post_1edd9d51.yaml

@@ -0,0 +1,49 @@
+interactions:
+- request:
+    body: '{"data": {"type": "experiments", "attributes": {"scope": "experiments",
+      "metrics": [{"metric_source": "custom", "span_id": "123", "trace_id": "456",
+      "timestamp_ms": 1234, "metric_type": "score", "label": "dummy_evaluator", "score_value":
+      0, "error": null, "tags": ["ddtrace.version:1.2.3", "experiment_id:9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"],
+      "experiment_id": "9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"}], "tags": ["ddtrace.version:1.2.3",
+      "experiment_id:9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"]}}}'
+    headers:
+      Accept:
+      - '*/*'
+      ? !!python/object/apply:multidict._multidict.istr
+      - Accept-Encoding
+      : - identity
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '494'
+      ? !!python/object/apply:multidict._multidict.istr
+      - Content-Type
+      : - application/json
+      User-Agent:
+      - python-requests/2.32.3
+    method: POST
+    uri: https://api.datadoghq.com/api/unstable/llm-obs/v1/experiments/9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8/events
+  response:
+    body:
+      string: ''
+    headers:
+      content-length:
+      - '0'
+      content-security-policy:
+      - frame-ancestors 'self'; report-uri https://logs.browser-intake-datadoghq.com/api/v2/logs?dd-api-key=pube4f163c23bbf91c16b8f57f56af9fc58&dd-evp-origin=content-security-policy&ddsource=csp-report&ddtags=site%3Adatadoghq.com
+      content-type:
+      - application/vnd.api+json
+      date:
+      - Thu, 18 Sep 2025 15:03:41 GMT
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+      x-frame-options:
+      - SAMEORIGIN
+    status:
+      code: 202
+      message: Accepted
+version: 1

tests/llmobs/llmobs_cassettes/datadog/datadog_api_unstable_llm-obs_v1_experiments_9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8_events_post_977047ce.yaml

@@ -0,0 +1,53 @@
+interactions:
+- request:
+    body: '{"data": {"type": "experiments", "attributes": {"scope": "experiments",
+      "metrics": [{"metric_source": "custom", "span_id": "123", "trace_id": "456",
+      "timestamp_ms": 1234, "metric_type": "score", "label": "dummy_evaluator", "score_value":
+      0, "error": null, "tags": ["ddtrace.version:1.2.3", "experiment_id:9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"],
+      "experiment_id": "9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"}, {"metric_source":
+      "summary", "span_id": "", "trace_id": "", "timestamp_ms": 1234, "metric_type":
+      "score", "label": "dummy_summary_evaluator", "score_value": 4, "error": null,
+      "tags": ["ddtrace.version:1.2.3", "experiment_id:9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"],
+      "experiment_id": "9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"}], "tags": ["ddtrace.version:1.2.3",
+      "experiment_id:9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8"]}}}'
+    headers:
+      Accept:
+      - '*/*'
+      ? !!python/object/apply:multidict._multidict.istr
+      - Accept-Encoding
+      : - identity
+      Connection:
+      - keep-alive
+      Content-Length:
+      - '816'
+      ? !!python/object/apply:multidict._multidict.istr
+      - Content-Type
+      : - application/json
+      User-Agent:
+      - python-requests/2.32.3
+    method: POST
+    uri: https://api.datadoghq.com/api/unstable/llm-obs/v1/experiments/9e046fc7-cf3f-4f01-b5ed-e5e7746fefa8/events
+  response:
+    body:
+      string: ''
+    headers:
+      content-length:
+      - '0'
+      content-security-policy:
+      - frame-ancestors 'self'; report-uri https://logs.browser-intake-datadoghq.com/api/v2/logs?dd-api-key=pube4f163c23bbf91c16b8f57f56af9fc58&dd-evp-origin=content-security-policy&ddsource=csp-report&ddtags=site%3Adatadoghq.com
+      content-type:
+      - application/vnd.api+json
+      date:
+      - Thu, 18 Sep 2025 21:00:41 GMT
+      strict-transport-security:
+      - max-age=31536000; includeSubDomains; preload
+      vary:
+      - Accept-Encoding
+      x-content-type-options:
+      - nosniff
+      x-frame-options:
+      - SAMEORIGIN
+    status:
+      code: 202
+      message: Accepted
+version: 1