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 5d8e8e813..5adef6d51 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py @@ -18,6 +18,7 @@ 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 resolve_base_url _log = logging.getLogger(__name__) @@ -113,7 +114,7 @@ class HttpxLangfuseClient: """Minimal Langfuse client using httpx — works on Python 3.14 (no Langfuse SDK needed).""" def __init__(self) -> None: - host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com").rstrip("/") + host = resolve_base_url() pub = os.environ.get("LANGFUSE_PUBLIC_KEY", "") sec = os.environ.get("LANGFUSE_SECRET_KEY", "") if not pub or not sec: diff --git a/packages/gooddata-eval/src/gooddata_eval/core/dataset/langfuse_source.py b/packages/gooddata-eval/src/gooddata_eval/core/dataset/langfuse_source.py index 90b56e05f..83c7d5f26 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/dataset/langfuse_source.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/dataset/langfuse_source.py @@ -8,7 +8,8 @@ Credentials are read from the standard Langfuse environment variables: LANGFUSE_PUBLIC_KEY — your public key (pk-lf-...) LANGFUSE_SECRET_KEY — your secret key (sk-lf-...) - LANGFUSE_HOST — base URL, e.g. https://us.cloud.langfuse.com (default) + LANGFUSE_BASE_URL — base URL, e.g. https://us.cloud.langfuse.com (preferred) + LANGFUSE_HOST — base URL, legacy alias for LANGFUSE_BASE_URL """ import base64 @@ -17,9 +18,9 @@ import httpx +from gooddata_eval.core.langfuse._env import resolve_base_url from gooddata_eval.core.models import DatasetItem, SummaryInput -_DEFAULT_HOST = "https://cloud.langfuse.com" _PAGE_SIZE = 100 _T = TypeVar("_T") @@ -27,7 +28,7 @@ def _make_client() -> httpx.Client: """Build an httpx client with Langfuse basic-auth headers.""" - host = os.environ.get("LANGFUSE_HOST", _DEFAULT_HOST).rstrip("/") + host = resolve_base_url() pub = os.environ.get("LANGFUSE_PUBLIC_KEY", "") sec = os.environ.get("LANGFUSE_SECRET_KEY", "") if not pub or not sec: diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/_env.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/_env.py new file mode 100644 index 000000000..eee5e89dd --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/_env.py @@ -0,0 +1,39 @@ +# (C) 2026 GoodData Corporation +"""Langfuse environment resolution: base URL and credentials, shared by all Langfuse call sites.""" + +from __future__ import annotations + +import base64 +import os + +import httpx + +_DEFAULT_BASE_URL = "https://us.cloud.langfuse.com" + + +def resolve_base_url() -> str: + """Resolve the Langfuse base URL: `LANGFUSE_BASE_URL` > `LANGFUSE_HOST` > the US cloud region GoodData uses.""" + base = os.environ.get("LANGFUSE_BASE_URL") or os.environ.get("LANGFUSE_HOST") or _DEFAULT_BASE_URL + return base.rstrip("/") + + +def credentials_present() -> bool: + return bool(os.environ.get("LANGFUSE_PUBLIC_KEY")) and bool(os.environ.get("LANGFUSE_SECRET_KEY")) + + +def basic_auth_header() -> str: + if not credentials_present(): + raise RuntimeError("Langfuse credentials not set. Export LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY.") + pub = os.environ["LANGFUSE_PUBLIC_KEY"] + sec = os.environ["LANGFUSE_SECRET_KEY"] + creds = base64.b64encode(f"{pub}:{sec}".encode()).decode() + return f"Basic {creds}" + + +def make_http_client(*, timeout: float, transport: httpx.BaseTransport | None = None) -> httpx.Client: + return httpx.Client( + base_url=resolve_base_url(), + headers={"Authorization": basic_auth_header()}, + timeout=timeout, + transport=transport, + ) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/experiment.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/experiment.py new file mode 100644 index 000000000..b3e9745d6 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/experiment.py @@ -0,0 +1,156 @@ +# (C) 2026 GoodData Corporation +"""Langfuse experiment root-span construction and score-target resolution.""" + +from __future__ import annotations + +import json +import uuid +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from gooddata_eval.core.langfuse.otlp import ( + ATTR_ENVIRONMENT, + ATTR_EXPERIMENT_DATASET_ID, + ATTR_EXPERIMENT_DESCRIPTION, + ATTR_EXPERIMENT_ID, + ATTR_EXPERIMENT_ITEM_EXPECTED_OUTPUT, + ATTR_EXPERIMENT_ITEM_ID, + ATTR_EXPERIMENT_ITEM_METADATA_PREFIX, + ATTR_EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, + ATTR_EXPERIMENT_METADATA_PREFIX, + ATTR_EXPERIMENT_NAME, + ATTR_OBSERVATION_INPUT, + ATTR_OBSERVATION_METADATA_PREFIX, + ATTR_OBSERVATION_OUTPUT, + ATTR_OBSERVATION_TYPE, + ATTR_SESSION_ID, + ATTR_TRACE_METADATA_PREFIX, + ATTR_TRACE_NAME, + ATTR_TRACE_TAGS, + ATTR_VERSION, + Span, + flatten_metadata, + new_span_id, + new_trace_id, + otlp_attribute, +) + +# Fixed namespace for deriving experiment ids from run names — arbitrary but stable across processes. +_EXPERIMENT_NAMESPACE = uuid.UUID("6f6e2f5a-2f0e-4f0c-9c1b-8f6f2a3b9d10") + + +def experiment_id_for(run_name: str) -> str: + return str(uuid.uuid5(_EXPERIMENT_NAMESPACE, run_name)) + + +@dataclass(frozen=True) +class ExperimentRun: + name: str + dataset_id: str + metadata: dict[str, Any] | None = None + description: str | None = None + + +@dataclass(frozen=True) +class ExperimentItem: + item_id: str + input: Any = None + output: Any = None + expected_output: Any = None + metadata: dict[str, Any] | None = None + + +def build_experiment_root_span( + run: ExperimentRun | None, + item: ExperimentItem, + *, + start: datetime, + end: datetime, + trace_name: str, + session_id: str | None = None, + version: str | None = None, + tags: tuple[str, ...] = (), + observation_metadata: dict[str, Any] | None = None, + trace_metadata: dict[str, Any] | None = None, + environment: str | None = None, +) -> Span: + """Build the single root span gd-eval emits per (dataset item, run). + + With `run=None` this is a plain observation span carrying no `langfuse.experiment.*` + attributes at all — used when there is no experiment to attach the item to. + """ + if end < start: + end = start + span_id = new_span_id() + + attributes: list[dict[str, Any]] = [otlp_attribute(ATTR_OBSERVATION_TYPE, "span")] + if item.input is not None: + attributes.append(otlp_attribute(ATTR_OBSERVATION_INPUT, json.dumps(item.input, default=str))) + if item.output is not None: + attributes.append(otlp_attribute(ATTR_OBSERVATION_OUTPUT, json.dumps(item.output, default=str))) + attributes.extend(flatten_metadata(ATTR_OBSERVATION_METADATA_PREFIX, observation_metadata)) + + attributes.append(otlp_attribute(ATTR_TRACE_NAME, trace_name)) + if session_id is not None: + attributes.append(otlp_attribute(ATTR_SESSION_ID, session_id)) + if version is not None: + attributes.append(otlp_attribute(ATTR_VERSION, version)) + if environment is not None: + attributes.append(otlp_attribute(ATTR_ENVIRONMENT, environment)) + if tags: + attributes.append(otlp_attribute(ATTR_TRACE_TAGS, list(tags))) + attributes.extend(flatten_metadata(ATTR_TRACE_METADATA_PREFIX, trace_metadata)) + + if run is not None: + attributes.append(otlp_attribute(ATTR_EXPERIMENT_ID, experiment_id_for(run.name))) + attributes.append(otlp_attribute(ATTR_EXPERIMENT_NAME, run.name)) + attributes.append(otlp_attribute(ATTR_EXPERIMENT_DATASET_ID, run.dataset_id)) + if run.description is not None: + attributes.append(otlp_attribute(ATTR_EXPERIMENT_DESCRIPTION, run.description)) + attributes.extend(flatten_metadata(ATTR_EXPERIMENT_METADATA_PREFIX, run.metadata)) + + attributes.append(otlp_attribute(ATTR_EXPERIMENT_ITEM_ID, item.item_id)) + attributes.append(otlp_attribute(ATTR_EXPERIMENT_ITEM_ROOT_OBSERVATION_ID, span_id)) + if item.expected_output is not None: + attributes.append( + otlp_attribute(ATTR_EXPERIMENT_ITEM_EXPECTED_OUTPUT, json.dumps(item.expected_output, default=str)) + ) + attributes.extend(flatten_metadata(ATTR_EXPERIMENT_ITEM_METADATA_PREFIX, item.metadata)) + + return Span(trace_id=new_trace_id(), span_id=span_id, name=trace_name, start=start, end=end, attributes=attributes) + + +class ScoreTarget(str): + """Where a score for one evaluated item is written: the gen-ai trace, gd-eval's own + experiment root observation, or both. + + The `str` value is the gen-ai trace id when present, else the experiment trace id, else + empty — so a `ScoreTarget` can be used directly wherever a plain trace id string was used. + """ + + gen_ai_trace_id: str | None + experiment_trace_id: str | None + experiment_span_id: str | None + + def __new__( + cls, + gen_ai_trace_id: str | None = None, + experiment_trace_id: str | None = None, + experiment_span_id: str | None = None, + ) -> ScoreTarget: + value = gen_ai_trace_id or experiment_trace_id or "" + instance = super().__new__(cls, value) + instance.gen_ai_trace_id = gen_ai_trace_id + instance.experiment_trace_id = experiment_trace_id + instance.experiment_span_id = experiment_span_id + return instance + + def destinations(self) -> list[tuple[str, str | None]]: + """Score write destinations as `(trace_id, observation_id)` pairs, gen-ai first.""" + targets: list[tuple[str, str | None]] = [] + if self.gen_ai_trace_id: + targets.append((self.gen_ai_trace_id, None)) + if self.experiment_trace_id: + targets.append((self.experiment_trace_id, self.experiment_span_id)) + return targets diff --git a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py new file mode 100644 index 000000000..ed94a89c0 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/otlp.py @@ -0,0 +1,166 @@ +# (C) 2026 GoodData Corporation +"""Pure OTLP/HTTP JSON encoding for the Langfuse ingestion endpoint. No HTTP here.""" + +from __future__ import annotations + +import json +import secrets +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any + +from gooddata_eval._version import __version__ + +if TYPE_CHECKING: + import httpx + +# Observation (span-level) attributes. +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. +ATTR_TRACE_NAME = "langfuse.trace.name" +ATTR_SESSION_ID = "langfuse.session.id" +ATTR_TRACE_TAGS = "langfuse.trace.tags" +ATTR_TRACE_METADATA_PREFIX = "langfuse.trace.metadata" +ATTR_VERSION = "langfuse.version" +ATTR_ENVIRONMENT = "langfuse.environment" + +# Experiment attributes, present on every span of an experiment item's trace. +ATTR_EXPERIMENT_ID = "langfuse.experiment.id" +ATTR_EXPERIMENT_NAME = "langfuse.experiment.name" +ATTR_EXPERIMENT_DATASET_ID = "langfuse.experiment.dataset.id" +ATTR_EXPERIMENT_DESCRIPTION = "langfuse.experiment.description" +ATTR_EXPERIMENT_METADATA_PREFIX = "langfuse.experiment.metadata" + +# Experiment item attributes, root span only. +ATTR_EXPERIMENT_ITEM_ID = "langfuse.experiment.item.id" +ATTR_EXPERIMENT_ITEM_ROOT_OBSERVATION_ID = "langfuse.experiment.item.root_observation_id" +ATTR_EXPERIMENT_ITEM_EXPECTED_OUTPUT = "langfuse.experiment.item.expected_output" +ATTR_EXPERIMENT_ITEM_METADATA_PREFIX = "langfuse.experiment.item.metadata" + +_SERVICE_NAME = "gooddata-eval" +_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + + +def new_trace_id() -> str: + return uuid.uuid4().hex + + +def new_span_id() -> str: + return secrets.token_hex(8) + + +def unix_nano(dt: datetime) -> str: + """Nanoseconds since the epoch as a decimal string. Naive datetimes are treated as UTC.""" + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + delta = dt - _EPOCH + nanos = (delta.days * 86400 + delta.seconds) * 1_000_000_000 + delta.microseconds * 1000 + return str(nanos) + + +def otlp_attribute(key: str, value: Any) -> dict[str, Any]: + """Encode one key/value pair as an OTLP attribute, typed by the Python value's type. + + `bool` is checked before `int` because `bool` is an `int` subclass. + """ + if isinstance(value, bool): + return {"key": key, "value": {"boolValue": value}} + if isinstance(value, int): + return {"key": key, "value": {"intValue": str(value)}} + if isinstance(value, float): + return {"key": key, "value": {"doubleValue": value}} + if isinstance(value, str): + return {"key": key, "value": {"stringValue": value}} + if isinstance(value, (list, tuple)) and all(isinstance(item, str) for item in value): + return {"key": key, "value": {"arrayValue": {"values": [{"stringValue": item} for item in value]}}} + return {"key": key, "value": {"stringValue": json.dumps(value, default=str)}} + + +def flatten_metadata(prefix: str, mapping: dict[str, Any] | None) -> list[dict[str, Any]]: + """Flatten a metadata mapping into dotted `prefix.` attributes, dropping `None` values. + + Scalars are typed via `otlp_attribute`; nested dicts and lists are encoded as a JSON string. + """ + if not mapping: + return [] + attributes = [] + for key, value in mapping.items(): + if value is None: + continue + full_key = f"{prefix}.{key}" + if isinstance(value, (dict, list, tuple)): + attributes.append({"key": full_key, "value": {"stringValue": json.dumps(value, default=str)}}) + else: + attributes.append(otlp_attribute(full_key, value)) + return attributes + + +@dataclass +class Span: + trace_id: str + span_id: str + name: str + start: datetime + end: datetime + attributes: list[dict[str, Any]] + status_code: int = 1 + + +def encode_export_request( + spans: list[Span], *, scope_name: str = _SERVICE_NAME, scope_version: str = __version__ +) -> dict[str, Any]: + """Build the OTLP/JSON export request body for `POST /api/public/otel/v1/traces`.""" + otlp_spans = [ + { + "traceId": span.trace_id, + "spanId": span.span_id, + "name": span.name, + "kind": 1, + "startTimeUnixNano": unix_nano(span.start), + "endTimeUnixNano": unix_nano(span.end), + "status": {"code": span.status_code}, + "attributes": span.attributes, + } + for span in spans + ] + return { + "resourceSpans": [ + { + "resource": {"attributes": [{"key": "service.name", "value": {"stringValue": _SERVICE_NAME}}]}, + "scopeSpans": [{"scope": {"name": scope_name, "version": scope_version}, "spans": otlp_spans}], + } + ] + } + + +def parse_export_response(resp: httpx.Response) -> None: + """Raise on a failed or partially-rejected export; return `None` on success.""" + if not (200 <= resp.status_code < 300): + raise RuntimeError(f"Langfuse OTLP export failed: {resp.status_code} {resp.text[:300]}") + if not resp.content: + return None + try: + body = resp.json() + except ValueError: + return None + if not isinstance(body, dict): + return None + partial = body.get("partialSuccess") + if isinstance(partial, dict) and _rejected_spans(partial) > 0: + raise RuntimeError(f"Langfuse OTLP export partially rejected: {partial.get('errorMessage', '')}") + return None + + +def _rejected_spans(partial: dict[str, Any]) -> int: + """`partialSuccess.rejectedSpans`, an int64 that OTLP/JSON may encode as a decimal string.""" + try: + return int(partial.get("rejectedSpans") or 0) + except (TypeError, ValueError): + return 0 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 ebb05af33..2058fb397 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/langfuse/sink.py @@ -12,6 +12,8 @@ import httpx +from gooddata_eval.core.langfuse._env import resolve_base_url + _MAX_LATENCY_S = 60.0 _QUALITY_WEIGHT = 0.6 _SPEED_WEIGHT = 0.2 @@ -62,7 +64,7 @@ def __init__( self._model_id = model_id self._provider_type = provider_type self._reasoning_effort = reasoning_effort - host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com").rstrip("/") + host = resolve_base_url() pub = os.environ.get("LANGFUSE_PUBLIC_KEY", "") sec = os.environ.get("LANGFUSE_SECRET_KEY", "") if not pub or not sec: diff --git a/packages/gooddata-eval/tests/test_langfuse_env.py b/packages/gooddata-eval/tests/test_langfuse_env.py new file mode 100644 index 000000000..bd6ae2dc5 --- /dev/null +++ b/packages/gooddata-eval/tests/test_langfuse_env.py @@ -0,0 +1,70 @@ +# (C) 2026 GoodData Corporation +from __future__ import annotations + +import base64 + +import pytest +from gooddata_eval.core.langfuse._env import basic_auth_header, credentials_present, make_http_client, resolve_base_url + + +def test_resolve_base_url_prefers_base_url_over_host(monkeypatch): + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://base.example.com") + monkeypatch.setenv("LANGFUSE_HOST", "https://host.example.com") + assert resolve_base_url() == "https://base.example.com" + + +def test_resolve_base_url_falls_back_to_host(monkeypatch): + monkeypatch.delenv("LANGFUSE_BASE_URL", raising=False) + monkeypatch.setenv("LANGFUSE_HOST", "https://host.example.com") + assert resolve_base_url() == "https://host.example.com" + + +def test_resolve_base_url_default(monkeypatch): + monkeypatch.delenv("LANGFUSE_BASE_URL", raising=False) + monkeypatch.delenv("LANGFUSE_HOST", raising=False) + assert resolve_base_url() == "https://us.cloud.langfuse.com" + + +def test_resolve_base_url_strips_trailing_slash(monkeypatch): + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://base.example.com/") + assert resolve_base_url() == "https://base.example.com" + + +def test_credentials_present_true(monkeypatch): + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") + assert credentials_present() is True + + +def test_credentials_present_false_when_missing(monkeypatch): + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") + assert credentials_present() is False + + +def test_basic_auth_header_missing_credentials_raises(monkeypatch): + monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) + monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False) + with pytest.raises(RuntimeError, match="credentials"): + basic_auth_header() + + +def test_basic_auth_header_encodes_credentials(monkeypatch): + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") + header = basic_auth_header() + assert header.startswith("Basic ") + assert base64.b64decode(header.removeprefix("Basic ")).decode() == "pk-test:sk-test" + + +def test_make_http_client_uses_resolved_base_url_and_auth(monkeypatch): + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://base.example.com") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") + + client = make_http_client(timeout=5) + try: + assert str(client.base_url).rstrip("/") == "https://base.example.com" + assert client.headers["Authorization"].startswith("Basic ") + finally: + client.close() diff --git a/packages/gooddata-eval/tests/test_langfuse_experiment.py b/packages/gooddata-eval/tests/test_langfuse_experiment.py new file mode 100644 index 000000000..bdafd20ac --- /dev/null +++ b/packages/gooddata-eval/tests/test_langfuse_experiment.py @@ -0,0 +1,206 @@ +# (C) 2026 GoodData Corporation +from __future__ import annotations + +import json +from datetime import datetime, timezone + +from gooddata_eval.core.langfuse.experiment import ( + ExperimentItem, + ExperimentRun, + ScoreTarget, + build_experiment_root_span, + experiment_id_for, +) + + +def _attr(span, key): + for attribute in span.attributes: + if attribute["key"] == key: + return attribute + return None + + +def test_experiment_id_for_is_stable(): + assert experiment_id_for("run0") == experiment_id_for("run0") + + +def test_experiment_id_for_differs_by_run_name(): + assert experiment_id_for("run0") != experiment_id_for("run1") + + +def test_build_experiment_root_span_root_observation_id_equals_span_id(): + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + attr = _attr(span, "langfuse.experiment.item.root_observation_id") + assert attr["value"]["stringValue"] == span.span_id + + +def test_build_experiment_root_span_dataset_id_only_when_run_given(): + item = ExperimentItem(item_id="item-1") + span_without_run = build_experiment_root_span( + None, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + assert _attr(span_without_run, "langfuse.experiment.dataset.id") is None + + run = ExperimentRun(name="run0", dataset_id="ds-1") + span_with_run = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + assert _attr(span_with_run, "langfuse.experiment.dataset.id")["value"]["stringValue"] == "ds-1" + + +def test_build_experiment_root_span_no_experiment_attrs_when_run_none(): + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + None, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + assert not any(a["key"].startswith("langfuse.experiment.") for a in span.attributes) + # still a valid observation span + assert _attr(span, "langfuse.observation.type")["value"]["stringValue"] == "span" + assert _attr(span, "langfuse.trace.name")["value"]["stringValue"] == "gd-eval: q" + + +def test_build_experiment_root_span_io_json_strings(): + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1", input={"question": "q"}, output={"passed": True}) + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + input_attr = _attr(span, "langfuse.observation.input") + output_attr = _attr(span, "langfuse.observation.output") + assert json.loads(input_attr["value"]["stringValue"]) == {"question": "q"} + assert json.loads(output_attr["value"]["stringValue"]) == {"passed": True} + + +def test_build_experiment_root_span_tags_array_value(): + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + tags=("gd-eval",), + ) + tags_attr = _attr(span, "langfuse.trace.tags") + assert tags_attr["value"] == {"arrayValue": {"values": [{"stringValue": "gd-eval"}]}} + + +def test_build_experiment_root_span_omits_tags_when_empty(): + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + assert _attr(span, "langfuse.trace.tags") is None + + +def test_build_experiment_root_span_clamps_end_before_start(): + start = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + end = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc) + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span(run, item, start=start, end=end, trace_name="gd-eval: q") + assert span.end == start + + +def test_build_experiment_root_span_omits_none_optional_attrs(): + run = ExperimentRun(name="run0", dataset_id="ds-1") + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + ) + assert _attr(span, "langfuse.session.id") is None + assert _attr(span, "langfuse.version") is None + assert _attr(span, "langfuse.environment") is None + assert _attr(span, "langfuse.experiment.description") is None + assert _attr(span, "langfuse.experiment.item.expected_output") is None + + +def test_build_experiment_root_span_full_wire_key_contract(): + run = ExperimentRun( + name="run0", + dataset_id="ds-1", + metadata={"model_version": "gpt", "testing_framework": "tavern-e2e"}, + ) + item = ExperimentItem(item_id="item-1") + span = build_experiment_root_span( + run, + item, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + trace_name="gd-eval: q", + session_id="c1", + version="gpt-5.2", + environment="staging", + observation_metadata={"gen_ai_trace_id": "abc", "conversation_id": "c1"}, + ) + keys = {a["key"] for a in span.attributes} + assert "langfuse.session.id" in keys + assert "langfuse.version" in keys + assert "langfuse.environment" in keys + assert "langfuse.observation.metadata.gen_ai_trace_id" in keys + assert "langfuse.observation.metadata.conversation_id" in keys + assert "langfuse.experiment.metadata.model_version" in keys + assert "langfuse.experiment.metadata.testing_framework" in keys + + +def test_score_target_str_value_prefers_gen_ai_trace_id(): + target = ScoreTarget("g", "t", "s") + assert target == "g" + assert target.gen_ai_trace_id == "g" + assert target.experiment_trace_id == "t" + assert target.experiment_span_id == "s" + + +def test_score_target_str_value_falls_back_to_experiment_trace_id(): + target = ScoreTarget(None, "t", "s") + assert str(target) == "t" + + +def test_score_target_destinations_order_and_skipping(): + target = ScoreTarget("g", "t", "s") + assert target.destinations() == [("g", None), ("t", "s")] + + +def test_score_target_destinations_skips_missing_gen_ai(): + target = ScoreTarget(None, "t", "s") + assert target.destinations() == [("t", "s")] + + +def test_score_target_destinations_empty_when_nothing_set(): + target = ScoreTarget() + assert target.destinations() == [] + assert str(target) == "" diff --git a/packages/gooddata-eval/tests/test_langfuse_otlp.py b/packages/gooddata-eval/tests/test_langfuse_otlp.py new file mode 100644 index 000000000..0c6dd30f0 --- /dev/null +++ b/packages/gooddata-eval/tests/test_langfuse_otlp.py @@ -0,0 +1,197 @@ +# (C) 2026 GoodData Corporation +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone + +import httpx +import pytest +from gooddata_eval.core.langfuse.otlp import ( + Span, + encode_export_request, + flatten_metadata, + new_span_id, + new_trace_id, + otlp_attribute, + parse_export_response, + unix_nano, +) + + +def test_new_trace_id_is_32_hex_chars(): + trace_id = new_trace_id() + assert re.fullmatch(r"[0-9a-f]{32}", trace_id) + + +def test_new_trace_id_is_unique(): + assert new_trace_id() != new_trace_id() + + +def test_new_span_id_is_16_hex_chars(): + span_id = new_span_id() + assert re.fullmatch(r"[0-9a-f]{16}", span_id) + + +def test_new_span_id_is_unique(): + assert new_span_id() != new_span_id() + + +def test_unix_nano_exact_value_for_known_utc_datetime(): + dt = datetime(2025, 9, 8, 10, 0, 0, tzinfo=timezone.utc) + assert unix_nano(dt) == "1757325600000000000" + + +def test_unix_nano_naive_datetime_treated_as_utc(): + naive = datetime(2025, 9, 8, 10, 0, 0) + aware = datetime(2025, 9, 8, 10, 0, 0, tzinfo=timezone.utc) + assert unix_nano(naive) == unix_nano(aware) + + +def test_unix_nano_includes_microsecond_precision(): + dt = datetime(2025, 9, 8, 10, 0, 0, 400000, tzinfo=timezone.utc) + assert unix_nano(dt) == "1757325600400000000" + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (True, {"boolValue": True}), + (False, {"boolValue": False}), + (42, {"intValue": "42"}), + (0, {"intValue": "0"}), + (0.5, {"doubleValue": 0.5}), + ("hello", {"stringValue": "hello"}), + ], +) +def test_otlp_attribute_typing_table(value, expected): + attr = otlp_attribute("k", value) + assert attr == {"key": "k", "value": expected} + + +def test_otlp_attribute_bool_checked_before_int(): + # bool is an int subclass; must map to boolValue, not intValue + attr = otlp_attribute("k", True) + assert attr["value"] == {"boolValue": True} + + +def test_otlp_attribute_list_of_str_is_array_value(): + attr = otlp_attribute("k", ["a", "b"]) + assert attr == {"key": "k", "value": {"arrayValue": {"values": [{"stringValue": "a"}, {"stringValue": "b"}]}}} + + +def test_otlp_attribute_dict_is_json_string(): + attr = otlp_attribute("k", {"a": 1}) + assert attr["value"]["stringValue"] == json.dumps({"a": 1}, default=str) + + +def test_flatten_metadata_skips_none_values(): + attrs = flatten_metadata("prefix", {"a": None, "b": "x"}) + assert attrs == [{"key": "prefix.b", "value": {"stringValue": "x"}}] + + +def test_flatten_metadata_types_scalars(): + attrs = flatten_metadata("prefix", {"n": 3, "f": 1.5, "flag": True}) + assert {"key": "prefix.n", "value": {"intValue": "3"}} in attrs + assert {"key": "prefix.f", "value": {"doubleValue": 1.5}} in attrs + assert {"key": "prefix.flag", "value": {"boolValue": True}} in attrs + + +def test_flatten_metadata_nested_dict_is_json_string(): + attrs = flatten_metadata("prefix", {"nested": {"x": 1}}) + assert attrs == [{"key": "prefix.nested", "value": {"stringValue": json.dumps({"x": 1}, default=str)}}] + + +def test_flatten_metadata_nested_list_is_json_string(): + attrs = flatten_metadata("prefix", {"items": ["a", "b"]}) + assert attrs == [{"key": "prefix.items", "value": {"stringValue": json.dumps(["a", "b"], default=str)}}] + + +def test_flatten_metadata_empty_mapping(): + assert flatten_metadata("prefix", {}) == [] + assert flatten_metadata("prefix", None) == [] + + +def test_encode_export_request_shape(): + span = Span( + trace_id="3f2a9c1e8b7d4e6f9a0b1c2d3e4f5a6b", + span_id="9a0b1c2d3e4f5a6b", + name="gd-eval: Show revenue by month", + start=datetime(2025, 9, 8, 10, 0, 0, tzinfo=timezone.utc), + end=datetime(2025, 9, 8, 10, 0, 18, 400000, tzinfo=timezone.utc), + attributes=[{"key": "langfuse.observation.type", "value": {"stringValue": "span"}}], + ) + request = encode_export_request([span]) + + resource_spans = request["resourceSpans"] + assert len(resource_spans) == 1 + resource = resource_spans[0]["resource"] + assert resource["attributes"] == [{"key": "service.name", "value": {"stringValue": "gooddata-eval"}}] + + scope_spans = resource_spans[0]["scopeSpans"] + assert len(scope_spans) == 1 + assert scope_spans[0]["scope"]["name"] == "gooddata-eval" + assert re.fullmatch(r"\d+\.\d+\.\d+", scope_spans[0]["scope"]["version"]) + + (otlp_span,) = scope_spans[0]["spans"] + assert otlp_span["traceId"] == span.trace_id + assert otlp_span["spanId"] == span.span_id + assert otlp_span["kind"] == 1 + assert otlp_span["status"] == {"code": 1} + assert otlp_span["startTimeUnixNano"].isdigit() + assert otlp_span["endTimeUnixNano"].isdigit() + assert int(otlp_span["startTimeUnixNano"]) <= int(otlp_span["endTimeUnixNano"]) + assert otlp_span["attributes"] == span.attributes + + +def test_encode_export_request_custom_scope(): + span = Span( + trace_id="a" * 32, + span_id="b" * 16, + name="s", + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 1, tzinfo=timezone.utc), + attributes=[], + ) + request = encode_export_request([span], scope_name="custom-scope", scope_version="9.9.9") + scope = request["resourceSpans"][0]["scopeSpans"][0]["scope"] + assert scope == {"name": "custom-scope", "version": "9.9.9"} + + +def test_parse_export_response_ok_empty_body(): + resp = httpx.Response(200, json={}) + assert parse_export_response(resp) is None + + +def test_parse_export_response_ok_no_content(): + resp = httpx.Response(200) + assert parse_export_response(resp) is None + + +def test_parse_export_response_partial_success_raises(): + resp = httpx.Response(200, json={"partialSuccess": {"rejectedSpans": 1, "errorMessage": "bad span"}}) + with pytest.raises(RuntimeError, match="bad span"): + parse_export_response(resp) + + +def test_parse_export_response_partial_success_string_count_raises(): + # OTLP/JSON encodes int64 fields as decimal strings. + resp = httpx.Response(200, json={"partialSuccess": {"rejectedSpans": "1", "errorMessage": "bad span"}}) + with pytest.raises(RuntimeError, match="bad span"): + parse_export_response(resp) + + +def test_parse_export_response_partial_success_zero_rejected_ok(): + resp = httpx.Response(200, json={"partialSuccess": {"rejectedSpans": 0}}) + assert parse_export_response(resp) is None + + +def test_parse_export_response_ok_non_json_body(): + resp = httpx.Response(200, text="not json") + assert parse_export_response(resp) is None + + +def test_parse_export_response_non_2xx_raises(): + resp = httpx.Response(400, text="bad request") + with pytest.raises(RuntimeError, match="400"): + parse_export_response(resp) diff --git a/packages/gooddata-eval/tests/test_langfuse_source.py b/packages/gooddata-eval/tests/test_langfuse_source.py index 7b61110d8..a9db1997e 100644 --- a/packages/gooddata-eval/tests/test_langfuse_source.py +++ b/packages/gooddata-eval/tests/test_langfuse_source.py @@ -2,7 +2,12 @@ from unittest.mock import MagicMock, patch import pytest -from gooddata_eval.core.dataset.langfuse_source import _infer_test_kind, _item_from_raw, load_langfuse_dataset +from gooddata_eval.core.dataset.langfuse_source import ( + _infer_test_kind, + _item_from_raw, + _make_client, + load_langfuse_dataset, +) def _raw_item(item_id, question, expected_output, dataset_name="ds"): @@ -97,6 +102,19 @@ def test_load_langfuse_dataset_calls_rest_api(monkeypatch): assert call_args[1]["params"]["datasetName"] == "my_dataset" +def test_make_client_prefers_langfuse_base_url_over_host(monkeypatch): + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test") + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://base.example.com") + monkeypatch.setenv("LANGFUSE_HOST", "https://host.example.com") + + client = _make_client() + try: + assert str(client.base_url).rstrip("/") == "https://base.example.com" + finally: + client.close() + + def test_load_langfuse_dataset_raises_on_missing_credentials(monkeypatch): monkeypatch.delenv("LANGFUSE_PUBLIC_KEY", raising=False) monkeypatch.delenv("LANGFUSE_SECRET_KEY", raising=False)