voice-ai-realtime-agent

Voice AI Realtime Agent (2026 Edition)

English | Bahasa Indonesia

---

English

Description

Expert guide for building ultra-low-latency (<300ms), bi-directional conversational voice AI applications. Covers WebRTC, full-duplex WebSocket audio streaming (PCM 24kHz), OpenAI Realtime API, Gemini Multimodal Live API, LiveKit Agents SDK, and smart interruption (barge-in) handling.

Trigger Conditions

• Applications requiring sub-second, spoken conversation with an AI agent.
• Voice customer service bots, verbal copilots, language tutors, and interactive voice assistants.
• Implementation of WebRTC audio streaming, full-duplex WebSocket audio (PCM 24kHz), and Silero VAD.
• Setting up OpenAI Realtime API (gpt-4o-realtime-preview) or Gemini Multimodal Live API.

---

1. Core Architecture: Full-Duplex Speech-to-Speech

Traditional voice pipelines chain STT ➔ LLM ➔ TTS with cumulative latency exceeding 1,200ms–2,500ms. Modern 2026 voice agents use native speech-to-speech or streamable full-duplex WebRTC pipelines achieving natural, human-like reaction times (~250–350ms).

User Mic ──► [WebRTC / WebSocket] ──► [VAD: Silero / WebRTC VAD]
                                                │
                                                ▼
User Speaks <── [Audio Output] ◄── [Native Audio Stream / Cartesia] ◄── [OpenAI Realtime / Gemini Live]
     │
     └── User Interrupts (Barge-in) ──► Instant Buffer Flush & Cancel Audio Frame Emission

---

2. Production Recipe: LiveKit Agents + OpenAI Realtime (Python)

# agent.py - Production Voice Agent Worker with LiveKit & OpenAI Realtime
import asyncio
import os
from livekit import rtc
from livekit.agents import (
    AutoSubscribe,
    JobContext,
    JobProcess,
    WorkerOptions,
    cli,
    llm,
)
from livekit.agents.pipeline import VoicePipelineAgent
from livekit.plugins import deepgram, openai, silero

async def entrypoint(ctx: JobContext):
    # Connect to room with audio only to minimize bandwidth & latency
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    
    # Wait for the user participant to join
    participant = await ctx.wait_for_participant()
    
    # Define agent instructions and tools
    initial_ctx = llm.ChatContext().append(
        role="system",
        text=(
            "You are a helpful, concise voice assistant. "
            "Respond naturally in 1-2 short sentences. Never output markdown, bullet points, or emojis."
        )
    )

    # Realtime Voice Pipeline: Deepgram (STT) + OpenAI (LLM) + Cartesia/OpenAI (TTS)
    # Or use native OpenAI Realtime Model: gpt-4o-realtime-preview
    agent = VoicePipelineAgent(
        vad=silero.VAD.load(
            min_speech_duration=0.1,
            min_silence_duration=0.3, # Snappy turn-taking
            prefix_padding_duration=0.2,
        ),
        stt=deepgram.STT(model="nova-2", language="id"), # Multi-language support
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(voice="alloy"),
        chat_ctx=initial_ctx,
        allow_interruptions=True, # Barge-in capability
        interrupt_speech_duration=0.3, # Immediate cutoff when user talks
    )

    agent.start(ctx.room, participant)
    
    # Greet user immediately
    await agent.say("Halo! Ada yang bisa saya bantu hari ini?", now=True)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))