Audio Language Models

Audio Language Models ()

Build real-time voice agents and audio processing using the latest native speech-to-speech models.

Overview

• Real-time voice assistants and agents
• Live conversational AI (phone agents, support bots)
• Audio transcription with speaker diarization
• Multilingual voice interactions
• Text-to-speech generation
• Voice-to-voice translation

Model Comparison (January )

Real-Time Voice (Speech-to-Speech)

Speech-to-Text Only

Grok Voice Agent API (xAI) - Fastest

import asyncio
import websockets
import json

async def grok_voice_agent():
    """Real-time voice agent with Grok - #1 on Big Bench Audio.

    Features:
    - <1 second time-to-first-audio (5x faster than competitors)
    - Native speech-to-speech (no transcription intermediary)
    - 100+ languages, $0.05/min
    - OpenAI Realtime API compatible
    """
    uri = "wss://api.x.ai/v1/realtime"
    headers = {"Authorization": f"Bearer {XAI_API_KEY}"}

    async with websockets.connect(uri, extra_headers=headers) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "grok-4-voice",
                "voice": "Aria",  # or "Eve", "Leo"
                "instructions": "You are a helpful voice assistant.",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {"type": "server_vad"}
            }
        }))

        # Stream audio in/out
        async def send_audio(audio_stream):
            async for chunk in audio_stream:
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode()
                }))

        async def receive_audio():
            async for message in ws:
                data = json.loads(message)
                if data["type"] == "response.audio.delta":
                    yield base64.b64decode(data["delta"])

        return send_audio, receive_audio

# Expressive voice with auditory cues
async def expressive_response(ws, text: str):
    """Use auditory cues for natural speech."""
    # Supports: [whisper], [sigh], [laugh], [pause]
    await ws.send(json.dumps({
        "type": "response.create",
        "response": {
            "instructions": "[sigh] Let me think about that... [pause] Here's what I found."
        }
    }))

Gemini Live API (Google) - Emotional Awareness

import google.generativeai as genai
from google.generativeai import live

genai.configure(api_key="YOUR_API_KEY")

async def gemini_live_voice():
    """Real-time voice with emotional understanding.

    Features:
    - 30 HD voices in 24 languages
    - Affective dialog (understands emotions)
    - Barge-in support (interrupt anytime)
    - Proactive audio (responds only when relevant)
    """
    model = genai.GenerativeModel("gemini-2.5-flash-live")

    config = live.LiveConnectConfig(
        response_modalities=["AUDIO"],
        speech_config=live.SpeechConfig(
            voice_config=live.VoiceConfig(
                prebuilt_voice_config=live.PrebuiltVoiceConfig(
                    voice_name="Puck"  # or Charon, Kore, Fenrir, Aoede
                )
            )
        ),
        system_instruction="You are a friendly voice assistant."
    )

    async with model.connect(config=config) as session:
        # Send audio
        async def send_audio(audio_chunk: bytes):
            await session.send(
                input=live.LiveClientContent(
                    realtime_input=live.RealtimeInput(
                        media_chunks=[live.MediaChunk(
                            data=audio_chunk,
                            mime_type="audio/pcm"
                        )]
                    )
                )
            )

        # Receive audio responses
        async for response in session.receive():
            if response.data:
                yield response.data  # Audio bytes

# With transcription
async def gemini_live_with_transcript():
    """Get both audio and text transcripts."""
    async with model.connect(config=config) as session:
        async for response in session.receive():
            if response.server_content:
                # Text transcript
                if response.server_content.model_turn:
                    for part in response.server_content.model_turn.parts:
                        if part.text:
                            print(f"Transcript: {part.text}")
            if response.data:
                yield response.data  # Audio

Gemini Audio Transcription (Long-Form)

import google.generativeai as genai

def transcribe_with_gemini(audio_path: str) -> dict:
    """Transcribe up to 9.5 hours of audio with speaker diarization.

    Gemini 2.5 Pro handles long-form audio natively.
    """
    model = genai.GenerativeModel("gemini-2.5-pro")

    # Upload audio file
    audio_file = genai.upload_file(audio_path)

    response = model.generate_content([
        audio_file,
        """Transcribe this audio with:
        1. Speaker labels (Speaker 1, Speaker 2, etc.)
        2. Timestamps for each segment
        3. Punctuation and formatting

        Format:
        [00:00:00] Speaker 1: First statement...
        [00:00:15] Speaker 2: Response..."""
    ])

    return {
        "transcript": response.text,
        "audio_duration": audio_file.duration
    }

Gemini TTS (Text-to-Speech)

def gemini_text_to_speech(text: str, voice: str = "Kore") -> bytes:
    """Generate speech with Gemini 2.5 TTS.

    Features:
    - Enhanced expressivity with style prompts
    - Precision pacing (context-aware speed)
    - Multi-speaker dialogue consistency
    """
    model = genai.GenerativeModel("gemini-2.5-flash-tts")

    response = model.generate_content(
        contents=text,
        generation_config=genai.GenerationConfig(
            response_mime_type="audio/mp3",
            speech_config=genai.SpeechConfig(
                voice_config=genai.VoiceConfig(
                    prebuilt_voice_config=genai.PrebuiltVoiceConfig(
                        voice_name=voice  # Puck, Charon, Kore, Fenrir, Aoede
                    )
                )
            )
        )
    )

    return response.audio

OpenAI GPT-4o-Transcribe

from openai import OpenAI

client = OpenAI()

def transcribe_openai(audio_path: str, language: str = None) -> dict:
    """Transcribe with GPT-4o-Transcribe (enhanced accuracy)."""
    with open(audio_path, "rb") as audio_file:
        response = client.audio.transcriptions.create(
            model="gpt-4o-transcribe",
            file=audio_file,
            language=language,
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
        )
    return {
        "text": response.text,
        "words": response.words,
        "segments": response.segments,
        "duration": response.duration
    }

AssemblyAI (Best Features)

import assemblyai as aai

aai.settings.api_key = "YOUR_API_KEY"

def transcribe_assemblyai(audio_url: str) -> dict:
    """Transcribe with speaker diarization, sentiment, entities."""
    config = aai.TranscriptionConfig(
        speaker_labels=True,
        sentiment_analysis=True,
        entity_detection=True,
        auto_highlights=True,
        language_detection=True
    )

    transcriber = aai.Transcriber()
    transcript = transcriber.transcribe(audio_url, config=config)

    return {
        "text": transcript.text,
        "speakers": transcript.utterances,
        "sentiment": transcript.sentiment_analysis,
        "entities": transcript.entities
    }

Real-Time Streaming Comparison

async def choose_realtime_provider(
    requirements: dict
) -> str:
    """Select best real-time voice provider."""

    if requirements.get("fastest_latency"):
        return "grok"  # <1s TTFA, 5x faster

    if requirements.get("emotional_understanding"):
        return "gemini"  # Affective dialog

    if requirements.get("openai_ecosystem"):
        return "openai"  # Compatible tools

    if requirements.get("lowest_cost"):
        return "grok"  # $0.05/min (half of OpenAI)

    return "gemini"  # Best overall for 

API Pricing (January )

Key Decisions

Common Mistakes

• Using STT+LLM+TTS pipeline instead of native speech-to-speech
• Not leveraging emotional understanding (Gemini)
• Ignoring barge-in support for natural conversations
• Using deprecated Whisper-1 instead of GPT-4o-Transcribe
• Not testing latency with real users

Related Skills

• vision-language-models - Image/video processing
• multimodal-rag - Audio + text retrieval
• streaming-api-patterns - WebSocket patterns

Capability Details

real-time-voice

Keywords: voice agent, real-time, conversational, live audio Solves:

• Build voice assistants
• Phone agents and support bots
• Interactive voice response (IVR)

speech-to-speech

Keywords: native audio, speech-to-speech, no transcription Solves:

• Low-latency voice responses
• Natural conversation flow
• Emotional voice interactions

transcription

Keywords: transcribe, speech-to-text, STT, convert audio Solves:

• Convert audio files to text
• Generate meeting transcripts
• Process long-form audio

voice-tts

Keywords: TTS, text-to-speech, voice synthesis Solves:

• Generate natural speech
• Multi-voice dialogue
• Expressive audio output

Audio Language Models Checklist

Real-Time Voice

• [ ] Grok Voice Agent WebSocket setup
• [ ] Gemini Live API connection
• [ ] Voice activity detection (VAD)
• [ ] Barge-in support
• [ ] Session management

Transcription

• [ ] Gemini 2.5 Pro for long audio (9.5hr)
• [ ] GPT-4o-Transcribe for accuracy
• [ ] Speaker diarization
• [ ] Timestamp generation
• [ ] Language detection

Text-to-Speech

• [ ] Gemini TTS with style prompts
• [ ] OpenAI TTS voice selection
• [ ] Grok expressive cues
• [ ] Multi-speaker dialogue
• [ ] Streaming audio output

Audio Processing

• [ ] Convert to 16kHz mono WAV
• [ ] Normalize audio levels
• [ ] Handle long audio chunking
• [ ] Support common formats (mp3, wav, m4a)

WebSocket Integration

• [ ] Connection establishment
• [ ] Audio streaming (input/output)
• [ ] Transcript events
• [ ] Reconnection logic
• [ ] Graceful shutdown

Error Handling

• [ ] Handle connection drops
• [ ] Audio format validation
• [ ] Rate limit handling
• [ ] Timeout management
• [ ] Fallback to STT+LLM+TTS if needed

Real-Time Voice Streaming (2026)

Patterns for building real-time voice agents using Grok Voice Agent API and Gemini Live API.

Provider Comparison

Grok Voice Agent (WebSocket)

import asyncio
import websockets
import json
import base64

class GrokVoiceAgent:
    """Real-time voice agent - #1 on Big Bench Audio.

    - <1 second time-to-first-audio
    - Native speech-to-speech (no transcription step)
    - $0.05/min (half of OpenAI)
    - OpenAI Realtime API compatible
    """

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.uri = "wss://api.x.ai/v1/realtime"
        self.ws = None

    async def connect(
        self,
        voice: str = "Aria",
        instructions: str = "You are a helpful assistant."
    ):
        """Establish WebSocket connection."""
        headers = {"Authorization": f"Bearer {self.api_key}"}

        self.ws = await websockets.connect(
            self.uri,
            extra_headers=headers
        )

        # Configure session
        await self.ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "grok-4-voice",
                "voice": voice,  # Aria, Eve, Leo
                "instructions": instructions,
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 500
                }
            }
        }))

    async def send_audio(self, audio_chunk: bytes):
        """Send audio chunk to the model."""
        await self.ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(audio_chunk).decode()
        }))

    async def receive(self):
        """Receive responses from the model."""
        async for message in self.ws:
            data = json.loads(message)

            if data["type"] == "response.audio.delta":
                yield {
                    "type": "audio",
                    "data": base64.b64decode(data["delta"])
                }
            elif data["type"] == "response.text.delta":
                yield {
                    "type": "transcript",
                    "data": data["delta"]
                }
            elif data["type"] == "input_audio_buffer.speech_started":
                yield {"type": "user_speaking"}
            elif data["type"] == "input_audio_buffer.speech_stopped":
                yield {"type": "user_stopped"}

    async def close(self):
        """Close the connection."""
        if self.ws:
            await self.ws.close()

# Usage
async def voice_assistant():
    agent = GrokVoiceAgent(api_key="YOUR_XAI_KEY")
    await agent.connect(
        voice="Aria",
        instructions="You are a friendly customer support agent."
    )

    # Stream microphone audio
    async for audio_chunk in get_microphone_stream():
        await agent.send_audio(audio_chunk)

    # Receive and play responses
    async for response in agent.receive():
        if response["type"] == "audio":
            play_audio(response["data"])
        elif response["type"] == "transcript":
            print(f"Assistant: {response['data']}")

Gemini Live API (Emotional AI)

import google.generativeai as genai
from google.generativeai import live

genai.configure(api_key="YOUR_API_KEY")

class GeminiLiveAgent:
    """Real-time voice with emotional understanding.

    - 30 HD voices in 24 languages
    - Affective dialog (understands user emotions)
    - Barge-in support
    - Proactive audio mode
    """

    def __init__(self):
        self.model = genai.GenerativeModel("gemini-2.5-flash-live")
        self.session = None

    async def connect(
        self,
        voice: str = "Puck",
        instructions: str = "You are a helpful assistant."
    ):
        """Connect to Gemini Live."""
        config = live.LiveConnectConfig(
            response_modalities=["AUDIO", "TEXT"],
            speech_config=live.SpeechConfig(
                voice_config=live.VoiceConfig(
                    prebuilt_voice_config=live.PrebuiltVoiceConfig(
                        voice_name=voice  # Puck, Charon, Kore, Fenrir, Aoede
                    )
                )
            ),
            system_instruction=instructions,
            # Enable emotional understanding
            enable_affective_dialog=True
        )

        self.session = await self.model.connect(config=config)

    async def send_audio(self, audio_chunk: bytes):
        """Send audio to Gemini."""
        await self.session.send(
            input=live.LiveClientContent(
                realtime_input=live.RealtimeInput(
                    media_chunks=[live.MediaChunk(
                        data=audio_chunk,
                        mime_type="audio/pcm"
                    )]
                )
            )
        )

    async def receive(self):
        """Receive audio and text responses."""
        async for response in self.session.receive():
            if response.data:
                yield {"type": "audio", "data": response.data}

            if response.server_content:
                if response.server_content.model_turn:
                    for part in response.server_content.model_turn.parts:
                        if part.text:
                            yield {"type": "transcript", "data": part.text}

    async def close(self):
        """Close the session."""
        if self.session:
            await self.session.close()

# Gemini voices
GEMINI_VOICES = {
    "Puck": "Playful, energetic",
    "Charon": "Deep, authoritative",
    "Kore": "Warm, friendly",
    "Fenrir": "Strong, confident",
    "Aoede": "Melodic, soothing"
}

FastAPI WebSocket Endpoint

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()

@app.websocket("/ws/voice")
async def voice_endpoint(websocket: WebSocket):
    """WebSocket endpoint for real-time voice."""
    await websocket.accept()

    # Choose provider based on requirements
    provider = websocket.query_params.get("provider", "grok")

    if provider == "grok":
        agent = GrokVoiceAgent(api_key=XAI_KEY)
    else:
        agent = GeminiLiveAgent()

    await agent.connect()

    async def receive_from_client():
        """Receive audio from client."""
        try:
            while True:
                audio = await websocket.receive_bytes()
                await agent.send_audio(audio)
        except Exception:
            pass

    async def send_to_client():
        """Send audio to client."""
        async for response in agent.receive():
            if response["type"] == "audio":
                await websocket.send_bytes(response["data"])
            elif response["type"] == "transcript":
                await websocket.send_json({
                    "type": "transcript",
                    "text": response["data"]
                })

    # Run both tasks
    await asyncio.gather(
        receive_from_client(),
        send_to_client()
    )

    await agent.close()

Browser Client (JavaScript)

class VoiceClient {
  constructor(wsUrl) {
    this.ws = new WebSocket(wsUrl);
    this.audioContext = new AudioContext({ sampleRate: 24000 });
    this.mediaRecorder = null;
  }

  async start() {
    // Get microphone
    const stream = await navigator.mediaDevices.getUserMedia({
      audio: {
        channelCount: 1,
        sampleRate: 16000,
        echoCancellation: true,
        noiseSuppression: true
      }
    });

    // Record and send audio
    this.mediaRecorder = new MediaRecorder(stream, {
      mimeType: 'audio/webm;codecs=opus'
    });

    this.mediaRecorder.ondataavailable = (e) => {
      if (e.data.size > 0 && this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(e.data);
      }
    };

    // Send chunks every 100ms for low latency
    this.mediaRecorder.start(100);

    // Handle incoming audio
    this.ws.onmessage = async (e) => {
      if (e.data instanceof Blob) {
        const arrayBuffer = await e.data.arrayBuffer();
        this.playAudio(arrayBuffer);
      } else {
        const data = JSON.parse(e.data);
        if (data.type === 'transcript') {
          this.onTranscript(data.text);
        }
      }
    };
  }

  playAudio(arrayBuffer) {
    this.audioContext.decodeAudioData(arrayBuffer, (buffer) => {
      const source = this.audioContext.createBufferSource();
      source.buffer = buffer;
      source.connect(this.audioContext.destination);
      source.start();
    });
  }

  onTranscript(text) {
    console.log('Assistant:', text);
  }

  stop() {
    this.mediaRecorder?.stop();
    this.ws?.close();
  }
}

// Usage
const client = new VoiceClient('wss://api.example.com/ws/voice?provider=grok');
await client.start();

Expressive Voice (Grok)

# Grok supports auditory cues for natural speech
AUDITORY_CUES = [
    "[whisper]",   # Soft, quiet speech
    "[sigh]",      # Exhalation
    "[laugh]",     # Laughter
    "[pause]",     # Brief pause
    "[excited]",   # Enthusiastic tone
    "[concerned]", # Worried tone
]

async def send_expressive_response(agent, text: str):
    """Send response with emotional cues."""
    await agent.ws.send(json.dumps({
        "type": "response.create",
        "response": {
            "modalities": ["text", "audio"],
            "instructions": text
        }
    }))

# Example: Empathetic response
await send_expressive_response(
    agent,
    "[concerned] I understand that must be frustrating. "
    "[pause] Let me help you resolve this issue."
)

Error Handling & Reconnection

async def resilient_voice_agent(
    agent_class,
    max_retries: int = 3
):
    """Voice agent with automatic reconnection."""
    retry_count = 0
    agent = None

    while retry_count < max_retries:
        try:
            agent = agent_class()
            await agent.connect()

            async for response in agent.receive():
                yield response
                retry_count = 0  # Reset on success

        except websockets.ConnectionClosed:
            retry_count += 1
            wait = 2 ** retry_count
            print(f"Connection lost, retrying in {wait}s...")
            await asyncio.sleep(wait)

        except Exception as e:
            print(f"Error: {e}")
            break

        finally:
            if agent:
                await agent.close()

Best Practices

1. Use native S2S: Avoid STT→LLM→TTS pipelines for latency 2. Enable VAD: Let server detect speech for natural turn-taking 3. Support barge-in: Allow users to interrupt at any time 4. Handle emotions: Use Gemini's affective dialog for empathy 5. Test latency: Measure TTFA with real users 6. Graceful degradation: Fall back to text if audio fails

Text-to-Speech Patterns (2026)

Implementing high-quality speech synthesis using Gemini TTS, OpenAI TTS, and ElevenLabs.

Provider Comparison (January 2026)

Gemini 2.5 TTS (Latest)

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

def gemini_tts(
    text: str,
    voice: str = "Kore",
    style: str = None
) -> bytes:
    """Generate speech with Gemini 2.5 TTS.

    Features:
    - 30 HD voices across 24 languages
    - Enhanced expressivity with style prompts
    - Precision pacing (context-aware speed)
    - Multi-speaker dialogue consistency

    Voices: Puck, Charon, Kore, Fenrir, Aoede, and 25 more
    """
    model = genai.GenerativeModel("gemini-2.5-flash-tts")

    # Optional style instruction
    content = text
    if style:
        content = f"[Style: {style}] {text}"

    response = model.generate_content(
        contents=content,
        generation_config=genai.GenerationConfig(
            response_mime_type="audio/mp3",
            speech_config=genai.SpeechConfig(
                voice_config=genai.VoiceConfig(
                    prebuilt_voice_config=genai.PrebuiltVoiceConfig(
                        voice_name=voice
                    )
                )
            )
        )
    )

    return response.audio

# Multi-speaker dialogue
def gemini_dialogue_tts(dialogue: list[dict]) -> bytes:
    """Generate multi-speaker dialogue with consistent voices.

    dialogue = [
        {"speaker": "Alice", "voice": "Kore", "text": "Hello!"},
        {"speaker": "Bob", "voice": "Charon", "text": "Hi there!"}
    ]
    """
    model = genai.GenerativeModel("gemini-2.5-flash-tts")

    formatted = "\n".join([
        f"[Voice: {d['voice']}] {d['speaker']}: {d['text']}"
        for d in dialogue
    ])

    response = model.generate_content(
        contents=formatted,
        generation_config=genai.GenerationConfig(
            response_mime_type="audio/mp3"
        )
    )

    return response.audio

# Gemini voice options
GEMINI_VOICES = {
    "Puck": "Playful, energetic",
    "Charon": "Deep, authoritative",
    "Kore": "Warm, friendly",
    "Fenrir": "Strong, confident",
    "Aoede": "Melodic, soothing"
}

OpenAI TTS

from openai import OpenAI

client = OpenAI()

def openai_tts(
    text: str,
    voice: str = "nova",
    model: str = "tts-1-hd"
) -> bytes:
    """Generate speech with OpenAI TTS.

    Voices: alloy, echo, fable, onyx, nova, shimmer
    Models: tts-1 (fast), tts-1-hd (quality)
    """
    response = client.audio.speech.create(
        model=model,
        voice=voice,
        input=text,
        response_format="mp3"
    )
    return response.content

# Streaming for immediate playback
async def stream_openai_tts(text: str, voice: str = "nova"):
    """Stream audio chunks for low-latency playback."""
    response = client.audio.speech.create(
        model="tts-1",  # Faster for streaming
        voice=voice,
        input=text
    )

    for chunk in response.iter_bytes(chunk_size=4096):
        yield chunk

# Voice characteristics
OPENAI_VOICES = {
    "alloy": "Neutral, balanced",
    "echo": "Male, warm",
    "fable": "Animated, storytelling",
    "onyx": "Male, deep (audiobooks)",
    "nova": "Female, warm (assistants)",
    "shimmer": "Female, clear (educational)"
}

ElevenLabs (Premium Quality)

from elevenlabs import generate, Voice, VoiceSettings, clone

def elevenlabs_tts(
    text: str,
    voice_id: str = "21m00Tcm4TlvDq8ikWAM",  # Rachel
    stability: float = 0.5,
    similarity: float = 0.8
) -> bytes:
    """High-quality TTS with ElevenLabs.

    Best voice quality, supports cloning.
    """
    audio = generate(
        text=text,
        voice=Voice(
            voice_id=voice_id,
            settings=VoiceSettings(
                stability=stability,
                similarity_boost=similarity,
                style=0.0,
                use_speaker_boost=True
            )
        ),
        model="eleven_turbo_v2_5"
    )
    return audio

# Clone a custom voice
def clone_voice(name: str, audio_samples: list[str]) -> str:
    """Create custom voice from 1-25 audio samples."""
    voice = clone(
        name=name,
        files=audio_samples,
        description=f"Cloned voice: {name}"
    )
    return voice.voice_id

Grok Voice (Real-Time TTS)

import websockets
import json
import base64

async def grok_tts(text: str, voice: str = "Aria") -> bytes:
    """TTS via Grok Voice Agent API.

    Supports expressive auditory cues:
    [whisper], [sigh], [laugh], [pause], [excited], [concerned]
    """
    uri = "wss://api.x.ai/v1/realtime"
    headers = {"Authorization": f"Bearer {XAI_API_KEY}"}

    async with websockets.connect(uri, extra_headers=headers) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "grok-4-voice",
                "voice": voice,  # Aria, Eve, Leo
                "modalities": ["audio"]
            }
        }))

        await ws.send(json.dumps({
            "type": "response.create",
            "response": {
                "modalities": ["audio"],
                "instructions": text
            }
        }))

        audio_chunks = []
        async for message in ws:
            data = json.loads(message)
            if data["type"] == "response.audio.delta":
                audio_chunks.append(base64.b64decode(data["delta"]))
            elif data["type"] == "response.done":
                break

        return b"".join(audio_chunks)

# Expressive example
text = "[excited] Great news! [pause] Your order has shipped. [whisper] It's a surprise."

Long-Form Audio Generation

import re
from io import BytesIO
from pydub import AudioSegment

def generate_audiobook(
    text: str,
    voice: str = "onyx",
    provider: str = "openai"
) -> bytes:
    """Generate audio for long text with chunking."""
    # Split at sentence boundaries
    sentences = re.split(r'(?<=[.!?])\s+', text)
    chunks = []
    current = ""

    for sentence in sentences:
        if len(current) + len(sentence) < 4000:
            current += sentence + " "
        else:
            chunks.append(current.strip())
            current = sentence + " "
    if current:
        chunks.append(current.strip())

    # Generate audio for each chunk
    segments = []
    for chunk in chunks:
        if provider == "gemini":
            audio = gemini_tts(chunk, voice)
        else:
            audio = openai_tts(chunk, voice)
        segments.append(AudioSegment.from_mp3(BytesIO(audio)))

    # Concatenate with pauses
    silence = AudioSegment.silent(duration=300)
    combined = segments[0]
    for seg in segments[1:]:
        combined += silence + seg

    output = BytesIO()
    combined.export(output, format="mp3")
    return output.getvalue()

Provider Selection

Best Practices

1. Match voice to content: Deep for audiobooks, warm for assistants 2. Use HD models: tts-1-hd for final output, tts-1 for drafts 3. Chunk long text: Stay under 4096 chars per request 4. Cache audio: Don't regenerate unchanged content 5. Style prompts: Use Gemini's style control for tone 6. Expressive cues: Use Grok's [whisper], [laugh] for emotion

Transcription Patterns (2026)

Comprehensive guide to audio transcription using Gemini 2.5 Pro, GPT-4o-Transcribe, and Whisper.

Model Comparison (January 2026)

Gemini 2.5 Pro Transcription (Best for Long Audio)

import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

def transcribe_with_gemini(audio_path: str) -> dict:
    """Transcribe up to 9.5 hours with speaker diarization.

    Gemini 2.5 Pro handles long-form audio natively without chunking.
    """
    model = genai.GenerativeModel("gemini-2.5-pro")

    # Upload audio file (supports wav, mp3, aac, etc.)
    audio_file = genai.upload_file(audio_path)

    response = model.generate_content([
        audio_file,
        """Transcribe this audio completely with:
        1. Speaker labels (Speaker 1, Speaker 2, etc.)
        2. Timestamps for each segment [HH:MM:SS]
        3. Proper punctuation and formatting
        4. Paragraph breaks for topic changes

        Format:
        [00:00:00] Speaker 1: First statement here...
        [00:00:15] Speaker 2: Response here..."""
    ])

    return {
        "transcript": response.text,
        "audio_duration": audio_file.duration,
        "model": "gemini-2.5-pro"
    }

# With structured output
def transcribe_structured(audio_path: str) -> dict:
    """Get structured JSON output."""
    model = genai.GenerativeModel("gemini-2.5-pro")
    audio_file = genai.upload_file(audio_path)

    response = model.generate_content([
        audio_file,
        """Transcribe this audio and return JSON:
        {
            "duration_seconds": <number>,
            "speakers": ["Speaker 1", "Speaker 2"],
            "segments": [
                {
                    "start": "00:00:00",
                    "end": "00:00:15",
                    "speaker": "Speaker 1",
                    "text": "..."
                }
            ],
            "summary": "Brief summary of the content"
        }"""
    ])

    import json
    return json.loads(response.text)

GPT-4o-Transcribe (Best Accuracy)

from openai import OpenAI

client = OpenAI()

def transcribe_openai(audio_path: str, language: str = None) -> dict:
    """Transcribe with GPT-4o-Transcribe for best accuracy.

    Replaces deprecated Whisper-1 API.
    """
    with open(audio_path, "rb") as audio_file:
        response = client.audio.transcriptions.create(
            model="gpt-4o-transcribe",
            file=audio_file,
            language=language,  # ISO 639-1: "en", "es", "ja"
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
        )

    return {
        "text": response.text,
        "words": response.words,
        "segments": response.segments,
        "duration": response.duration,
        "language": response.language
    }

# Generate SRT subtitles
def generate_srt(audio_path: str) -> str:
    """Generate SRT subtitle file."""
    with open(audio_path, "rb") as audio_file:
        response = client.audio.transcriptions.create(
            model="gpt-4o-transcribe",
            file=audio_file,
            response_format="srt"
        )
    return response

Whisper Self-Hosted (Cost-Effective)

import whisper
import torch

# Load model (GPU recommended)
model = whisper.load_model("large-v3")  # or "turbo" for 6x speed

def transcribe_local(audio_path: str, language: str = None) -> dict:
    """Transcribe with local Whisper model.

    Models: tiny, base, small, medium, large-v3, turbo
    """
    result = model.transcribe(
        audio_path,
        language=language,  # None for auto-detect
        task="transcribe",  # or "translate" for English
        word_timestamps=True,
        verbose=False
    )

    return {
        "text": result["text"],
        "segments": result["segments"],
        "language": result["language"]
    }

# Faster with Turbo
def transcribe_fast(audio_path: str) -> str:
    """6x faster with V3 Turbo, ~1% WER increase."""
    turbo_model = whisper.load_model("turbo")
    result = turbo_model.transcribe(audio_path)
    return result["text"]

Handling Long Audio (OpenAI)

from pydub import AudioSegment
import tempfile

def transcribe_long_audio_openai(
    audio_path: str,
    chunk_seconds: int = 600  # 10 minutes
) -> str:
    """Transcribe audio longer than 25MB limit."""
    audio = AudioSegment.from_file(audio_path)
    chunks = []

    chunk_ms = chunk_seconds * 1000
    for i in range(0, len(audio), chunk_ms):
        chunk = audio[i:i + chunk_ms]
        chunk_path = tempfile.mktemp(suffix=".wav")
        chunk.export(chunk_path, format="wav")
        chunks.append(chunk_path)

    # Transcribe with context chaining
    transcripts = []
    previous = ""

    for chunk_path in chunks:
        with open(chunk_path, "rb") as f:
            response = client.audio.transcriptions.create(
                model="gpt-4o-transcribe",
                file=f,
                prompt=previous[-224:]  # Context from previous
            )
        transcripts.append(response.text)
        previous = response.text

    return " ".join(transcripts)

Provider Selection Guide

def select_transcription_provider(
    audio_duration_hours: float,
    needs_diarization: bool,
    budget: str = "normal"
) -> str:
    """Select optimal transcription provider."""

    if audio_duration_hours > 1:
        return "gemini"  # 9.5hr limit, native diarization

    if needs_diarization and budget != "low":
        return "assemblyai"  # Best diarization features

    if budget == "low":
        return "whisper_local"  # Free, self-hosted

    return "openai"  # GPT-4o-Transcribe for accuracy

Best Practices

1. Long audio: Use Gemini 2.5 Pro (no chunking needed for 9.5hr) 2. Accuracy priority: GPT-4o-Transcribe with language hint 3. Cost-sensitive: Self-host Whisper Large V3 4. Speed priority: Whisper V3 Turbo (6x faster) 5. Features (sentiment, entities): AssemblyAI 6. Prompting: Include vocabulary for technical terms