chore: enable langfuse configuration through global variables

Author: italojohnny

src/backend/base/langflow/api/build.py

@@ -237,6 +237,7 @@ async def create_graph(fresh_session, flow_id_str: str, flow_name: str | None) -
             flow_name = result.first()
 
         return await build_graph_from_data(
+            session=fresh_session,
             flow_id=flow_id_str,
             payload=data.model_dump(),
             user_id=str(current_user.id),

src/backend/base/langflow/api/utils.py

@@ -149,7 +149,7 @@ async def _get_flow_name(flow_id: uuid.UUID) -> str:
     return flow.name
 
 
-async def build_graph_from_data(flow_id: uuid.UUID | str, payload: dict, **kwargs):
+async def build_graph_from_data(session: DbSession, flow_id: uuid.UUID | str, payload: dict, **kwargs):
     """Build and cache the graph."""
     # Get flow name
     if "flow_name" not in kwargs:
@@ -169,7 +169,7 @@ async def build_graph_from_data(flow_id: uuid.UUID | str, payload: dict, **kwarg
             vertex.update_raw_params({"session_id": session_id}, overwrite=True)
 
     graph.session_id = session_id
-    await graph.initialize_run()
+    await graph.initialize_run(session)
     return graph
 
 
@@ -180,7 +180,7 @@ async def build_graph_from_db_no_cache(flow_id: uuid.UUID, session: AsyncSession
         msg = "Invalid flow ID"
         raise ValueError(msg)
     kwargs["user_id"] = kwargs.get("user_id") or str(flow.user_id)
-    return await build_graph_from_data(flow_id, flow.data, flow_name=flow.name, **kwargs)
+    return await build_graph_from_data(session, flow_id, flow.data, flow_name=flow.name, **kwargs)
 
 
 async def build_graph_from_db(flow_id: uuid.UUID, session: AsyncSession, chat_service: ChatService, **kwargs):

src/backend/base/langflow/api/v1/chat.py

@@ -248,6 +248,7 @@ async def build_vertex(
     inputs: Annotated[InputValueRequest | None, Body(embed=True)] = None,
     files: list[str] | None = None,
     current_user: CurrentActiveUser,
+    session: DbSession,
 ) -> VertexBuildResponse:
     """Build a vertex instead of the entire graph.
 
@@ -258,6 +259,7 @@ async def build_vertex(
         inputs (Optional[InputValueRequest], optional): The input values for the vertex. Defaults to None.
         files (List[str], optional): The files to use. Defaults to None.
         current_user (Any, optional): The current user dependency. Defaults to Depends(get_current_active_user).
+        session (AsyncSession, optional): The session dependency. Defaults to Depends
 
     Returns:
         VertexBuildResponse: The response containing the built vertex information.
@@ -291,7 +293,7 @@ async def build_vertex(
             )
         else:
             graph = cache.get("result")
-            await graph.initialize_run()
+            await graph.initialize_run(session)
         vertex = graph.get_vertex(vertex_id)
 
         try:

src/backend/base/langflow/graph/graph/base.py

@@ -46,6 +46,8 @@
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator, Iterable
 
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
     from langflow.api.v1.schemas import InputValueRequest
     from langflow.custom.custom_component.component import Component
     from langflow.events.event_manager import EventManager
@@ -651,12 +653,13 @@ def set_run_id(self, run_id: uuid.UUID | None = None) -> None:
 
         self._run_id = str(run_id)
 
-    async def initialize_run(self) -> None:
+    async def initialize_run(self, session: AsyncSession) -> None:
         if not self._run_id:
             self.set_run_id()
         if self.tracing_service:
             run_name = f"{self.flow_name} - {self.flow_id}"
             await self.tracing_service.start_tracers(
+                session=session,
                 run_id=uuid.UUID(self._run_id),
                 run_name=run_name,
                 user_id=self.user_id,

src/backend/base/langflow/services/tracing/base.py

@@ -69,3 +69,8 @@ def end(
     @abstractmethod
     def get_langchain_callback(self) -> BaseCallbackHandler | None:
         raise NotImplementedError
+
+    @abstractmethod
+    def get_required_variable_names(self) -> list[str]:
+        """Returns a list of variable names required to configure the service."""
+        raise NotImplementedError

src/backend/base/langflow/services/tracing/langfuse.py

@@ -29,21 +29,44 @@ def __init__(
         trace_type: str,
         project_name: str,
         trace_id: UUID,
+        public_key: str | None = None,
+        secrect_key: str | None = None,
+        host: str | None = None,
         user_id: str | None = None,
         session_id: str | None = None,
     ) -> None:
         self.project_name = project_name
         self.trace_name = trace_name
         self.trace_type = trace_type
         self.trace_id = trace_id
+        self.public_key = public_key
+        self.secrect_key = secrect_key
+        self.host = host
         self.user_id = user_id
         self.session_id = session_id
         self.flow_id = trace_name.split(" - ")[-1]
         self.spans: dict = OrderedDict()  # spans that are not ended
 
+        if self.host is not None:
+            os.environ["LANGFUSE_HOST"] = self.host
+
+        if self.public_key is not None:
+            os.environ["LANGFUSE_PUBLIC_KEY"] = self.public_key
+
+        if self.secrect_key is not None:
+            os.environ["LANGFUSE_SECRET_KEY"] = self.secrect_key
+
         config = self._get_config()
         self._ready: bool = self.setup_langfuse(config) if config else False
 
+    @staticmethod
+    def get_required_variable_names():
+        return [
+            "LANGFUSE_SECRET_KEY",
+            "LANGFUSE_PUBLIC_KEY",
+            "LANGFUSE_HOST",
+        ]
+
     @property
     def ready(self):
         return self._ready

src/backend/base/langflow/services/tracing/service.py

@@ -6,15 +6,16 @@
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any
+from uuid import UUID
 
 from loguru import logger
 
 from langflow.services.base import Service
+from langflow.services.deps import get_variable_service
 
 if TYPE_CHECKING:
-    from uuid import UUID
-
     from langchain.callbacks.base import BaseCallbackHandler
+    from sqlmodel.ext.asyncio.session import AsyncSession
 
     from langflow.custom.custom_component.component import Component
     from langflow.graph.vertex.base import Vertex
@@ -162,13 +163,19 @@ def _initialize_langwatch_tracer(self, trace_context: TraceContext) -> None:
                 trace_id=trace_context.run_id,
             )
 
-    def _initialize_langfuse_tracer(self, trace_context: TraceContext) -> None:
+    async def _initialize_langfuse_tracer(self, trace_context: TraceContext, session: AsyncSession) -> None:
         langfuse_tracer = _get_langfuse_tracer()
+        variable_names = langfuse_tracer.get_required_variable_names()
+        variables = await self.get_varaibles_from_db(session, trace_context.user_id, variable_names)
+
         trace_context.tracers["langfuse"] = langfuse_tracer(
             trace_name=trace_context.run_name,
             trace_type="chain",
             project_name=trace_context.project_name,
             trace_id=trace_context.run_id,
+            public_key=variables.get("LANGFUSE_PUBLIC_KEY"),
+            secret_key=variables.get("LANGFUSE_SECRET_KEY"),
+            host=variables.get("LANGFUSE_HOST"),
             user_id=trace_context.user_id,
             session_id=trace_context.session_id,
         )
@@ -193,8 +200,17 @@ def _initialize_opik_tracer(self, trace_context: TraceContext) -> None:
             session_id=trace_context.session_id,
         )
 
+    async def get_varaibles_from_db(self, session, user_id, variable_names):
+        variable_service = get_variable_service()
+        result = {}
+        for var in await variable_service.get_all(UUID(user_id), session):
+            if var.name in variable_names:
+                result[var.name] = await variable_service.get_variable(UUID(user_id), var.name, "", session)
+        return result
+
     async def start_tracers(
         self,
+        session: AsyncSession,
         run_id: UUID,
         run_name: str,
         user_id: str | None,
@@ -213,11 +229,17 @@ async def start_tracers(
             project_name = project_name or os.getenv("LANGCHAIN_PROJECT", "Langflow")
             trace_context = TraceContext(run_id, run_name, project_name, user_id, session_id)
             trace_context_var.set(trace_context)
-            await self._start(trace_context)
+
+            await self._start(trace_context, session, user_id)
+
             self._initialize_langsmith_tracer(trace_context)
+
             self._initialize_langwatch_tracer(trace_context)
-            self._initialize_langfuse_tracer(trace_context)
+
+            await self._initialize_langfuse_tracer(trace_context, session)
+
             self._initialize_arize_phoenix_tracer(trace_context)
+
             self._initialize_opik_tracer(trace_context)
         except Exception as e:  # noqa: BLE001
             logger.debug(f"Error initializing tracers: {e}")