Building a Voice Agent with the Gemini Live API

The Gemini Live API enables low-latency, real-time voice and vision interactions. It processes continuous streams of audio, images, and text to deliver immediate, human-like spoken responses—creating natural conversational experiences.

Architecture Overview

A voice agent with Gemini Live API has three main components:

┌─────────────┐     WebSocket      ┌─────────────┐
│   Client    │ ◄────────────────► │   Gemini    │
│  (Browser)  │                    │  Live API   │
└─────────────┘                    └─────────────┘
       │                                 │
       │ Audio Stream                    │ Audio Response
       ▼                                 ▼
┌─────────────┐                    ┌─────────────┐
│   Microphone│                    │   Speaker   │
└─────────────┘                    └─────────────┘

Setting Up the Environment

Install Dependencies

pip install google-genai websockets asyncio

Basic Configuration

import asyncio
from google import genai

# Initialize client
client = genai.Client()

# Configuration
MODEL = "gemini-3.1-flash-live-preview"
CONFIG = {
    "response_modalities": ["AUDIO"],
    "speech_config": {
        "voice_name": "Kore",  # Expressive voice
        "language_code": "en-US"
    }
}

Creating a Voice Agent Session

Basic Session Setup

class VoiceAgent:
    def __init__(self):
        self.client = genai.Client()
        self.model = "gemini-3.1-flash-live-preview"
        self.config = {
            "response_modalities": ["AUDIO"],
            "speech_config": {
                "voice_name": "Kore",
                "language_code": "en-US"
            }
        }
        self.session = None
        
    async def start_session(self):
        """Start a Live API session."""
        self.session = await self.client.aio.live.connect(
            model=self.model,
            config=self.config
        )
        print("Session started")
        
    async def send_audio(self, audio_data: bytes):
        """Send audio data to the model."""
        await self.session.send_realtime_input(
            media=audio_data
        )
        
    async def receive_audio(self) -> AsyncGenerator[bytes, None]:
        """Receive audio responses from the model."""
        async for response in self.session.receive():
            if response.server_content and response.server_content.model_turn:
                for part in response.server_content.model_turn.parts:
                    if part.inline_data:
                        yield part.inline_data.data
                        
    async def close_session(self):
        """Close the session."""
        if self.session:
            await self.session.close()

End-to-End Example

import asyncio
import websockets
from google import genai

async def voice_agent_websocket():
    """Complete voice agent with WebSocket."""
    
    async with websockets.connect("ws://localhost:8080") as client_ws:
        # Connect to Gemini Live API
        async with client.aio.live.connect(
            model="gemini-3.1-flash-live-preview",
            config={"response_modalities": ["AUDIO"]}
        ) as session:
            
            # Set up bidirectional audio forwarding
            async def forward_to_gemini():
                async for message in client_ws:
                    await session.send_realtime_input(media=message)
                    
            async def forward_to_client():
                async for response in session.receive():
                    if response.server_content and response.server_content.model_turn:
                        for part in response.server_content.model_turn.parts:
                            if part.inline_data:
                                await client_ws.send(part.inline_data.data)
            
            # Run both directions concurrently
            await asyncio.gather(
                forward_to_gemini(),
                forward_to_client()
            )

Handling Voice Activity Detection (VAD)

Automatic VAD Configuration

class VoiceAgentWithVAD:
    def __init__(self):
        self.config = {
            "response_modalities": ["AUDIO"],
            "realtime_input_config": {
                "automatic_activity_detection": {
                    "start_of_speech_sensitivity": "HIGH",
                    "end_of_speech_sensitivity": "MEDIUM",
                    "prefix_padding_ms": 200,
                    "silence_duration_ms": 500
                }
            }
        }

Manual VAD Control

class VoiceAgentManualVAD:
    def __init__(self):
        self.config = {
            "response_modalities": ["AUDIO"],
            "realtime_input_config": {
                "automatic_activity_detection": {
                    "disabled": True
                }
            }
        }
        
    async def send_speech_start(self):
        """Manually signal start of speech."""
        await self.session.send_realtime_input(
            activity_start={}
        )
        
    async def send_speech_end(self):
        """Manually signal end of speech."""
        await self.session.send_realtime_input(
            activity_end={}
        )

Handling Interruptions

Barge-In Support

class InterruptibleVoiceAgent:
    def __init__(self):
        self.is_speaking = False
        
    async def handle_interruption(self):
        """Handle user interruption during model response."""
        # Gemini Live API automatically handles interruptions
        # When VAD detects user speech, model output is canceled
        
        # The server sends a BidiGenerateContentServerContent
        # message to report the interruption
        pass
        
    async def process_with_interruption(self):
        """Process audio with interruption support."""
        async for response in self.session.receive():
            if response.server_content:
                if response.server_content.interrupted:
                    print("Model was interrupted by user")
                    # Reset any buffered audio
                    self.clear_playback_buffer()
                elif response.server_content.model_turn:
                    # Process model response
                    await self.play_audio_response(
                        response.server_content.model_turn
                    )

Adding Tools and Function Calling

Define Tools

class VoiceAgentWithTools:
    def __init__(self):
        self.config = {
            "response_modalities": ["AUDIO"],
            "tools": [
                {
                    "function_declarations": [
                        {
                            "name": "get_weather",
                            "description": "Get current weather for a location",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "location": {
                                        "type": "string",
                                        "description": "City name"
                                    }
                                },
                                "required": ["location"]
                            }
                        }
                    ]
                }
            ]
        }
        
    async def handle_tool_call(self, tool_call):
        """Handle tool calls from the model."""
        if tool_call.function_call.name == "get_weather":
            location = tool_call.function_call.args["location"]
            weather_data = await self.get_weather(location)
            
            # Send tool response back
            await self.session.send_tool_response(
                function_responses=[{
                    "name": "get_weather",
                    "response": weather_data
                }]
            )

Audio Format Handling

PCM Audio Processing

class AudioProcessor:
    def __init__(self, input_sample_rate=16000, output_sample_rate=24000):
        self.input_sample_rate = input_sample_rate
        self.output_sample_rate = output_sample_rate
        
    def prepare_audio_for_gemini(self, audio_data: bytes) -> bytes:
        """Convert audio to Gemini's expected format."""
        # Gemini expects raw 16-bit PCM, 16kHz, little-endian
        # Ensure correct format
        return audio_data
        
    def process_gemini_response(self, audio_data: bytes) -> bytes:
        """Process Gemini's audio response."""
        # Gemini returns raw 16-bit PCM, 24kHz, little-endian
        return audio_data

Web Audio API Integration

// Client-side audio handling
class AudioStreamHandler {
    constructor() {
        this.audioContext = new AudioContext({ sampleRate: 16000 });
        this.analyser = this.audioContext.createAnalyser();
    }
    
    async startMicrophoneStream() {
        const stream = await navigator.mediaDevices.getUserMedia({
            audio: {
                sampleRate: 16000,
                channelCount: 1,
                echoCancellation: true,
                noiseSuppression: true
            }
        });
        
        const source = this.audioContext.createMediaStreamSource(stream);
        source.connect(this.analyser);
        
        return stream;
    }
    
    async sendAudioToWebSocket(audioData) {
        // Send raw PCM data to server
        await this.websocket.send(audioData);
    }
}

Session Management

Resumable Sessions

class ResumableVoiceAgent:
    def __init__(self):
        self.session_checkpoint = None
        
    async def save_session(self):
        """Save session state for resumption."""
        # Gemini Live API supports session resumption
        # via SessionResumptionUpdate messages
        pass
        
    async def resume_session(self, checkpoint: dict):
        """Resume from a saved checkpoint."""
        # Use checkpoint to resume session
        pass

Multi-Turn Conversations

class MultiTurnVoiceAgent:
    def __init__(self):
        self.conversation_history = []
        
    async def handle_conversation_turn(self, user_audio: bytes):
        """Handle a complete conversation turn."""
        # Send user audio
        await self.session.send_realtime_input(media=user_audio)
        
        # Receive model response
        response_text = ""
        response_audio = b""
        
        async for response in self.session.receive():
            if response.server_content and response.server_content.model_turn:
                for part in response.server_content.model_turn.parts:
                    if part.text:
                        response_text += part.text
                    if part.inline_data:
                        response_audio += part.inline_data.data
        
        # Store in history
        self.conversation_history.append({
            "user_audio": user_audio,
            "model_text": response_text,
            "model_audio": response_audio
        })

Production Considerations

Error Handling

class ProductionVoiceAgent:
    async def safe_session(self):
        """Handle connection errors gracefully."""
        try:
            async with self.client.aio.live.connect(
                model=self.model,
                config=self.config
            ) as session:
                return session
        except Exception as e:
            print(f"Session error: {e}")
            # Implement retry logic
            await self.retry_connection()

Rate Limiting

class RateLimitedAgent:
    def __init__(self, max_concurrent_sessions=10):
        self.semaphore = asyncio.Semaphore(max_concurrent_sessions)
        
    async def handle_session(self, session_id: str):
        """Handle session with rate limiting."""
        async with self.semaphore:
            # Process session
            pass

Monitoring and Logging

import logging
from dataclasses import dataclass
from datetime import datetime

@dataclass
class SessionMetrics:
    session_id: str
    start_time: datetime
    audio_duration_seconds: float
    interruption_count: int
    tool_calls: int

class MonitoredVoiceAgent:
    def __init__(self):
        self.logger = logging.getLogger("voice_agent")
        self.metrics = {}
        
    def log_session_start(self, session_id: str):
        self.logger.info(f"Session started: {session_id}")
        self.metrics[session_id] = SessionMetrics(
            session_id=session_id,
            start_time=datetime.now(),
            audio_duration_seconds=0,
            interruption_count=0,
            tool_calls=0
        )
        
    def log_interruption(self, session_id: str):
        self.metrics[session_id].interruption_count += 1

Complete Example

import asyncio
from google import genai

async def main():
    # Initialize
    client = genai.Client()
    
    # Configuration
    config = {
        "response_modalities": ["AUDIO"],
        "speech_config": {
            "voice_name": "Kore",
            "language_code": "en-US"
        },
        "system_instruction": {
            "parts": [{"text": "You are a helpful voice assistant."}]
        }
    }
    
    # Start session
    async with client.aio.live.connect(
        model="gemini-3.1-flash-live-preview",
        config=config
    ) as session:
        print("Connected to Gemini Live API")
        
        # Example: Send a text prompt (for testing)
        await session.send_client_content(
            turns={"role": "user", "parts": [{"text": "Hello!"}]}
        )
        
        # Receive response
        async for response in session.receive():
            if response.server_content and response.server_content.model_turn:
                for part in response.server_content.model_turn.parts:
                    if part.text:
                        print(f"Text: {part.text}")
                    if part.inline_data:
                        print(f"Audio chunk: {len(part.inline_data.data)} bytes")

if __name__ == "__main__":
    asyncio.run(main())

Conclusion

The Gemini Live API provides a powerful foundation for building voice agents:

1. WebSocket-based: Persistent connections for low-latency streaming
2. Automatic VAD: Built-in voice activity detection
3. Interrupt handling: Natural barge-in support
4. Tool integration: Function calling for external actions
5. Session memory: Context retention across turns

Key best practices:

• Use automatic VAD for simplicity, manual for control
• Handle interruptions gracefully
• Implement proper error handling and retries
• Monitor session metrics for production
• Test with real audio content

The API enables natural conversational experiences where users can interrupt, ask questions, and receive immediate spoken responses—making voice agents feel truly interactive.