LiveKit Integration with Future AGI for Voice Agent Tracing

1. Installation

Install the traceAI and LiveKit agent packages to enable voice agent capabilities with observability.

pip install traceAI-livekit
pip install livekit-agents livekit-plugins-openai livekit-plugins-silero
pip install python-dotenv

2. Set Environment Variables

Set up your environment variables to authenticate with both FutureAGI and LiveKit services.

# .env file
FI_API_KEY=your-futureagi-api-key
FI_SECRET_KEY=your-futureagi-secret-key
OPENAI_API_KEY=your-openai-api-key
LIVEKIT_API_KEY=your-livekit-api-key
LIVEKIT_API_SECRET=your-livekit-api-secret

3. Create Your Agent

Create a voice assistant agent by extending the LiveKit Agent class with your custom instructions.

import logging

from dotenv import load_dotenv
from livekit.agents import (
    Agent,
    AgentServer,
)

load_dotenv()

logger = logging.getLogger("traceai-example")

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a voice assistant created by Future AGI. Your interface with users will be voice.
            You should provide short and concise answers to user queries.
            """,
        )

4. Initialize Trace Provider

Set up the trace provider to create a new project in FutureAGI and establish telemetry data pipelines.

# TraceAI imports
from fi_instrumentation import FITracer
from fi_instrumentation import register
from fi_instrumentation.fi_types import ProjectType
from traceai_livekit import enable_http_attribute_mapping

# Initialize the trace provider
provider = register(
    project_name="LiveKit Agent Example",
    project_type=ProjectType.OBSERVE,
    set_global_tracer_provider=True,
)
enable_http_attribute_mapping()

5. Implement the Agent Session

Create the agent session with appropriate speech-to-text, language model, and text-to-speech components.

from livekit.agents import (
    JobContext,
    JobProcess,
    AgentSession,
    room_io,
)
from livekit.plugins import openai, silero

server = AgentServer()

def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    logger.info(f"connecting to room {ctx.room.name}")
    
    # Initialize TraceAI INSIDE the process to avoid multiprocessing pickling errors
    provider = register(
        project_name="LiveKit Agent Example",
        project_type=ProjectType.OBSERVE,
        set_global_tracer_provider=True,
    )
    enable_http_attribute_mapping()
    
    # Create the tracer helper
    tracer = FITracer(provider.get_tracer(__name__))
    
    # Use context manager for parent span instead of decorator
    # This ensures the span starts when this process is actually running.
    # "conversation", not "agent": the Voice tab lists a conversation-typed span with
    # no parent, and this span is opened before session.start() so it is that root.
    with tracer.start_as_current_span("LiveKit Agent Session", fi_span_kind="conversation") as parent_span:
        parent_span.set_input(f"Room: {ctx.room.name}")
    
        # Modern AgentSession setup
        session = AgentSession(
            stt=openai.STT(), # Requires OPENAI_API_KEY
            llm=openai.LLM(),   # Requires OPENAI_API_KEY
            tts=openai.TTS(),   # Requires OPENAI_API_KEY
            vad=ctx.proc.userdata["vad"],
            preemptive_generation=True,
        )

        await session.start(
            agent=Assistant(),
            room=ctx.room,
            room_options=room_io.RoomOptions(
                audio_input=room_io.AudioInputOptions(),
            ),
        )
        
        await ctx.connect()

6. Run Your Agent

Start the agent server with the CLI runner.

from livekit.agents import cli

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

Complete Example

Here’s a complete example that puts everything together:

import logging
import os

from dotenv import load_dotenv
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    cli,
    inference,
    room_io,
)
from livekit.plugins import openai, silero

# TraceAI Imports
from fi_instrumentation import FITracer
from fi_instrumentation import register
from fi_instrumentation.fi_types import ProjectType
from traceai_livekit import enable_http_attribute_mapping

load_dotenv()

logger = logging.getLogger("traceai-example")

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a voice assistant created by Future AGI. Your interface with users will be voice.
            You should provide short and concise answers to user queries.
            """,
        )

server = AgentServer()

def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    logger.info(f"connecting to room {ctx.room.name}")
    
    # Initialize TraceAI INSIDE the process to avoid multiprocessing pickling errors
    provider = register(
        project_name="LiveKit Agent Example",
        project_type=ProjectType.OBSERVE,
        set_global_tracer_provider=True,
    )
    enable_http_attribute_mapping()
    
    # Create the tracer helper
    tracer = FITracer(provider.get_tracer(__name__))
    
    # Use context manager for parent span instead of decorator
    # This ensures the span starts when this process is actually running.
    # "conversation", not "agent": the Voice tab lists a conversation-typed span with
    # no parent, and this span is opened before session.start() so it is that root.
    with tracer.start_as_current_span("LiveKit Agent Session", fi_span_kind="conversation") as parent_span:
        parent_span.set_input(f"Room: {ctx.room.name}")
    
        # Modern AgentSession setup
        session = AgentSession(
            stt=openai.STT(), # Requires OPENAI_API_KEY
            llm=openai.LLM(),   # Requires OPENAI_API_KEY
            tts=openai.TTS(),   # Requires OPENAI_API_KEY
            vad=ctx.proc.userdata["vad"],
            preemptive_generation=True,
        )

        await session.start(
            agent=Assistant(),
            room=ctx.room,
            room_options=room_io.RoomOptions(
                audio_input=room_io.AudioInputOptions(),
            ),
        )
        
        await ctx.connect()

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

Make the run appear in the Voice tab

The span kind above is what puts the call in the Voice tab: it lists a conversation-typed span with no parent, and this one is opened before session.start(), so it is the root. Typed anything else, or opened inside a running session, the call is correct in Traces and absent from every voice surface.

That span is also where the Duration, Turns, Talk ratio and transcript columns are read from, by name. The instrumentor does not write any of them.

Instrument and Verify a Voice Agent is the full path, with a checker that runs twelve gates against the spans your agent really sent.