@@ -366,3 +366,161 @@ async def fetch_langfuse_traces(
366366 raise
367367 except Exception as e :
368368 raise HTTPException (status_code = 500 , detail = f"Error fetching traces from Langfuse: { str (e )} )
369+
370+
371+ async def pointwise_fetch_langfuse_trace (
372+ config : ProxyConfig ,
373+ redis_client : redis .Redis ,
374+ request : Request ,
375+ params : TracesParams ,
376+ ):
377+ """
378+ Fetch the latest trace from Langfuse for the specified project.
379+
380+ Since insertion_ids are UUID v7 (time-ordered), we only fetch the last one
381+ as it contains all accumulated information from the pointwise evaluation.
382+
383+ Returns a single trace object or raises if not found.
384+ """
385+
386+ # Preprocess traces request
387+ if config .preprocess_traces_request :
388+ params = config .preprocess_traces_request (request , params )
389+
390+ tags = params .tags
391+ project_id = params .project_id
392+ user_id = params .user_id
393+ session_id = params .session_id
394+ name = params .name
395+ environment = params .environment
396+ version = params .version
397+ release = params .release
398+ fields = params .fields
399+ hours_back = params .hours_back
400+ from_timestamp = params .from_timestamp
401+ to_timestamp = params .to_timestamp
402+ sleep_between_gets = params .sleep_between_gets
403+ max_retries = params .max_retries
404+
405+ # Use default project if not specified
406+ if project_id is None :
407+ project_id = config .default_project_id
408+
409+ # Validate project_id
410+ if project_id not in config .langfuse_keys :
411+ raise HTTPException (
412+ status_code = 404 ,
413+ detail = f"Project ID '{ project_id } { list (config .langfuse_keys .keys ())} ,
414+ )
415+
416+ # Extract rollout_id from tags for Redis lookup
417+ rollout_id = _extract_tag_value (tags , "rollout_id:" )
418+
419+ try :
420+ # Import the Langfuse adapter
421+ from langfuse import Langfuse
422+
423+ # Create Langfuse client with the project's keys
424+ logger .debug (f"Connecting to Langfuse at { config .langfuse_host } { project_id } )
425+ langfuse_client = Langfuse (
426+ public_key = config .langfuse_keys [project_id ]["public_key" ],
427+ secret_key = config .langfuse_keys [project_id ]["secret_key" ],
428+ host = config .langfuse_host ,
429+ )
430+
431+ # Parse datetime strings if provided
432+ from_ts = None
433+ to_ts = None
434+ if from_timestamp :
435+ from_ts = datetime .fromisoformat (from_timestamp .replace ("Z" , "+00:00" ))
436+ if to_timestamp :
437+ to_ts = datetime .fromisoformat (to_timestamp .replace ("Z" , "+00:00" ))
438+
439+ # Determine time window: explicit from/to takes precedence over hours_back
440+ if from_ts is None and to_ts is None and hours_back :
441+ to_ts = datetime .now ()
442+ from_ts = to_ts - timedelta (hours = hours_back )
443+
444+ # Get expected insertion_ids from Redis for completeness checking
445+ expected_ids : Set [str ] = set ()
446+ if rollout_id :
447+ expected_ids = get_insertion_ids (redis_client , rollout_id )
448+ logger .info (f"Pointwise fetch for rollout_id '{ rollout_id } { len (expected_ids )} )
449+ if not expected_ids :
450+ logger .warning (
451+ f"No expected insertion_ids found in Redis for rollout '{ rollout_id }
452+ )
453+ raise HTTPException (
454+ status_code = 500 ,
455+ detail = f"No expected insertion_ids found in Redis for rollout '{ rollout_id } ,
456+ )
457+
458+ # Get the latest (last) insertion_id since UUID v7 is time-ordered
459+ latest_insertion_id = max (expected_ids ) # UUID v7 max = newest
460+ logger .info (f"Targeting latest insertion_id (last5): { latest_insertion_id [- 5 :]} { rollout_id } )
461+
462+ for retry in range (max_retries ):
463+ # Fetch trace list targeting the latest insertion_id
464+ traces = await _fetch_trace_list_with_retry (
465+ langfuse_client ,
466+ page = 1 ,
467+ limit = 1 , # Only need the one trace
468+ tags = [f"insertion_id:{ latest_insertion_id } ],
469+ user_id = user_id ,
470+ session_id = session_id ,
471+ name = name ,
472+ environment = environment ,
473+ version = version ,
474+ release = release ,
475+ fields = fields ,
476+ from_ts = from_ts ,
477+ to_ts = to_ts ,
478+ max_retries = max_retries ,
479+ )
480+
481+ if traces and traces .data :
482+ # Get the trace info
483+ trace_info = traces .data [0 ]
484+ logger .debug (f"Found trace { trace_info .id } { latest_insertion_id [- 5 :]} )
485+
486+ # Fetch full trace details
487+ trace_full = await _fetch_trace_detail_with_retry (
488+ langfuse_client ,
489+ trace_info .id ,
490+ max_retries ,
491+ )
492+
493+ if trace_full :
494+ trace_dict = _serialize_trace_to_dict (trace_full )
495+ logger .info (
496+ f"Successfully fetched latest trace for rollout '{ rollout_id } { latest_insertion_id [- 5 :]}
497+ )
498+ return LangfuseTracesResponse (
499+ project_id = project_id ,
500+ total_traces = 1 ,
501+ traces = [TraceResponse (** trace_dict )],
502+ )
503+
504+ # If not successful and not last retry, sleep and continue
505+ if retry < max_retries - 1 :
506+ wait_time = 2 ** retry
507+ logger .info (
508+ f"Pointwise fetch attempt { retry + 1 } { max_retries } { rollout_id } { latest_insertion_id [- 5 :]} { wait_time }
509+ )
510+ await asyncio .sleep (wait_time )
511+
512+ # After all retries failed
513+ logger .error (
514+ f"Failed to fetch latest trace for rollout '{ rollout_id } { latest_insertion_id [- 5 :]} { max_retries }
515+ )
516+ raise HTTPException (
517+ status_code = 404 ,
518+ detail = f"Failed to fetch latest trace for rollout '{ rollout_id } { max_retries } ,
519+ )
520+
521+ except ImportError :
522+ raise HTTPException (status_code = 500 , detail = "Langfuse SDK not installed. Install with: pip install langfuse" )
523+ except HTTPException :
524+ raise
525+ except Exception as e :
526+ raise HTTPException (status_code = 500 , detail = f"Error fetching latest trace from Langfuse: { str (e )} )
0 commit comments