feat: Added Traceloop SDK for collecting traces and metrics from Langflow (#9317)
* Added Traceloop SDK for colecting traces and metrics from Langflow * [autofix.ci] apply automated fixes * added test case for traceloop * Revert "[autofix.ci] apply automated fixes" This reverts commit 3a68113f0de65b2397ac88d1fc2cae8f786adbd7. * Updated logic for verifying returned callbacks list length * [autofix.ci] apply automated fixes * api key strip validation and updated logger warning * Add graceful fallback for Traceloop LangChain callback handler * Removed TraceloopLangChainCallbackHandler * default URL moved to configuration constant * add Timeout Protection to add trace method * add resource cleanup method for traceloop * fix(tracing): unify span lifecycle and fix async mismatch in TraceloopTracer - Remove incorrect asyncio.wait_for usage on synchronous method - Store active spans in _span_map to properly end them in end_trace - Prevent duplicate unrelated spans for same component - Use trace.get_tracer_provider().force_flush() for correct flushing - Add explicit close() method for manual flush at shutdown - Enforce HTTPS in TRACELOOP_BASE_URL validation - Improve reliability of cleanup in __del__ * stable uv.lock from upstream/main * Fix: ensure type-safe metadata handling * Block PYMYSQL intrumentor and relaxed https constraint * fix(tracing): handle invalid input/output type conversion for traceloop attributes * fix: Reduced cognitive complexity of get_trace_as_metadata method * fix: organize child spans under single root span and added type conversion methods * fix: Linting and formatting * uv.lock reset to upstream main * [autofix.ci] apply automated fixes --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Sandesh R <sandesh@ibm.com>
This commit is contained in:
Sandesh R
GitHub
parent
2f06fa9d4d
commit
3ecf9640b7
6 changed files with 880 additions and 4 deletions
|
|
@ -3,6 +3,7 @@ from __future__ import annotations
|
|||
import ast
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
from collections.abc import AsyncIterator, Iterator
|
||||
from copy import deepcopy
|
||||
from textwrap import dedent
|
||||
|
|
@ -1019,10 +1020,23 @@ class Component(CustomComponent):
|
|||
return {**predefined_inputs, **runtime_inputs}
|
||||
|
||||
def get_trace_as_metadata(self):
|
||||
def safe_list_values(items):
|
||||
return [v if isinstance(v, str | int | float | bool) or v is None else str(v) for v in items]
|
||||
|
||||
def safe_value(val):
|
||||
if isinstance(val, str | int | float | bool) or val is None:
|
||||
return val
|
||||
if isinstance(val, list | tuple):
|
||||
return safe_list_values(val)
|
||||
try:
|
||||
return json.dumps(val)
|
||||
except (TypeError, ValueError):
|
||||
return str(val)
|
||||
|
||||
return {
|
||||
input_.name: input_.value
|
||||
input_.name: safe_value(getattr(self, input_.name, input_.value))
|
||||
for input_ in self.inputs
|
||||
if hasattr(input_, "trace_as_metadata") and input_.trace_as_metadata
|
||||
if getattr(input_, "trace_as_metadata", False)
|
||||
}
|
||||
|
||||
async def _build_with_tracing(self):
|
||||
|
|
|
|||
|
|
@ -53,6 +53,12 @@ def _get_opik_tracer():
|
|||
return OpikTracer
|
||||
|
||||
|
||||
def _get_traceloop_tracer():
|
||||
from langflow.services.tracing.traceloop import TraceloopTracer
|
||||
|
||||
return TraceloopTracer
|
||||
|
||||
|
||||
trace_context_var: ContextVar[TraceContext | None] = ContextVar("trace_context", default=None)
|
||||
component_context_var: ContextVar[ComponentTraceContext | None] = ContextVar("component_trace_context", default=None)
|
||||
|
||||
|
|
@ -201,6 +207,19 @@ class TracingService(Service):
|
|||
session_id=trace_context.session_id,
|
||||
)
|
||||
|
||||
def _initialize_traceloop_tracer(self, trace_context: TraceContext) -> None:
|
||||
if self.deactivated:
|
||||
return
|
||||
traceloop_tracer = _get_traceloop_tracer()
|
||||
trace_context.tracers["traceloop"] = traceloop_tracer(
|
||||
trace_name=trace_context.run_name,
|
||||
trace_type="chain",
|
||||
project_name=trace_context.project_name,
|
||||
trace_id=trace_context.run_id,
|
||||
user_id=trace_context.user_id,
|
||||
session_id=trace_context.session_id,
|
||||
)
|
||||
|
||||
async def start_tracers(
|
||||
self,
|
||||
run_id: UUID,
|
||||
|
|
@ -227,6 +246,7 @@ class TracingService(Service):
|
|||
self._initialize_langfuse_tracer(trace_context)
|
||||
self._initialize_arize_phoenix_tracer(trace_context)
|
||||
self._initialize_opik_tracer(trace_context)
|
||||
self._initialize_traceloop_tracer(trace_context)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.debug(f"Error initializing tracers: {e}")
|
||||
|
||||
|
|
|
|||
245
src/backend/base/langflow/services/tracing/traceloop.py
Normal file
245
src/backend/base/langflow/services/tracing/traceloop.py
Normal file
|
|
@ -0,0 +1,245 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import types
|
||||
from datetime import datetime, timezone
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from loguru import logger
|
||||
from opentelemetry import trace
|
||||
from opentelemetry.trace import Span, use_span
|
||||
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
|
||||
from traceloop.sdk import Traceloop
|
||||
from traceloop.sdk.instruments import Instruments
|
||||
from typing_extensions import override
|
||||
|
||||
from langflow.services.tracing.base import BaseTracer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
from uuid import UUID
|
||||
|
||||
from langchain.callbacks.base import BaseCallbackHandler
|
||||
from opentelemetry.propagators.textmap import CarrierT
|
||||
from opentelemetry.trace import Span
|
||||
|
||||
from langflow.graph.vertex.base import Vertex
|
||||
from langflow.services.tracing.schema import Log
|
||||
|
||||
|
||||
class TraceloopTracer(BaseTracer):
|
||||
"""Traceloop tracer for Langflow."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
trace_name: str,
|
||||
trace_type: str,
|
||||
project_name: str,
|
||||
trace_id: UUID,
|
||||
user_id: str | None = None,
|
||||
session_id: str | None = None,
|
||||
):
|
||||
self.trace_id = trace_id
|
||||
self.trace_name = trace_name
|
||||
self.trace_type = trace_type
|
||||
self.project_name = project_name
|
||||
self.user_id = user_id
|
||||
self.session_id = session_id
|
||||
self.child_spans: dict[str, Span] = {}
|
||||
|
||||
if not self._validate_configuration():
|
||||
self._ready = False
|
||||
return
|
||||
|
||||
api_key = os.getenv("TRACELOOP_API_KEY", "").strip()
|
||||
try:
|
||||
Traceloop.init(
|
||||
block_instruments={Instruments.PYMYSQL},
|
||||
app_name=project_name,
|
||||
disable_batch=True,
|
||||
api_key=api_key,
|
||||
api_endpoint=os.getenv("TRACELOOP_BASE_URL", "https://api.traceloop.com"),
|
||||
)
|
||||
self._ready = True
|
||||
self._tracer = trace.get_tracer("langflow")
|
||||
self.propagator = TraceContextTextMapPropagator()
|
||||
self.carrier: CarrierT = {}
|
||||
|
||||
self.root_span = self._tracer.start_span(
|
||||
name=trace_name,
|
||||
start_time=self._get_current_timestamp(),
|
||||
)
|
||||
|
||||
with use_span(self.root_span, end_on_exit=False):
|
||||
self.propagator.inject(carrier=self.carrier)
|
||||
|
||||
except Exception: # noqa: BLE001
|
||||
logger.opt(exception=True).debug("Error setting up Traceloop tracer")
|
||||
self._ready = False
|
||||
|
||||
@property
|
||||
def ready(self) -> bool:
|
||||
return self._ready
|
||||
|
||||
def _validate_configuration(self) -> bool:
|
||||
api_key = os.getenv("TRACELOOP_API_KEY", "").strip()
|
||||
if not api_key:
|
||||
logger.warning("TRACELOOP_API_KEY not set or empty.")
|
||||
return False
|
||||
|
||||
base_url = os.getenv("TRACELOOP_BASE_URL", "https://api.traceloop.com")
|
||||
parsed = urlparse(base_url)
|
||||
if not parsed.netloc:
|
||||
logger.error(f"Invalid TRACELOOP_BASE_URL: {base_url}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _convert_to_traceloop_type(self, value):
|
||||
"""Recursively converts a value to a Traceloop compatible type."""
|
||||
from langchain.schema import BaseMessage, Document, HumanMessage, SystemMessage
|
||||
|
||||
from langflow.schema.message import Message
|
||||
|
||||
try:
|
||||
if isinstance(value, dict):
|
||||
value = {key: self._convert_to_traceloop_type(val) for key, val in value.items()}
|
||||
|
||||
elif isinstance(value, list):
|
||||
value = [self._convert_to_traceloop_type(v) for v in value]
|
||||
|
||||
elif isinstance(value, Message):
|
||||
value = value.text
|
||||
|
||||
elif isinstance(value, (BaseMessage | HumanMessage | SystemMessage)):
|
||||
value = str(value.content) if value.content is not None else ""
|
||||
|
||||
elif isinstance(value, Document):
|
||||
value = value.page_content
|
||||
|
||||
elif isinstance(value, (types.GeneratorType | types.NoneType)):
|
||||
value = str(value)
|
||||
|
||||
elif isinstance(value, float) and not math.isfinite(value):
|
||||
value = "NaN"
|
||||
|
||||
except (TypeError, ValueError) as e:
|
||||
logger.warning(f"Failed to convert value {value!r} to traceloop type: {e}")
|
||||
return str(value)
|
||||
else:
|
||||
return value
|
||||
|
||||
def _convert_to_traceloop_dict(self, io_dict: Any) -> dict[str, Any]:
|
||||
"""Ensure values are OTel-compatible. Dicts stay dicts, lists get JSON-serialized."""
|
||||
if isinstance(io_dict, dict):
|
||||
return {str(k): self._convert_to_traceloop_type(v) for k, v in io_dict.items()}
|
||||
if isinstance(io_dict, list):
|
||||
return {"list": json.dumps([self._convert_to_traceloop_type(v) for v in io_dict], default=str)}
|
||||
|
||||
return {"value": self._convert_to_traceloop_type(io_dict)}
|
||||
|
||||
@override
|
||||
def add_trace(
|
||||
self,
|
||||
trace_id: str,
|
||||
trace_name: str,
|
||||
trace_type: str,
|
||||
inputs: dict[str, Any],
|
||||
metadata: dict[str, Any] | None = None,
|
||||
vertex: Vertex | None = None,
|
||||
) -> None:
|
||||
if not self.ready:
|
||||
return
|
||||
|
||||
span_context = self.propagator.extract(carrier=self.carrier)
|
||||
child_span = self._tracer.start_span(
|
||||
name=trace_name,
|
||||
context=span_context,
|
||||
start_time=self._get_current_timestamp(),
|
||||
)
|
||||
|
||||
attributes = {
|
||||
"trace_id": trace_id,
|
||||
"trace_name": trace_name,
|
||||
"trace_type": trace_type,
|
||||
"inputs": json.dumps(self._convert_to_traceloop_dict(inputs), default=str),
|
||||
**self._convert_to_traceloop_dict(metadata or {}),
|
||||
}
|
||||
if vertex and vertex.id is not None:
|
||||
attributes["vertex_id"] = vertex.id
|
||||
|
||||
child_span.set_attributes(attributes)
|
||||
|
||||
self.child_spans[trace_id] = child_span
|
||||
|
||||
@override
|
||||
def end_trace(
|
||||
self,
|
||||
trace_id: str,
|
||||
trace_name: str,
|
||||
outputs: dict[str, Any] | None = None,
|
||||
error: Exception | None = None,
|
||||
logs: Sequence[Log | dict] = (),
|
||||
) -> None:
|
||||
if not self._ready or trace_id not in self.child_spans:
|
||||
return
|
||||
|
||||
child_span = self.child_spans.pop(trace_id)
|
||||
|
||||
if outputs:
|
||||
child_span.set_attribute("outputs", json.dumps(self._convert_to_traceloop_dict(outputs), default=str))
|
||||
if logs:
|
||||
child_span.set_attribute("logs", json.dumps(self._convert_to_traceloop_dict(list(logs)), default=str))
|
||||
if error:
|
||||
child_span.record_exception(error)
|
||||
|
||||
child_span.end()
|
||||
|
||||
@override
|
||||
def end(
|
||||
self,
|
||||
inputs: dict[str, Any],
|
||||
outputs: dict[str, Any],
|
||||
error: Exception | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
if not self.ready:
|
||||
return
|
||||
|
||||
safe_outputs = self._convert_to_traceloop_dict(outputs)
|
||||
safe_metadata = self._convert_to_traceloop_dict(metadata or {})
|
||||
|
||||
self.root_span.set_attributes(
|
||||
{
|
||||
"workflow_name": self.trace_name,
|
||||
"workflow_id": str(self.trace_id),
|
||||
"outputs": json.dumps(safe_outputs, default=str),
|
||||
**safe_metadata,
|
||||
}
|
||||
)
|
||||
if error:
|
||||
self.root_span.record_exception(error)
|
||||
|
||||
self.root_span.end()
|
||||
|
||||
@staticmethod
|
||||
def _get_current_timestamp() -> int:
|
||||
return int(datetime.now(timezone.utc).timestamp() * 1_000_000_000)
|
||||
|
||||
@override
|
||||
def get_langchain_callback(self) -> BaseCallbackHandler | None:
|
||||
return None
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
provider = trace.get_tracer_provider()
|
||||
if hasattr(provider, "force_flush"):
|
||||
provider.force_flush(timeout_millis=3000)
|
||||
except (ValueError, RuntimeError, OSError) as e:
|
||||
logger.warning(f"Error flushing spans: {e}")
|
||||
|
||||
def __del__(self):
|
||||
self.close()
|
||||
|
|
@ -139,6 +139,10 @@ def mock_tracers():
|
|||
"langflow.services.tracing.service._get_opik_tracer",
|
||||
return_value=MockTracer,
|
||||
),
|
||||
patch(
|
||||
"langflow.services.tracing.service._get_traceloop_tracer",
|
||||
return_value=MockTracer,
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
|
|
@ -169,6 +173,7 @@ async def test_start_end_tracers(tracing_service):
|
|||
assert "langwatch" in trace_context.tracers
|
||||
assert "langfuse" in trace_context.tracers
|
||||
assert "arize_phoenix" in trace_context.tracers
|
||||
assert "traceloop" in trace_context.tracers
|
||||
|
||||
await tracing_service.end_tracers(outputs)
|
||||
|
||||
|
|
@ -298,7 +303,8 @@ async def test_get_langchain_callbacks(tracing_service):
|
|||
assert tracer.get_langchain_callback_called
|
||||
|
||||
# Verify returned callbacks list length
|
||||
assert len(callbacks) == 5 # Five tracers
|
||||
expected = len(trace_context_var.get().tracers)
|
||||
assert len(callbacks) == expected
|
||||
|
||||
# Cleanup
|
||||
await tracing_service.end_tracers({})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue