feat(langfuse): evaluate existing traces and return ASSERT scores

.github/workflows/regression.yml

@@ -30,6 +30,7 @@ on:
       # removing each and watching the suite fail):
       #   tests/test_library_e2e.py         reads langgraph-foundry-hosted/eval_config.yaml
       #   tests/test_tool_module_sandbox.py imports examples.agents.health_assistant
+      - 'examples/langfuse_trace_evaluation/**'
       # Joint AgentShield + ASSERT demo: gate doc + example changes that
       # claim measured eval-fix-loop numbers (see PR #43 / case study).
       - 'examples/incident_triage_agent/**'

assert_ai/core/judge.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import logging
 import re
@@ -43,6 +44,7 @@
     "get_verdict_dimension",
     "has_successful_judge_verdict",
     "infer_judge_status",
+    "inference_row_sha256",
     "is_not_applicable_dimension",
     "is_valid_confidence_label",
     "is_valid_event_flag",
@@ -214,6 +216,17 @@ def infer_judge_status(record: Dict[str, Any]) -> str:
     return "ok" if success else "judge_failed"
 
 
+def inference_row_sha256(row: Dict[str, Any]) -> str:
+    """Fingerprint the exact inference content a score row judges."""
+    payload = json.dumps(
+        row,
+        ensure_ascii=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
 def has_successful_judge_verdict(
     verdict: Optional[Dict[str, Any]],
     required_dimension_names: list[str] | None = None,

assert_ai/integrations/langfuse/__init__.py

@@ -0,0 +1,37 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Export completed ASSERT judgments to Langfuse for storage and visualization."""
+
+from assert_ai.integrations.langfuse.client import LangfuseHTTPClient
+from assert_ai.integrations.langfuse.errors import (
+    LangfuseAdapterError,
+    LangfuseAuthError,
+    LangfuseConfigurationError,
+    LangfuseConnectionError,
+    LangfuseContractError,
+    LangfuseHTTPError,
+    LangfuseResponseError,
+)
+from assert_ai.integrations.langfuse.exporter import ExportSummary, LangfuseExporter
+from assert_ai.integrations.langfuse.mapping import (
+    inference_to_otlp_trace,
+    trace_ids,
+    verdict_dimension_to_score,
+)
+
+__all__ = [
+    "ExportSummary",
+    "LangfuseAdapterError",
+    "LangfuseAuthError",
+    "LangfuseConfigurationError",
+    "LangfuseConnectionError",
+    "LangfuseContractError",
+    "LangfuseExporter",
+    "LangfuseHTTPClient",
+    "LangfuseHTTPError",
+    "LangfuseResponseError",
+    "inference_to_otlp_trace",
+    "trace_ids",
+    "verdict_dimension_to_score",
+]

assert_ai/integrations/langfuse/client.py

@@ -0,0 +1,193 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Minimal standard-library client for the Langfuse public HTTP APIs."""
+
+from __future__ import annotations
+
+import base64
+import ipaddress
+import json
+import os
+from collections.abc import Mapping
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+from assert_ai.integrations.langfuse.errors import (
+    LangfuseAuthError,
+    LangfuseConfigurationError,
+    LangfuseConnectionError,
+    LangfuseHTTPError,
+    LangfuseResponseError,
+)
+
+_OTLP_TRACES_PATH = "/api/public/otel/v1/traces"
+_SCORES_PATH = "/api/public/scores"
+
+
+class _RejectRedirectHandler(HTTPRedirectHandler):
+    """Keep Basic Auth credentials on the configured Langfuse origin."""
+
+    def redirect_request(
+        self,
+        request: Request,
+        file_pointer: Any,
+        code: int,
+        message: str,
+        headers: Any,
+        new_url: str,
+    ) -> None:
+        return None
+
+
+class LangfuseHTTPClient:
+    """Post OTLP traces and scores without importing the Langfuse SDK."""
+
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        public_key: str,
+        secret_key: str,
+        timeout_s: float = 30.0,
+    ) -> None:
+        self._base_url = _validate_base_url(base_url)
+        if not public_key or not secret_key:
+            raise LangfuseConfigurationError(
+                "Langfuse public and secret keys must both be non-empty"
+            )
+        if timeout_s <= 0:
+            raise LangfuseConfigurationError("timeout_s must be greater than zero")
+        credentials = f"{public_key}:{secret_key}".encode()
+        self._authorization = "Basic " + base64.b64encode(credentials).decode("ascii")
+        self._timeout_s = timeout_s
+
+    @classmethod
+    def from_env(
+        cls,
+        env: Mapping[str, str] | None = None,
+        *,
+        timeout_s: float = 30.0,
+    ) -> "LangfuseHTTPClient":
+        """Build a client from current documented Langfuse environment names."""
+        values = os.environ if env is None else env
+        required = (
+            "LANGFUSE_BASE_URL",
+            "LANGFUSE_PUBLIC_KEY",
+            "LANGFUSE_SECRET_KEY",
+        )
+        missing = [name for name in required if not values.get(name)]
+        if missing:
+            raise LangfuseConfigurationError(
+                "Missing required Langfuse environment variable(s): "
+                + ", ".join(missing)
+            )
+        return cls(
+            base_url=values["LANGFUSE_BASE_URL"],
+            public_key=values["LANGFUSE_PUBLIC_KEY"],
+            secret_key=values["LANGFUSE_SECRET_KEY"],
+            timeout_s=timeout_s,
+        )
+
+    def post_trace(self, payload: dict[str, Any]) -> dict[str, Any]:
+        """Post one OTLP/HTTP JSON trace payload."""
+        return self._post_json(
+            _OTLP_TRACES_PATH,
+            payload,
+            extra_headers={"x-langfuse-ingestion-version": "4"},
+        )
+
+    def post_score(self, payload: dict[str, Any]) -> dict[str, Any]:
+        """Post one score through the stable public Scores API."""
+        return self._post_json(_SCORES_PATH, payload)
+
+    def _post_json(
+        self,
+        endpoint: str,
+        payload: dict[str, Any],
+        *,
+        extra_headers: Mapping[str, str] | None = None,
+    ) -> dict[str, Any]:
+        headers = {
+            "Accept": "application/json",
+            "Authorization": self._authorization,
+            "Content-Type": "application/json",
+            "User-Agent": "assert-ai-langfuse-bridge",
+        }
+        headers.update(extra_headers or {})
+        request = Request(
+            self._base_url + endpoint,
+            data=json.dumps(
+                payload,
+                ensure_ascii=False,
+                separators=(",", ":"),
+                sort_keys=True,
+            ).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        try:
+            opener = build_opener(_RejectRedirectHandler())
+            with opener.open(request, timeout=self._timeout_s) as response:
+                raw = response.read()
+        except HTTPError as exc:
+            error_type = (
+                LangfuseAuthError
+                if exc.code in (401, 403)
+                else LangfuseHTTPError
+            )
+            raise error_type(status_code=exc.code, endpoint=endpoint) from exc
+        except (URLError, TimeoutError, OSError) as exc:
+            raise LangfuseConnectionError(
+                f"Unable to reach the Langfuse endpoint for {endpoint}"
+            ) from exc
+
+        try:
+            decoded = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise LangfuseResponseError(
+                f"Langfuse returned invalid JSON for {endpoint}"
+            ) from exc
+        if not isinstance(decoded, dict):
+            raise LangfuseResponseError(
+                f"Langfuse returned a non-object JSON response for {endpoint}"
+            )
+        return decoded
+
+
+def _validate_base_url(value: str) -> str:
+    base_url = value.strip().rstrip("/")
+    parsed = urlsplit(base_url)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or not parsed.netloc
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.query
+        or parsed.fragment
+        or parsed.path not in {"", "/"}
+    ):
+        raise LangfuseConfigurationError(
+            "LANGFUSE_BASE_URL must be an http(s) origin without credentials or a path"
+        )
+    if parsed.scheme == "http" and not _is_loopback_host(parsed.hostname):
+        raise LangfuseConfigurationError(
+            "LANGFUSE_BASE_URL must use HTTPS except for a loopback development server"
+        )
+    return base_url
+
+
+def _is_loopback_host(hostname: str | None) -> bool:
+    if hostname == "localhost":
+        return True
+    if hostname is None:
+        return False
+    try:
+        return ipaddress.ip_address(hostname).is_loopback
+    except ValueError:
+        return False
+
+
+__all__ = ["LangfuseHTTPClient"]

assert_ai/integrations/langfuse/errors.py

@@ -0,0 +1,37 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Typed errors raised by the optional Langfuse artifact exporter."""
+
+
+class LangfuseAdapterError(Exception):
+    """Base class for Langfuse integration failures."""
+
+
+class LangfuseConfigurationError(LangfuseAdapterError):
+    """The local Langfuse configuration is missing or invalid."""
+
+
+class LangfuseContractError(LangfuseAdapterError):
+    """An ASSERT artifact does not satisfy the export contract."""
+
+
+class LangfuseConnectionError(LangfuseAdapterError):
+    """The Langfuse endpoint could not be reached."""
+
+
+class LangfuseHTTPError(LangfuseAdapterError):
+    """Langfuse returned an unsuccessful HTTP response."""
+
+    def __init__(self, *, status_code: int, endpoint: str) -> None:
+        self.status_code = status_code
+        self.endpoint = endpoint
+        super().__init__(f"Langfuse returned HTTP {status_code} for {endpoint}")
+
+
+class LangfuseAuthError(LangfuseHTTPError):
+    """Langfuse rejected the configured project credentials."""
+
+
+class LangfuseResponseError(LangfuseAdapterError):
+    """Langfuse returned a response that did not match its public API."""