# How the Realtime Voice Agent Works in AgentScope with TTS Integration

> Discover how AgentScope's realtime voice agent works with TTS integration. Learn how it streams audio using bidirectional queues for seamless speech-to-text and text-to-speech.

- Repository: [AgentScope-AI/agentscope](https://github.com/agentscope-ai/agentscope)
- Tags: how-to-guide
- Published: 2026-03-09

---

**The realtime voice agent in AgentScope uses the `RealtimeAgent` class to bridge WebSocket connections between a frontend and a realtime model, streaming audio through bidirectional queues that handle speech-to-text input and text-to-speech output via the `DashScopeRealtimeModel` integration.**

The **realtime voice agent** architecture in AgentScope enables low-latency voice conversations by combining a WebSocket-based frontend connection with streaming TTS (Text-to-Speech) capabilities. This implementation centers on the `RealtimeAgent` class in the agentscope-ai/agentscope repository, which orchestrates audio streaming between client microphones and realtime AI models like DashScope's `qwen3-omni-flash-realtime`.

## Architecture Overview of the Realtime Voice Agent

The realtime voice agent operates as a bidirectional streaming pipeline. At its core, the `RealtimeAgent` manages two primary data flows:

- **Inbound flow**: Client audio (microphone input) → `RealtimeAgent` → Realtime Model (STT processing)
- **Outbound flow**: Realtime Model (TTS generation) → `RealtimeAgent` → Frontend (speaker output)

This architecture relies on asyncio queues to handle concurrent streaming without blocking, enabling real-time conversation with sub-second latency.

## Initializing the RealtimeAgent Class

### Constructor and Internal Queues

The `RealtimeAgent` initialization establishes the core infrastructure for voice streaming. Located in [`src/agentscope/agent/_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_realtime_agent.py), the constructor creates two critical asyncio queues:

```python

# From _realtime_agent.py lines 65-92

def __init__(
    self,
    name: str,
    model: RealtimeModel,
    sys_prompt: str = "You are a helpful assistant.",
    toolkit: Optional[Toolkit] = None,
) -> None:
    self.name = name
    self.model = model
    self.sys_prompt = sys_prompt
    self.toolkit = toolkit
    
    # Internal queues for event handling

    self._incoming_queue = asyncio.Queue()  # From model to agent

    self._model_response_queue = asyncio.Queue()  # From agent to frontend

    
    self.id = str(uuid.uuid4())
    self._tasks = []

```

The `_incoming_queue` receives raw events from the realtime model, while `_model_response_queue` holds processed events ready for frontend transmission. This dual-queue design separates concerns between model communication and client delivery.

## Establishing the WebSocket Connection

### The start() Method and Async Tasks

The `start()` method in [`_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/_realtime_agent.py) (lines 102-124) activates the realtime voice agent by establishing the model connection and launching concurrent processing loops:

```python
async def start(self, outgoing_queue: asyncio.Queue) -> None:
    # Connect to the realtime model's WebSocket

    await self.model.connect()
    
    # Launch the forward loop (client → model)

    self._tasks.append(
        asyncio.create_task(self._forward_loop())
    )
    
    # Launch the response loop (model → client)

    self._tasks.append(
        asyncio.create_task(self._model_response_loop(outgoing_queue))
    )

```

This method creates two persistent asyncio tasks that run for the duration of the session. The `_forward_loop` handles incoming client audio, while `_model_response_loop` manages TTS output streaming.

## Processing Client Input and Audio Streaming

### Forwarding Events with _forward_loop()

The `_forward_loop` method processes client input from the frontend and forwards it to the realtime model. When the agent receives audio from the client, it packages the raw PCM data into an `AudioBlock` and transmits it via the model's `send()` method.

The implementation handles audio resampling when necessary. In [`_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/_realtime_agent.py), the code processes `ServerEvents` from the client:

```python

# Conceptual flow based on _realtime_agent.py implementation

async def _forward_loop(self):
    while True:
        event = await self._incoming_queue.get()
        
        if event.type == "client_audio":
            # Package into AudioBlock

            audio_block = AudioBlock(
                source=Base64Source(
                    data=event.audio_data,
                    media_type="audio/pcm"
                ),
                format={"type": "audio/pcm", "rate": 16000}
            )
            # Send to realtime model

            await self.model.send(audio_block)

```

This loop maintains the real-time stream from microphone to model, ensuring minimal latency for voice input processing.

## TTS Integration and Audio Generation

### Realtime Model Processing (DashScopeRealtimeModel)

The TTS integration occurs within the realtime model implementation. For DashScope integration, the `DashScopeRealtimeModel` class in [`src/agentscope/realtime/_dashscope_realtime_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/realtime/_dashscope_realtime_model.py) handles the WebSocket communication with the remote service.

The `send()` method (lines 31-88) formats client input into the DashScope protocol:

```python
async def send(self, block: Union[AudioBlock, TextBlock, ImageBlock]) -> None:
    if isinstance(block, AudioBlock):
        # Convert to DashScope audio format

        message = {
            "type": "input_audio",
            "audio": block.source.data,  # base64 encoded PCM

            "sample_rate": block.format["rate"]
        }
    elif isinstance(block, TextBlock):
        # Handle text input

        message = {"type": "input_text", "text": block.content}
    
    # Send via WebSocket

    await self.websocket.send(json.dumps(message))

```

### Parsing Audio Delta Events

When the remote model generates speech, it streams audio chunks as `response.audio.delta` events. The `parse_api_message` method in [`_dashscope_realtime_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/_dashscope_realtime_model.py) (lines 62-73) converts these into standardized `ModelResponseAudioDeltaEvent` objects:

```python
def parse_api_message(self, data: str) -> ModelEvents:
    msg = json.loads(data)
    
    if msg.get("type") == "response.audio.delta":
        return ModelEvents.ModelResponseAudioDeltaEvent(
            type="audio_delta",
            data=msg["delta"],  # base64 PCM data

            sample_rate=msg.get("sample_rate", 24000)
        )
    elif msg.get("type") == "response.audio.done":
        return ModelEvents.ModelResponseAudioDoneEvent(type="audio_done")

```

This parsing layer abstracts the vendor-specific protocol (DashScope) into generic events that the `RealtimeAgent` can process uniformly.

## Delivering Synthesized Speech to the Frontend

### Mapping Model Events to Server Events

The `_model_response_loop` in [`_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/_realtime_agent.py) bridges the model's output to the client-facing server. It translates internal `ModelEvents` into `ServerEvents` using the `from_model_event` factory method:

```python
async def _model_response_loop(self, outgoing_queue: asyncio.Queue):
    while True:
        # Get event from model

        model_event = await self._model_response_queue.get()
        
        # Convert to server event for frontend

        server_event = ServerEvents.from_model_event(model_event)
        
        # Queue for WebSocket transmission

        await outgoing_queue.put(server_event)

```

When the event is an `AgentResponseAudioDeltaEvent`, it contains the base64-encoded PCM audio chunk that the browser will decode and play. The `AgentResponseAudioDoneEvent` signals the end of the utterance.

## Complete Implementation Example

Here is a complete server implementation showing how to wire the realtime voice agent with TTS integration:

```python
from agentscope.agent import RealtimeAgent
from agentscope.realtime import DashScopeRealtimeModel
from fastapi import FastAPI, WebSocket
import asyncio
import os

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    # Initialize the realtime model with TTS capability

    model = DashScopeRealtimeModel(
        model_name="qwen3-omni-flash-realtime",
        api_key=os.getenv("DASHSCOPE_API_KEY"),
    )
    
    # Create the realtime voice agent

    agent = RealtimeAgent(
        name="VoiceAssistant",
        sys_prompt="You are a helpful voice assistant.",
        model=model,
    )
    
    # Queue for frontend-bound events

    outgoing = asyncio.Queue()
    
    # Start the agent (opens WebSocket to model, starts loops)

    await agent.start(outgoing)
    
    # Task: Forward agent events to frontend WebSocket

    async def send_to_frontend():
        while True:
            event = await outgoing.get()
            await websocket.send_json(event.model_dump())
    
    # Task: Receive from frontend and send to agent

    async def receive_from_frontend():
        while True:
            data = await websocket.receive_json()
            await agent.handle_input(data)
    
    # Run both directions concurrently

    await asyncio.gather(
        send_to_frontend(),
        receive_from_frontend()
    )

```

## Summary

- The **realtime voice agent** in AgentScope centers on the `RealtimeAgent` class in [`src/agentscope/agent/_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_realtime_agent.py), which manages bidirectional audio streaming through asyncio queues.
- **TTS integration** is handled by realtime model implementations like `DashScopeRealtimeModel` in [`src/agentscope/realtime/_dashscope_realtime_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/realtime/_dashscope_realtime_model.py), which stream base64-encoded PCM audio chunks via WebSocket.
- The agent uses two primary loops: `_forward_loop()` to send client audio to the model, and `_model_response_loop()` to receive synthesized speech and forward it to the frontend.
- Audio events flow through a standardized pipeline: `ModelResponseAudioDeltaEvent` → `ServerEvents.AgentResponseAudioDeltaEvent` → WebSocket → Browser playback.

## Frequently Asked Questions

### How does the RealtimeAgent handle audio format conversion?

The `RealtimeAgent` processes raw PCM audio through the `_forward_loop()` method in [`src/agentscope/agent/_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_realtime_agent.py). When receiving audio from the frontend, it packages the data into `AudioBlock` objects with base64 encoding. If resampling is required, the agent handles PCM delta adjustments before forwarding to the model's `send()` method. The DashScope realtime model expects specific sample rates (typically 16kHz for input), and the agent ensures compatibility before transmission.

### What is the difference between RealtimeAgent and standard TTS models in AgentScope?

The `RealtimeAgent` provides a unified, stateful conversation interface that combines both STT (Speech-to-Text) and TTS (Text-to-Speech) in a single WebSocket connection to a realtime model like `DashScopeRealtimeModel`. In contrast, standard TTS models in `src/agentscope/tts/` (such as [`_dashscope_realtime_tts_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/_dashscope_realtime_tts_model.py)) provide only text-to-speech conversion without the conversational state management or STT capabilities. The RealtimeAgent handles the full duplex communication, queue management, and event translation required for interactive voice applications.

### How does the TTS audio stream from the model to the browser?

The TTS audio stream flows through a three-stage pipeline. First, the `DashScopeRealtimeModel` receives `response.audio.delta` events from the remote API and parses them into `ModelResponseAudioDeltaEvent` objects containing base64-encoded PCM chunks. Second, the `RealtimeAgent._model_response_loop()` translates these into `ServerEvents.AgentResponseAudioDeltaEvent` objects and places them on the outgoing queue. Finally, the server (as shown in [`examples/agent/realtime_voice_agent/run_server.py`](https://github.com/agentscope-ai/agentscope/blob/main/examples/agent/realtime_voice_agent/run_server.py)) retrieves these events from the queue and transmits them via WebSocket to the browser, where JavaScript decodes the base64 PCM and plays it through the Web Audio API.

### Can I use a different realtime model provider instead of DashScope?

Yes, the AgentScope architecture supports pluggable realtime models through the abstract base class in [`src/agentscope/realtime/_base.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/realtime/_base.py). To integrate a different provider (such as OpenAI's Realtime API or Gemini), you would create a new class inheriting from the base realtime model class and implement the `connect()`, `send()`, and `parse_api_message()` methods. The `RealtimeAgent` in [`src/agentscope/agent/_realtime_agent.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_realtime_agent.py) interacts with these models through the standardized `ModelEvents` protocol, so switching providers requires no changes to the agent logic—only the model implementation needs to be swapped.