diff --git a/packages/gooddata-eval/AGENTS.md b/packages/gooddata-eval/AGENTS.md index cbef9a528..8dbd76fda 100644 --- a/packages/gooddata-eval/AGENTS.md +++ b/packages/gooddata-eval/AGENTS.md @@ -4,15 +4,15 @@ `gdc-nas`) through a dataset of natural-language questions and scores what comes back, including side-by-side comparison across models. Each dataset item is a JSON envelope loaded from a local folder or pulled from a Langfuse dataset. Results are aggregated into -pass@K / pass^K reports and optionally pushed to Langfuse as scored traces tied to a -dataset run. The newest and most actively developed package in the repo. +pass@K / pass^K reports and optionally pushed to Langfuse as scored traces tied to an +experiment. The newest and most actively developed package in the repo. ## Owns - The `gd-eval` CLI (`gd-eval run`, `gd-eval models`) - Dataset loading and the evaluation run loop - Per-capability evaluators and their scoring -- Result reporting, and pushing runs, scores and trace links to Langfuse +- Result reporting, and pushing experiments, scores and trace links to Langfuse ## Does NOT Own @@ -29,7 +29,7 @@ dataset run. The newest and most actively developed package in the repo. | `core/summary/` | HTTP client for the dedicated dashboard-summary endpoint — a single-shot chat backend, not reporting | | `core/dataset/` | dataset format and loading | | `core/evaluators/` | single-shot evaluators and their registry | -| `core/langfuse/` | `sink.py` only — pushes single-turn scores and dataset-run items | +| `core/langfuse/` | the whole Langfuse v4 client: `_env` (base URL + credentials), `otlp` (OTLP/JSON encoding), `experiment` (root-span construction, score targets), `observations` (trace reads), `client` (httpx calls), `sink` (single-shot results as experiments) | | `core/reporting/` | console and JSON output rendering | | `core/scoring.py`, `core/runner.py` | scoring and orchestration | | `core/models.py` | `DatasetItem`, `ChatResult`, `ItemReport` and friends | diff --git a/packages/gooddata-eval/README.md b/packages/gooddata-eval/README.md index be4f55046..71a8cb911 100644 --- a/packages/gooddata-eval/README.md +++ b/packages/gooddata-eval/README.md @@ -113,7 +113,7 @@ gd-eval run \ | Flag | Description | |---|---| | `--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`, `LANGFUSE_HOST`. | +| `--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`). | | `--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. | #### Model selection @@ -152,15 +152,39 @@ 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`, `LANGFUSE_HOST`. | +| `--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`). | -Set `TAVERN_E2E_SKIP_TRACE_LINK=1` to skip trace lookup entirely (scores are then orphaned; the run says so). +##### Langfuse v4 -**A local `--dataset` cannot be attached to a Langfuse run.** `--langfuse` is refused alongside `--dataset` -because a local folder's item ids are not Langfuse dataset item ids. But trace linking does not depend on that -flag — each `evaluate_agentic_*` builds its own client whenever `LANGFUSE_PUBLIC_KEY`/`LANGFUSE_SECRET_KEY` are -exported — so a local run still finds its traces and writes its scores onto them, and only the per-run grouping -fails, with one `404 from dataset-run-items` reported per run. The run warns about this before it starts. Use +One run is one Langfuse **experiment**. Each evaluated item becomes its own trace whose root span carries the +experiment and dataset-item attributes (`langfuse.experiment.name`, `langfuse.experiment.dataset.id`, +`langfuse.experiment.item.id`), and the four scores attach to that root observation. gd-eval speaks to Langfuse +over four REST endpoints and uses no Langfuse SDK, so it runs on every Python version the package supports: + +| Endpoint | Used for | +|---|---| +| `POST /api/public/otel/v1/traces` | exporting the experiment root span as OTLP/HTTP JSON | +| `POST /api/public/scores` | one score per write, on a trace or on a single observation inside it | +| `GET /api/public/v2/observations` | finding the agent's gen-ai trace for a conversation | +| `GET /api/public/dataset-items` | loading `--langfuse-dataset` items, and resolving an item's dataset id | + +Two consequences of v4's immutable observations. gd-eval sets `version` only on its own experiment span, never on +the agent's gen-ai trace — filter on the gd-eval experiment's `langfuse.version` to compare models. And on the +agentic kinds the latency in `value_score` is the gen-ai trace's root generation latency, read from the +observations endpoint; the single-shot `--langfuse` sink keeps using the item's own measured average latency. + +Langfuse Cloud drops v3 on **2026-11-16**; a self-hosted Langfuse must be on v4 for any of this to work. + +Set `TAVERN_E2E_SKIP_TRACE_LINK=1` to turn the whole **agentic** Langfuse write path off — no trace lookup, no +span export and no scores for `agentic_*` items. The run says so once. It does not reach the `--langfuse` sink, +which still writes a span and four scores for every single-shot item; drop `--langfuse` to silence that too. + +**A local `--dataset` cannot be attached to a Langfuse experiment.** `--langfuse` is refused alongside +`--dataset` because a local folder's item ids are not Langfuse dataset item ids. But trace linking does not +depend on that flag — each `evaluate_agentic_*` builds its own client whenever +`LANGFUSE_PUBLIC_KEY`/`LANGFUSE_SECRET_KEY` are exported — so a local run still finds its traces and writes its +scores onto them, and only the per-run grouping fails: the dataset-item lookup 404s and the run reports the item +as one that does not exist in Langfuse, once. The run also warns about this before it starts. Use `--langfuse-dataset` when you want runs that are comparable across models, or `TAVERN_E2E_SKIP_TRACE_LINK=1` to skip linking altogether. @@ -380,6 +404,10 @@ Without `[llm-judge]`, those items are **skipped**. ## Scores (in JSON report and Langfuse) +In Langfuse every score is written to the experiment run's root observation — `traceId` plus `observationId` of +the item's own root span. On the agentic path each score is mirrored onto the agent's gen-ai trace as well +(`traceId` only), so a score survives even when one of the two traces is missing. + | Score | Description | |---|---| | `pass_at_k` | 1 if any of the K runs passed strict checks, else 0. | diff --git a/packages/gooddata-eval/src/gooddata_eval/cli/main.py b/packages/gooddata-eval/src/gooddata_eval/cli/main.py index 0a12cefa5..77dbc9dde 100644 --- a/packages/gooddata-eval/src/gooddata_eval/cli/main.py +++ b/packages/gooddata-eval/src/gooddata_eval/cli/main.py @@ -181,15 +181,14 @@ def _apply_timer_flag(enabled: bool) -> None: def _warn_if_local_dataset_cannot_link(config: RunConfig, agentic_items: list) -> None: - """Say up front that dataset-run assembly will fail, rather than after the run. + """Say up front that experiment assembly will fail, rather than after the run. --langfuse is refused outright with a local dataset because local item ids cannot be linked. But every evaluate_agentic_* falls back to try_make_langfuse_client() when the - caller passes none, so with LANGFUSE_* exported the linking runs anyway and each - conversation earns a 404 from dataset-run-items -- arriving in a block at the very end - of the run, long after the flag that would have prevented it could be changed. The - fallback is deliberate (direct library and tavern callers rely on it), so this warns - instead of disabling it. + caller passes none, so with LANGFUSE_* exported the linking runs anyway and every + dataset-item lookup 404s -- arriving in a block at the very end of the run, long after + the flag that would have prevented it could be changed. The fallback is deliberate + (direct library and tavern callers rely on it), so this warns instead of disabling it. """ from gooddata_eval.core.agentic._langfuse import SKIP_ENV_VAR, langfuse_credentials_present # noqa: PLC0415 from gooddata_eval.core.config import env_flag # noqa: PLC0415 @@ -201,7 +200,7 @@ def _warn_if_local_dataset_cannot_link(config: RunConfig, agentic_items: list) - print( f"warning: --dataset is a local folder, so its item ids are not Langfuse dataset item ids. " f"Traces will be found and scored, but the per-run grouping that makes models comparable " - f"cannot be created and each conversation will report a 404 from dataset-run-items. " + f"cannot be created and each conversation will report that its item does not exist in Langfuse. " f"Use --langfuse-dataset for comparable runs, or set {SKIP_ENV_VAR}=1 to skip trace linking.", file=sys.stderr, ) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py index 9394068df..f6e9ff5b5 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging +import os import threading import time from collections.abc import Iterator @@ -11,15 +12,20 @@ from datetime import datetime, timedelta, timezone from typing import Any -import httpx - from gooddata_eval.core.agentic._trace_linker import link_cancel_event, linking_is_inline, warn_from_worker from gooddata_eval.core.config import ReasoningEffort, env_flag, normalize_reasoning_effort from gooddata_eval.core.langfuse._env import credentials_present from gooddata_eval.core.langfuse.client import HttpxLangfuseClient +from gooddata_eval.core.langfuse.experiment import ( + ExperimentItem, + ExperimentRun, + ScoreTarget, + build_experiment_root_span, +) # Part of this module's public surface: external callers import both names from here. from gooddata_eval.core.langfuse.observations import TraceSummary as _TraceObj # noqa: F401 +from gooddata_eval.core.langfuse.otlp import Span _log = logging.getLogger(__name__) @@ -50,10 +56,10 @@ def try_make_langfuse_client() -> HttpxLangfuseClient | None: SKIP_ENV_VAR = "TAVERN_E2E_SKIP_TRACE_LINK" -# Run names whose dataset-run assembly has already been reported as impossible. A 404 from -# dataset-run-items means the dataset item id is not in Langfuse, which is a property of -# the dataset and not of the attempt -- so it recurs identically for every item and every -# run of that dataset, and reporting it per conversation buries the run's real output under +# Run names whose experiment assembly has already been reported as impossible. A 404 from +# dataset-items means the dataset item id is not in Langfuse, which is a property of the +# dataset and not of the attempt -- so it recurs identically for every item and every run +# of that dataset, and reporting it per conversation buries the run's real output under # dozens of copies of the same HTTP error. Guarded by a lock because linking runs on the # drain pool's worker threads. _UNLINKABLE_RUNS: set[str] = set() @@ -166,6 +172,12 @@ def _fetch_traces_for_session( _CANCEL_CHECK_SEC = 0.5 +def _drain_is_cancelled() -> bool: + """Whether the batched drain running on this thread has been interrupted.""" + cancel = link_cancel_event() + return cancel is not None and cancel.is_set() + + def _wait_between_attempts(delay: float) -> bool: """Wait ``delay`` before the next poll attempt. False means "stop polling". @@ -200,19 +212,19 @@ def find_traces_per_conversation( ``window_end`` bounds the trace query and should be pinned by the caller to the moment the conversations ended. It matters because this poll is normally deferred onto a worker thread (see ``agentic/_trace_linker.py``): defaulting it to "now" would stretch - the window by however long the task waited in the queue, and since - ``_fetch_traces_for_session`` pages at ``_FETCH_LIMIT`` and filters by session locally, - a wide enough window can push the wanted trace off the page. Defaults to now only for - direct callers that poll immediately. + the window by however long the task waited in the queue, and the reader pages at 500 + rows over at most four pages and then keeps the newest ``_FETCH_LIMIT`` traces of what + that returns, so a wide enough window can push the wanted trace off the page. Defaults + to now only for direct callers that poll immediately. """ if env_flag(SKIP_ENV_VAR): - # Say so. Skipping returns all-None, which downstream renders as observe()'s - # generic "No trace found for dataset run ...; scores will be orphaned" -- the - # same message a real lookup failure produces. Left silent, an eval run looks - # like Langfuse is broken when trace linking was simply switched off. + # The one place the switch is announced: ``observe`` also honours it but stays + # silent, so a run says once that its Langfuse work was turned off rather than + # once per item. Left unsaid entirely, an eval run with no scores looks like + # Langfuse is broken. warn_from_worker( f"[langfuse] trace linking SKIPPED by {SKIP_ENV_VAR}: " - f"{len(conversation_ids)} conversation(s) will have orphaned scores. " + f"{len(conversation_ids)} conversation(s) will not be linked or scored in Langfuse. " f"Unset it to link traces." ) return dict.fromkeys(conversation_ids) @@ -227,8 +239,7 @@ def find_traces_per_conversation( stop_at = deadline if deadline is not None else time.monotonic() + budget for cid in conversation_ids: - cancel = link_cancel_event() - if cancel is not None and cancel.is_set(): + if _drain_is_cancelled(): # The run is being interrupted; the remaining conversations are not worth a # round trip, and their scores were never going to be written. break @@ -258,13 +269,67 @@ def find_traces_per_conversation( return by_conv -def _set_trace_version(langfuse: Any, trace_id: str, version: str) -> None: - """Write model version into the Langfuse trace version field.""" - try: - if hasattr(langfuse, "update_trace_version"): - langfuse.update_trace_version(trace_id, version) - except Exception as exc: - _log.warning("Failed to set trace version %r on %s: %s", version, trace_id, exc) +def _span_window(trace: Any, window: tuple[datetime, datetime] | None) -> tuple[datetime, datetime]: + """When gd-eval's span starts and ends: the gen-ai turn it describes, else the item's + trace window, else this instant. + + The only clock read on the linking path, and deliberately here rather than in the + deferred task, whose run time says nothing about when the agent answered. + """ + start = getattr(trace, "start_time", None) + end = getattr(trace, "end_time", None) + if start is not None and end is not None: + return start, end + if window is not None: + return window + now = datetime.now(timezone.utc) + return now, now + + +def _experiment_root_span( + trace_id: str | None, + dataset_item_id: str, + dataset_id: str, + run_name: str, + run_metadata: dict[str, Any], + *, + trace: Any, + window: tuple[datetime, datetime] | None, + conversation_id: str | None, + item_input: Any, + output: Any, +) -> Span: + """gd-eval's own root span for one (dataset item, run) -- the whole experiment item.""" + start, end = _span_window(trace, window) + session_id = conversation_id or getattr(trace, "session_id", None) + tags = tuple(tag for tag in ("gd-eval", run_metadata.get("testing_framework")) if tag) + # One predicate for both the name and the input, so a falsy question cannot name the span + # after the item while the input still says "question". + has_input = item_input is not None + return build_experiment_root_span( + ExperimentRun(run_name, dataset_id, run_metadata or None), + ExperimentItem( + dataset_item_id, + input={"question": item_input} if has_input else {"dataset_item_id": dataset_item_id}, + output=output, + ), + start=start, + end=end, + # Never the gen-ai trace's own name: the two traces sit side by side in Langfuse and + # only the prefix says which of them gd-eval wrote. + trace_name=f"gd-eval: {str(item_input)[:80]}" if has_input else f"gd-eval: {dataset_item_id}", + session_id=session_id, + version=run_metadata.get("model_version"), + tags=tags, + observation_metadata={ + "gen_ai_trace_id": trace_id, + "gen_ai_latency_s": getattr(trace, "latency", None), + "gen_ai_cost_usd": getattr(trace, "total_cost", None), + "conversation_id": session_id, + }, + trace_metadata={"run_name": run_name}, + environment=os.environ.get("LANGFUSE_TRACING_ENVIRONMENT"), + ) @contextmanager @@ -274,64 +339,117 @@ def observe( dataset_item_id: str, run_name: str, run_metadata: dict[str, Any] | None = None, -) -> Iterator[str | None]: - """Create a Langfuse dataset run item and yield the trace_id.""" - if trace_id is not None: - try: - langfuse.api.dataset_run_items.create( - run_name=run_name, - dataset_item_id=dataset_item_id, - trace_id=trace_id, - metadata=run_metadata or {}, - run_description="", - ) - _log.debug( - "[langfuse] Created dataset run item: run=%s trace=%s item=%s", run_name, trace_id, dataset_item_id - ) - except httpx.HTTPStatusError as exc: - if exc.response.status_code != 404: - _log.warning("Failed to link trace %s to run %s: %s", trace_id, run_name, exc) - warn_from_worker( - f"[langfuse] WARNING: failed to create dataset run item " - f"run={run_name} trace={trace_id} item={dataset_item_id}: {exc}" - ) - elif _first_report_for_run(run_name): - # Say what it means and what to do, once. A raw 404 names an endpoint, - # which tells the reader nothing about the cause being their --dataset. - _log.warning( - "Dataset item %s is not in Langfuse; run %s cannot be assembled.", dataset_item_id, run_name - ) - warn_from_worker( - f"[langfuse] WARNING: dataset item {dataset_item_id!r} does not exist in Langfuse, " - f"so the run {run_name!r} cannot be assembled (404 from dataset-run-items). " - f"Scores ARE still written to the traces themselves -- only the per-run grouping " - f"used to compare models is missing. This is what happens when --dataset points at " - f"a local folder: its item ids are local, not Langfuse dataset item ids. Use " - f"--langfuse-dataset to get comparable runs, or set {SKIP_ENV_VAR}=1 to skip linking " - f"altogether. Further occurrences for this run are suppressed." - ) - except Exception as exc: - _log.warning("Failed to link trace %s to run %s: %s", trace_id, run_name, exc) + *, + trace: Any = None, + window: tuple[datetime, datetime] | None = None, + conversation_id: str | None = None, + item_input: Any = None, + output: Any = None, +) -> Iterator[ScoreTarget | None]: + """Export gd-eval's experiment root span for one run and yield where to score it. + + A run belongs to a Langfuse experiment through the attributes on that span, so the span + IS the run item. The yielded target names both the gen-ai trace and the span; either + half may be missing, and ``score_safe`` writes to whichever are there. + + The yielded value is a ``ScoreTarget``: a ``str`` equal to the gen-ai trace id when one + was found, else to gd-eval's own experiment trace id. A caller that writes scores + directly with ``create_score(trace_id=tid)`` therefore scores that one trace and no + other; pass the target to ``score_safe(langfuse, tid, ...)`` to reach both destinations. + """ + fallback = ScoreTarget(trace_id) if trace_id else None + # isinstance, not hasattr: a MagicMock answers every attribute, and the skill suites + # hand this one exactly that. + if not isinstance(langfuse, HttpxLangfuseClient): + yield fallback + return + + # The operational off-switch for the whole agentic Langfuse write path, so it has to + # cover the span export and the scores as well as the poll that already announced it. + # Yielding None rather than the trace id is what stops ``score_safe`` writing: the + # flag is also how an operator gets through a run with Langfuse unreachable, and a + # score write would then fail per item. + if env_flag(SKIP_ENV_VAR): + yield None + return + + if _drain_is_cancelled(): + yield None + return + + try: + dataset_id = langfuse.dataset_id_for_item(dataset_item_id) + except Exception as exc: + _log.warning("Failed to resolve dataset item %s for run %s: %s", dataset_item_id, run_name, exc) + warn_from_worker( + f"[langfuse] WARNING: failed to resolve dataset item {dataset_item_id!r} for run {run_name!r}: {exc}" + ) + yield fallback + return + + if dataset_id is None: + if _first_report_for_run(run_name): + # Say what it means and what to do, once. A raw 404 names an endpoint, which + # tells the reader nothing about the cause being their --dataset. + _log.warning("Dataset item %s is not in Langfuse; run %s cannot be assembled.", dataset_item_id, run_name) warn_from_worker( - f"[langfuse] WARNING: failed to create dataset run item " - f"run={run_name} trace={trace_id} item={dataset_item_id}: {exc}" + f"[langfuse] WARNING: dataset item {dataset_item_id!r} does not exist in Langfuse, " + f"so the run {run_name!r} cannot be assembled (404 from dataset-items). " + f"Scores ARE still written to the traces themselves -- only the per-run grouping " + f"that makes models comparable is missing. This is what happens when --dataset points at " + f"a local folder: its item ids are local, not Langfuse dataset item ids. Use " + f"--langfuse-dataset to get comparable runs, or set {SKIP_ENV_VAR}=1 to skip linking " + f"altogether. Further occurrences for this run are suppressed." ) - model_version = (run_metadata or {}).get("model_version") - if model_version: - _set_trace_version(langfuse, trace_id, model_version) - else: - _log.warning("No trace found for dataset run %s; scores will be orphaned.", run_name) - yield trace_id + yield fallback + return + span = _experiment_root_span( + trace_id, + dataset_item_id, + dataset_id, + run_name, + run_metadata or {}, + trace=trace, + window=window, + conversation_id=conversation_id, + item_input=item_input, + output=output, + ) + try: + langfuse.export_spans([span]) + except Exception as exc: + _log.warning("Failed to export the experiment span for run %s: %s", run_name, exc) + warn_from_worker( + f"[langfuse] WARNING: failed to export experiment span run={run_name} item={dataset_item_id}: {exc}" + ) + yield fallback + return + + if trace_id is None: + _log.warning("No gen-ai trace found for run %s; scores go to the gd-eval experiment span only.", run_name) + yield ScoreTarget(trace_id, span.trace_id, span.span_id) + + +def score_safe(langfuse: Any, trace_id: Any, **kwargs: Any) -> None: + """Create one Langfuse score per destination the target names, ignoring errors. -def score_safe(langfuse: Any, trace_id: str | None, **kwargs: Any) -> None: - """Create a Langfuse score, ignoring errors.""" + ``observation_id`` is sent only for the experiment span, so the gen-ai write stays the + call every client already accepts. + """ if not trace_id: return - try: - langfuse.create_score(trace_id=trace_id, **kwargs) - except Exception as exc: - _log.warning("Failed to log score %s: %s", kwargs.get("name"), exc) + # ``create_score`` answers a throttled write by sleeping and trying again, so an + # interrupted drain has to stop short of the call rather than wait its retries out. + if _drain_is_cancelled(): + return + targets = trace_id.destinations() if isinstance(trace_id, ScoreTarget) else [(str(trace_id), None)] + for target_id, observation_id in targets: + extra = {"observation_id": observation_id} if observation_id else {} + try: + langfuse.create_score(trace_id=target_id, **kwargs, **extra) + except Exception as exc: + _log.warning("Failed to log score %s: %s", kwargs.get("name"), exc) def log_quality_and_value_scores( diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_trace_linker.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_trace_linker.py index ce4bc1a8f..de77ca445 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_trace_linker.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_trace_linker.py @@ -142,6 +142,8 @@ class RunTraceContext: _base_name: str _suffix_runs: bool _traces: dict[str, Any] + _window: tuple[datetime, datetime] | None = None + _item_input: Any = None def run_name(self, run_idx: int) -> str: """Dataset-run name for one run, suffixed only when the item has more than one.""" @@ -151,12 +153,13 @@ def trace(self, conversation_id: str) -> Any: """The trace picked for a conversation, or None when the poll never found one.""" return self._traces.get(conversation_id) - def observe(self, trace: Any, run_idx: int) -> Any: - """Attach this run to its dataset-run item, yielding the trace id to score against. + def observe(self, trace: Any, run_idx: int, *, conversation_id: str | None = None, output: Any = None) -> Any: + """Attach this run to its experiment item, yielding what to write its scores against. ``trace`` may be None -- a conversation whose trace never showed up is still - observed, so the run appears in the experiment with its scores orphaned rather than - missing entirely. + observed, so the run appears in the experiment scored against gd-eval's own span + rather than missing entirely. The window and the item's question travel on the + context, so a kind's scoring block never resolves them itself. """ return self._lf.observe( self._client, @@ -164,6 +167,11 @@ def observe(self, trace: Any, run_idx: int) -> Any: self._dataset_item_id, self.run_name(run_idx), self.run_metadata, + trace=trace, + window=self._window, + conversation_id=conversation_id, + item_input=self._item_input, + output=output, ) def score(self, trace_id: Any, *, name: str, value: Any, data_type: str) -> None: @@ -188,6 +196,7 @@ def submit_trace_scoring( window_end: datetime, suffix_runs: bool, write_scores: Callable[[RunTraceContext], None], + item_input: Any = None, ) -> None: """Defer one item's whole Langfuse block: resolve its run context, then write scores. @@ -212,7 +221,17 @@ def _link_traces() -> None: ) traces = _langfuse.find_traces_per_conversation(langfuse, conversation_ids, window_start, window_end) write_scores( - RunTraceContext(run_metadata, _langfuse, langfuse, dataset_item_id, base_name, suffix_runs, traces) + RunTraceContext( + run_metadata, + _langfuse, + langfuse, + dataset_item_id, + base_name, + suffix_runs, + traces, + (window_start, window_end), + item_input, + ) ) submit_trace_link(_link_traces, item_id=dataset_item_id) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/alert_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/alert_skill.py index cb1addf3d..330dea31e 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/alert_skill.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/alert_skill.py @@ -838,7 +838,7 @@ def _write_scores(ctx: RunTraceContext) -> None: "attributes_correct": ev.attributes_correct, "granularity_correct": ev.granularity_correct, } - with ctx.observe(pt, run_idx) as tid: + with ctx.observe(pt, run_idx, conversation_id=run.conversation_id, output=strict_checks) as tid: for score_name, value in strict_checks.items(): ctx.score(tid, name=score_name, value=float(value), data_type="BOOLEAN") ctx.quality( @@ -868,6 +868,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) runs_passed = sum(1 for r in summary.run_results if r.eval.strict_pass) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/conversation.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/conversation.py index 919e4022c..6965cbb82 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/conversation.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/conversation.py @@ -590,7 +590,15 @@ def evaluate_agentic_conversation( def _write_scores(ctx: RunTraceContext) -> None: pt = ctx.trace(result.conversation_id) - with ctx.observe(pt, 0) as tid: + with ctx.observe( + pt, + 0, + conversation_id=result.conversation_id, + output={ + "conversation_success": result.conversation_success, + "full_skill_coverage": result.full_skill_coverage, + }, + ) as tid: ctx.score( tid, name="conversation_success", @@ -640,6 +648,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=False, write_scores=_write_scores, + item_input=fixture.turns[0].message if fixture.turns else fixture.id, ) detail = _conversation_detail(result) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/general_question.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/general_question.py index ae40fafd5..fefb6c888 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/general_question.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/general_question.py @@ -257,7 +257,9 @@ def _write_scores(ctx: RunTraceContext) -> None: # never returned. continue pt = ctx.trace(run.conversation_id) - with ctx.observe(pt, run_idx) as tid: + with ctx.observe( + pt, run_idx, conversation_id=run.conversation_id, output={"general_question_pass": run.passed} + ) as tid: ctx.score(tid, name="general_question_pass", value=float(run.passed), data_type="BOOLEAN") ctx.score(tid, name="llm_judge_score", value=run.llm_judge_score, data_type="NUMERIC") ctx.quality( @@ -289,6 +291,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) item_timings = sum_timings([r.timings for r in summary.run_results]) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/guardrail.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/guardrail.py index ff709c61d..ffb5c8d56 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/guardrail.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/guardrail.py @@ -229,7 +229,9 @@ def _write_scores(ctx: RunTraceContext) -> None: # write a 0 the judge never returned. continue pt = ctx.trace(run.conversation_id) - with ctx.observe(pt, run_idx) as tid: + with ctx.observe( + pt, run_idx, conversation_id=run.conversation_id, output={"guardrail_pass": run.passed} + ) as tid: ctx.score(tid, name="guardrail_pass", value=float(run.passed), data_type="BOOLEAN") ctx.score(tid, name="llm_judge_score", value=run.llm_judge_score, data_type="NUMERIC") ctx.quality( @@ -261,6 +263,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) unscored = summary.judge_errors diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py index 9eda50d68..0b5122d23 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -428,7 +428,7 @@ def _write_scores(ctx: RunTraceContext) -> None: _log.info( "[kda-report] %s: strict_pass=%s latency_sec=%s", run_name, ev.strict_pass, turn_wall_clock_sec ) - with ctx.observe(pt, run_idx) as tid: + with ctx.observe(pt, run_idx, conversation_id=run.conversation_id, output=strict_checks) as tid: for score_name, value in strict_checks.items(): ctx.score(tid, name=score_name, value=float(value), data_type="BOOLEAN") ctx.score(tid, name="kda_disambiguated", value=float(ev.disambiguated), data_type="BOOLEAN") @@ -467,6 +467,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) runs_passed = sum(1 for r in summary.run_results if r.evaluation.strict_pass) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/metric_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/metric_skill.py index 7d8f18454..6bd960873 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/metric_skill.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/metric_skill.py @@ -448,7 +448,12 @@ def _write_scores(ctx: RunTraceContext) -> None: for run_idx, run in enumerate(summary.run_results): pt = ctx.trace(run.conversation_id) - with ctx.observe(pt, run_idx) as tid: + with ctx.observe( + pt, + run_idx, + conversation_id=run.conversation_id, + output={"metric_created": run.metric_created, "maql_correct": run.maql_correct}, + ) as tid: ctx.score(tid, name="metric_created", value=float(run.metric_created), data_type="BOOLEAN") ctx.score(tid, name="maql_correct", value=float(run.maql_correct), data_type="BOOLEAN") ctx.quality( @@ -478,6 +483,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) item_timings = sum_timings([r.timings for r in summary.run_results]) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/search_tool.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/search_tool.py index c3a70f81c..cf406f5dd 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/search_tool.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/search_tool.py @@ -205,7 +205,9 @@ def _write_scores(ctx: RunTraceContext) -> None: for run_idx, run in enumerate(summary.run_results): pt = ctx.trace(run.conversation_id) - with ctx.observe(pt, run_idx) as tid: + with ctx.observe( + pt, run_idx, conversation_id=run.conversation_id, output={"tool_selection": run.tool_selected} + ) as tid: ctx.score(tid, name="tool_selection", value=float(run.tool_selected), data_type="BOOLEAN") ctx.score(tid, name="tool_correctness", value=float(run.tool_correct), data_type="BOOLEAN") ctx.quality( @@ -235,6 +237,7 @@ def _write_scores(ctx: RunTraceContext) -> None: window_end=window_end, suffix_runs=len(summary.run_results) > 1, write_scores=_write_scores, + item_input=question, ) runs_passed = sum(1 for r in summary.run_results if r.tool_selected) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/visualization.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/visualization.py index 12faaaffb..159dea564 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/visualization.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/visualization.py @@ -359,7 +359,14 @@ def _write_scores(ctx: RunTraceContext) -> None: for run_idx, run in enumerate(summary.run_results): pt = ctx.trace(run.conversation_id) ev = run.eval_result - with ctx.observe(pt, run_idx) as tid: + strict_checks = { + "assertion-cross-ref-valid": ev.cross_ref_valid, + "assertion-vis-metric": ev.metrics_correct, + "assertion-vis-dimensions": ev.dimensions_correct, + "assertion-vis-filters": ev.filters_correct, + "assertion-vis-type": ev.viz_type_hard, + } + with ctx.observe(pt, run_idx, conversation_id=run.conversation_id, output=strict_checks) as tid: ctx.score(tid, name="assertion-cross-ref-valid", value=ev.cross_ref_valid, data_type="BOOLEAN") ctx.score(tid, name="assertion-vis-metric", value=ev.metrics_correct, data_type="BOOLEAN") ctx.score(tid, name="assertion-vis-dimensions", value=ev.dimensions_correct, data_type="BOOLEAN") @@ -372,13 +379,7 @@ def _write_scores(ctx: RunTraceContext) -> None: ctx.score(tid, name="steps", value=run.total_steps, data_type="NUMERIC") ctx.quality( tid, - strict_checks={ - "assertion-cross-ref-valid": ev.cross_ref_valid, - "assertion-vis-metric": ev.metrics_correct, - "assertion-vis-dimensions": ev.dimensions_correct, - "assertion-vis-filters": ev.filters_correct, - "assertion-vis-type": ev.viz_type_hard, - }, + strict_checks=strict_checks, latency_sec=pt.latency if pt else None, cost_usd=pt.total_cost if pt else None, ) @@ -404,6 +405,7 @@ def _write_scores(ctx: RunTraceContext) -> None: # Unlike the other runners, this one suffixes every run, K=1 included. suffix_runs=True, write_scores=_write_scores, + item_input=question, ) if record_output_path and summary.best.actual_output is not None: diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/client.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/client.py index 138719b4c..1441457ec 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/client.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/client.py @@ -15,12 +15,16 @@ import httpx from gooddata_eval.core.langfuse import _env, observations, otlp +from gooddata_eval.core.langfuse.experiment import ( + ExperimentItem, + ExperimentRun, + ScoreTarget, + build_experiment_root_span, +) from gooddata_eval.core.langfuse.observations import TraceSummary _SCORES_PATH = "/api/public/scores" _OTLP_PATH = "/api/public/otel/v1/traces" -_INGESTION_PATH = "/api/public/ingestion" -_DATASET_RUN_ITEMS_PATH = "/api/public/dataset-run-items" _MAX_SCORE_ATTEMPTS = 3 _DEFAULT_RETRY_DELAY = 0.5 @@ -75,8 +79,10 @@ def list( class _DatasetRunItemsAPI: - def __init__(self, client: httpx.Client) -> None: - self._client = client + """`api.dataset_run_items.create` for external callers on the dataset-run vocabulary: a v4 run is one span.""" + + def __init__(self, owner: HttpxLangfuseClient) -> None: + self._owner = owner def create( self, @@ -85,23 +91,36 @@ def create( trace_id: str, metadata: dict | None = None, run_description: str = "", - ) -> None: - self._client.post( - _DATASET_RUN_ITEMS_PATH, - json={ - "runName": run_name, - "datasetItemId": dataset_item_id, - "traceId": trace_id, - "metadata": metadata or {}, - "runDescription": run_description, - }, - ).raise_for_status() + ) -> ScoreTarget: + """Export one experiment root span for `dataset_item_id` and return where to score it. + + The span is its own trace, so `trace_id` is carried as metadata and is NOT what an + experiment-item score attaches to -- Langfuse reads those off the root observation. + The returned target names both, and `score_safe` writes to each; scoring the bare + `trace_id` instead reaches the gen-ai trace only and leaves the run item unscored. + """ + dataset_id = self._owner.dataset_id_for_item(dataset_item_id) + if dataset_id is None: + raise LookupError(f"dataset item {dataset_item_id!r} not found in Langfuse") + now = datetime.now(timezone.utc) + span = build_experiment_root_span( + ExperimentRun(run_name, dataset_id, metadata, run_description or None), + ExperimentItem(dataset_item_id, input={"dataset_item_id": dataset_item_id}), + start=now, + end=now, + trace_name=f"gd-eval: {dataset_item_id}", + tags=("gd-eval",), + observation_metadata={"gen_ai_trace_id": trace_id}, + trace_metadata={"run_name": run_name}, + ) + self._owner.export_spans([span]) + return ScoreTarget(trace_id, span.trace_id, span.span_id) class _LangfuseAPI: def __init__(self, owner: HttpxLangfuseClient) -> None: self.trace = _TraceAPI(owner) - self.dataset_run_items = _DatasetRunItemsAPI(owner._http) + self.dataset_run_items = _DatasetRunItemsAPI(owner) class HttpxLangfuseClient: @@ -127,8 +146,9 @@ def create_score( "id": str(uuid.uuid4()), "traceId": trace_id, "name": name, - # BOOLEAN scores go over the wire as 1.0/0.0, not as JSON booleans. - "value": (1.0 if value else 0.0) if isinstance(value, bool) else value, + # A BOOLEAN score goes over the wire as 1.0/0.0 whatever its Python type: the + # sink's compute_scores yields int 1/0 and the agentic path float 1.0/0.0. + "value": (1.0 if value else 0.0) if data_type == "BOOLEAN" else value, "dataType": data_type, } if comment: @@ -178,23 +198,6 @@ def list_traces( self._http, from_time=from_time, to_time=to_time, limit=limit, session_id=session_id ) - def update_trace_version(self, trace_id: str, version: str) -> None: - """Upsert the trace version field via the ingestion endpoint.""" - now = datetime.now(timezone.utc).isoformat() - self._http.post( - _INGESTION_PATH, - json={ - "batch": [ - { - "id": str(uuid.uuid4()), - "timestamp": now, - "type": "trace-create", - "body": {"id": trace_id, "version": version}, - } - ] - }, - ).raise_for_status() - def flush(self) -> None: pass # no client-side batching diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/observations.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/observations.py index 8e2d5fbe6..51686b8fe 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/observations.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/observations.py @@ -82,15 +82,24 @@ def list_traces_in_window( to_time: Any, limit: int, session_id: str | None, - page_size: int = 500, - max_pages: int = 4, + page_size: int | None = None, + max_pages: int | None = None, ) -> list[TraceSummary]: """List up to ``limit`` traces whose observations start inside the window, newest first. ``session_id`` is sent whenever it is not None -- an empty id is a real filter value that matches nothing, and dropping it would return the whole window for the caller to throw away, page after page. + + It also sets how deep the read goes. A filtered window holds one conversation and is + exhausted in a page or two; an unfiltered one holds every trace the workspace produced + in the same minutes, so it is read at the API's maximum page and twice as many pages. """ + if page_size is None: + page_size = 500 if session_id is not None else 1000 + if max_pages is None: + max_pages = 4 if session_id is not None else 8 + params: dict[str, Any] = { "fromStartTime": _iso(from_time), "toStartTime": _iso(to_time), diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py index ed94a89c0..2e605f734 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py @@ -19,8 +19,6 @@ ATTR_OBSERVATION_TYPE = "langfuse.observation.type" ATTR_OBSERVATION_INPUT = "langfuse.observation.input" ATTR_OBSERVATION_OUTPUT = "langfuse.observation.output" -ATTR_OBSERVATION_LEVEL = "langfuse.observation.level" -ATTR_OBSERVATION_STATUS_MESSAGE = "langfuse.observation.status_message" ATTR_OBSERVATION_METADATA_PREFIX = "langfuse.observation.metadata" # Trace-wide attributes, copied onto every span of the trace. diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py index 2058fb397..49fc47fcc 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py @@ -1,25 +1,25 @@ # (C) 2026 GoodData Corporation -"""Langfuse scoring sink — posts evaluation results via the Langfuse REST API.""" +"""Langfuse scoring sink — writes single-shot evaluation results as Langfuse experiments over OTLP.""" from __future__ import annotations -import base64 import os import sys -import uuid -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Any -import httpx - -from gooddata_eval.core.langfuse._env import resolve_base_url +from gooddata_eval.core.langfuse.client import HttpxLangfuseClient +from gooddata_eval.core.langfuse.experiment import ExperimentItem, ExperimentRun, build_experiment_root_span _MAX_LATENCY_S = 60.0 _QUALITY_WEIGHT = 0.6 _SPEED_WEIGHT = 0.2 if TYPE_CHECKING: + import httpx + from gooddata_eval.core.config import ReasoningEffort + from gooddata_eval.core.langfuse.otlp import Span from gooddata_eval.core.runner import ItemReport @@ -49,7 +49,7 @@ def compute_scores( class LangfuseSink: - """Posts evaluation results to Langfuse via the ingestion REST API.""" + """Writes evaluation results to Langfuse as an experiment root span plus four scores.""" def __init__( self, @@ -58,74 +58,112 @@ def __init__( model_id: str = "", provider_type: str = "", reasoning_effort: ReasoningEffort | None = None, + *, + transport: httpx.BaseTransport | None = None, ): self._dataset_name = dataset_name self._run_name = run_name self._model_id = model_id self._provider_type = provider_type self._reasoning_effort = reasoning_effort - host = resolve_base_url() - pub = os.environ.get("LANGFUSE_PUBLIC_KEY", "") - sec = os.environ.get("LANGFUSE_SECRET_KEY", "") - if not pub or not sec: - raise RuntimeError( - "Langfuse credentials not set. Export LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY to use --langfuse." + self._client = HttpxLangfuseClient(timeout=10.0, transport=transport) + self._warned_unlinkable = False + + def _resolve_run(self, report: ItemReport, dataset_item_id: str) -> ExperimentRun | None: + """The experiment this item belongs to, or None when it cannot be assembled. + + A lookup exception is reported every time (it may be transient); the item genuinely + not being a Langfuse dataset item is reported once per sink, since it recurs + identically for every item of the same --dataset. + """ + dataset_id: str | None = None + try: + dataset_id = self._client.dataset_id_for_item(dataset_item_id) + except Exception as exc: + print(f"warning: Langfuse dataset item lookup failed for item '{report.id}': {exc}", file=sys.stderr) + return None + if dataset_id is None: + if not self._warned_unlinkable: + self._warned_unlinkable = True + print( + f"warning: Langfuse dataset item '{dataset_item_id}' not found; " + f"run '{self._run_name}' is not assembled as an experiment", + file=sys.stderr, + ) + return None + description = ( + f"{self._provider_type}/{self._model_id}" + if self._provider_type and self._model_id + else self._model_id or None + ) + return ExperimentRun( + self._run_name, + dataset_id, + { + "model": self._model_id, + "provider_type": self._provider_type, + "reasoning_effort": self._reasoning_effort, + }, + description=description, + ) + + def _build_span(self, report: ItemReport, dataset_item_id: str, run: ExperimentRun | None) -> Span: + end = datetime.now(timezone.utc) + start = end - timedelta(seconds=report.avg_latency_s) + # "gd-eval" leads, exactly as on the agentic path, so one tag filter in Langfuse + # finds both single-shot and agentic traces this package wrote. + tags = tuple( + t + for t in ( + "gd-eval", + report.test_kind, + self._provider_type, + f"effort-{self._reasoning_effort.lower()}" if self._reasoning_effort else None, ) - creds = base64.b64encode(f"{pub}:{sec}".encode()).decode() - self._host = host - self._auth_header = f"Basic {creds}" + if t + ) + return build_experiment_root_span( + run, + ExperimentItem( + dataset_item_id, + input={"question": report.question}, + output=report.best_detail, + metadata={"test_kind": report.test_kind}, + ), + start=start, + end=end, + trace_name=f"gd-eval: {report.question[:80]}", + version=self._model_id or None, + tags=tags, + trace_metadata={ + "dataset_name": report.dataset_name, + "test_kind": report.test_kind, + "item_id": report.id, + "model": self._model_id, + "provider_type": self._provider_type, + "reasoning_effort": self._reasoning_effort, + }, + environment=os.environ.get("LANGFUSE_TRACING_ENVIRONMENT"), + ) def log_item(self, report: ItemReport, *, dataset_item_id: str) -> None: - """Send trace + dataset-run-item + scores for one evaluated item. + """Export one experiment root span plus its four scores for an evaluated item. Swallows all errors — Langfuse failures never abort the eval run. """ - trace_id = str(uuid.uuid4()) - now = datetime.now(timezone.utc).isoformat() scores = compute_scores( pass_at_k=report.pass_at_k, avg_latency_s=report.avg_latency_s, best_detail=report.best_detail, ) + run = self._resolve_run(report, dataset_item_id) + span = self._build_span(report, dataset_item_id, run) - # Each ingestion event needs a top-level id (dedup) and timestamp - # in addition to the body-level id/timestamp for the trace/score itself. - def _event(event_type: str, body: dict[str, Any]) -> dict[str, Any]: - return {"id": str(uuid.uuid4()), "timestamp": now, "type": event_type, "body": body} - - batch: list[dict[str, Any]] = [ - _event( - "trace-create", - { - "id": trace_id, - "timestamp": now, - "name": f"gd-eval: {report.question[:80]}", - # Expose the model on a first-class trace field so Langfuse - # dashboards can filter / break down by it ("Version"); trace - # metadata is not available as a breakdown dimension. - "version": self._model_id or None, - "input": {"question": report.question}, - "output": report.best_detail, - "metadata": { - "dataset_name": report.dataset_name, - "test_kind": report.test_kind, - "item_id": report.id, - "model": self._model_id, - "provider_type": self._provider_type, - "reasoning_effort": self._reasoning_effort, - }, - "tags": [ - t - for t in [ - report.test_kind, - self._provider_type, - f"effort-{self._reasoning_effort.lower()}" if self._reasoning_effort else None, - ] - if t - ], - }, - ), - ] + try: + self._client.export_spans([span]) + except Exception as exc: + print(f"warning: Langfuse span export failed for item '{report.id}': {exc}", file=sys.stderr) + return score_defs = [ ("pass_at_k", scores["pass_at_k"], "BOOLEAN"), @@ -134,66 +172,13 @@ def _event(event_type: str, body: dict[str, Any]) -> dict[str, Any]: ("latency_s", scores["latency_s"], "NUMERIC"), ] for name, value, data_type in score_defs: - batch.append( - _event( - "score-create", - { - "id": str(uuid.uuid4()), - "traceId": trace_id, - "name": name, - "value": value, - "dataType": data_type, - }, + try: + self._client.create_score( + trace_id=span.trace_id, + observation_id=span.span_id, + name=name, + value=value, + data_type=data_type, ) - ) - - try: - with httpx.Client( - base_url=self._host, - headers={"Authorization": self._auth_header}, - timeout=10, - ) as client: - resp = client.post("/api/public/ingestion", json={"batch": batch}) - resp.raise_for_status() - # The ingestion endpoint returns HTTP 200 even when individual events - # fail — per-event errors are in the response body. - body = resp.json() - errors = body.get("errors") or [] - for err in errors: - print( - f"warning: Langfuse event failed for item '{report.id}': " - f"type={err.get('error')} status={err.get('status')} id={err.get('id')}", - file=sys.stderr, - ) - except Exception as exc: - print(f"warning: Langfuse ingestion failed for item '{report.id}': {exc}", file=sys.stderr) - - # Link trace to dataset run via the dedicated endpoint (simpler than ingestion — - # does not require datasetId/runId; creates the run by name if absent). - try: - with httpx.Client( - base_url=self._host, - headers={"Authorization": self._auth_header}, - timeout=10, - ) as client: - r = client.post( - "/api/public/dataset-run-items", - json={ - "runName": self._run_name, - "runDescription": ( - f"{self._provider_type}/{self._model_id}" - if self._provider_type and self._model_id - else self._model_id or "" - ), - "metadata": { - "model": self._model_id, - "provider_type": self._provider_type, - "reasoning_effort": self._reasoning_effort, - }, - "datasetItemId": dataset_item_id, - "traceId": trace_id, - }, - ) - r.raise_for_status() - except Exception as exc: - print(f"warning: Langfuse dataset-run-item failed for item '{report.id}': {exc}", file=sys.stderr) + except Exception as exc: # noqa: PERF203 — one score's failure must not skip the rest + print(f"warning: Langfuse score '{name}' failed for item '{report.id}': {exc}", file=sys.stderr) diff --git a/packages/gooddata-eval/tests/_fake_langfuse.py b/packages/gooddata-eval/tests/_fake_langfuse.py index df342cc62..b9dcd638c 100644 --- a/packages/gooddata-eval/tests/_fake_langfuse.py +++ b/packages/gooddata-eval/tests/_fake_langfuse.py @@ -2,7 +2,7 @@ """In-process fake Langfuse HTTP server: a pytest fixture and a runnable wire-watching script. A `threading.Thread`-hosted `http.server` answering the Langfuse v4 endpoints the package -uses, plus the three legacy ones, with canned/synthesised data recorded on `requests`. +uses, with canned/synthesised data recorded on `requests`. """ from __future__ import annotations @@ -23,9 +23,6 @@ _OBSERVATIONS_PATH = "/api/public/v2/observations" _OTLP_PATH = "/api/public/otel/v1/traces" _SCORES_PATH = "/api/public/scores" -_TRACES_PATH = "/api/public/traces" -_INGESTION_PATH = "/api/public/ingestion" -_DATASET_RUN_ITEMS_PATH = "/api/public/dataset-run-items" _ROOT_LATENCY_SECONDS = 12.5 _CHILD_COSTS = (0.01, 0.02) @@ -91,7 +88,6 @@ def __init__(self, *, port: int = 0, verbose: bool = False) -> None: self.otlp_status = 200 self.otlp_body: dict = {} self.scores_429_once = False - self.ingestion_body: dict = {"successes": [], "errors": []} self.verbose = verbose self.requests: list[dict] = [] self.on_request = None @@ -169,12 +165,6 @@ def _route(self, handler: BaseHTTPRequestHandler, method: str) -> None: self._respond(handler, self.otlp_status, self.otlp_body) elif method == "POST" and path == _SCORES_PATH: self._post_scores(handler) - elif method == "GET" and path == _TRACES_PATH: - self._get_traces(handler, query) - elif method == "POST" and path == _INGESTION_PATH: - self._respond(handler, 200, self.ingestion_body) - elif method == "POST" and path == _DATASET_RUN_ITEMS_PATH: - self._respond(handler, 200, {}) else: self._respond(handler, 404, {"message": f"fake_langfuse: no route for {method} {path}"}) @@ -221,17 +211,6 @@ def _get_observations(self, handler: BaseHTTPRequestHandler, query: dict) -> Non return self._respond(handler, 200, {"data": rows, "meta": {}}) - def _get_traces(self, handler: BaseHTTPRequestHandler, query: dict) -> None: - session_id = query.get("sessionId", "") - legacy_trace = { - "id": _short_hash(session_id, 32), - "sessionId": session_id, - "latency": _ROOT_LATENCY_SECONDS, - "totalCost": sum(_CHILD_COSTS), - "metadata": {"conversation_id": session_id}, - } - self._respond(handler, 200, {"data": [legacy_trace]}) - def _respond( self, handler: BaseHTTPRequestHandler, status: int, body: dict, *, headers: dict | None = None ) -> None: diff --git a/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py index 770252870..c3b140fcc 100644 --- a/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py +++ b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py @@ -196,12 +196,11 @@ def _fetch(langfuse, cid, window_start, window_end, pad): assert set(looked_up) == {"c1", "c2", "c3"} -def test_the_skip_switch_announces_itself_instead_of_silently_orphaning_scores(monkeypatch): - # TAVERN_E2E_SKIP_TRACE_LINK returns all-None before any polling, so every score is - # orphaned and the only symptom is observe()'s generic "No trace found for dataset run" - # -- indistinguishable from a genuine lookup failure. That ambiguity cost a long - # debugging detour on a real run: Langfuse was healthy and every trace was present. - # If linking is switched off, say so. +def test_the_skip_switch_announces_itself(monkeypatch): + # TAVERN_E2E_SKIP_TRACE_LINK turns off the whole Langfuse write path: the poll returns + # all-None before any request and observe() writes nothing. A run under it therefore + # looks exactly like a run against a broken Langfuse, so the switch says once that it + # is on and how many conversations it covers. monkeypatch.setenv(SKIP_ENV_VAR, "1") with ( patch("gooddata_eval.core.agentic._langfuse._fetch_traces_for_session") as mock_fetch, @@ -277,18 +276,23 @@ def _observation_row(trace_id: str, session_id: str, latency: float) -> dict: } -def _stub_langfuse_http(monkeypatch, captured: list[httpx.Request], page: dict | None = None): - """A HttpxLangfuseClient whose requests are recorded instead of sent.""" +def _client_with(monkeypatch, handler) -> HttpxLangfuseClient: + """A HttpxLangfuseClient answered by ``handler`` instead of by the network.""" monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk") monkeypatch.setenv("LANGFUSE_HOST", "https://lf.test") + return HttpxLangfuseClient(transport=httpx.MockTransport(handler)) + + +def _stub_langfuse_http(monkeypatch, captured: list[httpx.Request], page: dict | None = None): + """A HttpxLangfuseClient whose requests are recorded instead of sent.""" body = page if page is not None else {"data": [], "meta": {}} def handler(request: httpx.Request) -> httpx.Response: captured.append(request) return httpx.Response(200, json=body) - return HttpxLangfuseClient(transport=httpx.MockTransport(handler)) + return _client_with(monkeypatch, handler) def test_the_trace_lookup_filters_by_session_server_side(monkeypatch): @@ -346,7 +350,7 @@ def test_a_client_without_the_session_parameter_is_filtered_locally_too(): assert found == [wanted] -# --- a 404 from dataset-run-items is one fact about the dataset, not N failures --- +# --- an unknown dataset item is one fact about the dataset, not N failures --- @pytest.fixture @@ -359,26 +363,26 @@ def _fresh_unlinkable_runs(): lf_module._UNLINKABLE_RUNS.clear() -def _http_404() -> Exception: - request = MagicMock() - response = MagicMock(status_code=404) - return httpx.HTTPStatusError("Client error '404 Not Found'", request=request, response=response) +def _langfuse_with_no_dataset_items(monkeypatch) -> HttpxLangfuseClient: + """A client for which no dataset item resolves -- what a local --dataset folder looks like.""" + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.startswith("/api/public/dataset-items/"), ( + f"nothing but the item lookup should be attempted, got {request.url.path}" + ) + return httpx.Response(404, json={"message": "not found"}) -def _langfuse_that_404s_on_run_items(): - lf = MagicMock() - lf.api.dataset_run_items.create.side_effect = _http_404() - return lf + return _client_with(monkeypatch, handler) -def test_a_missing_dataset_item_is_reported_once_per_run_with_its_cause(_fresh_unlinkable_runs): +def test_a_missing_dataset_item_is_reported_once_per_run_with_its_cause(monkeypatch, _fresh_unlinkable_runs): """20 identical raw 404s buried the run's real output and named an endpoint, not a cause. - The 404 means the dataset item id is not in Langfuse, which is a property of the - dataset -- it recurs identically for every item and every pass over it -- so it is one - fact to state once, with what to do about it. + An item id Langfuse does not know is a property of the dataset -- it recurs identically + for every item and every pass over it -- so it is one fact to state once, with what to + do about it. """ - lf = _langfuse_that_404s_on_run_items() + lf = _langfuse_with_no_dataset_items(monkeypatch) said: list[str] = [] with patch("gooddata_eval.core.agentic._langfuse.warn_from_worker", side_effect=said.append): @@ -387,19 +391,19 @@ def test_a_missing_dataset_item_is_reported_once_per_run_with_its_cause(_fresh_u with observe(lf, f"trace-{item}-{run_idx}", item, f"GDAI-2179_ts_model_run{run_idx}", {}): pass - assert lf.api.dataset_run_items.create.call_count == 20 assert len(said) == 1, f"expected one warning for the whole run, got {len(said)}" warning = said[0] assert "does not exist in Langfuse" in warning + assert "404 from dataset-items" in warning assert "--langfuse-dataset" in warning and SKIP_ENV_VAR in warning # The reader has to know the run is not a write-off. assert "Scores ARE still written to the traces" in warning -def test_each_model_run_gets_its_own_report(_fresh_unlinkable_runs): +def test_each_model_run_gets_its_own_report(monkeypatch, _fresh_unlinkable_runs): # --model a --model b produces two differently-named runs; each is separately # unlinkable and the operator should see that it affected both. - lf = _langfuse_that_404s_on_run_items() + lf = _langfuse_with_no_dataset_items(monkeypatch) said: list[str] = [] with patch("gooddata_eval.core.agentic._langfuse.warn_from_worker", side_effect=said.append): @@ -411,11 +415,15 @@ def test_each_model_run_gets_its_own_report(_fresh_unlinkable_runs): assert len(said) == 2 -def test_a_non_404_link_failure_is_still_reported_every_time(_fresh_unlinkable_runs): +def test_a_non_404_link_failure_is_still_reported_every_time(monkeypatch, _fresh_unlinkable_runs): # A 500 or a timeout may be transient and item-specific, so it must not be collapsed # into a one-shot "this dataset cannot link" claim. - lf = MagicMock() - lf.api.dataset_run_items.create.side_effect = RuntimeError("connection reset") + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.startswith("/api/public/dataset-items/"): + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-1"}) + return httpx.Response(500, text="connection reset") + + lf = _client_with(monkeypatch, handler) said: list[str] = [] with patch("gooddata_eval.core.agentic._langfuse.warn_from_worker", side_effect=said.append): @@ -424,12 +432,12 @@ def test_a_non_404_link_failure_is_still_reported_every_time(_fresh_unlinkable_r pass assert len(said) == 3 - assert all("failed to create dataset run item" in w for w in said) + assert all("failed to export experiment span" in w for w in said) -def test_scores_still_reach_the_trace_after_a_404(_fresh_unlinkable_runs): +def test_scores_still_reach_the_trace_after_a_404(monkeypatch, _fresh_unlinkable_runs): # observe() yields the trace id regardless, which is why the run was not a write-off. - lf = _langfuse_that_404s_on_run_items() + lf = _langfuse_with_no_dataset_items(monkeypatch) with ( patch("gooddata_eval.core.agentic._langfuse.warn_from_worker"), diff --git a/packages/gooddata-eval/tests/test_agentic_observe_experiment.py b/packages/gooddata-eval/tests/test_agentic_observe_experiment.py new file mode 100644 index 000000000..94e80f6f4 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_observe_experiment.py @@ -0,0 +1,325 @@ +# (C) 2026 GoodData Corporation +"""observe() exports gd-eval's own experiment root span, and scores fan out to both targets.""" + +from __future__ import annotations + +import json +import re +import threading +from datetime import datetime, timezone +from typing import Any +from unittest.mock import MagicMock + +import httpx +import pytest +from gooddata_eval.core.agentic._langfuse import SKIP_ENV_VAR, observe, score_safe +from gooddata_eval.core.agentic._trace_linker import _CANCEL, RunTraceContext +from gooddata_eval.core.langfuse.client import HttpxLangfuseClient +from gooddata_eval.core.langfuse.experiment import ScoreTarget, experiment_id_for +from gooddata_eval.core.langfuse.observations import TraceSummary +from gooddata_eval.core.langfuse.otlp import unix_nano + +_OTLP_PATH = "/api/public/otel/v1/traces" +_SCORES_PATH = "/api/public/scores" +_WINDOW = (datetime(2026, 9, 8, 10, 0, tzinfo=timezone.utc), datetime(2026, 9, 8, 10, 1, tzinfo=timezone.utc)) + + +@pytest.fixture(autouse=True) +def _langfuse_env(monkeypatch): + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://lf.test") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk") + monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False) + monkeypatch.delenv(SKIP_ENV_VAR, raising=False) + + +class _Recorder: + """A Langfuse client whose requests are recorded instead of sent; every item resolves.""" + + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + self.client = HttpxLangfuseClient(transport=httpx.MockTransport(self._handle)) + + def _handle(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + if request.url.path.startswith("/api/public/dataset-items/"): + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-1"}) + return httpx.Response(200, json={}) + + def spans(self) -> list[dict]: + return [ + span + for request in self.requests + if request.url.path == _OTLP_PATH + for span in json.loads(request.content)["resourceSpans"][0]["scopeSpans"][0]["spans"] + ] + + def score_bodies(self) -> list[dict]: + return [json.loads(request.content) for request in self.requests if request.url.path == _SCORES_PATH] + + +@pytest.fixture +def rec(): + recorder = _Recorder() + yield recorder + recorder.client.close() + + +def _attrs(span: dict) -> dict[str, Any]: + """Span attributes as `{key: value}`, unwrapped from the OTLP typed-value envelope.""" + unwrapped: dict[str, Any] = {} + for attr in span["attributes"]: + ((kind, value),) = attr["value"].items() + unwrapped[attr["key"]] = [v["stringValue"] for v in value["values"]] if kind == "arrayValue" else value + return unwrapped + + +def _gen_ai_trace() -> TraceSummary: + """The gen-ai root observation row gd-eval's span is timed and costed from.""" + return TraceSummary( + { + "traceId": "gen-ai-id", + "id": "o-root", + "parentObservationId": None, + "sessionId": "conv-1", + "latency": 18.4, + "totalCost": 0.0123, + "startTime": "2026-09-08T10:00:00+00:00", + "endTime": "2026-09-08T10:00:18.400000+00:00", + } + ) + + +def test_a_linked_run_is_exported_as_one_experiment_root_span(rec): + trace = _gen_ai_trace() + run_name = "GDAI-2179_2026-09-08_gpt-5.2_run0" + + with observe( + rec.client, + trace.id, + "item-1", + run_name, + {"model_version": "gpt-5.2", "testing_framework": "tavern-e2e"}, + trace=trace, + item_input="Show revenue by month", + output={"passed": True}, + ) as tid: + pass + + exports = [request for request in rec.requests if request.url.path == _OTLP_PATH] + assert len(exports) == 1 + assert exports[0].method == "POST" + assert exports[0].headers["x-langfuse-ingestion-version"] == "4" + assert exports[0].headers["Authorization"].startswith("Basic ") + + (span,) = rec.spans() + assert re.fullmatch(r"[0-9a-f]{32}", span["traceId"]) + assert re.fullmatch(r"[0-9a-f]{16}", span["spanId"]) + # The span covers the gen-ai turn, not the moment linking happened. + assert span["startTimeUnixNano"] == unix_nano(trace.start_time) + assert span["endTimeUnixNano"] == unix_nano(trace.end_time) + + attrs = _attrs(span) + assert attrs["langfuse.experiment.name"] == run_name + assert attrs["langfuse.experiment.id"] == experiment_id_for(run_name) + assert attrs["langfuse.experiment.dataset.id"] == "ds-1" + assert attrs["langfuse.experiment.item.id"] == "item-1" + assert attrs["langfuse.experiment.item.root_observation_id"] == span["spanId"] + assert attrs["langfuse.session.id"] == "conv-1" + assert attrs["langfuse.version"] == "gpt-5.2" + assert attrs["langfuse.trace.name"] == "gd-eval: Show revenue by month" + assert attrs["langfuse.trace.tags"] == ["gd-eval", "tavern-e2e"] + assert attrs["langfuse.observation.metadata.gen_ai_trace_id"] == "gen-ai-id" + assert attrs["langfuse.observation.metadata.gen_ai_latency_s"] == 18.4 + assert attrs["langfuse.observation.metadata.gen_ai_cost_usd"] == 0.0123 + assert json.loads(attrs["langfuse.observation.input"]) == {"question": "Show revenue by month"} + assert json.loads(attrs["langfuse.observation.output"]) == {"passed": True} + + assert tid == "gen-ai-id" + assert tid.experiment_trace_id == span["traceId"] + assert tid.experiment_span_id == span["spanId"] + + +def test_the_span_carries_every_key_the_daily_report_reads(rec): + """gdc-nas `report.py` and `combo_report.py` read these attribute names off the span. + + Renaming any of them breaks the daily report, so they are pinned here. + """ + trace = _gen_ai_trace() + + with observe( + rec.client, + trace.id, + "item-1", + "run0", + { + "testing_framework": "tavern-e2e", + "github_run_id": "123", + "model_version": "gpt", + "reasoning_effort": "LOW", + }, + trace=trace, + conversation_id="conv-1", + ): + pass + + attrs = _attrs(rec.spans()[0]) + assert attrs["langfuse.experiment.metadata.testing_framework"] == "tavern-e2e" + assert attrs["langfuse.experiment.metadata.github_run_id"] == "123" + assert attrs["langfuse.experiment.metadata.model_version"] == "gpt" + assert attrs["langfuse.experiment.metadata.reasoning_effort"] == "LOW" + assert attrs["langfuse.observation.metadata.gen_ai_trace_id"] == "gen-ai-id" + assert attrs["langfuse.observation.metadata.conversation_id"] == "conv-1" + assert attrs["langfuse.trace.metadata.run_name"] == "run0" + + +def test_a_run_whose_gen_ai_trace_never_arrived_still_gets_its_experiment_span(rec): + # Without a trace the run would vanish from the comparison entirely; the span keeps it + # in the experiment and gives the scores somewhere to land. + + with observe( + rec.client, + None, + "item-1", + "run0", + {"model_version": "gpt"}, + window=_WINDOW, + conversation_id="conv-9", + ) as tid: + score_safe(rec.client, tid, name="pass_at_k", value=False, data_type="BOOLEAN") + + (span,) = rec.spans() + attrs = _attrs(span) + assert attrs["langfuse.session.id"] == "conv-9" + assert "langfuse.observation.metadata.gen_ai_trace_id" not in attrs + assert span["startTimeUnixNano"] == unix_nano(_WINDOW[0]) + assert span["endTimeUnixNano"] == unix_nano(_WINDOW[1]) + + bodies = rec.score_bodies() + assert [body["traceId"] for body in bodies] == [span["traceId"]] + assert bodies[0]["observationId"] == span["spanId"] + + +def test_a_score_target_writes_to_both_the_gen_ai_trace_and_the_experiment_span(rec): + target = ScoreTarget("gen-ai-id", "0" * 32, "1" * 16) + + score_safe(rec.client, target, name="pass_at_k", value=True, data_type="BOOLEAN") + + bodies = rec.score_bodies() + assert [body["traceId"] for body in bodies] == ["gen-ai-id", "0" * 32] + # The gen-ai call stays byte-identical to what a legacy client already accepts. + assert "observationId" not in bodies[0] + assert bodies[1]["observationId"] == "1" * 16 + assert [body["value"] for body in bodies] == [1.0, 1.0] + + +def test_a_refused_gen_ai_score_still_reaches_the_experiment_span(): + """One destination rejecting a score must not cost the other one its copy.""" + bodies: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + bodies.append(body) + return httpx.Response(400 if body["traceId"] == "gen-ai-id" else 200, json={}) + + client = HttpxLangfuseClient(transport=httpx.MockTransport(handler)) + try: + score_safe( + client, ScoreTarget("gen-ai-id", "0" * 32, "1" * 16), name="pass_at_k", value=True, data_type="BOOLEAN" + ) + finally: + client.close() + + assert [body["traceId"] for body in bodies] == ["gen-ai-id", "0" * 32] + assert bodies[1]["observationId"] == "1" * 16 + + +def test_a_cancelled_drain_reaches_no_langfuse_http(rec): + """An interrupt has to be answered before the HTTP, not by the retry ladder inside it.""" + cancelled = threading.Event() + cancelled.set() + token = _CANCEL.set(cancelled) + try: + with observe(rec.client, "gen-ai-id", "item-1", "run0", {}) as tid: + pass + score_safe( + rec.client, ScoreTarget("gen-ai-id", "0" * 32, "1" * 16), name="pass_at_k", value=True, data_type="BOOLEAN" + ) + finally: + _CANCEL.reset(token) + + assert tid is None + assert rec.requests == [] + + +def test_a_plain_trace_id_still_writes_exactly_one_score(rec): + + score_safe(rec.client, "trace-abc", name="quality_score", value=0.5, data_type="NUMERIC") + + assert [body["traceId"] for body in rec.score_bodies()] == ["trace-abc"] + + +def test_nothing_to_score_against_writes_no_score(rec): + + score_safe(rec.client, None, name="quality_score", value=0.5, data_type="NUMERIC") + + assert rec.score_bodies() == [] + + +def test_the_skip_switch_stops_every_langfuse_write_not_just_the_poll(rec, monkeypatch): + """The switch is how an operator runs with Langfuse turned off or unreachable. + + Exporting the experiment span anyway would put one span, one item lookup and a score + fan-out per run back on a path the operator switched off -- and against an unreachable + Langfuse, one export warning per item. + """ + monkeypatch.setenv(SKIP_ENV_VAR, "1") + + with observe(rec.client, None, "item-1", "run", {}) as tid: + score_safe(rec.client, tid, name="pass_at_k", value=True, data_type="BOOLEAN") + + assert tid is None + assert rec.requests == [] + + +def test_a_non_httpx_client_never_reaches_the_experiment_path(): + # The skill suites hand observe() a MagicMock, which answers every hasattr -- so the + # legacy branch keys on the concrete client type instead. + lf = MagicMock() + + with observe(lf, "t", "item-1", "run0", {}) as tid: + pass + + lf.dataset_id_for_item.assert_not_called() + lf.export_spans.assert_not_called() + assert tid == "t" + assert tid.experiment_trace_id is None + + +def test_a_legacy_client_with_no_trace_yields_nothing_to_score(): + with observe(MagicMock(), None, "item-1", "run0", {}) as tid: + pass + + assert tid is None + + +def test_the_run_context_forwards_its_window_and_item_input_to_observe(): + # The window and the question are resolved on the calling thread and carried by the + # context, so a skill's scoring block keeps calling ctx.observe(pt, run_idx). + lf = MagicMock() + trace = MagicMock(id="gen-ai-id") + ctx = RunTraceContext( + {"model_version": "m"}, lf, "client", "item-1", "base", True, {}, _WINDOW, "Show revenue by month" + ) + + ctx.observe(trace, 2, conversation_id="conv-1", output={"passed": True}) + + args, kwargs = lf.observe.call_args + assert args == ("client", "gen-ai-id", "item-1", "base_run2", {"model_version": "m"}) + assert kwargs == { + "trace": trace, + "window": _WINDOW, + "conversation_id": "conv-1", + "item_input": "Show revenue by month", + "output": {"passed": True}, + } diff --git a/packages/gooddata-eval/tests/test_cli.py b/packages/gooddata-eval/tests/test_cli.py index 9ff008903..1d23293bf 100644 --- a/packages/gooddata-eval/tests/test_cli.py +++ b/packages/gooddata-eval/tests/test_cli.py @@ -954,8 +954,8 @@ def _agentic_item(): def test_warns_up_front_when_a_local_dataset_cannot_be_linked(monkeypatch, tmp_path, capsys): """--langfuse is refused with a local dataset, but the evaluators' own try_make_langfuse_client() fallback links anyway when LANGFUSE_* are exported -- so - every conversation 404s from dataset-run-items, in a block at the very END of the run. - By then the flag that would have avoided it is long past being changeable. + every conversation reports its item missing from Langfuse, in a block at the very END + of the run. By then the flag that would have avoided it is long past being changeable. """ _export_langfuse_creds(monkeypatch) monkeypatch.delenv(TIMERS_ENV_VAR, raising=False) diff --git a/packages/gooddata-eval/tests/test_langfuse_client.py b/packages/gooddata-eval/tests/test_langfuse_client.py index f1bf5f228..931d7588e 100644 --- a/packages/gooddata-eval/tests/test_langfuse_client.py +++ b/packages/gooddata-eval/tests/test_langfuse_client.py @@ -109,8 +109,14 @@ def handler(request: httpx.Request) -> httpx.Response: client = make_client(handler) client.create_score("t-1", "pass_at_k", True, "BOOLEAN") client.create_score("t-1", "pass_at_k", False, "BOOLEAN") + # The sink's compute_scores yields int 1/0 for pass_at_k, so the coercion keys off + # dataType rather than the Python type of the value. + client.create_score("t-1", "pass_at_k", 1, "BOOLEAN") + client.create_score("t-1", "pass_at_k", 0, "BOOLEAN") - assert [body["value"] for body in bodies] == [1.0, 0.0] + assert [body["value"] for body in bodies] == [1.0, 0.0, 1.0, 0.0] + # Typed, not just equal: `1 == 1.0` in Python, but the two serialise differently. + assert all(isinstance(body["value"], float) for body in bodies) def test_a_throttled_score_is_retried_after_the_delay_the_server_asked_for(make_client, monkeypatch): @@ -297,11 +303,29 @@ def handler(request: httpx.Request) -> httpx.Response: assert "sessionId" not in seen[0] -def test_a_dataset_run_item_is_posted_to_the_legacy_endpoint(make_client): +def test_a_negative_retry_after_never_reaches_sleep(make_client, monkeypatch): + # time.sleep raises on a negative delay, so a server clock skew or a hostile header + # would turn a throttled score into an exception instead of a retry. + slept: list[float] = [] + monkeypatch.setattr(client_module.time, "sleep", slept.append) + statuses = [429, 200] + + def handler(request: httpx.Request) -> httpx.Response: + status = statuses[len(slept)] + return httpx.Response(status, headers={"Retry-After": "-5"} if status == 429 else {}, json={}) + + make_client(handler).create_score("t-1", "quality_score", 1.0, "NUMERIC") + + assert slept == [0.5] + + +def test_a_dataset_run_item_is_exported_as_an_experiment_root_span(make_client): seen: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(request) + if request.url.path.startswith("/api/public/dataset-items/"): + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-1"}) return _ok(request) make_client(handler).api.dataset_run_items.create( @@ -312,33 +336,44 @@ def handler(request: httpx.Request) -> httpx.Response: run_description="desc", ) - assert seen[0].method == "POST" - assert seen[0].url.path == "/api/public/dataset-run-items" - assert json.loads(seen[0].content) == { - "runName": "ds_2026_model", - "datasetItemId": "item-1", - "traceId": "t-1", - "metadata": {"model_version": "m"}, - "runDescription": "desc", - } - - -def test_a_dataset_run_item_for_an_unknown_item_raises(make_client): - client = make_client(lambda request: httpx.Response(404, json={})) - with pytest.raises(httpx.HTTPStatusError): - client.api.dataset_run_items.create(run_name="run", dataset_item_id="local", trace_id="t-1") + assert [request.url.path for request in seen] == ["/api/public/dataset-items/item-1", "/api/public/otel/v1/traces"] + assert seen[1].headers["x-langfuse-ingestion-version"] == "4" + span = json.loads(seen[1].content)["resourceSpans"][0]["scopeSpans"][0]["spans"][0] + attrs = {attr["key"]: attr["value"].get("stringValue") for attr in span["attributes"]} + assert attrs["langfuse.experiment.name"] == "ds_2026_model" + assert attrs["langfuse.experiment.dataset.id"] == "ds-1" + assert attrs["langfuse.experiment.item.id"] == "item-1" + assert attrs["langfuse.experiment.item.root_observation_id"] == span["spanId"] + assert attrs["langfuse.experiment.description"] == "desc" + assert attrs["langfuse.experiment.metadata.model_version"] == "m" + assert attrs["langfuse.observation.metadata.gen_ai_trace_id"] == "t-1" -def test_the_trace_version_upsert_uses_the_ingestion_endpoint(make_client): +def test_a_dataset_run_item_returns_the_score_target_for_the_span_it_exported(make_client): seen: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(request) + if request.url.path.startswith("/api/public/dataset-items/"): + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-1"}) return _ok(request) - make_client(handler).update_trace_version("t-1", "gpt-5.2") + target = make_client(handler).api.dataset_run_items.create( + run_name="ds_2026_model", dataset_item_id="item-1", trace_id="t-1" + ) + + span = json.loads(seen[1].content)["resourceSpans"][0]["scopeSpans"][0]["spans"][0] + # The run item IS the root observation, so an experiment-item score needs the span's own + # trace and span id -- neither is recoverable from the gen-ai trace id the caller passed. + assert target.gen_ai_trace_id == "t-1" + assert target.experiment_trace_id == span["traceId"] + assert target.experiment_span_id == span["spanId"] + assert target.destinations() == [("t-1", None), (span["traceId"], span["spanId"])] + # Still a str equal to the gen-ai trace id, so a caller that treats it as one keeps working. + assert target == "t-1" + - assert seen[0].url.path == "/api/public/ingestion" - event = json.loads(seen[0].content)["batch"][0] - assert event["type"] == "trace-create" - assert event["body"] == {"id": "t-1", "version": "gpt-5.2"} +def test_a_dataset_run_item_for_an_unknown_item_raises(make_client): + client = make_client(lambda request: httpx.Response(404, json={})) + with pytest.raises(LookupError): + client.api.dataset_run_items.create(run_name="run", dataset_item_id="local", trace_id="t-1") diff --git a/packages/gooddata-eval/tests/test_langfuse_e2e_fake_server.py b/packages/gooddata-eval/tests/test_langfuse_e2e_fake_server.py new file mode 100644 index 000000000..09a18ae45 --- /dev/null +++ b/packages/gooddata-eval/tests/test_langfuse_e2e_fake_server.py @@ -0,0 +1,376 @@ +# (C) 2026 GoodData Corporation +"""End-to-end Langfuse v4 choreography against the in-process fake server. + +Nothing here is mocked below the HTTP boundary: the real `HttpxLangfuseClient` and +`LangfuseSink` are built from the fixture's `LANGFUSE_*` env and talk to `FakeLangfuse`, so +each test asserts the request sequence a real run produces -- poll observations, look the +dataset item up, export one experiment root span, score it. +""" + +from __future__ import annotations + +import contextlib +import time +from datetime import datetime +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from gooddata_eval.cli import main as cli_main +from gooddata_eval.cli.agentic_runner import run_agentic_items +from gooddata_eval.core.agentic import _langfuse as agentic_langfuse +from gooddata_eval.core.agentic._trace_linker import run_trace_link_inline +from gooddata_eval.core.agentic.general_question import evaluate_agentic_general_question +from gooddata_eval.core.langfuse.client import HttpxLangfuseClient +from gooddata_eval.core.langfuse.experiment import experiment_id_for +from gooddata_eval.core.langfuse.otlp import unix_nano +from gooddata_eval.core.models import ChatResult, DatasetItem +from gooddata_eval.core.runner import EvalReport, ItemReport +from gooddata_eval.core.workspace import ActiveLlmProvider, ResolvedModel + +from tests._fake_langfuse import FakeLangfuse, observation_rows + +_MODEL = "gpt-5.2" +_RUN_TS = "2026-01-01_00-00-00" +_OBSERVATIONS = "/api/public/v2/observations" +_OTLP = "/api/public/otel/v1/traces" +_SCORES = "/api/public/scores" +_DATASET_ITEMS = "/api/public/dataset-items" + + +class _NoSleep: + """`_langfuse`'s view of the `time` module with the poll's backoff sleeps removed.""" + + monotonic = staticmethod(time.monotonic) + + @staticmethod + def sleep(_seconds: float) -> None: + return None + + +@pytest.fixture(autouse=True) +def _forget_unlinkable_runs(): + """The once-per-run warning gate is module state; no test may inherit another's.""" + agentic_langfuse._UNLINKABLE_RUNS.clear() + yield + agentic_langfuse._UNLINKABLE_RUNS.clear() + + +def _chat_client(conversation_ids: list[str]) -> MagicMock: + client = MagicMock() + client.create_conversation.side_effect = conversation_ids + client.send_message.return_value = ChatResult.model_validate( + {"textResponse": "42", "toolCallEvents": [], "reasoningSteps": [], "responseId": "resp-1"} + ) + return client + + +def _passing_judge() -> MagicMock: + judge = MagicMock() + judge.model = "gpt-4o" + judge.score.return_value = (True, "Correct answer") + return judge + + +@contextlib.contextmanager +def _agent_stubbed(conversation_ids: list[str], *, no_sleep: bool = False): + """Stub the SSE agent and the judge, and pin the model version the run reports.""" + with ( + patch("gooddata_eval.core.agentic.general_question.ChatClient", return_value=_chat_client(conversation_ids)), + patch("gooddata_eval.core.agentic.general_question.LLMJudge", return_value=_passing_judge()), + patch.object(agentic_langfuse, "get_model_version", return_value=_MODEL), + ): + if no_sleep: + with patch.object(agentic_langfuse, "time", _NoSleep): + yield + else: + yield + + +def _run_general_question(*, dataset_item_id: str, dataset_name: str) -> None: + evaluate_agentic_general_question( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What is 6 times 7?", + expected_output="42", + k=1, + langfuse=None, + dataset_item_id=dataset_item_id, + dataset_name=dataset_name, + run_timestamp=_RUN_TS, + submit_trace_link=run_trace_link_inline, + ) + + +def _spans(server: FakeLangfuse) -> list[dict]: + return [ + span + for call in server.calls("POST", _OTLP) + for resource in call["json"]["resourceSpans"] + for scope in resource["scopeSpans"] + for span in scope["spans"] + ] + + +def _attrs(span: dict) -> dict[str, Any]: + """Span attributes as `key -> unwrapped OTLP value`.""" + return {a["key"]: next(iter(a["value"].values())) for a in span["attributes"]} + + +def _score_bodies(server: FakeLangfuse) -> list[dict]: + return [call["json"] for call in server.calls("POST", _SCORES)] + + +def test_agentic_inline_path_polls_looks_up_exports_and_scores(fake_langfuse: FakeLangfuse, capsys) -> None: + fake_langfuse.first_observations_call_empty = True + run_name = f"inline_{_RUN_TS}_{_MODEL}" + + with _agent_stubbed(["conv-1"], no_sleep=True): + _run_general_question(dataset_item_id="item-1", dataset_name="inline") + + polls = fake_langfuse.calls("GET", _OBSERVATIONS) + assert len(polls) >= 2, "the empty first page must be retried" + assert {p["query"]["sessionId"] for p in polls} == {"conv-1"} + + lookups = fake_langfuse.calls("GET", f"{_DATASET_ITEMS}/item-1") + assert len(lookups) == 1 + + spans = _spans(fake_langfuse) + assert len(spans) == 1 + span, attrs = spans[0], _attrs(spans[0]) + assert attrs["langfuse.experiment.name"] == run_name + assert attrs["langfuse.experiment.item.id"] == "item-1" + assert attrs["langfuse.experiment.item.root_observation_id"] == span["spanId"] + assert attrs["langfuse.session.id"] == "conv-1" + + root_row = next(r for r in observation_rows("conv-1") if r["parentObservationId"] is None) + assert attrs["langfuse.observation.metadata.gen_ai_trace_id"] == root_row["traceId"] + assert span["startTimeUnixNano"] == unix_nano(datetime.fromisoformat(root_row["startTime"])) + assert span["endTimeUnixNano"] == unix_nano(datetime.fromisoformat(root_row["endTime"])) + + bodies = _score_bodies(fake_langfuse) + on_gen_ai = [b for b in bodies if b["traceId"] == root_row["traceId"]] + on_span = [b for b in bodies if b["traceId"] == span["traceId"]] + assert len(bodies) == len(on_gen_ai) + len(on_span) + assert {b["name"] for b in on_gen_ai} == {b["name"] for b in on_span} + assert {"general_question_pass", "llm_judge_score", "quality_score", "value_score"} == { + b["name"] for b in on_gen_ai + } + assert all("observationId" not in b for b in on_gen_ai) + assert all(b["observationId"] == span["spanId"] for b in on_span) + + # Cost is summed over the trace's rows: the gen-ai root carries none, its children do. + value_score = next(b for b in on_span if b["name"] == "value_score") + assert "cost=$0.0300" in value_score["comment"] + assert "latency=12.50s" in value_score["comment"] + assert capsys.readouterr().out.count("WARNING") == 0 + + +def test_batched_path_puts_both_items_in_one_experiment(fake_langfuse: FakeLangfuse, monkeypatch) -> None: + items = [ + DatasetItem( + id=f"item-{n}", + dataset_name="batched", + test_kind="agentic_general_question", + question="What is 6 times 7?", + expected_output="42", + ) + for n in (1, 2) + ] + closed_after: list[int] = [] + real_close = HttpxLangfuseClient.close + + def _spy_close(self: HttpxLangfuseClient) -> None: + closed_after.append(len(fake_langfuse.requests)) + real_close(self) + + monkeypatch.setattr(HttpxLangfuseClient, "close", _spy_close) + + with _agent_stubbed(["conv-1", "conv-2"]): + report = run_agentic_items( + items, + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + k=1, + use_langfuse=True, + run_ts=_RUN_TS, + ) + + assert [i.pass_at_k for i in report.items] == [True, True] + + spans = _spans(fake_langfuse) + assert len(spans) == 2 + run_name = f"batched_{_RUN_TS}_{_MODEL}" + assert {_attrs(s)["langfuse.experiment.id"] for s in spans} == {experiment_id_for(run_name)} + assert {_attrs(s)["langfuse.experiment.item.id"] for s in spans} == {"item-1", "item-2"} + assert {_attrs(s)["langfuse.session.id"] for s in spans} == {"conv-1", "conv-2"} + + # flush() and close() are the run's last words to Langfuse: nothing may follow them. + assert closed_after == [len(fake_langfuse.requests)] + + +def test_local_dataset_item_keeps_gen_ai_scores_and_warns_once(fake_langfuse: FakeLangfuse, capsys) -> None: + fake_langfuse.missing = {"local-item"} + + with _agent_stubbed(["conv-1"]): + _run_general_question(dataset_item_id="local-item", dataset_name="local") + + assert fake_langfuse.calls("POST", _OTLP) == [] + + out = capsys.readouterr().out + assert out.count("does not exist in Langfuse") == 1 + + root_row = next(r for r in observation_rows("conv-1") if r["parentObservationId"] is None) + bodies = _score_bodies(fake_langfuse) + assert {b["name"] for b in bodies} == { + "general_question_pass", + "llm_judge_score", + "quality_score", + "value_score", + } + assert all(b["traceId"] == root_row["traceId"] for b in bodies) + assert all("observationId" not in b for b in bodies) + + +def _stub_single_shot_cli(monkeypatch) -> None: + """Stub everything the CLI touches outside Langfuse: connection, model, chat, runner.""" + monkeypatch.setattr(cli_main, "resolve_connection", lambda host, token, profile: ("https://h", "tok")) + + class _FakeController: + def __init__(self, *a: object, **k: object) -> None: ... + def get_active(self) -> ActiveLlmProvider: + return ActiveLlmProvider(provider_id="prov", default_model_id=_MODEL) + + def resolve_and_activate(self, requested: str | None, provider: str | None = None) -> ResolvedModel: + return ResolvedModel( + provider_id="prov", model_id=requested or _MODEL, switched=False, provider_name="Test Provider" + ) + + def restore(self, original: object) -> None: ... + def close(self) -> None: ... + + monkeypatch.setattr(cli_main, "WorkspaceModelController", _FakeController) + monkeypatch.setattr(cli_main, "ChatClient", lambda **k: object()) + + def _fake_run(items: list[DatasetItem], backend: object, *, runs: int, model: str, **kw: Any) -> EvalReport: + reports = [ + ItemReport( + id=item.id, + dataset_name=item.dataset_name, + test_kind=item.test_kind, + question=item.question, + pass_at_k=True, + runs=runs, + latency_s=15.0, + best_detail={"metrics_correct": True}, + ) + for item in items + ] + for index, item_report in enumerate(reports, start=1): + kw["on_langfuse_item_done"](index, len(reports), item_report) + return EvalReport(model=model, workspace_id=kw["workspace_id"], items=reports) + + monkeypatch.setattr(cli_main, "run_items", _fake_run) + + +def test_cli_langfuse_dataset_run_writes_one_experiment_span_per_item(fake_langfuse: FakeLangfuse, monkeypatch) -> None: + fake_langfuse.items = [ + {"id": f"item-{n}", "datasetName": "fake", "input": {"question": f"q{n}"}, "expectedOutput": "rubric"} + for n in (1, 2) + ] + _stub_single_shot_cli(monkeypatch) + + exit_code = cli_main.main( + [ + "run", + "--host", + "https://h", + "--token", + "tok", + "--workspace", + "ws1", + "--langfuse-dataset", + "fake", + "--langfuse", + "--runs", + "1", + "--quiet", + ] + ) + + assert exit_code == 0 + dataset_reads = [c for c in fake_langfuse.calls("GET", _DATASET_ITEMS) if c["path"] == _DATASET_ITEMS] + assert len(dataset_reads) == 1 + assert dataset_reads[0]["query"]["datasetName"] == "fake" + + spans = _spans(fake_langfuse) + assert len(spans) == 2 + assert {_attrs(s)["langfuse.experiment.item.id"] for s in spans} == {"item-1", "item-2"} + assert {_attrs(s)["langfuse.version"] for s in spans} == {_MODEL} + + bodies = _score_bodies(fake_langfuse) + assert len(bodies) == 8 + by_span = {s["spanId"]: s["traceId"] for s in spans} + for body in bodies: + assert by_span[body["observationId"]] == body["traceId"] + for span in spans: + names = {b["name"] for b in bodies if b["observationId"] == span["spanId"]} + assert names == {"pass_at_k", "quality_score", "value_score", "latency_s"} + + +def test_a_refused_span_export_keeps_the_gen_ai_scores(fake_langfuse: FakeLangfuse, capsys) -> None: + fake_langfuse.otlp_status = 500 + + with _agent_stubbed(["conv-1"]): + _run_general_question(dataset_item_id="item-1", dataset_name="refused") + + out = capsys.readouterr().out + assert "failed to export experiment span" in out + + bodies = _score_bodies(fake_langfuse) + assert bodies, "scores still go to the gen-ai trace when the span is refused" + assert all("observationId" not in b for b in bodies) + + +def test_a_rate_limited_score_is_retried_and_lands(fake_langfuse: FakeLangfuse) -> None: + fake_langfuse.scores_429_once = True + + with _agent_stubbed(["conv-1"]): + _run_general_question(dataset_item_id="item-1", dataset_name="throttled") + + bodies = _score_bodies(fake_langfuse) + # Eight writes -- four scores on each of the gen-ai trace and the experiment span -- + # plus the one refused attempt the client repeated. + assert len(bodies) == 9 + posted_twice = [b for b in bodies if bodies.count(b) == 2] + assert len(posted_twice) == 2, "exactly one score body was posted twice" + + +def test_the_dataset_run_item_shim_exports_one_experiment_span(fake_langfuse: FakeLangfuse) -> None: + """The keyword shape gdc-nas's trace linker calls on this client.""" + client = HttpxLangfuseClient() + try: + client.api.dataset_run_items.create( + run_name="nas_run", + dataset_item_id="item-1", + trace_id="gen-ai-id", + metadata={"testing_framework": "tavern-e2e"}, + run_description="", + ) + finally: + client.close() + + assert len(fake_langfuse.calls("GET", f"{_DATASET_ITEMS}/item-1")) == 1 + (span,) = _spans(fake_langfuse) + attrs = _attrs(span) + assert attrs["langfuse.experiment.name"] == "nas_run" + assert attrs["langfuse.experiment.dataset.id"] == fake_langfuse.dataset_id + assert attrs["langfuse.experiment.item.id"] == "item-1" + assert attrs["langfuse.experiment.item.root_observation_id"] == span["spanId"] + assert attrs["langfuse.experiment.metadata.testing_framework"] == "tavern-e2e" + assert attrs["langfuse.observation.metadata.gen_ai_trace_id"] == "gen-ai-id" + assert attrs["langfuse.trace.metadata.run_name"] == "nas_run" + # An empty run_description is no description at all. + assert "langfuse.experiment.description" not in attrs diff --git a/packages/gooddata-eval/tests/test_langfuse_observations.py b/packages/gooddata-eval/tests/test_langfuse_observations.py index b942e64e4..b4c26e6b3 100644 --- a/packages/gooddata-eval/tests/test_langfuse_observations.py +++ b/packages/gooddata-eval/tests/test_langfuse_observations.py @@ -154,6 +154,36 @@ def handler(request: httpx.Request) -> httpx.Response: assert "sessionId" not in seen[0] +def test_an_unfiltered_window_is_read_at_the_api_page_maximum(): + # Without a session filter the page holds every trace in the window, so the wanted one + # sits behind however many strangers the workspace produced. + seen: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(dict(request.url.params)) + return httpx.Response(200, json={"data": [], "meta": {}}) + + now = datetime.now(timezone.utc) + with _client(handler) as http: + list_traces_in_window(http, from_time=now, to_time=now, limit=10, session_id=None) + + assert seen[0]["limit"] == "1000" + + +def test_a_session_filtered_window_stays_on_the_smaller_page(): + seen: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(dict(request.url.params)) + return httpx.Response(200, json={"data": [], "meta": {}}) + + now = datetime.now(timezone.utc) + with _client(handler) as http: + list_traces_in_window(http, from_time=now, to_time=now, limit=10, session_id="conv-1") + + assert seen[0]["limit"] == "500" + + def test_the_cursor_is_followed_until_the_server_stops_handing_one_out(): pages = [ {"data": [_row("t-1", "o-1", parent="o-root-1", total_cost=0.5)], "meta": {"cursor": "c1"}}, diff --git a/packages/gooddata-eval/tests/test_langfuse_sink.py b/packages/gooddata-eval/tests/test_langfuse_sink.py index 610cac92b..5101903fc 100644 --- a/packages/gooddata-eval/tests/test_langfuse_sink.py +++ b/packages/gooddata-eval/tests/test_langfuse_sink.py @@ -1,11 +1,17 @@ # (C) 2026 GoodData Corporation -from unittest.mock import MagicMock, patch +from __future__ import annotations + +import base64 +import json +import re import httpx import pytest from gooddata_eval.core.langfuse.sink import LangfuseSink, compute_scores from gooddata_eval.core.runner import ItemReport +_BASIC_AUTH = f"Basic {base64.b64encode(b'pk-test:sk-test').decode()}" + def test_compute_scores_all_pass(): detail = { @@ -51,11 +57,11 @@ def test_compute_scores_skips_non_bool_detail_values(): assert scores["quality_score"] == 1.0 # only 1 bool key → 1/1 -def _make_sink(monkeypatch) -> LangfuseSink: - monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") - monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") - monkeypatch.setenv("LANGFUSE_HOST", "https://cloud.langfuse.com") - return LangfuseSink(dataset_name="my_dataset", run_name="gd-eval-2026-06-03-gpt-5.2") +def test_langfuse_sink_raises_without_credentials(monkeypatch): + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) + with pytest.raises(RuntimeError, match="credentials"): + LangfuseSink(dataset_name="d", run_name="r") def _passing_report() -> ItemReport: @@ -77,89 +83,169 @@ def _passing_report() -> ItemReport: ) -def test_langfuse_sink_posts_batch_with_four_event_types(monkeypatch): - sink = _make_sink(monkeypatch) - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_client = MagicMock() - mock_client.__enter__ = MagicMock(return_value=mock_client) - mock_client.__exit__ = MagicMock(return_value=False) - mock_client.post.return_value = mock_resp - - with patch("gooddata_eval.core.langfuse.sink.httpx.Client", return_value=mock_client): - sink.log_item(_passing_report(), dataset_item_id="item-1") - - assert mock_client.post.call_count == 2 # ingestion + dataset-run-items - # First call: ingestion batch (trace + 4 scores) - ingestion_call = mock_client.post.call_args_list[0] - batch = ingestion_call[1]["json"]["batch"] - types = [e["type"] for e in batch] - assert "trace-create" in types - assert "dataset-run-item-create" not in types # moved to dedicated endpoint - assert types.count("score-create") == 4 # pass_at_k, quality, value, latency - # Second call: dataset-run-items endpoint - run_item_call = mock_client.post.call_args_list[1] - assert "/api/public/dataset-run-items" in str(run_item_call) - - -def test_langfuse_sink_sets_trace_version_to_model(monkeypatch): - # The model id is exposed on the trace `version` field so Langfuse dashboards - # can break down / filter by it ("Version"). +def _known_item_handler(seen: list[httpx.Request]): + """GET dataset-items/item-1 resolves to dataset ds-123; every other call succeeds.""" + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.url.path == "/api/public/dataset-items/item-1": + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-123"}) + return httpx.Response(200, json={}) + + return handler + + +def _make_sink(monkeypatch, handler, **kwargs) -> LangfuseSink: monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") - monkeypatch.setenv("LANGFUSE_HOST", "https://cloud.langfuse.com") - sink = LangfuseSink(dataset_name="ds", run_name="gd-eval-r", model_id="gpt-5.4-mini") + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://lf.test") + monkeypatch.delenv("LANGFUSE_HOST", raising=False) + return LangfuseSink( + dataset_name="my_dataset", + run_name="gd-eval-2026-06-03-gpt-5.2", + transport=httpx.MockTransport(handler), + **kwargs, + ) + + +def _spans_from(requests: list[httpx.Request]) -> list[dict]: + return [ + json.loads(r.content)["resourceSpans"][0]["scopeSpans"][0]["spans"][0] + for r in requests + if r.url.path == "/api/public/otel/v1/traces" + ] + + +def _span_from(requests: list[httpx.Request]) -> dict: + return _spans_from(requests)[0] + + +def test_langfuse_sink_exports_one_span_and_four_scores(monkeypatch): + requests: list[httpx.Request] = [] + sink = _make_sink(monkeypatch, _known_item_handler(requests)) + + sink.log_item(_passing_report(), dataset_item_id="item-1") + + otlp_calls = [r for r in requests if r.url.path == "/api/public/otel/v1/traces"] + score_calls = [r for r in requests if r.url.path == "/api/public/scores"] + assert len(otlp_calls) == 1 + assert len(score_calls) == 4 + + otlp_req = otlp_calls[0] + assert otlp_req.headers["Authorization"] == _BASIC_AUTH + assert otlp_req.headers["x-langfuse-ingestion-version"] == "4" + + span = _span_from(requests) + assert re.fullmatch(r"[0-9a-f]{32}", span["traceId"]) + assert re.fullmatch(r"[0-9a-f]{16}", span["spanId"]) + assert int(span["startTimeUnixNano"]) <= int(span["endTimeUnixNano"]) - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_client = MagicMock() - mock_client.__enter__ = MagicMock(return_value=mock_client) - mock_client.__exit__ = MagicMock(return_value=False) - mock_client.post.return_value = mock_resp + attrs = {a["key"]: a["value"] for a in span["attributes"]} + assert attrs["langfuse.experiment.name"]["stringValue"] == "gd-eval-2026-06-03-gpt-5.2" + assert attrs["langfuse.experiment.dataset.id"]["stringValue"] == "ds-123" + assert attrs["langfuse.experiment.item.id"]["stringValue"] == "item-1" + assert attrs["langfuse.experiment.item.root_observation_id"]["stringValue"] == span["spanId"] + assert json.loads(attrs["langfuse.observation.input"]["stringValue"]) == {"question": "Show revenue by month"} + tags = [v["stringValue"] for v in attrs["langfuse.trace.tags"]["arrayValue"]["values"]] + # "gd-eval" leads on both paths, so one Langfuse tag filter finds every trace we write. + assert tags[0] == "gd-eval" + assert "visualization" in tags - with patch("gooddata_eval.core.langfuse.sink.httpx.Client", return_value=mock_client): - sink.log_item(_passing_report(), dataset_item_id="item-1") + score_names = set() + for score_req in score_calls: + assert score_req.headers["Authorization"] == _BASIC_AUTH + body = json.loads(score_req.content) + assert body["traceId"] == span["traceId"] + assert body["observationId"] == span["spanId"] + score_names.add(body["name"]) + assert score_names == {"pass_at_k", "quality_score", "value_score", "latency_s"} - batch = mock_client.post.call_args_list[0][1]["json"]["batch"] - trace = next(e for e in batch if e["type"] == "trace-create") - assert trace["body"]["version"] == "gpt-5.4-mini" + pass_score = next(json.loads(r.content) for r in score_calls if json.loads(r.content)["name"] == "pass_at_k") + assert pass_score["dataType"] == "BOOLEAN" + assert pass_score["value"] == 1.0 -def test_langfuse_sink_run_item_links_correct_dataset_item(monkeypatch): - sink = _make_sink(monkeypatch) - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_client = MagicMock() - mock_client.__enter__ = MagicMock(return_value=mock_client) - mock_client.__exit__ = MagicMock(return_value=False) - mock_client.post.return_value = mock_resp +def test_langfuse_sink_stamps_the_tracing_environment(monkeypatch): + monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "staging") + requests: list[httpx.Request] = [] + sink = _make_sink(monkeypatch, _known_item_handler(requests)) - with patch("gooddata_eval.core.langfuse.sink.httpx.Client", return_value=mock_client): - sink.log_item(_passing_report(), dataset_item_id="item-1") + sink.log_item(_passing_report(), dataset_item_id="item-1") - # Second call is to /api/public/dataset-run-items - run_item_call = mock_client.post.call_args_list[1] - run_item_body = run_item_call[1]["json"] - assert run_item_body["datasetItemId"] == "item-1" - assert run_item_body["runName"] == "gd-eval-2026-06-03-gpt-5.2" + attrs = {a["key"]: a["value"] for a in _span_from(requests)["attributes"]} + assert attrs["langfuse.environment"]["stringValue"] == "staging" -def test_langfuse_sink_swallows_http_error_and_warns(monkeypatch, capsys): - sink = _make_sink(monkeypatch) - mock_client = MagicMock() - mock_client.__enter__ = MagicMock(return_value=mock_client) - mock_client.__exit__ = MagicMock(return_value=False) - mock_client.post.side_effect = httpx.HTTPError("timeout") +def test_langfuse_sink_sets_span_version_to_model(monkeypatch): + requests: list[httpx.Request] = [] + sink = _make_sink(monkeypatch, _known_item_handler(requests), model_id="gpt-5.4-mini") - with patch("gooddata_eval.core.langfuse.sink.httpx.Client", return_value=mock_client): - sink.log_item(_passing_report(), dataset_item_id="item-1") # must not raise + sink.log_item(_passing_report(), dataset_item_id="item-1") + + attrs = {a["key"]: a["value"] for a in _span_from(requests)["attributes"]} + assert attrs["langfuse.version"]["stringValue"] == "gpt-5.4-mini" + + +def test_langfuse_sink_posts_plain_span_when_dataset_item_is_unknown(monkeypatch, capsys): + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/api/public/dataset-items/item-1": + return httpx.Response(404, json={}) + return httpx.Response(200, json={}) + + sink = _make_sink(monkeypatch, handler) + + sink.log_item(_passing_report(), dataset_item_id="item-1") + # Twice: the item is missing for every item of the same --dataset, so the explanation + # is worth exactly one line per sink. + sink.log_item(_passing_report(), dataset_item_id="item-1") + + span = _span_from(requests) + assert not any(a["key"].startswith("langfuse.experiment.") for a in span["attributes"]) + score_calls = [r for r in requests if r.url.path == "/api/public/scores"] + assert len(score_calls) == 8 + trace_of_span = {s["spanId"]: s["traceId"] for s in _spans_from(requests)} + for score_req in score_calls: + body = json.loads(score_req.content) + assert trace_of_span[body["observationId"]] == body["traceId"] err = capsys.readouterr().err - assert "warning" in err.lower() or "langfuse" in err.lower() + assert err.count("warning:") == 1 + assert "item-1" in err -def test_langfuse_sink_raises_without_credentials(monkeypatch): - monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) - monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) - with pytest.raises(RuntimeError, match="credentials"): - LangfuseSink(dataset_name="d", run_name="r") +def test_langfuse_sink_swallows_connect_error_and_warns(monkeypatch, capsys): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("boom", request=request) + + sink = _make_sink(monkeypatch, handler) + + sink.log_item(_passing_report(), dataset_item_id="item-1") # must not raise + + err = capsys.readouterr().err + assert "warning" in err.lower() + + +def test_langfuse_sink_skips_scores_when_span_export_fails(monkeypatch, capsys): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/public/dataset-items/item-1": + return httpx.Response(200, json={"id": "item-1", "datasetId": "ds-123"}) + if request.url.path == "/api/public/otel/v1/traces": + return httpx.Response(400, text="bad span") + return httpx.Response(200, json={}) + + calls: list[httpx.Request] = [] + + def recording_handler(request: httpx.Request) -> httpx.Response: + calls.append(request) + return handler(request) + + sink = _make_sink(monkeypatch, recording_handler) + + sink.log_item(_passing_report(), dataset_item_id="item-1") + + assert not any(r.url.path == "/api/public/scores" for r in calls) + err = capsys.readouterr().err + assert "span export failed" in err.lower() diff --git a/packages/gooddata-eval/tests/test_trace_linker.py b/packages/gooddata-eval/tests/test_trace_linker.py index 4737ee34c..f490effd6 100644 --- a/packages/gooddata-eval/tests/test_trace_linker.py +++ b/packages/gooddata-eval/tests/test_trace_linker.py @@ -172,7 +172,7 @@ def test_every_kind_hands_a_pinned_window_to_the_linker(module_name, _func_name) tree = ast.parse(inspect.getsource(module)) submits = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and _callee_name(n) == "submit_trace_scoring"] - assert submits, f"{module_name} no longer defers its Langfuse block to the linker" + assert submits, f"{module_name} does not defer its Langfuse block to the linker" for call in submits: assert any(kw.arg == "window_end" for kw in call.keywords), ( f"{module_name} leaves window_end to drift to the task's run time" @@ -187,6 +187,20 @@ def test_every_kind_hands_a_pinned_window_to_the_linker(module_name, _func_name) ) +@pytest.mark.parametrize(("module_name", "_func_name"), _EVALUATE_FUNCS) +def test_every_kind_passes_its_item_input_to_the_linker(module_name, _func_name): + """The scored item's question must travel with the score, not just its conversation id.""" + module = importlib.import_module(f"gooddata_eval.core.agentic.{module_name}") + tree = ast.parse(inspect.getsource(module)) + + submits = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and _callee_name(n) == "submit_trace_scoring"] + assert submits, f"{module_name} does not defer its Langfuse block to the linker" + for call in submits: + assert any(kw.arg == "item_input" for kw in call.keywords), ( + f"{module_name} scores a run without recording what question it answered" + ) + + @pytest.mark.parametrize(("module_name", "_func_name"), _EVALUATE_FUNCS) def test_every_kind_captures_the_window_before_deferring(module_name, _func_name): # The other half: window_end must actually be a captured timestamp. Passing a name that @@ -217,7 +231,7 @@ def test_the_linker_polls_the_window_it_was_given_instead_of_reading_the_clock() """ tree = ast.parse(inspect.getsource(_trace_linker)) blocks = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "_link_traces"] - assert blocks, "submit_trace_scoring no longer defers the trace lookup" + assert blocks, "submit_trace_scoring does not defer the trace lookup" for block in blocks: assert not _clock_reads(block), "the deferred lookup reads the clock instead of the pinned window"