Merge pull request #1954 from theopenconversationkit/upgrade/langfusev3/DERCBOT-1746

[DERCBOT-1746] Langfuse V3 Callback Fixes

Author: assouktim

gen-ai/orchestrator-server/src/main/python/server/poetry.lock

@@ -19,9 +19,9 @@ opensearch-py = "^2.8.0"
 path = "^17.1.0"
 colorlog = "^6.9.0"
 boto3 = "^1.35.96"
-urllib3 = "^2.3.0"
+urllib3 = "^2.6.0"
 jinja2 = "^3.1.5"
-langfuse = "^3.3.0"
+langfuse = "^3.10.6"
 httpx-auth-awssigv4 = "^0.1.4"
 langchain-postgres = "^0.0.12"
 google-cloud-secret-manager = "^2.22.0"

gen-ai/orchestrator-server/src/main/python/server/pyproject.toml

@@ -18,7 +18,7 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
-from langfuse.langchain.CallbackHandler import LangchainCallbackHandler
+from langfuse.langchain import CallbackHandler as LangfuseCallbackHandler
 from pydantic import BaseModel
 
 from gen_ai_orchestrator.models.observability.observability_setting import (
@@ -34,7 +34,7 @@ class LangChainCallbackHandlerFactory(ABC, BaseModel):
     setting: BaseObservabilitySetting
 
     @abstractmethod
-    def get_callback_handler(self, **kwargs: Any) -> LangchainCallbackHandler:
+    def get_callback_handler(self, **kwargs: Any) -> LangfuseCallbackHandler:
         """
         Fabric a callback handler.
         :return: LangchainCallbackHandler.

gen-ai/orchestrator-server/src/main/python/server/src/gen_ai_orchestrator/services/langchain/factories/callback_handlers/callback_handlers_factory.py

@@ -25,6 +25,7 @@
 from langfuse import Langfuse
 from langfuse.api.core import ApiError
 from langfuse.langchain import CallbackHandler as LangfuseCallbackHandler
+from pydantic import PrivateAttr
 
 from gen_ai_orchestrator.configurations.environment.settings import (
     application_settings,
@@ -59,30 +60,82 @@ class LangfuseCallbackHandlerFactory(LangChainCallbackHandlerFactory):
 
     setting: ObservabilitySetting
 
+    # Internal client cache
+    _langfuse_client: Optional[Langfuse] = PrivateAttr(default=None)
+
+    def _get_langfuse_client(self) -> Langfuse:
+        """
+        Create or return the initialized Langfuse client
+        """
+        if self._langfuse_client is None:
+            settings = self._fetch_settings()
+            self._langfuse_client = Langfuse(
+                public_key=settings['public_key'],
+                secret_key=settings['secret_key'],
+                base_url=settings['base_url'],
+                timeout=settings['timeout'],
+                httpx_client=self._get_httpx_client(),
+            )
+        return self._langfuse_client
+
     def get_callback_handler(self, **kwargs: Any) -> LangfuseCallbackHandler:
-        return LangfuseCallbackHandler(**self._fetch_settings(), httpx_client=self._get_httpx_client(), **kwargs)
+        """
+        Create Langfuse CallbackHandler
+        """
+        self._get_langfuse_client()
+
+        # Ignore Langfuse V2 parameters to stay backward-compatible
+        if kwargs:
+            logger.debug(
+                'Ignoring unsupported Langfuse CallbackHandler kwargs in V3: %s',
+                list(kwargs.keys()),
+            )
+
+        # Langfuse SDK maintains an internal map / pool of clients based on there public key, that why the client isn't passed to the callbackhandler constructor.
+        return LangfuseCallbackHandler(
+            public_key=self.setting.public_key,
+        )
 
     def check_observability_setting(self) -> bool:
         """Check if the provided credentials (public and secret key) are valid,
         while tracing a sample phrase"""
         try:
-            self.get_callback_handler().auth_check()
-            Langfuse(**self._fetch_settings(), httpx_client=self._get_httpx_client()).trace(
-                name=ObservabilityTrace.CHECK_OBSERVABILITY_SETTINGS.value, output='Check observability setting trace')
+            client = self._get_langfuse_client()
+            logger.debug('Lang')
+
+            if not client.auth_check():
+                logger.error('Langfuse auth_check() returned False')
+                raise GenAIObservabilityErrorException(
+                    'Langfuse authentication check failed'
+                )
+
+            with client.start_as_current_observation(
+                as_type='span',
+                name=ObservabilityTrace.CHECK_OBSERVABILITY_SETTINGS.value,
+                input={'message': 'Check observability setting'},
+            ) as span:
+                span.update(output='Check observability setting trace')
+
+            client.flush()
+
         except ApiError as exc:
             logger.error(exc)
             raise GenAIObservabilityErrorException(
                 create_error_info_langfuse(exc)
             )
         return True
 
-    def _fetch_settings(self):
+    def _fetch_settings(self) -> dict:
+        """
+        Fetch necessary parameters to initialise Langfuse client.
+        """
         return {
-            'host': str(self.setting.url),
+            'base_url': str(self.setting.url),
             'public_key': self.setting.public_key,
             'secret_key': fetch_secret_key_value(self.setting.secret_key),
             'timeout': application_settings.observability_provider_timeout,
-            'max_retries': application_settings.observability_provider_max_retries
+            # kept for backward-compatibility, not used anymore
+            'max_retries': application_settings.observability_provider_max_retries,
         }
 
     def _get_httpx_client(self) -> Optional[Client]:

gen-ai/orchestrator-server/src/main/python/server/src/gen_ai_orchestrator/services/langchain/factories/callback_handlers/langfuse_callback_handler_factory.py

@@ -41,7 +41,7 @@
     RunnableSerializable,
 )
 from langchain_core.vectorstores import VectorStoreRetriever
-from langfuse.langchain import CallbackHandler as LangfuseCallbackHandler
+from langfuse import get_client, propagate_attributes
 from typing_extensions import Any
 
 from gen_ai_orchestrator.errors.exceptions.exceptions import (
@@ -131,7 +131,7 @@ async def execute_rag_chain(
                 message_history.add_ai_message(msg.text)
         session_id = (request.dialog.dialog_id,)
         user_id = (request.dialog.user_id,)
-        tags = (request.dialog.tags,)
+        tags = (request.dialog.tags,) or []
 
     logger.debug(
         'RAG chain - Use chat history: %s',
@@ -160,16 +160,23 @@ async def execute_rag_chain(
         # Langfuse callback handler
         observability_handler = create_observability_callback_handler(
             observability_setting=request.observability_setting,
-            trace_name=ObservabilityTrace.RAG.value,
-            session_id=session_id,
-            user_id=user_id,
-            tags=tags,
         )
         callback_handlers.append(observability_handler)
 
+    metadata = {}
+    if user_id is not None:
+        metadata['langfuse_user_id'] = str(user_id)
+    if session_id is not None:
+        metadata['langfuse_session_id'] = str(session_id)
+    if tags:
+        metadata['langfuse_tags'] = list(tags)
+
     response = await conversational_retrieval_chain.ainvoke(
         input=inputs,
-        config={'callbacks': callback_handlers},
+        config={
+            'callbacks': callback_handlers,
+            'metadata': metadata,
+        },
     )
 
     # RAG Guard
@@ -204,7 +211,10 @@ async def execute_rag_chain(
                 )
             ),
         ),
-        observability_info=get_observability_info(observability_handler),
+        observability_info=get_observability_info(
+            observability_handler,
+            ObservabilityTrace.RAG.value if observability_handler is not None else None,
+        ),
         debug=get_rag_debug_data(request, records_callback_handler, rag_duration)
         if debug
         else None,

gen-ai/orchestrator-server/src/main/python/server/src/gen_ai_orchestrator/services/langchain/rag_chain.py

@@ -17,6 +17,7 @@
 import logging
 from typing import Optional
 
+from langfuse import get_client
 from langfuse.langchain import CallbackHandler as LangfuseCallbackHandler
 
 from gen_ai_orchestrator.models.observability.observability_type import (
@@ -45,13 +46,20 @@ def check_observability_setting(setting: ObservabilitySetting) -> bool:
     return get_callback_handler_factory(setting).check_observability_setting()
 
 
-def get_observability_info(observability_handler) -> Optional[ObservabilityInfo]:
+def get_observability_info(observability_handler, trace_name: Optional[str] = None) -> Optional[ObservabilityInfo]:
     """Get the observability Information"""
-    if isinstance(observability_handler, LangfuseCallbackHandler):
-        return ObservabilityInfo(
-            trace_id=observability_handler.trace.id,
-            trace_name=observability_handler.trace_name,
-            trace_url=observability_handler.get_trace_url()
-        )
-    else:
-        return None
+    if not isinstance(observability_handler, LangfuseCallbackHandler):
+        return None
+
+    trace_id = getattr(observability_handler, 'last_trace_id', None)
+    if trace_id is None:
+        return None
+
+    langfuse_client = observability_handler.client
+    trace_url = langfuse_client.get_trace_url(trace_id=trace_id)
+
+    return ObservabilityInfo(
+        trace_id=trace_id,
+        trace_name=trace_name,
+        trace_url=trace_url,
+    )

gen-ai/orchestrator-server/src/main/python/server/src/gen_ai_orchestrator/services/observability/observabilty_service.py

@@ -13,7 +13,7 @@
 #   limitations under the License.
 #
 import os
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
 import pytest
 from langchain_core.documents import Document
@@ -184,7 +184,10 @@ async def test_rag_chain(
     # Assert qa chain is ainvoke()d with the expected settings from request
     mocked_chain.ainvoke.assert_called_once_with(
         input=inputs,
-        config={'callbacks': [mocked_callback, mocked_langfuse_callback]},
+        config={
+            'callbacks': [mocked_callback, mocked_langfuse_callback],
+            'metadata': ANY,
+        },
     )
     # Assert the response is build using the expected settings
     mocked_rag_response.assert_called_once_with(
@@ -430,7 +433,7 @@ def test_rag_guard_accepts_no_answer_even_with_docs(mocked_log):
         'documents': ['a doc as a string'],
     }
     rag_chain.rag_guard(inputs, response, documents_required=True)
-    assert response['documents'] == ['a doc as a string']
+    assert response['documents'] == []
 
 
 @patch('gen_ai_orchestrator.services.langchain.rag_chain.rag_log')