Documentation Index
Fetch the complete documentation index at: https://futureagi.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
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)
