feat(llma): support per-item offline evaluation models

.sampo/changesets/jovial-shaman-kullervo.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add optional per-item model_version to offline evaluations and dataset conversion, preserving model attribution across serialization, retries, and recovery.

examples/example-offline-evaluations/README.md

@@ -65,6 +65,7 @@ with OfflineEvaluations(
         output="Paris",
         expected_output="Paris",
         case_key="capital-france",
+        model_version="model-b",
     )
     receipt = experiment.upload_result(
         item=item,
@@ -77,6 +78,8 @@ with OfflineEvaluations(
 
 Use one `EvaluationItem` for the application's execution and reuse it across scorer results. `upload_results([EvaluationResult(...), ...])` uploads a finite sequence, eagerly and sequentially. It splits at 1,000 items/results or 5 MiB of UTF-8 JSON per request, whichever comes first. Item payloads are limited to 1 MiB; result payloads to 256 KiB. Payload JSON supports at most 32 nested containers. The SDK raises for invalid values and oversized payloads without truncating them. Memory use scales with the sequence you supply, so submit caller-sized batches for large datasets.
 
+Set `model_version` on each item to compare executions from different models in one experiment. It identifies the application's model and overrides the default supplied to `create_experiment(model_version=...)`. Omit it or pass `None` to inherit that default on the server; without either value, the model remains unknown. Explicit model identifiers must be nonempty strings of at most 255 characters and are preserved without trimming. The declared model is immutable item content: changing, adding, or removing it after acceptance causes a `content_conflict`, even if the effective model is unchanged. Create a new item for a different model execution, and preserve the original declaration for retries.
+
 `status="ok"` means the evaluator ran successfully. Its boolean, number, or category-list `value` is uploaded with its original polarity. Numeric scores use finite double-precision (binary64) values; integer inputs are rejected if conversion would change their value. Use `error`, `skipped`, or `not_applicable` without a value for other outcomes. `error_code` and a non-null `error_message` are valid only with `status="error"`.
 
 Omitted payload fields, explicit `None`, and empty payloads are distinct:
@@ -154,6 +157,8 @@ This example uses string inputs. Replace `capital_answer` and the equality check
 
 `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`.
 
+Pass `EvaluationItem.from_dataset_item(case, output=output, model_version="model-b")` to attribute a dataset execution to a model. Calling it again with another model creates a new execution UUID while preserving the same case key and dataset-item version. Omitting the model or passing `None` leaves experiment fallback to the server; dataset metadata and source traces never supply the execution model.
+
 ### 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.

posthog/ai/evaluations/_types.py

@@ -83,6 +83,8 @@ class EvaluationItem:
     Payload values are copied as JSON. Omitted fields differ from explicit None.
     Use ``payload={}`` for an explicitly empty payload. Save ``to_dict()`` to
     restore exactly the same declaration after restarting the process.
+    ``model_version`` identifies this execution's model and overrides the
+    experiment default. Omit it or pass None to inherit the default on the server.
     """
 
     id: str
@@ -102,6 +104,7 @@ def __init__(
         dataset_item_version_identifier: str | None = None,
         dataset_item_version_id: str | UUID | None = None,
         application_trace_id: str | None = None,
+        model_version: str | None = None,
         payload: dict[str, JSONValue] | _Unset = _UNSET,
     ) -> None:
         identity = _uuid(uuid4() if id is None else id)
@@ -112,6 +115,7 @@ def __init__(
             "dataset_item_identifier": dataset_item_identifier,
             "dataset_item_version_identifier": dataset_item_version_identifier,
             "application_trace_id": application_trace_id,
+            "model_version": model_version,
         }.items():
             if value is not None:
                 data[key] = _identifier(value, key)
@@ -146,6 +150,7 @@ def from_dataset_item(
         case_key: str | None = None,
         trial: str | None = None,
         application_trace_id: str | None = None,
+        model_version: str | None = None,
     ) -> EvaluationItem:
         """Declare one execution of a dataset case, preserving its exact version.
 
@@ -156,6 +161,8 @@ def from_dataset_item(
         ``case_key`` defaults to the caller-owned client item ID, or stable item
         UUID. Omit output for an execution that failed before producing one.
         Use the owning snapshot's revision_id when creating the experiment.
+        Supply ``model_version`` for this execution's model; omitted or None
+        inherits the experiment default, never dataset metadata or source traces.
         """
         if not isinstance(item, DatasetItem):
             raise TypeError("Expected a DatasetItem.")
@@ -173,6 +180,7 @@ def from_dataset_item(
             dataset_item_version_identifier=item.version_id,
             dataset_item_version_id=item.version_id,
             application_trace_id=application_trace_id,
+            model_version=model_version,
         )
 
     @classmethod

posthog/test/ai/test_evaluations.py

@@ -276,7 +276,9 @@ def test_resume_is_local_and_supports_references_without_scorer_discovery(client
 
 def test_same_item_is_declared_once_for_multiple_scorers(client):
     evaluations, transport = client
-    item = EvaluationItem(input={"question": "Paris?"}, output="France")
+    item = EvaluationItem(
+        input={"question": "Paris?"}, output="France", model_version="model-b"
+    )
     results = [
         result(item),
         EvaluationResult(item=item, scorer_version_id=OTHER_VERSION_ID, value=True),
@@ -286,11 +288,136 @@ def test_same_item_is_declared_once_for_multiple_scorers(client):
     )
     body = json.loads(transport.request.call_args.kwargs["body"])
     assert body["items"] == [item.to_dict()]
+    assert body["items"][0]["model_version"] == "model-b"
+    assert all("model_version" not in row for row in body["results"])
     assert len(receipt.chunks) == 1
     assert len(receipt.chunks[0].results) == 2
     assert len(receipt.chunks[0].items) == 1
 
 
+@pytest.mark.parametrize("experiment_options", [{}, {"model_version": "model-a"}, None])
+def test_item_models_are_declared_without_resolving_experiment_defaults(
+    client, experiment_options
+):
+    evaluations, transport = client
+    if experiment_options is None:
+        experiment = evaluations.resume_experiment(EXPERIMENT_ID)
+        transport.request.assert_not_called()
+    else:
+        experiment = invoke(
+            evaluations.create_experiment, name="Models", **experiment_options
+        )
+        creation = json.loads(transport.request.call_args.kwargs["body"])
+        assert creation.get("model_version") == experiment_options.get("model_version")
+        transport.request.reset_mock()
+    items = [
+        EvaluationItem(model_version=model)
+        for model in ["model-b", "model-c", "x" * 255, " model with spaces ", None]
+    ] + [EvaluationItem()]
+    invoke(experiment.upload_results, [result(item) for item in items])
+    transport.request.assert_called_once()
+    assert transport.request.call_args.args == (
+        "POST",
+        BASE + experiment.id + "/upload/",
+    )
+    body = json.loads(transport.request.call_args.kwargs["body"])
+    assert body["items"] == [item.to_dict() for item in items]
+    assert [item.get("model_version") for item in body["items"]] == [
+        "model-b",
+        "model-c",
+        "x" * 255,
+        " model with spaces ",
+        None,
+        None,
+    ]
+    assert all("model_version" not in row for row in body["results"])
+    assert all("payload" not in item for item in body["items"])
+    if experiment_options is not None:
+        assert experiment.submission == creation
+
+
+@pytest.mark.parametrize("model", ["", 1, True, [], {}, "x" * 256, "with\x00null"])
+def test_invalid_item_models_are_rejected_before_io(client, model):
+    _, transport = client
+    with pytest.raises(ValueError, match="model_version"):
+        EvaluationItem(model_version=model)
+    with pytest.raises(ValueError, match="model_version"):
+        EvaluationItem.from_dict({"id": str(uuid4()), "model_version": model})
+    transport.request.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "original_model,changed_model",
+    [("model-b", "model-c"), ("model-b", None), (None, "model-a")],
+)
+def test_conflicting_item_models_are_rejected_before_any_bulk_chunk(
+    client, monkeypatch, original_model, changed_model
+):
+    evaluations, transport = client
+    monkeypatch.setattr(_client, "_MAX_RESULTS", 1)
+    item = EvaluationItem(model_version=original_model)
+    changed = EvaluationItem(id=item.id, model_version=changed_model)
+    results = [
+        result(item),
+        EvaluationResult(item=changed, scorer_version_id=OTHER_VERSION_ID, value=True),
+    ]
+    with pytest.raises(ValueError, match="Conflicting"):
+        invoke(evaluations.resume_experiment(EXPERIMENT_ID).upload_results, results)
+    transport.request.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "model_options", [{}, {"model_version": None}, {"model_version": "model-b"}]
+)
+def test_saved_item_models_replay_and_allow_another_scorer_by_uuid(
+    client, model_options
+):
+    evaluations, transport = client
+    experiment = invoke(
+        evaluations.create_experiment, name="Models", model_version="model-a"
+    )
+    item = EvaluationItem(**model_options)
+    original = result(item)
+    first = invoke(experiment.upload_results, [original]).chunks[0]
+    original_body = transport.request.call_args.kwargs["body"]
+    saved = item.to_dict()
+    saved.update(model_options)
+    restored_item = EvaluationItem.from_dict(saved)
+    restored = EvaluationResult.from_dict(original.to_dict(), item=restored_item)
+    second_scorer = EvaluationResult(
+        item=item, scorer_version_id=OTHER_VERSION_ID, value=True
+    )
+    restored_second = EvaluationResult.from_dict(
+        second_scorer.to_dict(), item=restored_item
+    )
+    saved["model_version"] = "changed"
+    assert restored_item.to_dict() == item.to_dict()
+    assert restored.item is restored_second.item is restored_item
+    assert restored.item_id == restored_second.item_id == item.id
+    transport.request.side_effect = lambda method, path, *, body: upload_response(
+        body, created=False
+    )
+    replayed = invoke(experiment.upload_results, [restored]).chunks[0]
+    assert transport.request.call_args.kwargs["body"] == original_body
+    for accepted, replay in zip(
+        first.items + first.results, replayed.items + replayed.results
+    ):
+        assert replay.id == accepted.id
+        assert replay.accepted_at == accepted.accepted_at
+        assert replay.created is False
+    transport.request.reset_mock()
+    invoke(
+        experiment.upload_result,
+        item=item.id,
+        scorer_version_id=OTHER_VERSION_ID,
+        value=True,
+    )
+    transport.request.assert_called_once()
+    body = json.loads(transport.request.call_args.kwargs["body"])
+    assert body["items"] == []
+    assert "model_version" not in body["results"][0]
+
+
 def test_real_result_count_boundary_splits_without_dropping_declarations(client):
     evaluations, transport = client
     results = [result() for _ in range(1001)]
@@ -321,6 +448,57 @@ def test_utf8_byte_boundary_splits_below_request_limit(client):
     )
 
 
+def test_item_models_count_towards_the_encoded_request_limit(client, monkeypatch):
+    evaluations, transport = client
+    experiment = evaluations.resume_experiment(EXPERIMENT_ID)
+    items = [EvaluationItem(), EvaluationItem()]
+    invoke(experiment.upload_results, [result(item) for item in items])
+    limit = len(transport.request.call_args.kwargs["body"])
+    monkeypatch.setattr(_client, "_MAX_REQUEST_BYTES", limit)
+    transport.request.reset_mock()
+    modeled = [EvaluationItem(id=item.id, model_version="🙂" * 20) for item in items]
+    receipt = invoke(experiment.upload_results, [result(item) for item in modeled])
+    assert len(receipt.chunks) == 2
+    for call, item in zip(transport.request.call_args_list, modeled):
+        assert len(call.kwargs["body"]) <= limit
+        assert json.loads(call.kwargs["body"])["items"] == [item.to_dict()]
+
+
+def test_shared_item_model_survives_partial_bulk_failure_and_recovery(client):
+    evaluations, transport = client
+    item = EvaluationItem(model_version="model-b")
+    results = [
+        EvaluationResult(
+            item=item, scorer_version_id=uuid5(UUID(VERSION_ID), str(index)), value=True
+        )
+        for index in range(1001)
+    ]
+
+    def fail_second(method, path, *, body):
+        if transport.request.call_count == 2:
+            raise EvaluationAPIError(persistence="unknown")
+        return upload_response(body)
+
+    transport.request.side_effect = fail_second
+    experiment = evaluations.resume_experiment(EXPERIMENT_ID)
+    with pytest.raises(BulkUploadError) as caught:
+        invoke(experiment.upload_results, results)
+    error = caught.value
+    assert len(error.completed.chunks[0].results) == 1000
+    assert error.failed_index == 1000
+    assert error.pending_results == (results[-1],)
+    assert error.failed_chunk["items"] == [item.to_dict()]
+    failed_body = transport.request.call_args.kwargs["body"]
+    error.failed_chunk["items"][0]["model_version"] = "changed"
+    assert error.failed_chunk["items"] == [item.to_dict()]
+    transport.request.side_effect = respond
+    invoke(experiment.upload_results, error.pending_results)
+    assert transport.request.call_args.kwargs["body"] == failed_body
+    assert transport.request.call_count == 3
+    for call in transport.request.call_args_list:
+        assert json.loads(call.kwargs["body"])["items"] == [item.to_dict()]
+
+
 def test_shared_item_is_redeclared_in_each_independently_replayable_chunk(
     client, monkeypatch
 ):

posthog/test/ai/test_evaluations_datasets.py

@@ -648,6 +648,57 @@ def test_dataset_to_execution_copies_payload_and_keeps_trials_independent(client
         EvaluationItem.from_dataset_item(replace(case, archived=True))
 
 
+def test_dataset_executions_keep_explicit_models_and_never_infer_them(client):
+    evaluations, transport = client
+    transport.request.return_value = item_response(
+        metadata={"model_version": "source-model"}
+    )
+    case = invoke(evaluations.datasets.get_item, ITEM_ID)
+    transport.request.reset_mock()
+    items = [
+        EvaluationItem.from_dataset_item(case, output="Paris", model_version="model-b"),
+        EvaluationItem.from_dataset_item(case, output="Lyon", model_version="model-c"),
+        EvaluationItem.from_dataset_item(case),
+        EvaluationItem.from_dataset_item(case, model_version=None),
+    ]
+    assert len({item.id for item in items}) == 4
+    for item in items:
+        data = item.to_dict()
+        assert data["case_key"] == "capital-france"
+        assert data["dataset_item_version_id"] == VERSION_ID
+        assert data["payload"]["metadata"] == {"model_version": "source-model"}
+        assert EvaluationItem.from_dict(data).to_dict() == data
+    for invalid in ["", 1, True, "x" * 256]:
+        with pytest.raises(ValueError, match="model_version"):
+            EvaluationItem.from_dataset_item(case, model_version=invalid)
+    transport.request.assert_not_called()
+    transport.request.side_effect = respond_to_experiment
+    experiment = invoke(
+        evaluations.create_experiment, name="Models", model_version="model-a"
+    )
+    invoke(
+        experiment.upload_results,
+        [
+            EvaluationResult(item=item, scorer_version_id=OTHER_ITEM_ID, value=True)
+            for item in items
+        ],
+    )
+    body = json.loads(transport.request.call_args.kwargs["body"])
+    assert body["items"] == [item.to_dict() for item in items]
+    assert [item.get("model_version") for item in body["items"]] == [
+        "model-b",
+        "model-c",
+        None,
+        None,
+    ]
+    assert [item["payload"].get("output") for item in body["items"]] == [
+        "Paris",
+        "Lyon",
+        None,
+        None,
+    ]
+
+
 def test_dataset_scorer_experiment_flow_pins_both_versions_and_reuses_execution(client):
     evaluations, transport = client
     bodies = []