Export traces

import base64
import os

from dotenv import load_dotenv
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.util.types import AttributeValue

from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    cli,
    inference,
)
from livekit.agents.telemetry import set_tracer_provider

load_dotenv(".env.local")


def setup_langfuse(metadata: dict[str, AttributeValue] | None = None) -> TracerProvider:
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.sdk.trace.export import BatchSpanProcessor

    public_key = os.environ.get("LANGFUSE_PUBLIC_KEY")
    secret_key = os.environ.get("LANGFUSE_SECRET_KEY")
    base_url = os.environ.get("LANGFUSE_BASE_URL")

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

    langfuse_auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"{base_url.rstrip('/')}/api/public/otel"
    os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = (
        f"Authorization=Basic {langfuse_auth},x-langfuse-ingestion-version=4"
    )

    trace_provider = TracerProvider()
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    set_tracer_provider(trace_provider, metadata=metadata)
    return trace_provider


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful voice AI assistant.",
            llm=inference.LLM(model="openai/gpt-5.2-chat-latest"),
        )


server = AgentServer()


@server.rtc_session(agent_name="my-agent")
async def entrypoint(ctx: JobContext):
    # Route spans to Langfuse before the session starts.
    trace_provider = setup_langfuse(metadata={"langfuse.session.id": ctx.room.name})

    # Flush any remaining spans before the process exits.
    async def flush_trace():
        trace_provider.force_flush()

    ctx.add_shutdown_callback(flush_trace)

    session = AgentSession(
        stt=inference.STT(model="deepgram/nova-3", language="multi"),

        tts=inference.TTS(model="inworld/inworld-tts-2"),

        preemptive_generation=True,
    )

    await session.start(agent=Assistant(), room=ctx.room)
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(server)

pnpm add @opentelemetry/api @opentelemetry/sdk-trace-node @opentelemetry/exporter-trace-otlp-http

import {
  type JobContext,
  ServerOptions,
  cli,
  defineAgent,
  inference,
  telemetry,
  voice,
} from '@livekit/agents';
import { type Attributes } from '@opentelemetry/api';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { BatchSpanProcessor, NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import dotenv from 'dotenv';
import { fileURLToPath } from 'node:url';

dotenv.config({ path: '.env.local' });

function setupLangfuse(metadata?: Attributes): NodeTracerProvider {
  const publicKey = process.env.LANGFUSE_PUBLIC_KEY;
  const secretKey = process.env.LANGFUSE_SECRET_KEY;
  const baseUrl = process.env.LANGFUSE_BASE_URL;

  if (!publicKey || !secretKey || !baseUrl) {
    throw new Error('LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_BASE_URL must be set');
  }

  const auth = Buffer.from(`${publicKey}:${secretKey}`).toString('base64');
  const traceExporter = new OTLPTraceExporter({
    url: `${baseUrl.replace(/\/$/, '')}/api/public/otel/v1/traces`,
    headers: { Authorization: `Basic ${auth}`, 'x-langfuse-ingestion-version': '4' },
  });

  // A provider takes its span processors at construction. Include a FanoutSpanProcessor and
  // hand its `add` method to setTracerProvider so the framework can attach the processor that
  // applies `metadata` to every span.
  const fanout = new telemetry.FanoutSpanProcessor();
  const traceProvider = new NodeTracerProvider({
    spanProcessors: [new BatchSpanProcessor(traceExporter), fanout],
  });

  traceProvider.register();
  telemetry.setTracerProvider(traceProvider, {
    metadata,
    registerSpanProcessor: (processor) => fanout.add(processor),
  });
  return traceProvider;
}

export default defineAgent({
  entry: async (ctx: JobContext) => {
    // Route spans to Langfuse before the session starts.
    const traceProvider = setupLangfuse({ 'langfuse.session.id': ctx.room.name });

    // Flush any remaining spans before the process exits.
    ctx.addShutdownCallback(async () => {
      await traceProvider.shutdown();
    });

    const session = new voice.AgentSession({
      stt: new inference.STT({ model: 'deepgram/nova-3', language: 'multi' }),
      llm: new inference.LLM({ model: 'openai/gpt-5.2-chat-latest' }),
      tts: new inference.TTS({
        model: 'inworld/inworld-tts-2',
        voice: 'Ashley',
      }),
    });

    await session.start({
      agent: voice.Agent.create({ instructions: 'You are a helpful voice AI assistant.' }),
      room: ctx.room,
    });

    await ctx.connect();
  },
});

cli.runApp(new ServerOptions({ agent: fileURLToPath(import.meta.url), agentName: 'my-agent' }));