Livekit integration with langfuse

Agents
, , , ,
1

Hey community!!
I have integrated langfuse with livekit(using python), but an issue which i am facing is that after the agent terminal is run for the first time, the telemetry data of the very first voice call conducted is being shown correctly as the new session in the langfuse dashboard but any subsequent calls conducted are not shown in the dashboard.

Below given is the entrypoint code related to langfuse:

async def entrypoint(ctx: agents.JobContext):

    # Setup Langfuse locally per-job to properly bind LiveKit's internal telemetry

    try:

        trace_provider = setup_langfuse(metadata={"langfuse.session.id": ctx.room.name})

        logger.info(f"Langfuse tracing bound to session: {ctx.room.name}")

    except Exception as e:

        logger.error(f"Failed to setup Langfuse tracing: {e}")

        trace_provider = None

    if trace_provider:

        tracer = trace.get_tracer(__name__, tracer_provider=trace_provider)

    else:

        tracer = trace.get_tracer(__name__)

    # Create root span

    span = tracer.start_span("voice_agent_session")

    span.set_attribute("langfuse.trace.name", "voice_call")

    # Also set the session id on the root span explicitly for safety

    span.set_attribute("langfuse.session.id", ctx.room.name)

    span_ctx = trace.use_span(span, end_on_exit=False)

    span_ctx.__enter__()

    async def _cleanup():

        logger.info(f"🔄 [LANGFUSE] _cleanup called for {ctx.room.name}, ending root span...")

        # Exit the span context first to detach from OpenTelemetry context

        try:

            span_ctx.__exit__(None, None, None)

        except Exception:

            pass

        span.end()

        if trace_provider:

            try:

                logger.info("🔄 [LANGFUSE] Forcing flush of Langfuse traces...")

                # Small delay to let BatchSpanProcessor pick up the final span

                await asyncio.sleep(0.2)

                trace_provider.force_flush()

                logger.info("âś… [LANGFUSE] Trace flush successful!")

            except Exception as e:

                logger.error(f"❌ [LANGFUSE] Trace flush failed: {e}")

        else:

            logger.warning("⚠️ [LANGFUSE] No trace_provider found during cleanup!")

    ctx.add_shutdown_call

back(_cleanup)

    logger.info(f"Connecting to room: {ctx.room.name}")

    await ctx.connect()

and below, i have attached the .py file responsible for initializing the setup_langfuse function , TracerProvider :

langfuse_tracing.py (2.4 KB)

please help me understand if there’s anything wrong.
Thank you

2

trace.get_tracer(name, tracer_provider=...) only attaches your local tracer variable to the new provider. LiveKit’s internal telemetry (the spans for AgentSession, llm_request, tts_request, etc.) reads from the provider set via livekit.agents.telemetry.set_tracer_provider. Switch to that and the per-job binding will hold.

What’s likely happening: on the first job, your setup_langfuse is calling stock OpenTelemetry’s trace.set_tracer_provider(...) globally, so LiveKit’s runtime picks it up and traces flow. On every subsequent job, OTel silently refuses to override the global provider (logs a warning, keeps the first one), so the new provider you create is orphaned; only your manual voice_agent_session span gets exported through it. LiveKit’s session/llm/tts spans keep going to the original (now stale) provider, which is why nothing shows up in Langfuse for calls 2..N.

The official pattern from the agents repo:

# livekit-agents==1.5.8
from livekit.agents.telemetry import set_tracer_provider
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

def setup_langfuse(metadata: dict | None = None) -> TracerProvider:
    # set LANGFUSE_HOST / OTEL_EXPORTER_OTLP_ENDPOINT / OTEL_EXPORTER_OTLP_HEADERS env
    trace_provider = TracerProvider()
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    set_tracer_provider(trace_provider, metadata=metadata)   # LiveKit's, not OTel's
    return trace_provider

async def entrypoint(ctx: agents.JobContext):
    setup_langfuse(metadata={"langfuse.session.id": ctx.room.name})
    # no manual root span / span_ctx / force_flush needed — LiveKit's job spans are the root
    await ctx.connect()

Canonical implementation: agents/examples/voice_agents/langfuse_trace.py at main · livekit/agents · GitHub on main.

livekit.agents.telemetry.set_tracer_provider is LiveKit’s own wrapper; it rebinds the agent runtime on every call (unlike stock OTel) and accepts metadata= so session-level attributes like langfuse.session.id ride along on every span.

You can drop the manual root-span / span_ctx.__enter__() / force_flush() plumbing entirely; the framework owns that lifecycle. If you still see split traces after the fix, Multiple traces created for a single LiveKit call instead of one unified trace · Issue #3319 · livekit/agents · GitHub is the open thread on multiple-traces-per-session for Langfuse worth tracking.

Cc: @Yash_Zinzuwadiya