LiveKit

LiveKit Agents is an open-source framework for Python and Node.js that handles production-grade voice and multimodal AI applications. The framework provides abstractions for routing realtime media through AI pipelines, supporting STT-LLM-TTS voice workflows and speech-to-speech models across providers.

This guide covers routing telemetry from LiveKit Agents to InteractiveAI for monitoring, debugging, and evaluating realtime voice AI applications.

Prerequisites

• InteractiveAI account with API credentials

• LiveKit account and API credentials

• LLM provider credentials (OpenAI, Ollama, or other supported provider)

---

Configuration

Set your InteractiveAI credentials as environment variables:

INTERACTIVEAI_PUBLIC_KEY="pk-ia-..."
INTERACTIVEAI_SECRET_KEY="sk-ia-..."
INTERACTIVEAI_HOST="https://app.interactiveai.com"

---

Enabling Telemetry

LiveKit Agents includes native OpenTelemetry support. Configure a tracer provider using set_tracer_provider in your entrypoint function to route spans to InteractiveAI.

import base64
import os

from livekit.agents.telemetry import set_tracer_provider

def setup_interactiveai(
    host: str | None = None, 
    public_key: str | None = None, 
    secret_key: str | None = None
):
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor

    public_key = public_key or os.getenv("INTERACTIVEAI_PUBLIC_KEY")
    secret_key = secret_key or os.getenv("INTERACTIVEAI_SECRET_KEY")
    host = host or os.getenv("INTERACTIVEAI_HOST")

    if not public_key or not secret_key or not host:
        raise ValueError(
            "INTERACTIVEAI_PUBLIC_KEY, INTERACTIVEAI_SECRET_KEY, and INTERACTIVEAI_HOST must be set"
        )

    auth_string = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"{host.rstrip('/')}/api/public/otel"
    os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {auth_string}"

    trace_provider = TracerProvider()
    trace_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    set_tracer_provider(trace_provider)

async def entrypoint(ctx: JobContext):
    setup_interactiveai()
    
    # Agent session setup continues here
    # ...

---

Complete Example

A full implementation with agent, tool usage, and session metadata:

import base64
import logging
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,
    RunContext,
    cli,
    inference,
    metrics,
)
from livekit.agents.llm import FallbackAdapter as FallbackLLMAdapter, function_tool
from livekit.agents.stt import FallbackAdapter as FallbackSTTAdapter
from livekit.agents.telemetry import set_tracer_provider
from livekit.agents.tts import FallbackAdapter as FallbackTTSAdapter, StreamAdapter
from livekit.agents.voice import MetricsCollectedEvent
from livekit.plugins import openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

logger = logging.getLogger("interactiveai-trace-example")

load_dotenv()


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

    public_key = public_key or os.getenv("INTERACTIVEAI_PUBLIC_KEY")
    secret_key = secret_key or os.getenv("INTERACTIVEAI_SECRET_KEY")
    host = host or os.getenv("INTERACTIVEAI_HOST")

    if not public_key or not secret_key or not host:
        raise ValueError(
            "INTERACTIVEAI_PUBLIC_KEY, INTERACTIVEAI_SECRET_KEY, and INTERACTIVEAI_HOST must be set"
        )

    auth_string = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
    os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"{host.rstrip('/')}/api/public/otel"
    os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {auth_string}"

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


@function_tool
async def get_current_temperature(context: RunContext, location: str) -> str:
    """Called when the user asks about temperature or weather conditions.

    Args:
        location: The location being queried
    """
    logger.info(f"Fetching temperature for {location}")
    return "currently 72 degrees with clear skies."


class AssistantAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="Your name is Assistant.",
            llm=FallbackLLMAdapter(
                llm=[
                    inference.LLM("openai/gpt-4o-mini"),
                    inference.LLM("google/gemini-2.5-flash"),
                ]
            ),
            stt=FallbackSTTAdapter(stt=[inference.STT("deepgram"), inference.STT("cartesia")]),
            tts=FallbackTTSAdapter(
                tts=[
                    StreamAdapter(tts=openai.TTS()),
                    inference.TTS("cartesia"),
                    inference.TTS("elevenlabs"),
                ]
            ),
            turn_detection=MultilingualModel(),
            tools=[get_current_temperature],
        )

    async def on_enter(self):
        logger.info("Assistant is entering the session")
        self.session.generate_reply()


server = AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    trace_provider = setup_interactiveai(
        metadata={
            "interactiveai.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=AssistantAgent(), room=ctx.room)


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

---

Trace Visibility

The InteractiveAI platform displays:

• Complete conversation flows with turn-by-turn breakdown

• LLM requests with prompts, completions, and latency

• Speech-to-text and text-to-speech conversion times

• Function tool invocations and results

• Session metadata and performance metrics