How to Build an STT-LLM-TTS Voice Pipeline (Python, 100 lines)

The Architecture

[microphone] --> Realtime STT --> Realtime Router --> Realtime TTS --> [speaker]
                          \                            /
                           \---- one WebSocket -------/

Implementation 1: Sync Three-Step Pipeline (50 lines)

import requests
import base64
import json

API = "https://api.inworld.ai"
AUTH = {"Authorization": "Basic <your-api-key>"}

def stt(audio_path: str) -> str:
    with open(audio_path, "rb") as f:
        audio_b64 = base64.b64encode(f.read()).decode()
    r = requests.post(
        f"{API}/stt/v1/transcribe",
        headers=AUTH,
        json={
            "transcribeConfig": {
                "modelId": "groq/whisper-large-v3",
                "audioEncoding": "AUTO_DETECT",
                "language": "en-US"
            },
            "audioData": {"content": audio_b64}
        }
    )
    return r.json()["transcription"]["transcript"]

def llm(prompt: str) -> str:
    r = requests.post(
        f"{API}/v1/chat/completions",
        headers=AUTH,
        json={
            "model": "gpt-5.5",
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    return r.json()["choices"][0]["message"]["content"]

def tts(text: str, out_path: str) -> None:
    r = requests.post(
        f"{API}/tts/v1/voice",
        headers=AUTH,
        json={
            "text": text,
            "voiceId": "Sarah",
            "modelId": "inworld-tts-1.5-max",
            "audioConfig": {
                "audioEncoding": "MP3",
                "sampleRateHertz": 24000
            }
        }
    )
    audio = base64.b64decode(r.json()["audioContent"])
    with open(out_path, "wb") as f:
        f.write(audio)

# Run the full pipeline
transcript = stt("user_audio.wav")
print(f"User: {transcript}")
reply = llm(transcript)
print(f"Assistant: {reply}")
tts(reply, "assistant_reply.mp3")

Implementation 2: Streaming Loop (~100 lines)

import requests
import base64
import json

API = "https://api.inworld.ai"
AUTH = {"Authorization": "Basic <your-api-key>"}

def stream_llm(prompt: str):
    """Yield tokens from the Realtime Router as they arrive."""
    r = requests.post(
        f"{API}/v1/chat/completions",
        headers=AUTH,
        json={
            "model": "gpt-5.5",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True
        },
        stream=True
    )
    for line in r.iter_lines():
        if not line:
            continue
        if line.startswith(b"data: "):
            data = line[6:]
            if data == b"[DONE]":
                break
            chunk = json.loads(data)
            delta = chunk["choices"][0]["delta"].get("content", "")
            if delta:
                yield delta

def stream_tts(text: str):
    """Yield PCM audio chunks from the Realtime TTS streaming endpoint."""
    with requests.post(
        f"{API}/tts/v1/voice:stream",
        headers=AUTH,
        json={
            "text": text,
            "voiceId": "Sarah",
            "modelId": "inworld-tts-1.5-mini",
            "audioConfig": {
                "audioEncoding": "PCM",
                "sampleRateHertz": 24000
            }
        },
        stream=True
    ) as r:
        for line in r.iter_lines():
            if not line:
                continue
            audio = base64.b64decode(
                json.loads(line)["result"]["audioContent"]
            )
            yield audio

# In a real loop, you would buffer LLM tokens by sentence boundary
# and start TTS as soon as the first sentence is ready.
def voice_reply(user_text: str, audio_callback):
    sentence_buffer = ""
    for token in stream_llm(user_text):
        sentence_buffer += token
        # Naive sentence split; replace with proper segmenter in production.
        if any(p in token for p in ".!?"):
            for chunk in stream_tts(sentence_buffer.strip()):
                audio_callback(chunk)
            sentence_buffer = ""
    if sentence_buffer.strip():
        for chunk in stream_tts(sentence_buffer.strip()):
            audio_callback(chunk)

# audio_callback writes PCM bytes to your audio output
# (PortAudio, sounddevice, WebSocket to browser, etc.)

Implementation 3: Production-Ready Realtime API (~80 lines)

import asyncio
import json
import base64
import websockets

URL = (
    "wss://api.inworld.ai/api/v1/realtime/session"
    "?key=<session-id>&protocol=realtime"
)
AUTH_HEADERS = {"Authorization": "Basic <your-api-key>"}

async def voice_session(mic_stream, speaker_callback):
    async with websockets.connect(URL, extra_headers=AUTH_HEADERS) as ws:
        # Configure the session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "gpt-5.5",
                "instructions": "You are a helpful voice assistant. Be concise.",
                "output_modalities": ["audio", "text"],
                "audio": {
                    "input": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "turn_detection": {
                            "type": "semantic_vad",
                            "eagerness": "auto"
                        }
                    },
                    "output": {
                        "voice": "Sarah",
                        "model": "inworld-tts-1.5-mini",
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "speed": 1.0
                    }
                }
            }
        }))

        async def send_mic():
            async for chunk in mic_stream:
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(chunk).decode()
                }))

        async def receive_audio():
            async for raw in ws:
                event = json.loads(raw)
                if event["type"] == "response.audio.delta":
                    audio_bytes = base64.b64decode(event["delta"])
                    speaker_callback(audio_bytes)
                elif event["type"] == "response.done":
                    pass  # turn complete; ready for next input

        await asyncio.gather(send_mic(), receive_audio())

# mic_stream is an async generator yielding PCM16 24kHz chunks (e.g., 100ms each).
# speaker_callback writes PCM16 bytes to your audio output.

What the Realtime API Adds Over the Sync Pipeline

• Voice activity detection. The server detects when the user finishes speaking using semantic VAD, not just silence. No client-side push-to-talk required.
• Interruption handling. When the user starts talking while the assistant is speaking, the Realtime API stops audio output and starts processing the new input within milliseconds.
• Pipelined latency. STT, LLM, and TTS overlap. The first audible word reaches the user within ~200ms of the user finishing.
• Voice-aware routing. STT acoustic signals (emotion, hesitation, speaker profile) are passed to the Router for model selection.
• Tool calling inside the audio loop. Function calls happen without dropping the audio stream.

FAQ

What is the simplest way to build a voice agent in Python?

How do I get sub-1-second response time?

What audio format does the Realtime API expect?

Can I run the LLM on a different provider?

Do I need to handle interruptions and turn-taking myself?