feat(llma): add datasets to offline evaluations

.sampo/changesets/ai-evaluation-datasets.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Manage AI observability datasets and items through `OfflineEvaluations.datasets` and `AsyncOfflineEvaluations.datasets`, including history, tags, and archive/restore operations. Pin a dataset snapshot for lazy, consistent reads and use `EvaluationItem.from_dataset_item()` to preserve case provenance across offline experiments and scorers. Item creates with a caller-supplied `client_item_id` support safe retries; updates use explicit version checks.

examples/example-offline-evaluations/README.md

@@ -1,6 +1,6 @@
 # Offline evaluations
 
-Upload evaluations you have already run, manage reusable scorers, and explicitly finish an experiment. The SDK returns server acknowledgments for every operation. It does not execute your application or judges, create traces, or upload in the background.
+Upload evaluations you have already run, manage reusable scorers and datasets, and explicitly finish an experiment. The SDK returns server acknowledgments for every operation. It does not execute your application or judges, create traces, or upload in the background.
 
 ## Credentials and setup
 
@@ -11,11 +11,15 @@ Offline evaluations must be enabled for your PostHog project. Use the project's
 | Create experiments, upload results, complete/fail | Personal API key or project secret key with `offline_evaluation_ingestion:write` |
 | Fetch scorers and versions | Personal API key with `llm_analytics:read` |
 | Create/update scorers and create versions | Personal API key with `llm_analytics:write` |
+| Fetch datasets, items, and history | Personal API key with `dataset:read` |
+| Create/update/archive/restore datasets and items | Personal API key with `dataset:write` |
 | Read experiment tags | Personal API key with `evaluation:read` |
 | Replace experiment tags | Personal API key with `evaluation:write` |
 
 A personal key can cover both scorer management and uploads. Project secret keys can upload using a known scorer-version UUID without scorer read access. All operations use the explicit project/environment; scorer names are not unique.
 
+Dataset operations also require the datasets feature to be enabled. Dataset reads, scorer reads, and experiment uploads use their own scopes; an upload-only credential cannot fetch a dataset. Keep personal keys in server-side scripts or CI secrets.
+
 From this repository's root:
 
 ```bash
@@ -85,6 +89,163 @@ EvaluationItem(input=None)       # Payload contains a null input.
 
 The client context manager releases connections. Call `complete()` explicitly after uploads succeed, or `fail()` when you deliberately want to close the experiment as failed. Neither exceptions nor async cancellation automatically close an experiment. Expected counts include every outcome status; the server checks them on completion.
 
+## Datasets and reproducible evaluations
+
+Use `evaluations.datasets` to manage PostHog-hosted datasets with the same client configuration as scorers and experiments. Dataset names are unique within a project; keep the returned UUID for subsequent operations.
+
+Provision a dataset once with a personal key that has `dataset:write`:
+
+```python
+dataset = evaluations.datasets.create(
+    name="Capital city regression cases",
+    description="Cases for the offline evaluation example",
+    tags=["regression"],
+)
+case = evaluations.datasets.create_item(
+    dataset.id,
+    client_item_id="capital-france",
+    input="Capital of France?",
+    expected_output="Paris",
+    metadata={"category": "geography"},
+)
+print(dataset.id)
+```
+
+An identical item create with the same `client_item_id` returns the existing active item. Reusing that key with different current content, or for an archived item, raises `EvaluationAPIError` with `code="client_item_id_conflict"`. Choose a stable key per case and retain it across process restarts. Creating the dataset itself is not idempotent; keep its ID instead of provisioning it on each evaluation run.
+
+Pin both the dataset and scorer before running your application:
+
+```python
+from posthog.ai.evaluations import EvaluationItem, OfflineEvaluations
+
+def capital_answer(question):
+    return {"Capital of France?": "Paris", "Capital of Italy?": "Rome"}.get(
+        question, "I don't know"
+    )
+
+with OfflineEvaluations(project_id=123, secret_key=personal_api_key) as evaluations:
+    snapshot = evaluations.datasets.snapshot(dataset_id)  # Resolve latest once.
+    scorer = evaluations.scorers.get(scorer_id)  # A boolean exact-match scorer.
+    version_id = scorer.current_version_id
+    experiment = evaluations.create_experiment(
+        name="Capital city regression",
+        dataset_revision_id=snapshot.revision_id,
+        application_version="your-commit-sha",
+        run_source="ci",
+    )
+
+    for case in snapshot.iter_items():
+        output = capital_answer(case.input)
+        item = EvaluationItem.from_dataset_item(case, output=output)
+        if case.expected_output is None:
+            experiment.upload_result(
+                item=item, scorer_version_id=version_id, status="not_applicable",
+            )
+        else:
+            experiment.upload_result(
+                item=item, scorer_version_id=version_id,
+                value=output == case.expected_output,
+            )
+
+    experiment.complete()
+```
+
+This example uses string inputs. Replace `capital_answer` and the equality check with your own application and judge. The SDK does not run scorer definitions. For multiple scorers, reuse the same `EvaluationItem` and upload one result per scorer-version UUID. For independent trials, call `from_dataset_item` again and pass a distinct `trial`, such as `trial="2"`; each call generates a new execution UUID. The helper defaults `case_key` to `client_item_id`, falling back to the stable dataset item UUID, and accepts an explicit override.
+
+`from_dataset_item` copies input, expected output, metadata, and immutable item provenance. It never uses `source_output` as a new execution's output or expected answer, and never uses the original source trace as the new application's trace. Pass `application_trace_id` explicitly for the new execution. Omit `output` if execution failed before producing one, then upload an error outcome. Omitted output differs from `output=None`.
+
+### Snapshot and pagination behavior
+
+- `snapshot(dataset_id, revision=12)` selects a historical **number**. The returned `revision_id` is the **UUID** to pass to `create_experiment`. The backend derives the remaining hosted-dataset identifiers from that UUID.
+- Every item request uses the pinned revision, even if another process edits or archives items during iteration. `DatasetItem.dataset_revision_id` identifies the revision that introduced that item version; it can differ from the snapshot's revision ID. Do not use it as the experiment's revision.
+- `snapshot()` fetches metadata only. `snapshot.iter_items(page_size=25)` lazily fetches active items, one page at a time. Keep the client open. Repeating iteration starts again at the first page without caching; read errors propagate rather than silently returning a partial dataset.
+- A new dataset has no revision until its first item is added, so `snapshot()` raises `ValueError`. An existing revision with zero active items yields an empty iterator. Archived datasets remain readable.
+- `snapshot.dataset` contains current descriptive metadata observed when the snapshot was resolved. Names, descriptions, tags, and dataset metadata are not historical item content.
+
+For explicit paging, each `list*` method returns typed `results`, `count`, and `next_offset`. Continue with the same filters until `next_offset is None`. Dataset/revision pages allow at most 100 entries, item pages 25, and item-history pages 100. `list_items` without a revision reads live state on each request; use a snapshot for an evaluation run. A historical snapshot may make multiple metadata requests to find its revision UUID.
+
+| Method | Purpose |
+| --- | --- |
+| `list(search=..., tags=..., tag_match="any", archived=False, order_by="-created_at")` | Find datasets; `archived=True` selects archived datasets only |
+| `get(dataset_id)` | Read current dataset metadata and revision references |
+| `create(name=..., description=..., metadata=..., tags=...)` | Create an empty dataset |
+| `update(dataset_id, name=..., description=..., metadata=..., tags=...)` | Replace supplied metadata fields without creating an item revision |
+| `archive(dataset_id)` / `restore(dataset_id)` | Change dataset archive state; restore leaves individual item states unchanged |
+| `list_revisions(dataset_id)` | Read revision history, newest first |
+| `list_items(dataset_id, revision=..., archived=False)` | Read one page of current or historical items |
+| `get_item(item_id, revision=...)` | Read one stable item at the current or selected dataset revision |
+| `create_item(dataset_id, input=..., ...)` | Add a case and its first version |
+| `update_item(item_id, base_version=..., input=..., expected_output=..., metadata=...)` | Update editable content with optimistic concurrency |
+| `archive_item(item_id, base_version=...)` | Archive an active item with a new version |
+| `restore_item(item_id, base_version=..., source_version=...)` | Restore an archived item, optionally copying a historical item version |
+| `list_item_versions(item_id)` | Read item history, newest first |
+
+Dataset and item metadata are JSON objects. Inputs accept any non-null JSON value; expected/source outputs can be null. Writes require finite numbers, JSON string keys, and at most 32 nested containers. The server enforces dataset, item, version, and payload size limits; failures retain their codes and details in `EvaluationAPIError`. JSON values returned by reads are local copies. Editing them does not write to PostHog.
+
+To save source provenance on creation, supply `source_trace_id` together with a timezone-aware `source_timestamp`; optionally add `source_event_id`. Source fields and `client_item_id` cannot be edited later.
+
+### Updates and recovery
+
+```python
+case = evaluations.datasets.get_item(item_id)
+updated = evaluations.datasets.update_item(
+    case.id,
+    base_version=case.version,
+    expected_output=None,  # Clear the answer; omitting it preserves it.
+    metadata={},          # Clear metadata; omitting it preserves it.
+)
+```
+
+Pass the **observed item version number** as `base_version` for updates, archive, and restore. A stale version raises a 409 `stale_version` conflict with details in `error.response`, including `current_version`. The SDK does not fetch a newer version and overwrite it automatically. Changed content creates a version and dataset revision; identical content is a no-op. Dataset descriptions clear with `""`, metadata with `{}`, and tags with `[]`; those fields do not accept `None`.
+
+Reads retry within the configured transport limits. Item creates retry only when you supply a `client_item_id`. Other mutations never retry automatically. After an unknown write outcome, inspect the current state before issuing another mutation. Dataset errors use the same `EvaluationAPIError` as scorer and experiment operations, including `status`, `code`, `response`, and `persistence`.
+
+Run the supplied example against the provisioned dataset:
+
+```bash
+uv run python examples/example-offline-evaluations/main.py \
+  --dataset-id '<dataset-uuid>' --state dataset-run.json
+
+# Optional: select an earlier numeric revision for a new run.
+uv run python examples/example-offline-evaluations/main.py \
+  --dataset-id '<dataset-uuid>' --dataset-revision 1 --state historical-run.json
+```
+
+The example prepares and saves the pinned revision, execution identities, scorer version, and results **before the first experiment write**. If the state file already exists, replay uses that file without fetching a newer dataset, selecting another scorer version, or running the application again. Dataset selection arguments apply only when preparing a new state file. Choose a new file for a new run. This small example materializes its results; for larger workloads, use caller-sized batches and persist each prepared batch before upload.
+
+### Async datasets
+
+Await management methods and snapshot creation. Use `async for` on `iter_items()`; iteration itself is an async iterator, not an awaitable list.
+
+```python
+from posthog.ai.evaluations import AsyncOfflineEvaluations, EvaluationItem
+
+async with AsyncOfflineEvaluations(
+    project_id=123, secret_key=personal_api_key,
+) as evaluations:
+    snapshot = await evaluations.datasets.snapshot(dataset_id)
+    scorer = await evaluations.scorers.get(scorer_id)
+    experiment = await evaluations.create_experiment(
+        name="Async regression", dataset_revision_id=snapshot.revision_id,
+    )
+    async for case in snapshot.iter_items():
+        output = capital_answer(case.input)
+        item = EvaluationItem.from_dataset_item(case, output=output)
+        if case.expected_output is None:
+            await experiment.upload_result(
+                item=item, scorer_version_id=scorer.current_version_id,
+                status="not_applicable",
+            )
+        else:
+            await experiment.upload_result(
+                item=item, scorer_version_id=scorer.current_version_id,
+                value=output == case.expected_output,
+            )
+    await experiment.complete()
+```
+
+Install `posthog[async]` for async support. Cancellation propagates and does not automatically complete or fail an experiment.
+
 ## Typed scorer configuration and polarity
 
 Configurations are public `TypedDict` types: editors suggest fields and type checkers connect `create(kind=...)` to the correct configuration. They are ordinary dictionaries at runtime; server validation remains authoritative.

examples/example-offline-evaluations/main.py

@@ -14,6 +14,7 @@
     EvaluationAPIError,
     EvaluationItem,
     EvaluationResult,
+    JSONValue,
     OfflineEvaluations,
 )
 
@@ -33,8 +34,39 @@ def resolve_version(evaluations: OfflineEvaluations) -> str:
     return scorer.current_version_id
 
 
+def capital_answer(input: JSONValue) -> str:
+    """A deterministic application to replace with your own model or agent."""
+    if not isinstance(input, str):
+        raise ValueError("The capital-city example expects string dataset inputs.")
+    return {"Capital of France?": "Paris", "Capital of Italy?": "Milan"}.get(
+        input, "I don't know"
+    )
+
+
+def exact_match_result(item: EvaluationItem, version_id: str) -> EvaluationResult:
+    payload = item.to_dict()["payload"]
+    if payload.get("expected_output") is None:
+        return EvaluationResult(
+            item=item,
+            scorer_version_id=version_id,
+            status="not_applicable",
+            reasoning="This case has no expected answer.",
+        )
+    return EvaluationResult(
+        item=item,
+        scorer_version_id=version_id,
+        value=payload["output"] == payload["expected_output"],
+        reasoning="Exact string comparison",
+    )
+
+
 def prepare_state(
-    evaluations: OfflineEvaluations, project_id: int, host: str
+    evaluations: OfflineEvaluations,
+    project_id: int,
+    host: str,
+    *,
+    dataset_id: str | None = None,
+    dataset_revision: int | None = None,
 ) -> dict[str, Any]:
     version_id = resolve_version(evaluations)
     cases = [
@@ -51,24 +83,24 @@ def prepare_state(
             "expected": "Rome",
         },
     ]
-    items = [
-        EvaluationItem(
-            case_key=case["key"],
-            input=case["input"],
-            output=case["output"],
-            expected_output=case["expected"],
-        )
-        for case in cases
-    ]
-    results = [
-        EvaluationResult(
-            item=item,
-            scorer_version_id=version_id,
-            value=case["output"] == case["expected"],
-            reasoning="Exact string comparison",
-        )
-        for item, case in zip(items, cases)
-    ]
+    snapshot = None
+    if dataset_id is not None:
+        snapshot = evaluations.datasets.snapshot(dataset_id, revision=dataset_revision)
+        items = [
+            EvaluationItem.from_dataset_item(case, output=capital_answer(case.input))
+            for case in snapshot.iter_items()
+        ]
+    else:
+        items = [
+            EvaluationItem(
+                case_key=case["key"],
+                input=case["input"],
+                output=case["output"],
+                expected_output=case["expected"],
+            )
+            for case in cases
+        ]
+    results = [exact_match_result(item, version_id) for item in items]
     return {
         "project_id": project_id,
         "host": host,
@@ -79,6 +111,11 @@ def prepare_state(
             "run_source": "local",
             "expected_item_count": len(items),
             "expected_result_count": len(results),
+            **(
+                {"dataset_revision_id": snapshot.revision_id}
+                if snapshot is not None
+                else {}
+            ),
         },
         "items": [item.to_dict() for item in items],
         "results": [result.to_dict() for result in results],
@@ -108,7 +145,17 @@ def main() -> None:
         default=Path("offline-evaluation-state.json"),
         help="Create a new prepared run here, or replay the existing run exactly.",
     )
+    parser.add_argument(
+        "--dataset-id", help="Run the capital-city application on a PostHog dataset."
+    )
+    parser.add_argument(
+        "--dataset-revision",
+        type=int,
+        help="Pin this numeric revision; defaults to latest once.",
+    )
     args = parser.parse_args()
+    if args.dataset_revision is not None and args.dataset_id is None:
+        parser.error("--dataset-revision requires --dataset-id")
     project_id = int(os.environ["POSTHOG_PROJECT_ID"])
     host = os.environ.get("POSTHOG_HOST", "https://us.posthog.com")
     with OfflineEvaluations(
@@ -133,7 +180,13 @@ def main() -> None:
                     "The saved run belongs to a different project or host."
                 )
         else:
-            state = prepare_state(evaluations, project_id, host)
+            state = prepare_state(
+                evaluations,
+                project_id,
+                host,
+                dataset_id=args.dataset_id,
+                dataset_revision=args.dataset_revision,
+            )
             save_state(args.state, state)
 
         items = {item["id"]: EvaluationItem.from_dict(item) for item in state["items"]}

posthog/ai/evaluations/__init__.py

@@ -1,4 +1,4 @@
-"""Typed offline evaluation uploads, experiment lifecycle, and scorer management.
+"""Typed offline evaluation uploads, experiments, scorers, and dataset management.
 
 from posthog.ai.evaluations import OfflineEvaluations, EvaluationItem
 
@@ -10,6 +10,17 @@
     # experiment.complete()
 """
 
+from ._async_datasets import AsyncDatasets, AsyncDatasetSnapshot
+from ._dataset_types import (
+    Dataset,
+    DatasetItem,
+    DatasetItemPage,
+    DatasetOrder,
+    DatasetPage,
+    DatasetRevision,
+    DatasetRevisionPage,
+)
+from ._datasets import Datasets, DatasetSnapshot
 from ._client import (
     AsyncExperiment,
     AsyncOfflineEvaluations,
@@ -51,6 +62,8 @@
 )
 
 __all__ = [
+    "AsyncDatasets",
+    "AsyncDatasetSnapshot",
     "AsyncExperiment",
     "AsyncOfflineEvaluations",
     "AsyncScorers",
@@ -60,6 +73,15 @@
     "CategoricalPassingRule",
     "CategoricalScorerConfig",
     "CategoricalScorerOption",
+    "Dataset",
+    "DatasetItem",
+    "DatasetItemPage",
+    "DatasetOrder",
+    "DatasetPage",
+    "DatasetRevision",
+    "DatasetRevisionPage",
+    "Datasets",
+    "DatasetSnapshot",
     "EvaluationAPIError",
     "EvaluationItem",
     "EvaluationResult",