https://livekit.io logo
Join Slack
Powered by
I need to have langfuse observability in place for...
# ask-ai
b
I need to have langfuse observability in place for realtime models. guide me how with manual instrumentation.
t
Thanks for reaching out, Samuel. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’ll walk you through manual instrumentation to send Realtime model traces to Langfuse. I’ll assume you’re using OpenTelemetry in your LiveKit Agent/process; if not, tell me which language/runtime you’re using and I’ll tailor the steps. High-level approach • Create a custom OpenTelemetry span processor that injects 
user_id
and 
session_id
into every span. • Attach that processor to your tracer provider. • Configure an OTLP exporter (OTLP/HTTP or OTLP/gRPC) to send traces to Langfuse’s ingest endpoint. Minimal Python example (custom span processor + OTLP exporter) 1) Custom span processor to add attributes:
Copy code
from opentelemetry.sdk.trace import SpanProcessor

class CustomSpanProcessor(SpanProcessor):
    def __init__(self, user_id, session_id):
        self.user_id = user_id
        self.session_id = session_id

    def on_start(self, span, parent_context):
        span.set_attribute("user_id", self.user_id)
        span.set_attribute("session_id", self.session_id)

    def on_end(self, span):
        pass
2) Attach processor and configure OTLP exporter:
Copy code
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# replace with your Langfuse OTLP endpoint + auth if required
otlp_exporter = OTLPSpanExporter(endpoint="https://<langfuse-otlp-endpoint>:4317", insecure=False)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

# add your custom processor (example values; set dynamically per session/user)
provider.add_span_processor(CustomSpanProcessor(user_id="123", session_id="abc"))

trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
What to ensure • Set 
user_id
and 
session_id
per agent session (inject dynamic values when a user connects). • Exporter endpoint and auth must match your Langfuse ingest configuration (use OTLP to Langfuse). • Use a batch/span exporter for production to avoid performance issues. If you want: • I can provide the same example in Node.js, Go, or Java. • I can show how to set `user_id`/`session_id` dynamically per LiveKit session (tell me how you identify users: token, session metadata, etc.). Would you like the Node/Go/JS example or a snippet showing dynamic injection per LiveKit session? Sources: Setting custom user_id and session_id in Langfuse traces
b
what about tracking tool calls & using Langfuse Evaluators which need input and output.
t
Thanks — I’m responding to your question about tracking tool calls and supplying input/output for Langfuse Evaluators. Short answer: LiveKit Agents expose tool calls and their metadata in Agent Observability (trace view), but there is no built‑in, documented integration that automatically emits tool inputs/outputs in the exact format Langfuse Evaluators expect. You can achieve this by manually instrumenting tool calls: capture inputs/outputs and add them to OpenTelemetry spans (or attach them as events/attributes) so the OTLP exporter sends them to Langfuse. Suggested approach (high-level) • Wrap each tool invocation so you can capture: ◦ tool name, input payload, output/result, timestamps, errors ◦ session and user identifiers (
user_id
, 
session_id
) • Record those details on the span that represents the tool call: ◦ set attributes for metadata (tool name, session/user IDs, status) ◦ add span events or attributes with serialized 
input
and 
output
(or store them in a linked log/DB and include a reference/ID in the span) • Ensure your OTLP exporter sends these spans/events to Langfuse. Concrete patterns you can use • Add to custom span processor (Python example shown earlier): on tool start, create a child span, set 
tool_name
, 
user_id
, `session_id`; on end, set 
status
, and attach 
input
and 
output
as attributes or as an event. • If payloads are large or sensitive, store them externally (secure store/DB) and put a reference/short hash in the span instead of full content. • Use consistent attribute names so Langfuse Evaluators can query input/output pairs (for example: 
tool.name
, 
tool.input_id
or 
tool.input
, 
tool.output
). Example sketch (Python pseudo)
Copy code
with tracer.start_as_current_span("tool.call", attributes={"tool.name": name, "user_id": uid, "session_id": sid}) as span:
    span.add_event("tool.input", {"input": serialize(input)})
    try:
        result = call_tool(input)
        span.add_event("tool.output", {"output": serialize(result)})
        span.set_attribute("tool.status", "ok")
    except Exception as e:
        span.set_attribute("tool.status", "error")
        span.add_event("tool.error", {"error": str(e)})
        raise
Next steps I can help with • Provide a concrete code example for Node.js, Go, or Java. • Show a pattern that stores large inputs/outputs externally and links them from spans. • Suggest attribute naming conventions that make evaluation easier in Langfuse. Which language/runtime do you want an example for, and do you plan to store full inputs/outputs in traces or keep them external and reference them from spans? Sources: Tool definition and use | LiveKit docs | https://blog.livekit.io/streamline-troubleshooting-with-agent-observability/#/portal/signup| | Setting custom user_id and session_id in Langfuse traces
2 Views
Open in Slack
PreviousNext
Powered by