From 5b918b69cae5321c49da24de3907f852b5dbb0ea Mon Sep 17 00:00:00 2001 From: Radu Raicea Date: Mon, 26 Oct 2026 15:05:28 -0400 Subject: [PATCH] feat(llma): support per-item offline evaluation models --- .sampo/changesets/jovial-shaman-kullervo.md | 5 + .../example-offline-evaluations/README.md | 5 + posthog/ai/evaluations/_types.py | 8 + posthog/test/ai/test_evaluations.py | 180 +++++++++++++++++- posthog/test/ai/test_evaluations_datasets.py | 51 +++++ posthog/test/ai/test_evaluations_transport.py | 91 +++++++++ references/public_api_snapshot.txt | 4 +- 7 files changed, 341 insertions(+), 3 deletions(-) create mode 100644 .sampo/changesets/jovial-shaman-kullervo.md diff --git a/.sampo/changesets/jovial-shaman-kullervo.md b/.sampo/changesets/jovial-shaman-kullervo.md new file mode 100644 index 000000000..1138e5cc2 --- /dev/null +++ b/.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. diff --git a/examples/example-offline-evaluations/README.md b/examples/example-offline-evaluations/README.md index 7da7dd1c8..00dfec78e 100644 --- a/examples/example-offline-evaluations/README.md +++ b/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. diff --git a/posthog/ai/evaluations/_types.py b/posthog/ai/evaluations/_types.py index 7992ac1c9..595f06920 100644 --- a/posthog/ai/evaluations/_types.py +++ b/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 diff --git a/posthog/test/ai/test_evaluations.py b/posthog/test/ai/test_evaluations.py index b2e76d0eb..99bcd962c 100644 --- a/posthog/test/ai/test_evaluations.py +++ b/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 ): diff --git a/posthog/test/ai/test_evaluations_datasets.py b/posthog/test/ai/test_evaluations_datasets.py index e90d34005..dfc40597c 100644 --- a/posthog/test/ai/test_evaluations_datasets.py +++ b/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 = [] diff --git a/posthog/test/ai/test_evaluations_transport.py b/posthog/test/ai/test_evaluations_transport.py index 3f62fe932..8f247f762 100644 --- a/posthog/test/ai/test_evaluations_transport.py +++ b/posthog/test/ai/test_evaluations_transport.py @@ -12,9 +12,16 @@ import pytest import requests +from posthog.ai.evaluations import ( + AsyncOfflineEvaluations, + EvaluationItem, + OfflineEvaluations, +) from posthog.ai.evaluations._errors import EvaluationAPIError from posthog.ai.evaluations._transport import AsyncTransport, SyncTransport +from .test_evaluations import EXPERIMENT_ID, VERSION_ID, invoke, upload_response + PATH = "/api/projects/123/ai_observability/offline_experiments/" BODY = b'{"id":"stable-id","name":"evaluation data"}' @@ -102,6 +109,90 @@ def test_lost_response_retries_identical_body_and_identity(transport): sleep.assert_called_once() +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "original_model,changed_model", + [("model-b", "model-c"), ("model-b", None), (None, "model-a")], +) +def test_item_models_survive_transport_retries_and_content_conflicts( + asynchronous, original_model, changed_model +): + evaluations = (AsyncOfflineEvaluations if asynchronous else OfflineEvaluations)( + project_id=123, secret_key="phx_test_secret" + ) + experiment = evaluations.resume_experiment(EXPERIMENT_ID) + item = EvaluationItem(model_version=original_model) + conflict = { + "code": "content_conflict", + "detail": "Item content differs from its accepted declaration.", + "attr": "items.0.model_version", + } + attempts = [] + + def respond(method, url, **kwargs): + body = kwargs["content" if asynchronous else "data"] + attempts.append((method, url, body)) + if len(attempts) == 1: + raise ( + httpx.ReadTimeout("lost response") + if asynchronous + else requests.ReadTimeout() + ) + status, payload = ( + (201, upload_response(body)) if len(attempts) == 2 else (409, conflict) + ) + return ( + httpx.Response(status, json=payload) + if asynchronous + else response(status, payload) + ) + + target = ( + evaluations._transport._client + if asynchronous + else evaluations._transport._session + ) + try: + with ( + patch.object(target, "request", side_effect=respond), + patch("posthog.ai.evaluations._transport.time.sleep"), + patch( + "posthog.ai.evaluations._transport.asyncio.sleep", + new_callable=AsyncMock, + ), + ): + receipt = invoke( + experiment.upload_result, + item=item, + scorer_version_id=VERSION_ID, + value=True, + ) + assert receipt.items[0].id == item.id + assert len(attempts) == 2 + assert attempts[0] == attempts[1] + assert json.loads(attempts[0][2])["items"] == [item.to_dict()] + changed = EvaluationItem(id=item.id, model_version=changed_model) + with pytest.raises(EvaluationAPIError) as caught: + invoke( + experiment.upload_result, + item=changed, + scorer_version_id=VERSION_ID, + value=True, + ) + assert caught.value.status == 409 + assert caught.value.code == "content_conflict" + assert caught.value.response == conflict + assert caught.value.persistence == "rejected" + assert len(attempts) == 3 + assert json.loads(attempts[2][2])["items"] == [changed.to_dict()] + assert all( + method == "POST" and url.endswith(f"/{EXPERIMENT_ID}/upload/") + for method, url, _ in attempts + ) + finally: + invoke(evaluations.aclose if asynchronous else evaluations.close) + + @pytest.mark.parametrize("failure", [requests.ReadTimeout(), response(503)]) def test_non_idempotent_mutation_never_replays(transport, failure): with ( diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index feb33578b..25cbaa418 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -1209,7 +1209,7 @@ class posthog.ai.evaluations.DatasetRevisionPage(count: int, results: tuple[Data class posthog.ai.evaluations.DatasetSnapshot(dataset: Dataset, revision: int, revision_id: str, _datasets: Datasets) class posthog.ai.evaluations.Datasets(transport: SyncTransport, project_id: int) class posthog.ai.evaluations.EvaluationAPIError(*, status: int | None = None, code: str | None = None, detail: Any = None, attr: str | None = None, errors: Any = None, response: dict[str, Any] | None = None, retry_after: float | None = None, persistence: Literal['rejected', 'unknown', 'not_sent'] = 'unknown', submission: dict[str, Any] | None = None) -class posthog.ai.evaluations.EvaluationItem(*, id: str | UUID | None = None, input: JSONValue | _Unset = _UNSET, output: JSONValue | _Unset = _UNSET, expected_output: JSONValue | _Unset = _UNSET, metadata: dict[str, JSONValue] | None | _Unset = _UNSET, case_key: str | None = None, trial: str | None = None, dataset_item_identifier: str | None = None, dataset_item_version_identifier: str | None = None, dataset_item_version_id: str | UUID | None = None, application_trace_id: str | None = None, payload: dict[str, JSONValue] | _Unset = _UNSET) +class posthog.ai.evaluations.EvaluationItem(*, id: str | UUID | None = None, input: JSONValue | _Unset = _UNSET, output: JSONValue | _Unset = _UNSET, expected_output: JSONValue | _Unset = _UNSET, metadata: dict[str, JSONValue] | None | _Unset = _UNSET, case_key: str | None = None, trial: str | None = None, dataset_item_identifier: str | None = None, 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) class posthog.ai.evaluations.EvaluationResult(*, item: EvaluationItem | str | UUID, scorer_version_id: str | UUID, value: ScoreValue | None = None, status: ResultStatus = 'ok', error_code: str | None = None, reasoning: str | None | _Unset = _UNSET, error_message: str | None | _Unset = _UNSET, metadata: dict[str, JSONValue] | None | _Unset = _UNSET, evaluator_trace_id: str | None = None, evaluated_at: datetime | str | None = None, payload: dict[str, JSONValue] | _Unset = _UNSET) class posthog.ai.evaluations.Experiment(transport: SyncTransport, project_id: int, id: str, *, receipt: ExperimentReceipt | None = None, submission: bytes | None = None) class posthog.ai.evaluations.ExperimentOptions @@ -1622,7 +1622,7 @@ method posthog.ai.evaluations.Datasets.restore_item(item_id: str | UUID, *, base method posthog.ai.evaluations.Datasets.snapshot(dataset_id: str | UUID, *, revision: int | None = None) method posthog.ai.evaluations.Datasets.update(dataset_id: str | UUID, *, name: str | _Unset = _UNSET, description: str | _Unset = _UNSET, metadata: dict[str, JSONValue] | _Unset = _UNSET, tags: Sequence[str] | _Unset = _UNSET) method posthog.ai.evaluations.Datasets.update_item(item_id: str | UUID, *, base_version: int, input: JSONValue | _Unset = _UNSET, expected_output: JSONValue | _Unset = _UNSET, metadata: dict[str, JSONValue] | _Unset = _UNSET) -method posthog.ai.evaluations.EvaluationItem.from_dataset_item(item: DatasetItem, *, output: JSONValue | _Unset = _UNSET, id: str | UUID | None = None, case_key: str | None = None, trial: str | None = None, application_trace_id: str | None = None) +method posthog.ai.evaluations.EvaluationItem.from_dataset_item(item: DatasetItem, *, output: JSONValue | _Unset = _UNSET, id: str | UUID | None = None, case_key: str | None = None, trial: str | None = None, application_trace_id: str | None = None, model_version: str | None = None) method posthog.ai.evaluations.EvaluationItem.from_dict(data: Mapping[str, Any]) method posthog.ai.evaluations.EvaluationItem.to_dict() method posthog.ai.evaluations.EvaluationResult.from_dict(data: Mapping[str, Any], *, item: EvaluationItem | None = None)