Speech to Speech vs Cascade Voice Agent Architecture: Complete Technical & Benchmark Guide (2026)

+-----------------------------------------------------------------------------------+
|                                 QUICK SUMMARY                                     |
+-----------------------------------------------------------------------------------+
| Feature               | Speech to Speech (S2S)       | Modular Cascade            |
+-----------------------+------------------------------+----------------------------+
| Average Latency       | 280ms to 450ms               | 500ms to 850ms             |
| Emotional Expressiveness| Exceptional (Native Audio) | Moderate (SSML Dependent)  |
| Tool Calling Reliability| Moderate (Prompt Dependent)  | High (Structured JSON Schema)|
| Cost per 1,000 Minutes| $35 to $70                   | $12 to $28                 |
| Modular Flexibility   | Low (Single Vendor Lock-in)  | High (Swap STT/LLM/TTS)    |
+-----------------------+------------------------------+----------------------------+

[User Mic] ---> (WebRTC Audio Stream) ---> [Unified S2S Neural Net] ---> (WebRTC Audio Stream) ---> [User Speaker]
                                                  |
                                                  +---> [External Tool Function (Async)]

[User Mic] ---> [Streaming STT] ---> [Text Transcript] ---> [Orchestrator LLM] ---> [Text Tokens] ---> [Streaming TTS] ---> [User Speaker]
                                                                  |
                                                                  +---> [Structured JSON Tool Call]

pip install asyncio websockets python-dotenv

import asyncio
import json
import time
from typing import AsyncGenerator

class VoiceAgentPipeline:
    def __init__(self, stt_api_key: str, llm_api_key: str, tts_api_key: str):
        self.stt_api_key = stt_api_key
        self.llm_api_key = llm_api_key
        self.tts_api_key = tts_api_key
        self.is_interrupted = False
        self.audio_queue = asyncio.Queue()

    async def process_speech_to_text(self, audio_chunk_stream: AsyncGenerator[bytes, None]) -> str:
        """
        Step 1: Stream raw PCM audio bytes to Speech-to-Text engine.
        Simulates WebSocket streaming transcription.
        """
        print("[STT] Streaming audio chunks to STT engine...")
        start_time = time.time()

        # Simulating audio processing time for incoming stream
        await asyncio.sleep(0.120)
        transcript = "Can you schedule a demo for tomorrow at 2 PM?"

        stt_duration = round((time.time() - start_time) * 1000, 2)
        print(f"[STT Completed] Transcript: '{transcript}' in {stt_duration}ms")
        return transcript

    async def process_language_model(self, transcript: str) -> AsyncGenerator[str, None]:
        """
        Step 2: Stream text transcript to LLM and yield token chunks.
        Supports parallel tool calling execution.
        """
        print(f"[LLM] Processing transcript: '{transcript}'")
        start_time = time.time()

        # Simulating first token response
        await asyncio.sleep(0.180)
        first_token_time = round((time.time() - start_time) * 1000, 2)
        print(f"[LLM First Token] TTFB: {first_token_time}ms")

        response_chunks = [
            "I would be happy to help with that. ",
            "Let me check the availability ",
            "for tomorrow at 2 PM right now."
        ]

        for chunk in response_chunks:
            if self.is_interrupted:
                print("[LLM] Generation halted due to user interruption.")
                break
            yield chunk
            await asyncio.sleep(0.05)

    async def process_text_to_speech(self, text_stream: AsyncGenerator[str, None]):
        """
        Step 3: Convert text tokens into streaming audio frames.
        """
        print("[TTS] Initializing text-to-speech synthesis stream...")
        start_tts_time = time.time()
        first_frame = True

        async for text_chunk in text_stream:
            if self.is_interrupted:
                print("[TTS] Clearing audio playback buffer immediately.")
                break

            # Synthesize audio frame
            await asyncio.sleep(0.090)
            if first_frame:
                ttfa = round((time.time() - start_tts_time) * 1000, 2)
                print(f"[TTS First Audio Frame] Time to First Audio: {ttfa}ms")
                first_frame = False

            audio_frame = b"\x00\x00\xff\xff" * 100
            await self.audio_queue.put(audio_frame)
            print(f"[TTS Output] Synthesized audio frame for chunk: '{text_chunk.strip()}'")

    async def trigger_user_interruption(self):
        """
        Interruption Handler: Instantly flushes queues when user barge-in occurs.
        """
        print("[Interruption Event] User started speaking! Flushing buffers...")
        self.is_interrupted = True
        while not self.audio_queue.empty():
            try:
                self.audio_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
        print("[Interruption Event] Audio playback queue completely cleared.")

    async def run_pipeline(self, mock_audio_stream):
        total_start = time.time()

        # 1. Transcribe Audio
        transcript = await self.process_speech_to_text(mock_audio_stream)

        # 2. Generate LLM Stream
        llm_stream = self.process_language_model(transcript)

        # 3. Synthesize Speech Stream
        await self.process_text_to_speech(llm_stream)

        total_duration = round((time.time() - total_start) * 1000, 2)
        print(f"[Pipeline Success] Total roundtrip completed in {total_duration}ms")


async def main():
    agent = VoiceAgentPipeline(stt_api_key="demo", llm_api_key="demo", tts_api_key="demo")

    async def mock_audio():
        for _ in range(5):
            yield b"\x01\x02\x03\x04"
            await asyncio.sleep(0.02)

    print("--- Starting Production Modular Cascade Test ---")
    await agent.run_pipeline(mock_audio())

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