feat(gooddata-eval): select Langfuse dataset items by id or metadata

packages/gooddata-eval/README.md

@@ -117,8 +117,17 @@ gd-eval run \
 |---|---|
 | `--dataset PATH` | Flat folder of JSON files — one question per file. |
 | `--langfuse-dataset NAME` | Pull items by name from a Langfuse dataset. Requires `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY` and `LANGFUSE_BASE_URL` (or the legacy `LANGFUSE_HOST`). |
+| `--langfuse-item-id ITEM_ID` | Run only this item from the Langfuse dataset. Repeatable. An id the dataset does not have is an error, not a skip — asking for twelve items and evaluating nine, silently, reads downstream as a pass-rate change. Requires `--langfuse-dataset`. |
+| `--langfuse-item-metadata KEY=VALUE` | Run only the items whose Langfuse item metadata carries this key and value, e.g. `--langfuse-item-metadata language=es`. Repeatable; several are ANDed, and values compare as strings. Combined with `--langfuse-item-id`, an item must satisfy both. Requires `--langfuse-dataset`. |
 | `--kind TEST_KIND` | Fallback `test_kind` for dataset items that do not embed one. Defaults to `visualization`; use e.g. `agentic_metric_skill` for multi-turn agentic evaluation. Items that declare their own `test_kind` ignore this. |
 
+Both item filters are applied after the fetch — the Langfuse dataset-items endpoint selects
+by dataset but not by item id or item metadata. The fetch costs the same either way; what
+the filters save is the part that actually costs, evaluating an item against a live agent.
+They exist so one published dataset can hold every variant of a question (its locales, say)
+and still be dispatched a slice at a time, instead of being split into datasets that
+Langfuse can then no longer compare against each other.
+
 #### Model selection
 
 | Flag | Description |
@@ -159,6 +168,7 @@ interleaves when K > 1, and per-item latencies rise, so they stop being clean si
 | Flag | Description |
 |---|---|
 | `--langfuse` | Log scores and traces to Langfuse after each item. Requires `--langfuse-dataset`. Names each experiment run `{dataset_name}_{timestamp}_{model}`, suffixed `_effort-{level}` when `--reasoning-effort` is set (so runs differing only by effort stay separate) and `_run{N}` per run when `--runs` > 1 — e.g. `general_question_2026-09-02-11-13_gpt-5.2_run0`. Requires `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY` and `LANGFUSE_BASE_URL` (or the legacy `LANGFUSE_HOST`). |
+| `--run-metadata-json PATH` | A JSON object merged into the Langfuse **dataset-run** metadata. Per run, not per item — for what is constant across the run and otherwise only recoverable from the repo that launched it: the agent id and its personality text, the workspace actually evaluated, the memory snapshot in force. A payload that is not an object is rejected up front, since run metadata merges key by key and a list would be dropped after the run had already cost its agent calls. Merged under `--tag`, which wins on a `tags` key. |
 
 ##### Langfuse v4
 

packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py

@@ -5,6 +5,7 @@
 
 import sys
 import time
+from collections.abc import Mapping
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from typing import Any, TypedDict
@@ -419,6 +420,7 @@ def run_agentic_items(
     concurrency: int = 1,
     gate: EvalGate = DEFAULT_GATE,
     tags: tuple[str, ...] = (),
+    run_metadata: Mapping[str, Any] | None = None,
 ) -> EvalReport:
     """Run agentic items through evaluate_agentic_* and return an EvalReport.
 
@@ -431,10 +433,16 @@ def run_agentic_items(
     automatic ones. They reach Langfuse whenever trace linking runs, with or without
     ``use_langfuse``.
 
+    ``run_metadata`` is merged into the Langfuse dataset-run metadata, once per run rather
+    than once per item. ``tags`` wins on a "tags" key, since that one is derived here.
+
     Results are collected in dataset order regardless of completion order.
     """
     langfuse = make_langfuse_client() if use_langfuse else None
-    run_metadata_extra: dict[str, Any] | None = {"tags": list(tags)} if tags else None
+    extra: dict[str, Any] = dict(run_metadata or {})
+    if tags:
+        extra["tags"] = list(tags)
+    run_metadata_extra: dict[str, Any] | None = extra or None
 
     report = EvalReport(model=model_version, gate=normalize_gate(gate))
     total = len(items)

packages/gooddata-eval/src/gooddata_eval/cli/main.py

@@ -2,12 +2,13 @@
 """`gd-eval` command-line entry point."""
 
 import argparse
+import json
 import os
 import sys
 import threading
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import get_args
+from typing import Any, get_args
 
 import httpx
 from gooddata_api_client.exceptions import ApiException, ApiTypeError
@@ -78,6 +79,35 @@ def _build_parser() -> argparse.ArgumentParser:
     source = run.add_mutually_exclusive_group(required=True)
     source.add_argument("--dataset", help="Path to a folder of dataset JSON files.")
     source.add_argument("--langfuse-dataset", dest="langfuse_dataset", help="Langfuse dataset name.")
+    run.add_argument(
+        "--langfuse-item-id",
+        dest="langfuse_item_ids",
+        action="append",
+        default=[],
+        metavar="ITEM_ID",
+        help="Run only this Langfuse dataset item. Repeat for several. An id the dataset does "
+        "not have is an error, not a skip. Requires --langfuse-dataset.",
+    )
+    run.add_argument(
+        "--langfuse-item-metadata",
+        dest="langfuse_item_metadata",
+        action="append",
+        default=[],
+        type=_parse_item_metadata,
+        metavar="KEY=VALUE",
+        help="Run only Langfuse items whose metadata carries this key and value (e.g. "
+        "--langfuse-item-metadata language=es). Repeat to require several; values compare as "
+        "strings. Combined with --langfuse-item-id, an item must satisfy both. "
+        "Requires --langfuse-dataset.",
+    )
+    run.add_argument(
+        "--run-metadata-json",
+        dest="run_metadata_json",
+        metavar="PATH",
+        help="JSON object merged into the Langfuse dataset-run metadata. Per run, not per item "
+        "-- use it for what is constant across the run (agent id, personality, workspace, "
+        "memory snapshot id), not for anything that varies by question.",
+    )
     run.add_argument(
         "--kind",
         dest="kind",
@@ -396,6 +426,30 @@ def _parse_tag(val: str) -> str:
     return tag
 
 
+def _parse_item_metadata(val: str) -> tuple[str, str]:
+    """Split one --langfuse-item-metadata KEY=VALUE pair."""
+    key, sep, value = val.partition("=")
+    if not sep or not key.strip():
+        raise argparse.ArgumentTypeError(f"expected KEY=VALUE, got {val!r}")
+    return key.strip(), value.strip()
+
+
+def _load_run_metadata(path: str) -> dict[str, Any]:
+    """Read --run-metadata-json.
+
+    A non-object payload is rejected here rather than at the Langfuse call: dataset-run
+    metadata is merged key by key, so a list or a bare string has nowhere to go and would
+    otherwise be dropped silently after the run had already cost its agent calls.
+    """
+    try:
+        payload = json.loads(Path(path).read_text(encoding="utf-8"))
+    except json.JSONDecodeError as e:
+        raise ValueError(f"--run-metadata-json {path}: not valid JSON ({e})") from e
+    if not isinstance(payload, dict):
+        raise ValueError(f"--run-metadata-json {path}: expected a JSON object, got {type(payload).__name__}")
+    return payload
+
+
 def _make_progress_callbacks(console: Console):
     """Build (on_item_start, on_run_done, on_item_done) callbacks that stream progress.
 
@@ -446,7 +500,12 @@ def _load_dataset(config: RunConfig):
 
     if config.langfuse_dataset is None:  # pragma: no cover - argparse mutually-exclusive group guarantees one is set
         raise ValueError("Either --dataset or --langfuse-dataset is required.")
-    return load_langfuse_dataset(config.langfuse_dataset, default_test_kind=config.kind)
+    return load_langfuse_dataset(
+        config.langfuse_dataset,
+        default_test_kind=config.kind,
+        item_ids=config.item_ids or None,
+        metadata_filter=config.item_metadata_filter or None,
+    )
 
 
 def _list_models(host: str, token: str, workspace_id: str | None) -> int:
@@ -510,6 +569,15 @@ def _run(config: RunConfig) -> int:
             file=sys.stderr,
         )
         return _EXIT_OPERATIONAL_ERROR
+    if (config.item_ids or config.item_metadata_filter) and config.langfuse_dataset is None:
+        # Refused rather than ignored: a run that quietly evaluated the whole local folder
+        # after being asked for one language would be read as that language's pass rate.
+        print(
+            "error: --langfuse-item-id / --langfuse-item-metadata require --langfuse-dataset "
+            "(a local dataset folder has no item metadata to select on).",
+            file=sys.stderr,
+        )
+        return _EXIT_OPERATIONAL_ERROR
 
     items = _load_dataset(config)
     agentic_items = [i for i in items if i.test_kind in AGENTIC_TEST_KINDS]
@@ -595,6 +663,7 @@ def on_langfuse_item_done(
                     reasoning_effort=config.reasoning_effort,
                     use_langfuse=config.log_to_langfuse,
                     tags=config.tags,
+                    run_metadata=config.run_metadata,
                     run_ts=run_ts,
                     on_item_start=on_item_start,
                     on_item_done=on_item_done,
@@ -754,6 +823,9 @@ def main(argv: list[str] | None = None) -> int:
             turn_timeout_s=args.turn_timeout,
             item_timeout_s=args.item_timeout,
             tags=tuple(dict.fromkeys(args.tags)),
+            item_ids=tuple(dict.fromkeys(args.langfuse_item_ids)),
+            item_metadata_filter=dict(args.langfuse_item_metadata),
+            run_metadata=_load_run_metadata(args.run_metadata_json) if args.run_metadata_json else {},
         )
         return _run(config)
     except (

packages/gooddata-eval/src/gooddata_eval/core/config.py

@@ -4,7 +4,7 @@
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal, cast, get_args
+from typing import Any, Literal, cast, get_args
 
 # Values that mean "off" when read from the environment. Needed because every non-empty
 # string is truthy in Python, so a bare bool() on an env var reads FOO=0 -- the obvious way
@@ -99,3 +99,9 @@ class RunConfig:
     gate: EvalGate = DEFAULT_GATE
     tags: tuple[str, ...] = ()
     """Custom Langfuse trace tags, added after the automatic ones."""
+    item_ids: tuple[str, ...] = ()
+    """Langfuse dataset items to run, by id. Empty runs them all. Ignored for a local dataset."""
+    item_metadata_filter: dict[str, str] = field(default_factory=dict)
+    """Run only Langfuse items whose metadata carries all these key/values. Ignored for a local dataset."""
+    run_metadata: dict[str, Any] = field(default_factory=dict)
+    """Extra key/values written into the Langfuse dataset-run metadata, per run, not per item."""

packages/gooddata-eval/src/gooddata_eval/core/dataset/langfuse_source.py

@@ -12,6 +12,7 @@
   LANGFUSE_HOST         — base URL, legacy alias for LANGFUSE_BASE_URL
 """
 
+from collections.abc import Collection, Mapping
 from typing import Any, TypeVar, cast
 
 import httpx
@@ -24,6 +25,19 @@
 _T = TypeVar("_T")
 
 
+class DatasetItemNotFoundError(LookupError):
+    """Named item ids were asked for and the dataset does not have them.
+
+    Raised rather than silently returning the ids that did exist: asking for twelve items
+    and evaluating nine, with nothing said, reads downstream as a pass-rate change.
+    """
+
+    def __init__(self, dataset_name: str, missing: Collection[str]) -> None:
+        self.dataset_name = dataset_name
+        self.missing = sorted(missing)
+        super().__init__(f"Langfuse dataset {dataset_name!r} has no item(s) with id: {', '.join(self.missing)}")
+
+
 def _make_client() -> httpx.Client:
     """Build an httpx client with Langfuse basic-auth headers."""
     return make_http_client(timeout=30)
@@ -124,21 +138,52 @@ def _item_from_raw(raw: dict, *, dataset_name: str, test_kind: str) -> DatasetIt
         expected_output=expected_output,
         summary_input=_summary_input_from_raw(raw, expected_output),
         user_context=_user_context_from_raw(raw),
+        metadata=metadata if isinstance(metadata := raw.get("metadata"), dict) else {},
     )
 
 
-def load_langfuse_dataset(name: str, *, default_test_kind: str = "visualization") -> list[DatasetItem]:
-    """Pull all items from a Langfuse dataset by name via the REST API.
+def _matches_metadata(item: DatasetItem, metadata_filter: Mapping[str, str]) -> bool:
+    """True when the item's metadata carries every wanted key with the wanted value.
+
+    Values are compared as strings because the filter arrives from a CLI `KEY=VALUE` pair,
+    where everything is a string, while the item side may hold a number or a bool. A key
+    the item does not have never matches -- an item with no `language` is not an item in
+    every language.
+    """
+    return all(key in item.metadata and str(item.metadata[key]) == value for key, value in metadata_filter.items())
+
+
+def load_langfuse_dataset(
+    name: str,
+    *,
+    default_test_kind: str = "visualization",
+    item_ids: Collection[str] | None = None,
+    metadata_filter: Mapping[str, str] | None = None,
+) -> list[DatasetItem]:
+    """Pull items from a Langfuse dataset by name via the REST API.
+
+    Both filters are applied client-side, after the fetch: the Langfuse dataset-items
+    endpoint can select by dataset but not by item id or item metadata, so narrowing has
+    to happen here. The fetch cost is the whole dataset either way; what the filters save
+    is the part that actually costs -- evaluating an item against a live agent.
 
     Args:
         name: The Langfuse dataset name (as shown in the Langfuse UI).
         default_test_kind: Fallback test_kind when the item doesn't specify one.
+        item_ids: Keep only these item ids. Every id must exist, or it raises -- see
+            `DatasetItemNotFoundError`.
+        metadata_filter: Keep only items whose metadata carries all of these key/value
+            pairs, compared as strings.
+
+    Given both, an item has to satisfy both.
 
     Returns:
-        Parsed dataset items.
+        Parsed dataset items, in the order Langfuse returned them.
 
     Raises:
         RuntimeError: Missing Langfuse credentials or dataset not found.
+        ValueError: The dataset is empty, or the filters selected nothing from it.
+        DatasetItemNotFoundError: `item_ids` named an id the dataset does not have.
     """
     items: list[dict] = []
     page = 1
@@ -165,4 +210,21 @@ def load_langfuse_dataset(name: str, *, default_test_kind: str = "visualization"
     if not items:
         raise ValueError(f"Langfuse dataset '{name}' exists but contains no items.")
 
-    return [_item_from_raw(raw, dataset_name=name, test_kind=default_test_kind) for raw in items]
+    parsed = [_item_from_raw(raw, dataset_name=name, test_kind=default_test_kind) for raw in items]
+
+    if item_ids is not None:
+        wanted = set(item_ids)
+        if missing := wanted - {item.id for item in parsed}:
+            raise DatasetItemNotFoundError(name, missing)
+        parsed = [item for item in parsed if item.id in wanted]
+
+    if metadata_filter:
+        parsed = [item for item in parsed if _matches_metadata(item, metadata_filter)]
+
+    if not parsed:
+        raise ValueError(
+            f"Langfuse dataset '{name}' has items, but none matched the requested filter "
+            f"(metadata_filter={dict(metadata_filter or {})!r})."
+        )
+
+    return parsed

packages/gooddata-eval/src/gooddata_eval/core/models.py

@@ -483,3 +483,10 @@ class DatasetItem(BaseModel):
     # that schema (a discriminated union of view/widget descriptors), so re-modelling it here
     # would only create a second copy to keep in sync.
     user_context: dict[str, Any] | None = None
+    # The source's own item metadata, carried through verbatim. A Langfuse dataset item can
+    # hold whatever its publisher put there -- language, complexity, workspace id, the git
+    # revision the fixture was published from -- and `extra="ignore"` above used to drop all
+    # of it on load, so a caller could neither select on it nor report it. Not interpreted
+    # here: `test_kind`, `summary_input` and `user_context` are still read out into their own
+    # fields by the loader, and this is the rest of it.
+    metadata: dict[str, Any] = Field(default_factory=dict)

packages/gooddata-eval/tests/test_agentic_runner.py

@@ -1004,6 +1004,36 @@ def test_run_agentic_items_passes_custom_tags_as_run_metadata(tags, expected_ext
     assert mock_eval.call_args.kwargs["run_metadata_extra"] == expected_extra
 
 
+@pytest.mark.parametrize(
+    ("run_metadata", "tags", "expected_extra"),
+    [
+        # --run-metadata-json alone: reaches the dataset-run metadata with no tags key.
+        ({"agent_id": "a-1", "child_workspace_id": "ws-1"}, (), {"agent_id": "a-1", "child_workspace_id": "ws-1"}),
+        # Both: merged, not one replacing the other.
+        ({"agent_id": "a-1"}, ("nightly",), {"agent_id": "a-1", "tags": ["nightly"]}),
+        # A "tags" key in the JSON file loses -- tags is derived here from --tag.
+        ({"tags": ["from-json"]}, ("nightly",), {"tags": ["nightly"]}),
+        # Neither: unchanged from before this argument existed.
+        ({}, (), None),
+    ],
+)
+def test_run_agentic_items_merges_run_metadata_with_tags(run_metadata, tags, expected_extra):
+    with patch(
+        "gooddata_eval.cli.agentic_runner.evaluate_agentic_alert_skill",
+        return_value=AgenticEvalOutcome(detail={"alert_created": True}),
+    ) as mock_eval:
+        run_agentic_items(
+            [_item()],
+            host="http://host",
+            token="tok",
+            workspace_id="ws1",
+            run_ts="2026-01-01",
+            tags=tags,
+            run_metadata=run_metadata,
+        )
+    assert mock_eval.call_args.kwargs["run_metadata_extra"] == expected_extra
+
+
 class _FirstSend(BaseException):
     """Ends a run at its first chat turn; BaseException so no kind's error handling keeps it."""