Add Langfuse trace retrieval and scoring for user feedback

- Implement `get_langfuse_trace_id()` to fetch recent traces
- Update `vote()` function to log user feedback scores to Langfuse
- Add error handling and retry logic for trace retrieval
- Import necessary Langfuse and logging dependencies

Author: strickvl

llm-complete-guide/deployment_hf.py

@@ -5,8 +5,13 @@
 from constants import SECRET_NAME
 from utils.llm_utils import process_input_with_retrieval
 from zenml.client import Client
+from datetime import datetime, timedelta, UTC
+from langfuse import Langfuse
+import time
+from rich import print
+
+langfuse = Langfuse()
 
-# Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -24,25 +29,51 @@
     raise RuntimeError(f"Application startup failed: {e}")
 
 
-# def predict(message, history):
-#     try:
-#         return process_input_with_retrieval(
-#             input=message,
-#             n_items_retrieved=20,
-#             use_reranking=True,
-#             tracing_tags=["gradio", "web-interface", APP_ENVIRONMENT],
-#         )
-#     except Exception as e:
-#         logger.error(f"Error processing message: {e}")
-#         return f"Sorry, I encountered an error: {str(e)}"
+def get_langfuse_trace_id() -> str | None:
+    """Get the trace from Langfuse.
 
+    This is a very naive implementation. It simply returns the id of the first trace
+    in the last 60 seconds. Will retry up to 3 times if no traces are found or if
+    there's an error.
 
-# # Launch the Gradio interface
-# interface = gr.ChatInterface(
-#     predict,
-#     title="ZenML Documentation Assistant",
-#     description="Ask me anything about ZenML!",
-# )
+    Returns:
+        str | None: The trace ID if found, None otherwise
+    """
+    logger.info("Getting trace from Langfuse")
+    retries = 0
+    max_retries = 3
+    while retries < max_retries:
+        try:
+            # Wait 5 seconds before making the API call
+            time.sleep(5)
+            traces = langfuse.fetch_traces(
+                limit=1, order_by="timestamp.desc"
+            ).data
+            if not traces:
+                retries += 1
+                if retries == max_retries:
+                    logger.error(
+                        f"No traces found after {max_retries} attempts"
+                    )
+                    return None
+                logger.warning(
+                    f"No traces found (attempt {retries}/{max_retries})"
+                )
+                time.sleep(10)
+                continue
+            return traces[0].id
+        except Exception as e:
+            retries += 1
+            if retries == max_retries:
+                logger.error(
+                    f"Error fetching traces after {max_retries} attempts: {e}"
+                )
+                return None
+            logger.warning(
+                f"Error fetching traces (attempt {retries}/{max_retries}): {e}"
+            )
+            time.sleep(10)
+    return None
 
 
 def vote(data: gr.LikeData):
@@ -51,13 +82,25 @@ def vote(data: gr.LikeData):
     Args:
         data (gr.LikeData): The vote data.
     """
-    trace = logger.info("Getting trace from Langfuse")
 
+    trace_id = get_langfuse_trace_id()
     logger.info(f"Vote data: {data}")
     if data.liked:
         logger.info("Vote up")
+        langfuse.score(
+            trace_id=trace_id,
+            name="user-explicit-feedback",
+            value="like",
+            comment="I like this response",
+        )
     else:
         logger.info("Vote down")
+        langfuse.score(
+            trace_id=trace_id,
+            name="user-explicit-feedback",
+            value="dislike",
+            comment="I don't like the response",
+        )
 
 
 def predict(message, history):