feat(llma): support composite offline scorers

.github/scripts/check_strict_types.sh

@@ -20,9 +20,11 @@ from posthog.ai.evaluations import (
     AsyncOfflineEvaluations,
     BooleanScorerConfig,
     EvaluationItem,
+    EvaluationResult,
     NumericPassingRule,
     NumericScorerConfig,
     OfflineEvaluations,
+    ScorerComposition,
 )
 
 client = Posthog("phc_test")
@@ -73,6 +75,21 @@ async def offline_evaluation_types(
     async_experiment = await async_client.create_experiment(name="async run")
     await async_experiment.upload_result(item=item, scorer_version_id=boolean.current_version_id, value=False)
     _ = config
+    composition: ScorerComposition = {
+        "expression": "relevance",
+        "inputs": [{"alias": "relevance", "scorer_version_id": numeric.current_version_id}],
+    }
+    overall = sync_client.scorers.create(
+        name="Overall", kind="numeric", config=NumericScorerConfig(min=0, max=1),
+        calculation="composite", composition=composition,
+    )
+    valid: bool = await async_client.scorers.validate_composition(kind="numeric", composition=composition)
+    sync_client.scorers.list(calculation="composite", input_definition_id=numeric.id)
+    experiment.upload_results([
+        EvaluationResult(item=item, scorer_version_id=numeric.current_version_id, value=0.9),
+        EvaluationResult(item=item, scorer_version_id=overall.current_version_id, calculate=True),
+    ])
+    _ = valid
 PY
 
 cat > "$tmp/invalid_types.py" <<'PY'

.sampo/changesets/composite-offline-scorers.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Support versioned composite scorers for offline evaluations, including formula validation, dependency metadata, explicit server calculations, stored outcome receipts, and composite-safe bulk uploads.

examples/example-offline-evaluations/README.md

@@ -10,6 +10,8 @@ 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` |
+| Validate a composite formula without saving | Personal API key with `llm_analytics:read` |
+| Calculate composites using previously stored inputs | Personal API key with ingestion access, `evaluation:read`, and `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` |
@@ -92,6 +94,79 @@ 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.
 
+## Composite item scores
+
+Composite scorers calculate one score per item on the PostHog server from exact reported scorer versions. They can return numeric, boolean, or categorical values. This requires a backend with composite scorer support; ordinary uploads remain compatible with older backends.
+
+Provision a composite once, using a personal key with scorer write access. This example assumes `correctness_version_id` identifies a reported boolean scorer and `relevance_version_id` a reported numeric scorer:
+
+```python
+from posthog.ai.evaluations import ScorerComposition
+
+composition: ScorerComposition = {
+    "expression": "(number(correct) + relevance) / 2",
+    "inputs": [
+        {"alias": "correct", "scorer_version_id": correctness_version_id},
+        {"alias": "relevance", "scorer_version_id": relevance_version_id},
+    ],
+}
+# Optional; read scope is sufficient and no scorer/version is created.
+evaluations.scorers.validate_composition(
+    kind="numeric", composition=composition, config={"min": 0, "max": 1},
+)
+overall = evaluations.scorers.create(
+    name="Overall",
+    kind="numeric",
+    calculation="composite",
+    config={"min": 0, "max": 1, "passing_rule": {"operator": "gte", "threshold": 0.8}},
+    composition=composition,
+)
+overall_version_id = overall.current_version_id
+```
+
+Save that version UUID for your runs. Creating scorers/versions is not idempotent, so a lost response requires inspecting server state before repeating creation. Update a formula, output configuration, or pinned input through `create_version(scorer_id, config=..., composition=..., base_version=...)`. Supply the complete composition each time. New source versions never update a composite automatically.
+
+Upload the reported inputs and explicitly request the calculation:
+
+```python
+experiment = evaluations.create_experiment(
+    name="Composite example", expected_item_count=1, expected_result_count=3,
+)
+item = EvaluationItem(input="Capital of France?", output="Paris")
+receipt = experiment.upload_results([
+    EvaluationResult(item=item, scorer_version_id=correctness_version_id, value=True),
+    EvaluationResult(item=item, scorer_version_id=relevance_version_id, value=0.8),
+    EvaluationResult(item=item, scorer_version_id=overall_version_id, calculate=True),
+])
+for chunk in receipt.chunks:
+    for outcome in chunk.results:
+        print(outcome.scorer_version_id, outcome.status, outcome.error_code)
+experiment.complete()
+```
+
+Calculation declarations contain only the item reference, scorer version, and `calculate=True`. Do not supply a value, reasoning, metadata, evaluator trace, or evaluation time. An item may also be referenced by its already accepted UUID.
+
+Project secret keys and ingestion-only personal keys must supply **every pinned input result in the same HTTP request**, including on retries. Previously uploaded inputs may be resubmitted with their original values and metadata. Reusing stored inputs without resubmitting them requires both `evaluation:read` and `llm_analytics:read`, in addition to ingestion access. Composite requests and retries also require current access to the composite and its inputs.
+
+The SDK keeps all supplied results for a composite-bearing item together without fetching scorer definitions. It preserves order, so interleaved composite items can form a larger indivisible group. Group each item's results together for large batches. If a group exceeds 1,000 items/results or 5 MiB, the entire call fails locally before any upload. The SDK cannot check that your supplied inputs match the pinned formula without read access; the server validates this. Missing inputs reject the whole HTTP request. Inputs can appear before or after their composite in that request.
+
+HTTP success acknowledges persistence, not successful calculation. Input errors become `composite_input_error`; formula errors or invalid output values become `composite_formula_error`. Input statuses propagate in the order `error`, `skipped`, `not_applicable`. Check each receipt's `status` and `error_code`; no result-read permission is needed. For ordinary results on an older backend, `status=None` means the backend omitted the outcome. Composite acknowledgments require an outcome.
+
+All accepted composite outcomes count toward `expected_result_count`, including errors and skips. Derived results add no uploaded-result billing usage. Omitting a composite leaves it unrequested. Exact retries return the stored outcome, including after completion; formulas are not recalculated on retry or read. Persist declarations with `to_dict` and restore them with `from_dict` for exact recovery.
+
+Formulas support 1–16 reported inputs and at most 1,024 characters; aliases start with a lowercase letter and contain lowercase letters, digits, or underscores, up to 32 characters. Nested composites are unsupported. Use `number(flag)` for explicit boolean conversion and `"good" in categories` for categorical membership. Boolean inputs retain their raw meaning regardless of passing rules. Numeric composites support bounds and passing thresholds, but no non-null `step`. Formula syntax, input access, and output types are validated by the server; runtime values and bounds are checked during ingestion.
+
+Read `scorer.calculation`, `scorer.composition`, and `scorer.input_scorers` to inspect saved inputs and their pinned/latest versions. `dependent_composite_count` counts accessible active dependents; `scorers.list(calculation="composite", input_definition_id=source_id)` retrieves them. Historical versions retain their saved composition. All operations also exist on `AsyncOfflineEvaluations`.
+
+The runnable capital-city example can include a composite consuming only its configured boolean scorer version. Provision one with expression `number(correct)`, pin that boolean version as `correct`, and pass its composite version UUID when creating a new state file:
+
+```bash
+uv run python examples/example-offline-evaluations/main.py \
+  --state composite-run.json --composite-version-id 'your-composite-version-uuid'
+```
+
+The state file includes both reported and calculation declarations and the combined expected count. Reusing the file replays that saved run regardless of newly supplied scorer flags.
+
 ## 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.

examples/example-offline-evaluations/main.py

@@ -29,8 +29,8 @@ def resolve_version(evaluations: OfflineEvaluations) -> str:
             "Set POSTHOG_SCORER_ID or POSTHOG_SCORER_VERSION_ID, or run --create-scorer once."
         )
     scorer = evaluations.scorers.get(scorer_id)
-    if scorer.kind != "boolean":
-        raise ValueError("This example requires a boolean scorer.")
+    if scorer.kind != "boolean" or scorer.calculation != "reported":
+        raise ValueError("This example requires a reported boolean scorer.")
     return scorer.current_version_id
 
 
@@ -67,6 +67,7 @@ def prepare_state(
     *,
     dataset_id: str | None = None,
     dataset_revision: int | None = None,
+    composite_version_id: str | None = None,
 ) -> dict[str, Any]:
     version_id = resolve_version(evaluations)
     cases = [
@@ -100,7 +101,17 @@ def prepare_state(
             )
             for case in cases
         ]
-    results = [exact_match_result(item, version_id) for item in items]
+    results = []
+    for item in items:
+        results.append(exact_match_result(item, version_id))
+        if composite_version_id is not None:
+            results.append(
+                EvaluationResult(
+                    item=item,
+                    scorer_version_id=composite_version_id,
+                    calculate=True,
+                )
+            )
     return {
         "project_id": project_id,
         "host": host,
@@ -153,6 +164,10 @@ def main() -> None:
         type=int,
         help="Pin this numeric revision; defaults to latest once.",
     )
+    parser.add_argument(
+        "--composite-version-id",
+        help="Also calculate this composite version, which must pin the configured boolean scorer version.",
+    )
     args = parser.parse_args()
     if args.dataset_revision is not None and args.dataset_id is None:
         parser.error("--dataset-revision requires --dataset-id")
@@ -186,6 +201,7 @@ def main() -> None:
                 host,
                 dataset_id=args.dataset_id,
                 dataset_revision=args.dataset_revision,
+                composite_version_id=args.composite_version_id,
             )
             save_state(args.state, state)
 
@@ -215,6 +231,16 @@ def main() -> None:
         print(
             f"Acknowledged {sum(len(chunk.results) for chunk in receipt.chunks)} results across {len(receipt.chunks)} chunks."
         )
+        errors = [
+            result
+            for chunk in receipt.chunks
+            for result in chunk.results
+            if result.status == "error"
+        ]
+        if errors:
+            print(
+                f"Accepted {len(errors)} error outcomes: {[result.error_code for result in errors]}"
+            )
         print(
             f"Keep {args.state} for exact replay; choose a different --state path for a new experiment."
         )

posthog/ai/evaluations/__init__.py

@@ -36,7 +36,11 @@
     NumericPassingRule,
     NumericScorerConfig,
     Scorer,
+    ScorerCalculation,
+    ScorerComposition,
+    ScorerCompositionInput,
     ScorerConfig,
+    ScorerInput,
     ScorerKind,
     ScorerOrder,
     ScorerPage,
@@ -99,7 +103,11 @@
     "RunSource",
     "ScoreValue",
     "Scorer",
+    "ScorerCalculation",
+    "ScorerComposition",
+    "ScorerCompositionInput",
     "ScorerConfig",
+    "ScorerInput",
     "ScorerKind",
     "ScorerOrder",
     "ScorerPage",

posthog/ai/evaluations/_client.py

@@ -95,6 +95,30 @@ def _upload_body(items: list[bytes], results: list[bytes]) -> bytes:
     )
 
 
+def _group_ends(results: tuple[EvaluationResult, ...]) -> list[int]:
+    # A calculation only names a version, so ingestion-only clients cannot
+    # discover its dependencies. Keep every supplied result for that item in
+    # one request. Merge overlapping spans to retain the original order.
+    first: dict[str, int] = {}
+    last: dict[str, int] = {}
+    composite_items: set[str] = set()
+    for index, result in enumerate(results):
+        first.setdefault(result.item_id, index)
+        last[result.item_id] = index + 1
+        if result.calculate:
+            composite_items.add(result.item_id)
+    spans = {first[item_id]: last[item_id] for item_id in composite_items}
+    ends: list[int] = []
+    index = 0
+    while index < len(results):
+        end = index + 1
+        while index < end:
+            end = max(end, spans.get(index, index + 1))
+            index += 1
+        ends.append(end)
+    return ends
+
+
 def _chunks(results: tuple[EvaluationResult, ...]) -> list[_Chunk]:
     # Validate the finite submission before sending any chunk. Receipts and
     # prepared bodies use memory proportional to this caller-sized sequence.
@@ -120,38 +144,50 @@ def _chunks(results: tuple[EvaluationResult, ...]) -> list[_Chunk]:
     encoded_results: list[bytes] = []
     start = 0
     size = len(_upload_body([], []))
-    for index, result in enumerate(results):
-        # Include declarations wherever their results occur. This makes every
-        # chunk independently replayable even after a lost acknowledgment.
-        item = (
-            declarations.get(result.item_id) if result.item_id not in item_ids else None
-        )
-        extra = len(result._json) + bool(encoded_results)
-        if item is not None:
-            extra += len(item) + bool(items)
-        if encoded_results and (
-            len(encoded_results) >= _MAX_RESULTS
-            or len(items) + (item is not None) > _MAX_ITEMS
-            or size + extra > _MAX_REQUEST_BYTES
-        ):
+    group_start = 0
+    for end in _group_ends(results):
+        group = results[group_start:end]
+        group_results = [result._json for result in group]
+        group_items = {
+            result.item_id: declarations[result.item_id]
+            for result in group
+            if result.item_id in declarations
+        }
+        while True:
+            new_items = [body for id, body in group_items.items() if id not in item_ids]
+            extra = (
+                sum(map(len, group_results))
+                + len(group_results)
+                - (not encoded_results)
+            )
+            if new_items:
+                extra += sum(map(len, new_items)) + len(new_items) - (not items)
+            fits = (
+                len(encoded_results) + len(group) <= _MAX_RESULTS
+                and len(items) + len(new_items) <= _MAX_ITEMS
+                and size + extra <= _MAX_REQUEST_BYTES
+            )
+            if fits:
+                break
+            if not encoded_results:
+                raise ValueError(
+                    "One result or composite item group exceeds the upload request limit."
+                )
             chunks.append(
                 _Chunk(
-                    start, results[start:index], _upload_body(items, encoded_results)
+                    start,
+                    results[start:group_start],
+                    _upload_body(items, encoded_results),
                 )
             )
-            start, size = index, len(_upload_body([], []))
+            start, size = group_start, len(_upload_body([], []))
             item_ids, items, encoded_results = set(), [], []
-            item = declarations.get(result.item_id)
-            extra = len(result._json) + (len(item) if item is not None else 0)
-        if size + extra > _MAX_REQUEST_BYTES:
-            raise ValueError(
-                "One result and its item exceed the upload request byte limit."
-            )
-        if item is not None:
-            item_ids.add(result.item_id)
-            items.append(item)
-        encoded_results.append(result._json)
+        # Repeat available declarations in every request for exact recovery.
+        item_ids.update(group_items)
+        items.extend(new_items)
+        encoded_results.extend(group_results)
         size += extra
+        group_start = end
     if encoded_results:
         chunks.append(
             _Chunk(start, results[start:], _upload_body(items, encoded_results))
@@ -245,6 +281,11 @@ def upload_results(self, results: Sequence[EvaluationResult]) -> BulkUploadRecei
         Returns per-chunk receipts; an empty sequence makes no requests. Memory
         scales with the caller's sequence. For large datasets, call in batches.
         BulkUploadError exposes committed chunks and pending_results to retry.
+        Items with calculations keep all their supplied results in one request.
+        Interleaved items may form a larger indivisible group; an oversized
+        group fails before any I/O. Group each item's results together when
+        preparing large batches. Stored inputs require evaluation:read and
+        llm_analytics:read; ingestion-only keys must supply every pinned input.
         """
         prepared = tuple(results)
         chunks = _chunks(prepared)