jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

ref: Refactor tracing service (#7011)

Browse source 
* Refactor tracing service

* Remove start, flush and stop
This commit is contained in:
Christophe Bornet 2025-03-13 11:01:39 +01:00 • committed by GitHub
commit 3f427555f4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 7 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

37
src/backend/base/langflow/services/tracing/service.py
View file

@ -60,35 +60,10 @@ class TracingService(Service):
self.project_name: str | None = None
self._tracers: dict[str, BaseTracer] = {}
self._logs: dict[str, list[Log | dict[Any, Any]]] = defaultdict(list)
self.running = False
self.end_trace_tasks: set[asyncio.Task] = set()
self.deactivated = self.settings_service.settings.deactivate_tracing
self.session_id: str | None = None
async def start(self) -> None:
if self.running:
return
try:
self.running = True
except Exception: # noqa: BLE001
logger.exception("Error starting tracing service")
async def flush(self) -> None:
# This method is kept for API compatibility but now has minimal implementation
try:
# Wait for any pending trace tasks to complete
if self.end_trace_tasks:
await asyncio.gather(*self.end_trace_tasks, return_exceptions=True)
except Exception: # noqa: BLE001
logger.exception("Error flushing logs")
async def stop(self) -> None:
try:
self.running = False
await self.flush()
except Exception: # noqa: BLE001
logger.exception("Error stopping tracing service")
def _reset_io(self) -> None:
self.inputs = defaultdict(dict)
self.inputs_metadata = defaultdict(dict)
@ -99,14 +74,12 @@ class TracingService(Service):
if self.deactivated:
return
try:
await self.start()
self._initialize_langsmith_tracer()
self._initialize_langwatch_tracer()
self._initialize_langfuse_tracer()
self._initialize_arize_phoenix_tracer()
except Exception as e: # noqa: BLE001
logger.debug(f"Error initializing tracers: {e}")
except Exception: # noqa: BLE001
logger.opt(exception=True).debug("Error initializing tracers")
def _initialize_langsmith_tracer(self) -> None:
project_name = os.getenv("LANGCHAIN_PROJECT", "Langflow")
@ -202,7 +175,11 @@ class TracingService(Service):
async def end(self, outputs: dict, error: Exception | None = None) -> None:
await asyncio.to_thread(self._end_all_traces, outputs, error)
await self.stop()
try:
# Wait for any pending trace tasks to complete
await asyncio.gather(*self.end_trace_tasks)
except Exception: # noqa: BLE001
logger.exception("Error flushing logs")
def add_log(self, trace_name: str, log: Log) -> None:
self._logs[trace_name].append(log)