LiveKit Integration with Future AGI for Voice Agent Tracing

Trace a LiveKit voice agent session and inspect it in Future AGI.

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_URL=your-livekit-websocket-url
LIVEKIT_API_KEY=your-livekit-api-key
LIVEKIT_API_SECRET=your-livekit-api-secret

3. Create Your Agent

Add the Python blocks in steps 3 through 6, in order, to the same file. Create a voice assistant by extending the LiveKit Agent class with your custom instructions.

import asyncio
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

Import the tracing tools. Initialize the provider inside the agent process in the next step.

from fi_instrumentation import FITracer, register
from fi_instrumentation.fi_types import ProjectType
from traceai_livekit import 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,
        )

        # Keep the parent span open until the AgentSession closes.
        closed = asyncio.Event()

        @session.on("close")
        def on_close(_event):
            closed.set()

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

LiveKit emits close when the session ends. Waiting for that event keeps the conversation span open beyond startup. The session normally closes when its linked participant disconnects; configure session shutdown if your application keeps rooms open.

6. Run Your Agent

Start the agent server with the CLI runner.

from livekit.agents import cli

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.

This setup does not populate the Voice tab’s Duration, Turns, Talk ratio or transcript columns. Those fields require named attributes on the conversation span; its elapsed span time alone is not enough. Add application-level conversation metrics and transcripts only when you intend to store them. See Instrument and Verify a Voice Agent for those fields.

In Future AGI, select the LiveKit Agent Example project and open the conversation in the Voice tab, or inspect its model spans in Observe.