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)