hi

Thanks for reaching out, Arun. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

from datetime import datetime import json import time from dotenv import load_dotenv from livekit import api from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool from livekit.agents.voice import AgentSession from livekit.plugins import silero from agents.functions.function_builder import function_builder from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config, inactivity_handler from agents.utils import WELCOME_MESSAGE, OUTBOUND_MESSAGE, END_CALL_PROMPT from agents.utils import get_all_metadata, get_all_instructions, substitute_variables, get_interruption_params from agents.services.call_recording import CallRecordingService from agents.api import end_call from agents.rag import get_embed_model load_dotenv() def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() proc.userdata["embedding_model"] = get_embed_model() class VoiceAgent(Agent): def __init__(self, ctx, meta_data, variables=None, call_metadata=None): self.ctx = ctx self.rag = meta_data['rag'] super().__init__( instructions=get_all_instructions(meta_data['agent'], variables), stt=get_stt_from_config(meta_data['stt']), llm=get_llm_from_config(meta_data['llm']), tts=get_tts_from_config(meta_data['tts']), tools=function_builder(meta_data['functions'], call_metadata) ) async def on_user_turn_completed(self, turn_ctx, new_message) -> None: if not self.rag: return query = new_message.text_content if not query or len(query) < 3: return rag_content = await self.rag.query(query) if rag_content: turn_ctx.add_message( role="assistant", content=f"[Knowledge Base Context]: {rag_content}" ) @function_tool(description=f"{END_CALL_PROMPT}") async def end_call_on_user_request(self) -> str: self.ctx.shutdown() return json.dumps({"status": "session shutdown initiated"}, indent=2) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) dial_info = meta_data.get('dial_info', {}) to_number = dial_info.get('to_number') from_number = dial_info.get('from_number') sip_trunk_id = dial_info.get('sip_trunk_id') variables = dial_info.get('variables', {}) is_outbound = bool(to_number) if is_outbound: if not sip_trunk_id: ctx.shutdown() return try: await ctx.api.sip.create_sip_participant( api.CreateSIPParticipantRequest( room_name=ctx.room.name, sip_trunk_id=sip_trunk_id, sip_call_to=to_number, participant_identity=to_number, participant_name=f"Call to {to_number}", krisp_enabled=True, wait_until_answered=True, ) ) except api.TwirpError as e: ctx.shutdown() return is_telephony = True channel = "PHONE" else: await ctx.connect() is_telephony = False for participant in ctx.room.remote_participants.values(): if hasattr(participant, 'attributes') and participant.attributes: attrs = participant.attributes sip_phone = attrs.get('sip.phoneNumber') if sip_phone: from_number = sip_phone to_number = attrs.get('sip.trunkPhoneNumber') is_telephony = True break channel = "PHONE" if is_telephony else "WEB" call_metadata = { "from_number": from_number if is_telephony else None, "to_number": to_number if is_telephony else None, "call_type": channel, "direction": "OUTBOUND" if is_outbound else "INBOUND" if is_telephony else None, } agent = VoiceAgent(ctx, meta_data, variables, call_metadata) agent_data = meta_data['agent'] interruption_sensitivity = agent_data.get('interruption_sensitivity', 0.5) min_interruption_duration, false_interruption_timeout = get_interruption_params(interruption_sensitivity) session = AgentSession( vad=ctx.proc.userdata.get("vad"), preemptive_generation=False, min_interruption_duration=min_interruption_duration, false_interruption_timeout=false_interruption_timeout, ) inactivity_handler(session, ctx) start_time = datetime.now() recorder = CallRecordingService() await recorder.start_recording( room_name=ctx.room.name, filename=f"{ctx.room.name}_{int(time.time())}.ogg", ) await session.start(agent=agent, room=ctx.room) if is_telephony and is_outbound: greeting = agent_data.get('greeting_message', OUTBOUND_MESSAGE) else: greeting = agent_data.get('greeting_message', WELCOME_MESSAGE) greeting = substitute_variables(greeting, variables) await session.say(greeting) async def on_session_end(): recording_url = await recorder.stop_and_wait() end_call_data = { "session_id": ctx.job.id, "started_at": start_time.isoformat(), "ended_at": datetime.now().isoformat(), "transcript": json.dumps(session.history.to_dict() if hasattr(session, "history") else {"items": []}, indent=2), "agent_id": agent_data.get('id'), "client_id": agent_data.get('client_id'), "recording_url": recording_url, "direction": "OUTBOUND" if is_outbound else "INBOUND" if is_telephony else None, "channel": channel, "from_number": from_number if is_telephony else None, "to_number": to_number if is_telephony else None, } try: await end_call(end_call_data) except Exception as e: print(f"Error logging call: {e}") try: ctx.delete_room(ctx.room.name) except Exception as e: print(f"Error deleting room: {e}") ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="voice-agent", port=8082, num_idle_processes=2, ) ) if name == "__main__": run() waht will bet the minimim latence for this excpcetdd

Thanks — I see you pasted your VoiceAgent code and asked: “what will be the minimum latency for this?” I’m happy to help — I’ll explain what we can and cannot determine from the code and what affects end-to-end latency, then ask a couple quick clarifying questions so I can give a tighter estimate. Short answer • The code alone does not determine a single minimum latency. The realistic lower bound for a well‑optimized pipeline is roughly 700 ms (about 500 ms for model start + ~200 ms for first audio bytes), and an end‑to‑end target under 1.5 seconds is what teams typically aim for for natural conversational flow. • Actual latency you see will vary based on STT/LLM/TTS provider choices, network proximity, and runtime configuration. Why the code can’t give a precise minimum • Your code wires together STT, LLM, and TTS plus LiveKit networking and optional SIP. The time spent in each external service (STT -> LLM -> TTS) and the network round trips dominate latency, not the Python control flow. • Local factors the code doesn’t specify: which STT/LLM/TTS providers or models, whether those services are local vs cloud, region of deployment, and client ↔️ LiveKit network path. Key components that determine latency • STT (speech-to-text): how quickly the STT returns a stable transcription after user speech ends (varies by provider/model). • LLM: Time to First Token (TTFT) — time from transcription to first generated token. • TTS: Time to First Byte (TTFB) — time from start of audio generation to first audio bytes available to play. • Network: round-trip times between your agent, the STT/LLM/TTS services, and the client or SIP trunk. • Agent settings: you set

preemptive_generation=False

— enabling preemptive or streaming generation can reduce perceived latency. • VAD and interruption handling: VAD decisions and interruption thresholds add small delays to avoid false triggers. • Recording and logging: synchronous waits for recording or downstream logging can add time at session end (not in-turn latency, but relevant to throughput). Practical lower-bound example (optimized) • STT stable result: ~500 ms • LLM TTFT: < 500 ms (can be <200–300 ms with very low-latency models/inference hardware) • TTS TTFB: ~100–200 ms with low-latency TTS Combined optimistic lower bound: ~700 ms — 1.0 s (this is achievable when using low-latency providers, co‑located services, and streaming generation). What you can do to reduce latency • Co-locate/region-match your agent with STT/LLM/TTS endpoints to reduce RTT. • Choose low-latency providers or on-prem/inference instances (examples: real‑time STT, low-latency model inference). • Enable streaming or preemptive generation where safe (revisit

preemptive_generation

). • Tune VAD/interruption parameters to reduce wait time without creating false triggers. • Use smaller/optimized LLMs for faster TTFT when full model fidelity is not required. Quick clarifying questions so I can give a tighter estimate 1. Which STT, LLM, and TTS providers and models are you using (or planning to use)? 2. Will the agent and those services be deployed in the same region/cloud provider as LiveKit, or are they remote? 3. Is this for telephony (SIP) or web calls in production with carrier network hops? 4. Do you want minimum latency per turn (user speech → assistant audio) or total system latency including recording/logging? If you share those details I’ll give a more specific latency range and targeted optimizations for your setup. Sources: examples/bank-ivr | livekit/livekit_composite | DeepWiki | Agents (Python) | livekit/livekit_composite | DeepWiki | examples/drive-thru | livekit/livekit_composite | DeepWiki | How can I reduce latency in voice agents using STT, TTS and LLM? | LiveKit | Introducing LiveKit Inference: A unified model interface for voice AI