# 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:

```env
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.

```python
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:

```python
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


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.interactive.ai/integrations/ai-frameworks/livekit.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.