LiveKit

1. Installation

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

pip install traceai-livekit
pip install livekit
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.otel import register, ProjectType, Transport
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
    with tracer.start_as_current_span("LiveKit Agent Session", fi_span_kind="agent") 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.otel import register, ProjectType, Transport
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
    with tracer.start_as_current_span("LiveKit Agent Session", fi_span_kind="agent") 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)