Merge pull request #1794 from gooddata/jt/langfuse-v4-otlp-foundations

feat(gooddata-eval): Langfuse v4 foundations — OTLP encoder, experiment span builder, LANGFUSE_BASE_URL

Author: tychtjan

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:

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,17 +18,17 @@
 
 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")
 
 
 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:

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,
+    )

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