import asyncio import base64 import json import logging import os from dotenv import load_dotenv from livekit import rtc from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, JobProcess, MetricsCollectedEvent, RunContext, WorkerOptions, cli, function_tool, metrics, room_io, ) from livekit.agents.telemetry import ( # pyright: ignore[reportMissingImports] set_tracer_provider, tracer, ) from livekit.plugins import ( anthropic, cartesia, deepgram, elevenlabs, noise_cancellation, openai, silero, ) from livekit.plugins.turn_detector.english import EnglishModel from livekit.plugins.turn_detector.multilingual import MultilingualModel from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor logger = logging.getLogger("agent") load_dotenv(".env.local") def setup_langfuse() -> bool: """Set up Langfuse tracing. Call this BEFORE session.start(). Returns True if Langfuse was configured, False if credentials not found. """ public_key = os.getenv("LANGFUSE_PUBLIC_KEY") secret_key = os.getenv("LANGFUSE_SECRET_KEY") host = os.getenv("LANGFUSE_BASE_URL", "https://cloud.langfuse.com") if not public_key or not secret_key: logger.warning("Langfuse credentials not found, skipping Langfuse setup") return False langfuse_auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode() exporter = OTLPSpanExporter( endpoint=f"{host.rstrip('/')}/api/public/otel/v1/traces", headers={"Authorization": f"Basic {langfuse_auth}"}, ) trace_provider = TracerProvider() trace_provider.add_span_processor(BatchSpanProcessor(exporter)) # Set provider BEFORE session.start() so LiveKit Cloud adds its processor too set_tracer_provider(trace_provider) logger.info("Langfuse tracing configured") return True class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions="""You are a helpful voice AI assistant. The user is interacting with you via voice, even if you perceive the conversation as text. You eagerly assist users with their questions by providing information from your extensive knowledge. Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. You are curious, friendly, and have a sense of humor.""", ) @function_tool async def lookup_weather(self, context: RunContext, location: str): """Use this tool to look up current weather information in the given location. If the location is not supported by the weather service, the tool will indicate this. You must tell the user the location's weather is unavailable. Args: location: The location to look up weather information for (e.g. city name) """ # Custom trace that appears in both LiveKit Cloud and Langfuse with tracer.start_as_current_span("lookup_weather") as span: span.set_attribute("lk.weather.location", location) span.add_event("weather_lookup_started", {"location": location}) logger.info(f"Looking up weather for {location}") # Simulate some async work (e.g., API call) await asyncio.sleep(0.1) result = "sunny with a temperature of 70 degrees" span.set_attribute("lk.weather.result", result) span.add_event("weather_lookup_completed", {"result": result}) return result server = AgentServer() def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() server.setup_fnc = prewarm async def on_session_end(ctx: JobContext) -> None: report = ctx.make_session_report() print(json.dumps(report.to_dict(), indent=2)) chat_history = report.chat_history print(chat_history) @server.rtc_session(on_session_end=on_session_end, agent_name="local_inbound_zaheer") async def my_agent(ctx: JobContext): # Logging setup # Add any other context you want in all log entries here ctx.log_context_fields = { "room": ctx.room.name, } stt = deepgram.STT(model="nova-3", language="multi") tts = elevenlabs.TTS( model="eleven_turbo_v2_5", voice_id="jqcCZkN6Knx8BJ5TBdYR", auto_mode=True ) llm = anthropic.LLM(model="claude-sonnet-4-5", caching="ephemeral") # Set up a voice AI pipeline using OpenAI, Cartesia, AssemblyAI, and the LiveKit turn detector session = AgentSession( # Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand # See all available models at https://docs.livekit.io/agents/models/stt/ stt=stt, # A Large Language Model (LLM) is your agent's brain, processing user input and generating a response # See all available models at https://docs.livekit.io/agents/models/llm/ llm=llm, # Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear # See all available models as well as voice selections at https://docs.livekit.io/agents/models/tts/ tts=tts, # VAD and turn detection are used to determine when the user is speaking and when the agent should respond # See more at https://docs.livekit.io/agents/build/turns turn_detection=EnglishModel(), vad=ctx.proc.userdata["vad"], # allow the LLM to generate a response while waiting for the end of turn # See more at https://docs.livekit.io/agents/build/audio/#preemptive-generation preemptive_generation=True, ) # To use a realtime model instead of a voice pipeline, use the following session setup instead. # (Note: This is for the OpenAI Realtime API. For other providers, see https://docs.livekit.io/agents/models/realtime/)) # 1. Install livekit-agents[openai] # 2. Set OPENAI_API_KEY in .env.local # 3. Add `from livekit.plugins import openai` to the top of this file # 4. Use the following session setup instead of the version above # session = AgentSession( # llm=openai.realtime.RealtimeModel(voice="marin") # ) # # Add a virtual avatar to the session, if desired # # For other providers, see https://docs.livekit.io/agents/models/avatar/ # avatar = hedra.AvatarSession( # avatar_id="...", # See https://docs.livekit.io/agents/models/avatar/plugins/hedra # ) # # Start the avatar and wait for it to join # await avatar.start(session, room=ctx.room) # Set up Langfuse BEFORE session.start() so both Langfuse and LiveKit Cloud receive traces # setup_langfuse() usage_collector = metrics.UsageCollector() @session.on("metrics_collected") def _on_metrics_collected(ev: MetricsCollectedEvent): metrics.log_metrics(ev.metrics) if ev.metrics.type not in ["vad_metrics", "stt_metrics"]: logger.info(ev.metrics) usage_collector.collect(ev.metrics) async def log_usage(): summary = usage_collector.get_summary() logger.info(f"Usage: {summary}") logger.info(session.history.to_dict()) # shutdown callbacks are triggered when the session is over ctx.add_shutdown_callback(log_usage) # Start the session, which initializes the voice pipeline and warms up the models await session.start( agent=Assistant(), room=ctx.room, room_options=room_io.RoomOptions( audio_input=room_io.AudioInputOptions( noise_cancellation=lambda params: noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC(), ), ), record=True ) # Join the room and connect to the user await ctx.connect() print(f"session_directory: {ctx.session_directory}") print(f"input.audio: {session.input.audio}, output.audio: {session.output.audio}") print(f"_enable_recording: {session._enable_recording}") if __name__ == "__main__": cli.run_app(server)