Trace LiveKit Agents with Langfuse

pip install langfuse livekit-agents livekit-plugins-openai livekit-plugins-silero livekit-plugins-turn-detector opentelemetry-sdk python-dotenv

npm install @langfuse/otel @opentelemetry/sdk-trace-node @livekit/agents @livekit/agents-plugin-openai @livekit/agents-plugin-silero

LANGFUSE_SECRET_KEY = "sk-lf-..."
LANGFUSE_PUBLIC_KEY = "pk-lf-..."
LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region
# Other Langfuse data regions include 🇺🇸 US: https://us.cloud.langfuse.com, 🇯🇵 Japan: https://jp.cloud.langfuse.com and ⚕️ HIPAA: https://hipaa.cloud.langfuse.com

LANGFUSE_SECRET_KEY = "sk-lf-..."
LANGFUSE_PUBLIC_KEY = "pk-lf-..."
LANGFUSE_BASE_URL = "https://cloud.langfuse.com" # 🇪🇺 EU region
# Other Langfuse data regions include 🇺🇸 US: https://us.cloud.langfuse.com, 🇯🇵 Japan: https://jp.cloud.langfuse.com and ⚕️ HIPAA: https://hipaa.cloud.langfuse.com

import os

from langfuse import Langfuse
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.util.types import AttributeValue

from livekit.agents.telemetry import set_tracer_provider


def setup_langfuse(
    metadata: dict[str, AttributeValue] | None = None,
    *,
    host: str | None = None,
    public_key: str | None = None,
    secret_key: str | None = None,
) -> TracerProvider:
    public_key = public_key or os.getenv("LANGFUSE_PUBLIC_KEY")
    secret_key = secret_key or os.getenv("LANGFUSE_SECRET_KEY")
    host = host or os.getenv("LANGFUSE_HOST") or os.getenv("LANGFUSE_BASE_URL")

    if not public_key or not secret_key or not host:
        raise ValueError(
            "LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_HOST (or LANGFUSE_BASE_URL) must be set"
        )

    trace_provider = TracerProvider()
    set_tracer_provider(trace_provider, metadata=metadata)

    Langfuse(
        public_key=public_key,
        secret_key=secret_key,
        base_url=host,
        tracer_provider=trace_provider,
        should_export_span=lambda span: True,
    )

    return trace_provider

import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { telemetry } from "@livekit/agents";

export function setupLangfuse(): NodeTracerProvider {
  const tracerProvider = new NodeTracerProvider({
    spanProcessors: [
      // LiveKit emits many non-LLM spans; forward all of them to Langfuse
      new LangfuseSpanProcessor({ shouldExportSpan: () => true }),
    ],
  });

  // Register as the global OpenTelemetry provider so that spans you create in
  // your own code (via the OTel API or @langfuse/tracing helpers) are also
  // exported to Langfuse, not just LiveKit's internal spans.
  tracerProvider.register();

  // Pass the provider to LiveKit WITHOUT the metadata option (see warnings above).
  // The cast bridges the SDK v1/v2 type mismatch; it is safe at runtime.
  telemetry.setTracerProvider(
    tracerProvider as unknown as Parameters<
      typeof telemetry.setTracerProvider
    >[0],
  );

  return tracerProvider;
}

from livekit.agents import AgentSession, JobContext, metrics
from livekit.agents.voice import MetricsCollectedEvent
from livekit.plugins import silero


async def entrypoint(ctx: JobContext):
    trace_provider = setup_langfuse(
        metadata={
            "langfuse.session.id": ctx.room.name,
        }
    )

    async def flush_trace():
        trace_provider.force_flush()

    ctx.add_shutdown_callback(flush_trace)

    session = AgentSession(vad=silero.VAD.load())

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        metrics.log_metrics(ev.metrics)

    await session.start(agent=MyAgent(), room=ctx.room)

import { type JobContext, defineAgent } from "@livekit/agents";

export default defineAgent({
  entry: async (ctx: JobContext) => {
    const tracerProvider = setupLangfuse();

    // Flush buffered spans when the job shuts down
    ctx.addShutdownCallback(async () => {
      await tracerProvider.forceFlush();
    });

    // ... start your AgentSession as usual (see the LiveKit Agents docs)
  },
});