33
44from __future__ import annotations
55
6- import base64
76import logging
8- import os
97import threading
108import time
11- import uuid
129from collections .abc import Iterator
1310from contextlib import contextmanager
1411from datetime import datetime , timedelta , timezone
1815
1916from gooddata_eval .core .agentic ._trace_linker import link_cancel_event , linking_is_inline , warn_from_worker
2017from gooddata_eval .core .config import ReasoningEffort , env_flag , normalize_reasoning_effort
21- from gooddata_eval .core .langfuse ._env import resolve_base_url
18+ from gooddata_eval .core .langfuse ._env import credentials_present
19+ from gooddata_eval .core .langfuse .client import HttpxLangfuseClient
2220
23- _log = logging .getLogger (__name__ )
24-
25- # ---------------------------------------------------------------------------
26- # httpx-based Langfuse client — Python 3.14 safe (no Langfuse SDK required)
27- # ---------------------------------------------------------------------------
21+ # Part of this module's public surface: external callers import both names from here.
22+ from gooddata_eval .core .langfuse .observations import TraceSummary as _TraceObj # noqa: F401
2823
29-
30- class _TraceObj :
31- """Duck-type wrapper around a raw Langfuse trace dict."""
32-
33- def __init__ (self , raw : dict ) -> None :
34- self .id : str = raw .get ("id" , "" )
35- self .metadata : dict = raw .get ("metadata" ) or {}
36- self .session_id : str | None = raw .get ("sessionId" ) or raw .get ("session_id" )
37- self .latency : float = float (raw .get ("latency" ) or 0.0 )
38- self .total_cost : float = float (raw .get ("totalCost" ) or raw .get ("total_cost" ) or 0.0 )
39-
40-
41- class _TraceListResult :
42- def __init__ (self , data : list [_TraceObj ]) -> None :
43- self .data = data
44-
45-
46- class _TraceAPI :
47- def __init__ (self , client : httpx .Client ) -> None :
48- self ._client = client
49-
50- def list (
51- self , from_timestamp : Any , to_timestamp : Any , limit : int , session_id : str | None = None
52- ) -> _TraceListResult :
53- """List traces in a window, optionally narrowed to one session server-side.
54-
55- ``session_id`` is what makes ``limit`` a non-issue. Without it the endpoint returns
56- every trace in the window newest-first, so an eval workspace busy enough to put more
57- than ``limit`` traces inside one item's window pushes that item's OWN (oldest) trace
58- off the page -- it then polls its whole retry budget against a page that can never
59- contain it, and the score orphans with only a generic "no trace found" line to show
60- for it. Concurrency makes that likelier by overlapping every item's window. Named
61- ``session_id`` because ``_fetch_traces_for_session`` probes for exactly that
62- parameter; gen-ai sets sessionId = conversationId.
63- """
64-
65- def _ts (v : Any ) -> str :
66- return v .isoformat () if hasattr (v , "isoformat" ) else str (v )
67-
68- params : dict [str , Any ] = {
69- "fromTimestamp" : _ts (from_timestamp ),
70- "toTimestamp" : _ts (to_timestamp ),
71- "limit" : limit ,
72- }
73- # `is not None`, not truthiness: an empty id is a real filter value that matches
74- # nothing. Dropped here, the query would return the whole padded window for the
75- # caller's post-check to throw away, page after page, for the poll's whole budget.
76- if session_id is not None :
77- params ["sessionId" ] = session_id
78- resp = self ._client .get ("/api/public/traces" , params = params )
79- resp .raise_for_status ()
80- return _TraceListResult ([_TraceObj (t ) for t in resp .json ().get ("data" , [])])
81-
82-
83- class _DatasetRunItemsAPI :
84- def __init__ (self , client : httpx .Client ) -> None :
85- self ._client = client
86-
87- def create (
88- self ,
89- run_name : str ,
90- dataset_item_id : str ,
91- trace_id : str ,
92- metadata : dict | None = None ,
93- run_description : str = "" ,
94- ) -> None :
95- self ._client .post (
96- "/api/public/dataset-run-items" ,
97- json = {
98- "runName" : run_name ,
99- "datasetItemId" : dataset_item_id ,
100- "traceId" : trace_id ,
101- "metadata" : metadata or {},
102- "runDescription" : run_description ,
103- },
104- ).raise_for_status ()
105-
106-
107- class _LangfuseAPI :
108- def __init__ (self , client : httpx .Client ) -> None :
109- self .trace = _TraceAPI (client )
110- self .dataset_run_items = _DatasetRunItemsAPI (client )
111-
112-
113- class HttpxLangfuseClient :
114- """Minimal Langfuse client using httpx — works on Python 3.14 (no Langfuse SDK needed)."""
115-
116- def __init__ (self ) -> None :
117- host = resolve_base_url ()
118- pub = os .environ .get ("LANGFUSE_PUBLIC_KEY" , "" )
119- sec = os .environ .get ("LANGFUSE_SECRET_KEY" , "" )
120- if not pub or not sec :
121- raise RuntimeError (
122- "Langfuse credentials not set. "
123- "Export LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY before using --langfuse."
124- )
125- creds = base64 .b64encode (f"{ pub } :{ sec } " .encode ()).decode ()
126- self ._http = httpx .Client (
127- base_url = host ,
128- headers = {"Authorization" : f"Basic { creds } " },
129- timeout = 10 ,
130- )
131- self .api = _LangfuseAPI (self ._http )
132-
133- def create_score (
134- self ,
135- trace_id : str ,
136- name : str ,
137- value : float ,
138- data_type : str ,
139- comment : str | None = None ,
140- ) -> None :
141- now = datetime .now (timezone .utc ).isoformat ()
142- # Langfuse API requires numeric value for BOOLEAN type (1.0/0.0), not JSON booleans
143- if isinstance (value , bool ):
144- value = 1.0 if value else 0.0
145- body : dict [str , Any ] = {
146- "id" : str (uuid .uuid4 ()),
147- "traceId" : trace_id ,
148- "name" : name ,
149- "value" : value ,
150- "dataType" : data_type ,
151- }
152- if comment :
153- body ["comment" ] = comment
154- self ._http .post (
155- "/api/public/ingestion" ,
156- json = {"batch" : [{"id" : str (uuid .uuid4 ()), "timestamp" : now , "type" : "score-create" , "body" : body }]},
157- ).raise_for_status ()
158-
159- def update_trace_version (self , trace_id : str , version : str ) -> None :
160- """Upsert the trace version field via the ingestion endpoint."""
161- now = datetime .now (timezone .utc ).isoformat ()
162- self ._http .post (
163- "/api/public/ingestion" ,
164- json = {
165- "batch" : [
166- {
167- "id" : str (uuid .uuid4 ()),
168- "timestamp" : now ,
169- "type" : "trace-create" ,
170- "body" : {"id" : trace_id , "version" : version },
171- }
172- ]
173- },
174- ).raise_for_status ()
175-
176- def flush (self ) -> None :
177- pass # no client-side batching
178-
179- def close (self ) -> None :
180- self ._http .close ()
24+ _log = logging .getLogger (__name__ )
18125
18226
18327def make_langfuse_client () -> HttpxLangfuseClient :
@@ -191,7 +35,7 @@ def langfuse_credentials_present() -> bool:
19135 Separate from ``try_make_langfuse_client`` so a caller can ask the question without
19236 opening an httpx client it does not intend to use.
19337 """
194- return bool ( os . environ . get ( "LANGFUSE_PUBLIC_KEY" )) and bool ( os . environ . get ( "LANGFUSE_SECRET_KEY" ) )
38+ return credentials_present ( )
19539
19640
19741def try_make_langfuse_client () -> HttpxLangfuseClient | None :
@@ -418,11 +262,7 @@ def _set_trace_version(langfuse: Any, trace_id: str, version: str) -> None:
418262 """Write model version into the Langfuse trace version field."""
419263 try :
420264 if hasattr (langfuse , "update_trace_version" ):
421- # HttpxLangfuseClient path
422265 langfuse .update_trace_version (trace_id , version )
423- elif hasattr (langfuse , "trace" ):
424- # Langfuse Python SDK path (v2+)
425- langfuse .trace (id = trace_id , version = version )
426266 except Exception as exc :
427267 _log .warning ("Failed to set trace version %r on %s: %s" , version , trace_id , exc )
428268
0 commit comments