How the Realtime Voice Agent Works in AgentScope with TTS Integration

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, the constructor creates two critical asyncio queues:

# 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 (lines 102-124) activates the realtime voice agent by establishing the model connection and launching concurrent processing loops:

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, the code processes ServerEvents from the client:

# 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 handles the WebSocket communication with the remote service.

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

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 (lines 62-73) converts these into standardized ModelResponseAudioDeltaEvent objects:

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 bridges the model's output to the client-facing server. It translates internal ModelEvents into ServerEvents using the from_model_event factory method:

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:

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, 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, 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. 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) 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) 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. 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}