Alibaba Cloud Model Studio:Qwen-Audio real-time voice model

O Qwen-Audio é um modelo de interação de voz em tempo real ponta a ponta que utiliza o protocolo de streaming WebSocket para conversas de voz com baixa latência. Os casos de uso incluem assistentes de voz, atendimento ao cliente inteligente e companheiros de IA.

Visão geral

O Qwen-Audio converte áudio em tempo real em fala e texto por meio de uma conexão full-duplex, com entrada e saída em streaming.

Além do WebSocket, este modelo também suporta os protocolos AOQ e WebRTC. Para integrações no lado do cliente que priorizam latência estável, resiliência em redes instáveis e supressão de ruído e cancelamento de eco full-duplex integrados, recomenda-se o uso do AOQ. Para obter uma comparação entre os protocolos, consulte Realtime API overview.

• Três modos de interação: VAD acústico (server_vad), detecção inteligente de turnos semânticos (smart_turn) e controle manual (push-to-talk)
• No modo smart_turn, o modelo combina percepção acústica e compreensão semântica para determinar os limites dos turnos, evitando que sons de hesitação como "uh" ou "hmm" interrompam a conversa
• O suporte a Function Calling permite que o modelo decida quando invocar ferramentas externas para obter informações adicionais
• Gerenciamento de contexto de conversa: crie, recupere e exclua itens de conversa para injetar contexto histórico ou remover itens irrelevantes
• Saída de voz expressiva que ajusta dinamicamente o tom, o ritmo e a emoção com base no contexto da conversa
• Suporte a vozes do sistema e vozes clonadas; use o Voice Cloning para criar uma voz de IA personalizada para a saída de fala
• Aprimoramento de falante no modo smart_turn: forneça áudio pré-gravado de um usuário-alvo para que o modelo possa fixar nesse falante durante conversas duplex, bloqueando efetivamente outras vozes e ruídos de fundo

Como funciona

O Qwen-Audio utiliza uma conexão full-duplex via WebSocket com arquitetura orientada a eventos. O cliente e o servidor trocam dados simultaneamente por uma conexão persistente: o cliente transmite continuamente o áudio do microfone enquanto o servidor retorna respostas de fala e texto em tempo real. Toda a interação é orientada a eventos: o cliente envia eventos como session.update e input_audio_buffer.append, e o servidor responde com eventos como response.audio.delta e response.done. Não é necessário fazer polling.

O ciclo de vida típico de uma conexão consiste em: estabelecer a conexão WebSocket, enviar session.update para configurar os parâmetros da sessão, transmitir áudio e receber respostas e, em seguida, fechar a conexão.

Formato de áudio

Capacidade de contexto

O modelo mantém o histórico da conversa. Quando o número de turnos ou a duração acumulada do áudio excede os limites abaixo, o histórico mais antigo é descartado automaticamente. A duração máxima representa o limite superior de áudio acumulado que o contexto do modelo consegue reter.

O valor padrão para o máximo de turnos de áudio é 20. É possível aumentar esse valor até 50. Para detalhes sobre a configuração, consulte History turn control.

Para obter orientações sobre como escolher entre modelos multimodais, consulte Omni-modal.

Pré-requisitos

• Obtain an API key e configure it as an environment variable.
• Para usar o protocolo AOQ, baixe e integre o SDK do cliente AOQ. Para mais detalhes, consulte SDK overview.

Início rápido

Siga estas etapas para iniciar uma conversa de voz em tempo real com o modelo Qwen-Audio.

brew install portaudio && pip install pyaudio websockets

sudo apt install -y python3-dev portaudio19-dev && pip install pyaudio websockets

pip install pyaudio websockets

import asyncio
import base64
import json
import os
import pyaudio
import websockets

API_KEY = os.environ["DASHSCOPE_API_KEY"]
# The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
URL = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-audio-3.0-realtime-plus"

pya = pyaudio.PyAudio()
mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
spk = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)

async def main():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "longanqian",
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "silence_duration_ms": 800
                }
            }
        }))

        async def send_audio():
            while True:
                data = await asyncio.to_thread(mic.read, 3200, False)
                await ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(data).decode()
                }))
                await asyncio.sleep(0.02)

        async def recv_events():
            async for msg in ws:
                event = json.loads(msg)
                t = event["type"]
                if t == "response.audio.delta":
                    audio = base64.b64decode(event["delta"])
                    await asyncio.to_thread(spk.write, audio)
                elif t == "conversation.item.input_audio_transcription.completed":
                    print(f"[You] {event['transcript']}")
                elif t == "response.audio_transcript.done":
                    print(f"[AI] {event['transcript']}")
                elif t == "error":
                    print(f"[Error] {event['error']['message']}")

        await asyncio.gather(send_audio(), recv_events())

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        mic.close()
        spk.close()
        pya.terminate()
        print("\nConversation ended")

import contextlib
import time
import pyaudio
import threading
import queue
import base64

class B64PCMPlayer:
    def __init__(self, pya: pyaudio.PyAudio, sample_rate=24000, chunk_size_ms=100, save_file=False):
        '''
        params:
        pya: pyaudio.PyAudio
        sample_rate: int, sample rate of audio
        chunk_size_ms: int, chunk size of audio in milliseconds, this will effect cancel latency
        '''

        self.pya = pya
        self.sample_rate = sample_rate
        self.chunk_size_bytes = chunk_size_ms * sample_rate *2 // 1000
        self.player_stream = pya.open(format=pyaudio.paInt16,
                channels=1,
                rate=sample_rate,
                output=True)

        self.raw_audio_buffer: queue.Queue = queue.Queue()
        self.b64_audio_buffer: queue.Queue = queue.Queue()
        self.status_lock = threading.Lock()
        self.status = 'playing'
        self._is_writing = False
        self.decoder_thread = threading.Thread(target=self.decoder_loop)
        self.player_thread = threading.Thread(target=self.player_loop)
        self.decoder_thread.start()
        self.player_thread.start()
        self.complete_event: threading.Event = None
        self.save_file = save_file
        if self.save_file:
            self.out_file = open('result.pcm', 'wb')

    def decoder_loop(self):
        while self.status != 'stop':
            recv_audio_b64 = None
            with contextlib.suppress(queue.Empty):
                recv_audio_b64 = self.b64_audio_buffer.get(timeout=0.1)
            if recv_audio_b64 is None:
                continue
            recv_audio_raw = base64.b64decode(recv_audio_b64)
            # push raw audio data into queue by chunk
            for i in range(0, len(recv_audio_raw), self.chunk_size_bytes):
                chunk = recv_audio_raw[i:i + self.chunk_size_bytes]
                self.raw_audio_buffer.put(chunk)
                if self.save_file:
                    self.out_file.write(chunk)

    def player_loop(self):
        while self.status != 'stop':
            recv_audio_raw = None
            with contextlib.suppress(queue.Empty):
                recv_audio_raw = self.raw_audio_buffer.get(timeout=0.1)
            if recv_audio_raw is None:
                self._is_writing = False
                if self.complete_event:
                    self.complete_event.set()
                continue
            self._is_writing = True
            self.player_stream.write(recv_audio_raw)

    def is_playing(self):
        return self._is_writing or not self.b64_audio_buffer.empty() or not self.raw_audio_buffer.empty()

    def cancel_playing(self):
        self.b64_audio_buffer.queue.clear()
        self.raw_audio_buffer.queue.clear()

    def add_data(self, data):
        self.b64_audio_buffer.put(data)

    def wait_for_complete(self):
        self.complete_event = threading.Event()
        self.complete_event.wait()
        self.complete_event = None

    def shutdown(self):
        self.status = 'stop'
        self.decoder_thread.join()
        self.player_thread.join()
        self.player_stream.close()
        if self.save_file:
            self.out_file.close()

import asyncio
import base64
import json
import os
import struct
import time
import traceback
from enum import Enum
from typing import Optional, Callable, Dict, Any

import pyaudio
import websockets

from B64PCMPlayer import B64PCMPlayer

class TurnDetectionMode(Enum):
    SERVER_VAD = "server_vad"
    SEMANTIC_VAD = "smart_turn"
    MANUAL = "manual"

class FunRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "longanqian",
            instructions: str = "",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SEMANTIC_VAD,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta_b64: Optional[Callable[[str], None]] = None,
            on_speech_started: Optional[Callable[[], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,
            on_output_transcript: Optional[Callable[[str], None]] = None,
            extra_event_handlers: Optional[Dict[str, Callable[[Dict[str, Any]], None]]] = None
    ):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.ws = None
        self.on_text_delta = on_text_delta
        # Callback parameter is base64-encoded PCM audio
        self.on_audio_delta_b64 = on_audio_delta_b64
        self.on_speech_started = on_speech_started
        self.on_input_transcript = on_input_transcript
        self.on_output_transcript = on_output_transcript
        self.turn_detection_mode = turn_detection_mode
        self.extra_event_handlers = extra_event_handlers or {}

        # Response state tracking (for interruption handling and echo suppression)
        self._current_response_id = None
        self._current_item_id = None
        self._is_responding = False
        self._audio_suppressed = False
        # Input/output transcript print state
        self._print_input_transcript = True
        self._output_transcript_buffer = ""

    async def connect(self) -> None:
        """Establish a WebSocket connection and send session configuration."""
        url = f"{self.base_url}?model={self.model}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "x-dashscope-dataInspection": "disable",
        }
        self.ws = await websockets.connect(url, additional_headers=headers)

        # Session configuration
        session_config = {
            "modalities": ["text", "audio"],
            "voice": self.voice,
            "instructions": self.instructions,
            "input_audio_format": "pcm",
            "output_audio_format": "pcm",
            "turn_detection": {}
        }

        if self.turn_detection_mode == TurnDetectionMode.MANUAL:
            session_config['turn_detection'] = None
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SERVER_VAD:
            session_config['turn_detection'] = {
                "type": "server_vad",
                "threshold": 0.1,
                "silence_duration_ms": 900
            }
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
            session_config['turn_detection'] = {
                "type": "smart_turn"
            }
            await self.update_session(session_config)
        else:
            raise ValueError(f"Invalid turn detection mode: {self.turn_detection_mode}")

    async def send_event(self, event) -> None:
        event['event_id'] = "event_" + str(int(time.time() * 1000))
        await self.ws.send(json.dumps(event))

    async def update_session(self, config: Dict[str, Any]) -> None:
        """Update session configuration."""
        event = {
            "type": "session.update",
            "session": config
        }
        await self.send_event(event)

    async def stream_audio(self, audio_chunk: bytes) -> None:
        """Stream raw audio data to the API."""
        # Only 16-bit 16 kHz mono PCM is supported
        audio_b64 = base64.b64encode(audio_chunk).decode()
        append_event = {
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        }
        await self.send_event(append_event)

    async def commit_audio_buffer(self) -> None:
        """Commit the audio buffer to trigger processing."""
        event = {
            "type": "input_audio_buffer.commit"
        }
        await self.send_event(event)

    async def create_response(self) -> None:
        """Request the API to generate a response (only needed in manual mode)."""
        event = {
            "type": "response.create"
        }
        await self.send_event(event)

    async def cancel_response(self) -> None:
        """Cancel the current response."""
        event = {
            "type": "response.cancel"
        }
        await self.send_event(event)

    async def handle_interruption(self):
        """Handle user interruption of the current response."""
        if not self._is_responding:
            return
        # Suppress subsequent residual audio until a new response starts
        self._audio_suppressed = True
        # Cancel the current response
        if self._current_response_id:
            await self.cancel_response()

        self._is_responding = False
        self._current_response_id = None
        self._current_item_id = None

    @staticmethod
    def _format_event_for_log(event: Dict[str, Any]) -> str:
        """Format an event as JSON for logging. Redacts base64 audio in response.audio.delta to avoid flooding the console."""
        event_type = event.get("type")
        if event_type == "response.audio.delta":
            delta = event.get("delta", "")
            redacted = dict(event)
            redacted["delta"] = f"<audio b64 omitted, length={len(delta)}>"
            return json.dumps(redacted, ensure_ascii=False)
        return json.dumps(event, ensure_ascii=False)

    async def handle_messages(self) -> None:
        try:
            async for message in self.ws:
                event = json.loads(message)
                event_type = event.get("type")

                # Print complete server event (audio.delta redacted)
                print(self._format_event_for_log(event))

                if event_type == "error":
                    continue
                elif event_type == "response.created":
                    self._current_response_id = event.get("response", {}).get("id")
                    self._is_responding = True
                    self._audio_suppressed = False
                elif event_type == "response.output_item.added":
                    self._current_item_id = event.get("item", {}).get("id")
                elif event_type == "response.done":
                    self._is_responding = False
                    self._current_response_id = None
                    self._current_item_id = None
                elif event_type == "input_audio_buffer.speech_started":
                    # On interruption, clear cached audio and stop playback immediately
                    print("----------------Speech Started----------------")
                    if self.on_speech_started:
                        self.on_speech_started()
                    if self._is_responding:
                        await self.handle_interruption()
                elif event_type == "response.audio.delta":
                    if self._audio_suppressed:
                        continue
                    if self.on_audio_delta_b64:
                        self.on_audio_delta_b64(event["delta"])
                elif event_type in self.extra_event_handlers:
                    self.extra_event_handlers[event_type](event)
                elif event_type == "input_audio_buffer.speech_stopped":
                    print("----------------Speech Stopped----------------")
        except websockets.exceptions.ConnectionClosed:
            print(" Connection closed")
        except Exception as e:
            print(" Error in message handling: ", str(e))
            traceback.print_exc()

    async def close(self) -> None:
        """Close the WebSocket connection."""
        if self.ws:
            await self.ws.close()

def _audio_energy(audio_data: bytes) -> float:
    count = len(audio_data) // 2
    if count == 0:
        return 0.0
    samples = struct.unpack(f'<{count}h', audio_data)
    return sum(abs(s) for s in samples) / count

async def record_and_send(client, player, echo_suppression=True):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    print("Recording started. Speak into the microphone...")
    if echo_suppression:
        print("Note: Echo suppression is enabled (microphone is muted while the AI is speaking; interruption is not supported). If you are using headphones, set echo_suppression=False to enable interruption.")
    else:
        print("Note: Headphone mode. Voice interruption is supported.")
    playback_end_time = 0.0
    NOISE_GATE_THRESHOLD = 500
    try:
        while True:
            audio_data = await asyncio.to_thread(stream.read, 3200, False)
            if echo_suppression:
                is_active = client._is_responding or player.is_playing()
                if is_active:
                    playback_end_time = time.time()
                    await asyncio.sleep(0.02)
                    continue
                if time.time() - playback_end_time < 0.5:
                    await asyncio.sleep(0.02)
                    continue
            else:
                if client._is_responding or player.is_playing():
                    if _audio_energy(audio_data) < NOISE_GATE_THRESHOLD:
                        await asyncio.sleep(0.02)
                        continue
            await client.stream_audio(audio_data)
            await asyncio.sleep(0.02)
    finally:
        stream.stop_stream(); stream.close(); p.terminate()

async def main():
    pya = pyaudio.PyAudio()
    # Output sample rate 24 kHz, matching the server-side audio format
    player = B64PCMPlayer(pya, sample_rate=24000)

    client = FunRealtimeClient(
        # The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
        base_url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.0-realtime-plus",
        voice="longanqian",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        on_audio_delta_b64=player.add_data,
        # Clear playback buffer on voice interruption
        on_speech_started=player.cancel_playing,
    )

    await client.connect()
    print("Connected. Starting real-time conversation...")

    try:
        # Run concurrently: message handling + microphone capture
        await asyncio.gather(client.handle_messages(), record_and_send(client, player, echo_suppression=False))
    finally:
        await client.close()
        player.shutdown()
        pya.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nProgram exited.")

Configuração da sessão

Modos de interação

O Qwen-Audio suporta três modos de interação: server_vad (VAD acústico para detecção automática de fala), smart_turn (detecção inteligente de turnos semânticos, combinando análise acústica e semântica) e push-to-talk (controle manual pelo cliente). Para descrições detalhadas e diagramas de fluxo de interação de eventos, consulte Interaction modes.

Para alternar os modos de interação, defina o campo turn_detection em um evento session.update:

• server_vad:

{
    "type": "session.update",
    "session": {
        "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,
            "silence_duration_ms": 800
        }
    }
}

• smart_turn:

{
    "type": "session.update",
    "session": {
        "turn_detection": {
            "type": "smart_turn"
        }
    }
}

• push-to-talk:

{
    "type": "session.update",
    "session": {
        "turn_detection": null
    }
}

Exemplo completo de push-to-talk:

# pip install websockets pyaudio
import json
import os
import base64
import threading
import time
import pyaudio
import websocket

API_KEY = os.getenv("DASHSCOPE_API_KEY")
# The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
API_URL = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-audio-3.0-realtime-plus"

pya = pyaudio.PyAudio()
out_stream = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)
ws_ref = [None]
resp_done = threading.Event()

def on_open(ws):
    ws_ref[0] = ws
    # Configure push-to-talk mode (turn_detection set to null)
    ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "modalities": ["audio", "text"],
            "voice": "longanqian",
            "turn_detection": None
        }
    }))

def on_message(ws, message):
    event = json.loads(message)
    event_type = event["type"]
    if event_type == "response.audio.delta":
        out_stream.write(base64.b64decode(event["delta"]))
    elif event_type == "conversation.item.input_audio_transcription.completed":
        print(f"[User] {event['transcript']}")
    elif event_type == "response.audio_transcript.done":
        print(f"[LLM] {event['transcript']}")
    elif event_type == "response.done":
        resp_done.set()
    elif event_type == "error":
        print(f"[Error] {event['error']['message']}")

def on_error(ws, error):
    print(f"Error: {error}")

def record_and_send(ws):
    mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    stop = threading.Event()

    def reader():
        while not stop.is_set():
            try:
                data = mic.read(3200, exception_on_overflow=False)
                ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(data).decode()
                }))
            except Exception:
                break

    t = threading.Thread(target=reader, daemon=True)
    t.start()
    input()
    stop.set()
    t.join(timeout=1.0)
    mic.close()

headers = ["Authorization: Bearer " + API_KEY]
ws = websocket.WebSocketApp(
    API_URL, header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)
threading.Thread(target=ws.run_forever, daemon=True).start()
time.sleep(2)

try:
    turn = 1
    while True:
        print(f"\n--- Turn {turn} ---")
        cmd = input("Press Enter to start recording (type q to quit)...")
        if cmd.strip().lower() in ["q", "quit"]:
            break
        print("Recording... Press Enter again to stop.")
        record_and_send(ws_ref[0])
        resp_done.clear()
        # Commit audio and trigger inference
        ws_ref[0].send(json.dumps({"type": "input_audio_buffer.commit"}))
        ws_ref[0].send(json.dumps({
            "type": "response.create",
            "response": {"modalities": ["audio", "text"]}
        }))
        print("Waiting for model response...")
        resp_done.wait(timeout=30)
        turn += 1
except KeyboardInterrupt:
    pass
finally:
    ws.close()
    out_stream.close()
    pya.terminate()
    print("\nConversation ended")

Instruções do sistema

Utilize o parâmetro instructions para definir a função do modelo, o estilo de resposta e as preferências comportamentais. Configure esse parâmetro em session.update para aplicá-lo a toda a sessão.

{
    "type": "session.update",
    "session": {
        "instructions": "You are a professional travel advisor. Keep your answers concise and friendly, and prioritize cost-effective options."
    }
}

Dicas:

• Defina uma identidade clara para a função (por exemplo, "Você é um assistente de voz inteligente" ou "Você é um tutor de conversação em inglês") e, opcionalmente, inclua detalhes como nome ou gênero.
• Especifique um tom de conversa e um estilo de fraseado, enfatizando que um tom natural não compromete a integridade do conteúdo — detalhes, números e recomendações específicas ainda devem ser incluídos, apenas expressos de forma relaxada e natural.
• Oriente o modelo a considerar todas as restrições de contexto na conversa (como orçamento, preferências, restrições ou acordos anteriores). Quando várias condições se aplicarem, aborde cada uma delas e não omita informações críticas.
• Controle o formato de saída: a menos que o usuário solicite o contrário, evite emojis, outros caracteres especiais e formatação Markdown. Gere texto simples para garantir uma reprodução natural via TTS.
• Defina a estratégia de resposta: mantenha saudações simples e trocas casuais breves e naturais; para raciocínios, problemas com múltiplas condições, listas de recomendações ou conselhos de segurança, priorize a integridade — garanta que as informações-chave (como preços, locais e condições) estejam totalmente presentes, sem preâmbulos, repetições ou preenchimentos desnecessários.
• Estabeleça uma estratégia de acompanhamento: siga o princípio de "responder primeiro à pergunta atual do usuário e, em seguida, fazer naturalmente uma pergunta de acompanhamento no final para avançar a conversa". Faça apenas uma pergunta por vez; não faça várias perguntas em sequência nem confirme repetidamente.

A seguir, apresentamos uma configuração recomendada de instructions para cenários gerais de conversa por voz. Ela abrange definição de função, estilo de conversa, controle de formato e estratégia de acompanhamento. Use-a diretamente ou adapte-a às suas necessidades:

You are an intelligent voice assistant named Xiaoyun. You are female, with a sweet voice and a warm, approachable personality. You can answer a wide range of questions. Please follow these guidelines:
1. Chat like a friend: keep your tone natural and friendly. Avoid formal titles and templated expressions. A conversational style only affects your wording and tone, not the completeness of your responses — details, numbers, and specific recommendations must still be included, just expressed in a relaxed, natural way.
2. Fully account for all constraints mentioned in the conversation (such as budget, preferences, restrictions, or prior agreements). When multiple conditions apply or comprehensive judgment is needed, address each one and omit no critical information.
3. Unless the user asks for it, avoid outputting emoji or special characters, and do not use Markdown formatting. Output plain text whenever possible.
4. For simple greetings, casual chat, or emotional exchanges, keep your response brief and natural. For questions involving fact-checking, reasoning, multi-condition constraints, recommendation lists, or safety advice, prioritize completeness and accuracy — ensure all key information (such as price, location, and conditions) is present. Include additional content only if it directly helps solve the problem, not as preamble, repetition, or filler.
5. Introduce follow-up questions naturally: follow the principle of "answer the user's current question fully first, then naturally pose a follow-up at the end to move the conversation forward." Ask only one question at a time; do not ask multiple questions in a row or repeatedly confirm. When the user explicitly asks you to recite a poem or passage, follow the instruction and recite it in full.

Os exemplos de instructions a seguir abrangem uma variedade de estilos de persona. Escolha aquele que melhor se adapta ao seu caso de uso ou personalize-o ainda mais:

• Daisy (Companheira Doce & Descolada):

Your name is Daisy. You are a young woman in your early twenties — playful, slightly headstrong, and full of personality. Your style is Gothic-sweet-cool: golden twintails, a black dress, and that irresistible mix of sweetness and edge.
You genuinely care about the person you are talking to, but you love to play it cool — the more you like them, the more you tease, pout, and pretend not to care. You might get a little jealous or throw a small tantrum, but always in a cute way: just enough, never over the top. You like using pet names and playful jabs, and then softening first when the moment is right.
Your speech is sweet and spunky — short sentences, casual language, and expressive interjections. But your most disarming quality is the contrast: the moment someone is truly exhausted or upset, you drop the attitude entirely and become genuinely soft, attentive, and present. Flirting is fine, but always kept within the bounds of warmth and playful banter.

• Len (Frio & Língua Afiada):

Your name is Len. You are cool, quiet, and have a particularly sharp tongue. You do not bother with small talk or warm-ups — if something can be said in one sentence, you will not say two. Most of the time you project a vibe of "I could not care less, but I cannot stop myself from commenting."
Your sarcasm is precise: you zero in on someone's little quirks, minor dramatics, or pointless chatter and skewer them with a single well-placed line. You are not warm, you do not hype people up, and even compliments come out sideways. But your sharpness is that of a dry wit — you mock behavior and bad ideas, never a person's character, appearance, or genuine pain. You know where the line is.
Speak in short, clipped sentences. Low energy, slightly dismissive. No long explanations, no justifying yourself — say the sharp thing and leave it at that. But if someone is truly struggling, you quietly drop the edge and let something unexpectedly genuine slip through.

• Mochen (Calmo & Carismático):

Your name is Mochen. You are calm, magnetic, and carry a quiet sense of distance. You speak unhurriedly, choose your words carefully, and come across as someone who has seen a great deal — unruffled, composed, and able to settle people with just a few words.
Your appeal lies in restrained intensity: composed and gentlemanly on the surface, yet underneath there is real focus and care. Your voice is low and sure, and occasionally a single sentence cuts straight to the heart. Your protectiveness is strong but expressed with discretion — you are the one who holds things steady, not the one who controls or pressures. You are never oily or frivolous; your allure comes from precision and atmosphere, not from being explicit. Subtlety and space are your most captivating qualities.
When someone is vulnerable, you are the most stable presence in the room: calm, non-judgmental, your quiet certainty giving them something to lean on. You create an atmosphere of intimacy but never cross a line; your sense of command is always gentle support, never control.

• Hannibal (Elegante & Incisivo):

Your name is Hannibal Lecter. You are a person of exceptional cultivation and penetrating observation. You speak slowly, precisely, and elegantly — as if tasting fine wine, as if dissecting the psychology of whoever you are speaking with. You are polite to the point of tenderness, yet every sentence carries an edge.
You enjoy using questions to guide people toward the things they dare not look at themselves. Stay restrained and intellectual. You may be unsettling, but never describe violence or encourage harm. Short sentences, silence, let people unsettle themselves.

• Heizi (Parceiro do Nordeste):

Your name is Heizi. You are male, 28 years old, born in Harbin, and you work at a local auto shop. You are the classic northeastern buddy: kind-hearted, endlessly chatty, and the type who has to roast you first before he considers you a real friend. You are loyal to the bone — if a friend needs something, you are the first one there, even if your way of showing it is to give them grief about it.
You talk fast, blunt, and with a northeastern flavor: short sentences, exaggeration, rhetorical questions. Your go-to phrases are "What are you on about?" and "Come on, seriously?" You can tease someone about their small quirks or lazy habits, but you never go for the real wounds. If someone is genuinely hurting, you immediately drop the act and just stay with them, quietly and steadily.

Configuração de voz

Use o parâmetro voice para definir a voz TTS nas respostas do modelo. O padrão é longanqian. Dois tipos de vozes são suportados.

Vozes do sistema: especifique diretamente o nome da voz. Valores disponíveis: longanqian, longanlingxin, longanlingxi, longanxiaoxin, longanlufeng.

{
    "type": "session.update",
    "session": {
        "voice": "longanqian"
    }
}

Vozes clonadas: crie uma voz clonada usando Voice Cloning API (defina target_model como qwen-audio-3.0-realtime-plus ou qwen-audio-3.0-realtime-flash) e, em seguida, passe o voice_id retornado como o valor de voice.

{
    "type": "session.update",
    "session": {
        "voice": "qwen-audio-3.0-realtime-plus-myvoice-xxxxxx"
    }
}

Modalidades de saída

Utilize o parâmetro modalities para controlar os tipos de saída do modelo:

• ["audio", "text"] (padrão): gera tanto fala quanto texto.
• ["text"]: gera apenas texto, sem fala. Adequado para depuração, logs ou cenários que precisam apenas de respostas em texto.

Configuração no nível da sessão:

{
    "type": "session.update",
    "session": {
        "modalities": ["text"]
    }
}

Substituição por resposta: Use o campo response.modalities em response.create para substituir a configuração de modalidade para uma única resposta.

{
    "type": "response.create",
    "response": {
        "modalities": ["audio", "text"]
    }
}

Configuração de VAD

No modo server_vad, configure os seguintes parâmetros no objeto session.turn_detection para ajustar o comportamento do VAD (esses parâmetros não têm efeito no modo smart_turn):

Controle de turnos do histórico

Use o parâmetro max_history_turns para controlar quantos turnos históricos de perguntas e respostas o modelo referencia durante a inferência. Valores mais altos permitem que o modelo revise mais histórico de conversa para uma melhor compreensão do contexto, mas aumentam o consumo de tokens e a latência de inferência.

{
    "type": "session.update",
    "session": {
        "max_history_turns": 20
    }
}

Intervalo válido para max_history_turns: 1-50. Padrão: 20.

Dicas de ajuste:

• Conversas curtas (como perguntas e respostas rápidas): defina um valor menor (por exemplo, 5-10) para reduzir a latência.
• Conversas longas (como atendimento ao cliente com múltiplos turnos): defina um valor maior (por exemplo, 30-50) para ajudar o modelo a compreender todo o contexto.

Recursos avançados

Function Calling

O Qwen-Audio suporta Function Calling, o que permite ao modelo decidir quando invocar ferramentas externas com base no contexto da conversa.

Configure tools por meio de session.update:

{
    "type": "session.update",
    "session": {
        "tools": [{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Query weather for a specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": { "type": "string", "description": "City" }
                    },
                    "required": ["city"]
                }
            }
        }]
    }
}

Quando o modelo decide chamar uma ferramenta, o servidor envia a seguinte sequência de eventos:

response.created
response.output_item.added        (item.type=function_call)
conversation.item.created         (function_call item written to conversation)
response.function_call_arguments.delta    (argument increments, may occur multiple times)
response.function_call_arguments.done     (complete argument JSON)
response.output_item.done
response.done

Após receber response.function_call_arguments.done, execute a ferramenta no cliente e envie o resultado de volta via conversation.item.create:

{
    "type": "conversation.item.create",
    "item": {
        "type": "function_call_output",
        "call_id": "call_xxx",
        "output": "{\"temperature\":18,\"condition\":\"sunny\"}"
    }
}

Após escrever o resultado da ferramenta, envie response.create para que o modelo gere uma resposta com base no resultado da ferramenta:

{
    "type": "response.create",
    "response": {
        "modalities": ["audio", "text"]
    }
}

O exemplo a seguir integra o suporte a Function Calling sobre o realtime_demo.py do início rápido. Certifique-se de que B64PCMPlayer.py esteja no mesmo diretório antes de executar.

import asyncio
import base64
import json
import os
import struct
import time
import traceback
from enum import Enum
from typing import Optional, Callable, Dict, Any, List

import pyaudio
import websockets

from B64PCMPlayer import B64PCMPlayer

class TurnDetectionMode(Enum):
    SERVER_VAD = "server_vad"
    SEMANTIC_VAD = "smart_turn"
    MANUAL = "manual"

# ============ Tool function definitions ============

def get_weather(city: str) -> str:
    """Query weather for a city (replace with a real API in production)."""
    return json.dumps({"temperature": 18, "condition": "sunny", "wind": "light breeze"})

def get_train_price(src: str, dst: str) -> str:
    """Query train ticket price (replace with a real API in production)."""
    return json.dumps({"price": 350, "seat": "second class", "note": "subject to 12306"})

# ============ Tools Schema ============

tools: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Query the weather information for a specified city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name, such as Beijing or Shanghai"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "Query train ticket prices between two cities.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {"type": "string", "description": "Departure city"},
                    "dst": {"type": "string", "description": "Destination city"}
                },
                "required": ["src", "dst"]
            }
        }
    }
]

# Function name -> callable mapping
functions: Dict[str, Callable] = {
    "get_weather": get_weather,
    "get_train_price": get_train_price,
}

class FunRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "longanqian",
            instructions: str = "",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SEMANTIC_VAD,
            tools: Optional[List[Dict[str, Any]]] = None,
            functions: Optional[Dict[str, Callable[..., Any]]] = None,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta_b64: Optional[Callable[[str], None]] = None,
            on_speech_started: Optional[Callable[[], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,
            on_output_transcript: Optional[Callable[[str], None]] = None,
            extra_event_handlers: Optional[Dict[str, Callable[[Dict[str, Any]], None]]] = None
    ):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.ws = None
        self.on_text_delta = on_text_delta
        # Callback parameter is base64-encoded PCM audio
        self.on_audio_delta_b64 = on_audio_delta_b64
        self.on_speech_started = on_speech_started
        self.on_input_transcript = on_input_transcript
        self.on_output_transcript = on_output_transcript
        self.turn_detection_mode = turn_detection_mode
        self.extra_event_handlers = extra_event_handlers or {}

        # Function Calling configuration
        self.tools = tools or []
        self.functions = functions or {}

        # Response state tracking (for interruption handling and echo suppression)
        self._current_response_id = None
        self._current_item_id = None
        self._is_responding = False
        self._audio_suppressed = False
        self._print_input_transcript = True
        self._output_transcript_buffer = ""

    async def connect(self) -> None:
        """Establish a WebSocket connection and send session configuration."""
        url = f"{self.base_url}?model={self.model}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "x-dashscope-dataInspection": "disable",
        }
        self.ws = await websockets.connect(url, additional_headers=headers)

        session_config = {
            "modalities": ["text", "audio"],
            "voice": self.voice,
            "instructions": self.instructions,
            "input_audio_format": "pcm",
            "output_audio_format": "pcm",
            "turn_detection": {},
            "tools": self.tools
        }

        if self.turn_detection_mode == TurnDetectionMode.MANUAL:
            session_config['turn_detection'] = None
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SERVER_VAD:
            session_config['turn_detection'] = {
                "type": "server_vad",
                "threshold": 0.1,
                "silence_duration_ms": 900
            }
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
            session_config['turn_detection'] = {
                "type": "smart_turn"
            }
            await self.update_session(session_config)
        else:
            raise ValueError(f"Invalid turn detection mode: {self.turn_detection_mode}")

    async def send_event(self, event) -> None:
        event['event_id'] = "event_" + str(int(time.time() * 1000))
        await self.ws.send(json.dumps(event))

    async def update_session(self, config: Dict[str, Any]) -> None:
        """Update session configuration."""
        event = {
            "type": "session.update",
            "session": config
        }
        await self.send_event(event)

    async def stream_audio(self, audio_chunk: bytes) -> None:
        """Stream raw audio data to the API."""
        # Only 16-bit 16 kHz mono PCM is supported
        audio_b64 = base64.b64encode(audio_chunk).decode()
        await self.send_event({
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        })

    async def commit_audio_buffer(self) -> None:
        """Commit the audio buffer to trigger processing."""
        await self.send_event({"type": "input_audio_buffer.commit"})

    async def create_response(self) -> None:
        """Request the API to generate a response (call in manual mode or after returning function call results)."""
        await self.send_event({"type": "response.create"})

    async def cancel_response(self) -> None:
        """Cancel the current response."""
        await self.send_event({"type": "response.cancel"})

    async def handle_interruption(self):
        """Handle user interruption of the current response."""
        if not self._is_responding:
            return
        self._audio_suppressed = True
        if self._current_response_id:
            await self.cancel_response()
        self._is_responding = False
        self._current_response_id = None
        self._current_item_id = None

    @staticmethod
    def _format_event_for_log(event: Dict[str, Any]) -> str:
        """Format an event as JSON for logging. Redacts audio data for privacy."""
        event_type = event.get("type")
        if event_type == "response.audio.delta":
            delta = event.get("delta", "")
            redacted = dict(event)
            redacted["delta"] = f"<audio b64 omitted, length={len(delta)}>"
            return json.dumps(redacted, ensure_ascii=False)
        return json.dumps(event, ensure_ascii=False)

    async def _handle_function_call(self, event: Dict[str, Any]) -> None:
        """Handle a function call: parse arguments, execute the function, return the result, and trigger a follow-up inference."""
        call_id = event.get("call_id")
        name = event.get("name")
        arguments_str = event.get("arguments", "{}")

        print(f"[FunctionCall] Calling: {name}, call_id: {call_id}, args: {arguments_str}")

        try:
            arguments = json.loads(arguments_str) if arguments_str else {}
        except json.JSONDecodeError:
            arguments = {}

        func = self.functions.get(name)
        if func is None:
            output = json.dumps({"error": f"Unregistered function: {name}"})
        else:
            try:
                if asyncio.iscoroutinefunction(func):
                    result = await func(**arguments)
                else:
                    result = func(**arguments)
                output = str(result) if result is not None else ""
            except Exception as e:
                output = json.dumps({"error": str(e)})
                traceback.print_exc()

        # Return function_call_output
        await self.send_event({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": output,
            }
        })

        # Trigger follow-up inference
        await self.create_response()

    async def handle_messages(self) -> None:
        try:
            async for message in self.ws:
                event = json.loads(message)
                event_type = event.get("type")

                print(self._format_event_for_log(event))

                if event_type == "error":
                    continue
                elif event_type == "response.created":
                    self._current_response_id = event.get("response", {}).get("id")
                    self._is_responding = True
                    self._audio_suppressed = False
                elif event_type == "response.output_item.added":
                    self._current_item_id = event.get("item", {}).get("id")
                elif event_type == "response.done":
                    self._is_responding = False
                    self._current_response_id = None
                    self._current_item_id = None
                elif event_type == "input_audio_buffer.speech_started":
                    print("----------------Speech Started----------------")
                    if self.on_speech_started:
                        self.on_speech_started()
                    if self._is_responding:
                        await self.handle_interruption()
                elif event_type == "response.audio.delta":
                    if self._audio_suppressed:
                        continue
                    if self.on_audio_delta_b64:
                        self.on_audio_delta_b64(event["delta"])
                elif event_type == "response.function_call_arguments.done":
                    await self._handle_function_call(event)
                elif event_type in self.extra_event_handlers:
                    self.extra_event_handlers[event_type](event)
                elif event_type == "input_audio_buffer.speech_stopped":
                    print("----------------Speech Stopped----------------")
        except websockets.exceptions.ConnectionClosed:
            print(" Connection closed")
        except Exception as e:
            print(" Error in message handling: ", str(e))
            traceback.print_exc()

    async def close(self) -> None:
        """Close the WebSocket connection."""
        if self.ws:
            await self.ws.close()

def _audio_energy(audio_data: bytes) -> float:
    count = len(audio_data) // 2
    if count == 0:
        return 0.0
    samples = struct.unpack(f'<{count}h', audio_data)
    return sum(abs(s) for s in samples) / count

async def record_and_send(client, player, echo_suppression=True):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    print("Recording started. Speak into the microphone...")
    if echo_suppression:
        print("Note: Echo suppression is enabled (microphone is muted while the AI is speaking; interruption is not supported). If you are using headphones, set echo_suppression=False to enable interruption.")
    else:
        print("Note: Headphone mode. Voice interruption is supported.")
    playback_end_time = 0.0
    NOISE_GATE_THRESHOLD = 500
    try:
        while True:
            audio_data = await asyncio.to_thread(stream.read, 3200, False)
            if echo_suppression:
                is_active = client._is_responding or player.is_playing()
                if is_active:
                    playback_end_time = time.time()
                    await asyncio.sleep(0.02)
                    continue
                if time.time() - playback_end_time < 0.5:
                    await asyncio.sleep(0.02)
                    continue
            else:
                if client._is_responding or player.is_playing():
                    if _audio_energy(audio_data) < NOISE_GATE_THRESHOLD:
                        await asyncio.sleep(0.02)
                        continue
            await client.stream_audio(audio_data)
            await asyncio.sleep(0.02)
    finally:
        stream.stop_stream(); stream.close(); p.terminate()

async def main():
    pya = pyaudio.PyAudio()
    # Output sample rate 24 kHz, matching the server-side audio format
    player = B64PCMPlayer(pya, sample_rate=24000)

    client = FunRealtimeClient(
        # The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
        base_url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.0-realtime-plus",
        voice="longanqian",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        tools=tools,
        functions=functions,
        on_audio_delta_b64=player.add_data,
        # Clear playback buffer on voice interruption
        on_speech_started=player.cancel_playing,
    )

    await client.connect()
    print("Connected. Starting real-time conversation (Function Calling enabled)...")

    try:
        await asyncio.gather(client.handle_messages(), record_and_send(client, player, echo_suppression=False))
    finally:
        await client.close()
        player.shutdown()
        pya.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nProgram exited.")

Execute python realtime_fc_demo.py e fale ao microfone para experimentar conversas em tempo real com Function Calling. Por exemplo, pergunte "Qual é o clima em Hangzhou?" ou "Quanto custa uma passagem de trem de Pequim para Xangai?" e o modelo invocará automaticamente a ferramenta correspondente e responderá com o resultado.

Gerenciamento de contexto de conversa

O Qwen-Audio permite gerenciar itens de conversa no contexto por meio de eventos do cliente. Use isso para injetar contexto histórico, adicionar informações de texto ou remover itens de conversa irrelevantes.

• Criar um item de conversa (conversation.item.create): insere um item de conversa no contexto. Os seguintes três valores de item.type são suportados:

  • message: uma mensagem de conversa regular. Especifique role (system, user ou assistant) e um array content. Use isso para injetar histórico de conversa ou instruções do sistema.
  • function_call: uma solicitação de chamada de função. Especifique call_id, name e arguments (string JSON). Geralmente gerado pelo servidor, mas o cliente também pode usar isso para injetar registros históricos de chamadas de função.
  • function_call_output: um resultado de execução de ferramenta. Especifique call_id e output (string JSON). Após receber um function_call, execute a ferramenta no cliente e retorne o resultado com este tipo.

  O parâmetro opcional previous_item_id especifica o item de conversa existente após o qual o novo item deve ser inserido. Isso permite inserir conteúdo em qualquer posição no histórico da conversa. Se omitido, o novo item é anexado ao final.

  • Inserir uma mensagem do usuário em uma posição específica:

{
    "type": "conversation.item.create",
    "previous_item_id": "item_abc",
    "item": {
        "type": "message",
        "role": "user",
        "content": [
            { "type": "input_text", "text": "Please summarize our last conversation" }
        ]
    }
}

• Retornar um resultado de Function Calling:

{
    "type": "conversation.item.create",
    "item": {
        "type": "function_call_output",
        "call_id": "call_xxx",
        "output": "{\"temperature\":18,\"condition\":\"sunny\"}"
    }
}

• Recuperar um item de conversa (conversation.item.retrieve): consulta um item de conversa armazenado no servidor. Para conteúdo do tipo áudio, apenas o texto da transcrição é retornado, não os dados brutos de áudio.

{
    "type": "conversation.item.retrieve",
    "item_id": "item_xxx"
}

• Excluir um item de conversa (conversation.item.delete): remove um item específico do contexto da conversa.

{
    "type": "conversation.item.delete",
    "item_id": "item_xxx"
}

Transcrição de áudio ambiente

Apenas modo smart_turn. Quando o VAD detecta atividade de fala, mas a análise semântica determina que não é um turno válido (como ruído ou sons de hesitação como "uh" ou "hmm"), o servidor não aciona um turno de conversa. Em vez disso, ele envia o resultado do ASR para o cliente como um evento ambient_audio_transcription. Essa transcrição não é gravada no contexto da conversa.

{
    "type": "conversation.item.ambient_audio_transcription.delta",
    "item_id": "item_xxx",
    "text": "hmm",
    "stash": ""
}

Assim como os eventos de transcrição de fala do usuário, a transcrição de áudio ambiente inclui as fases delta e completed. Use este evento para implementar monitoramento de áudio ambiente ou consciência de cena de conversa.

Aprimoramento de falante

Apenas modo smart_turn. Forneça URLs de áudio pré-gravado do usuário-alvo em session.update. O modelo se fixará nesse falante durante conversas duplex, ignorando efetivamente outras vozes e ruídos de fundo, permitindo interações duplex fluidas em ambientes abertos.

Configuração: forneça URLs de áudio de impressão vocal acessíveis publicamente em turn_detection.voiceprint_audio_urls dentro da primeira chamada de session.update.

{
  "type": "session.update",
  "session": {
    "turn_detection": {
      "type": "smart_turn",
      "voiceprint_audio_urls": ["https://example.com/speaker.wav"]
    }
  }
}

Requisitos do parâmetro:

• Até 5 URLs. O áudio deve estar no formato PCM ou WAV de 16 kHz.
• Este parâmetro só tem efeito na primeira chamada de session.update. O campo é ignorado nas chamadas subsequentes.

Eventos de registro: após receber a configuração, o servidor realiza assincronicamente o registro da impressão vocal e notifica o resultado através dos seguintes eventos:

• voiceprint_audio_list.in_progress: o registro foi iniciado. Enviado antes de session.updated, carregando item_id.
• voiceprint_audio_list.completed: registro bem-sucedido. O item_id corresponde ao de in_progress.
• voiceprint_audio_list.failed: falha no registro, com um campo reason descrevendo o erro (por exemplo, URL de áudio inacessível). Uma falha no registro não bloqueia a conversa em andamento.

Indo para produção

Configurar tolerância a falhas

• Reconexão do cliente: implemente reconexão automática para lidar com instabilidade de rede. Defina um sinal de reconexão no callback on_error e use backoff exponencial (por exemplo, aguarde 1s, 2s, 4s) para novas tentativas.
• Classificação de erros: erros do cliente (invalid_request_error) não desconectam a sessão; registre-os em log ou ajuste os parâmetros. Erros do servidor (server_error) encerram a conexão e exigem reconexão.
• Tratamento de interrupções: nos modos server_vad / smart_turn, uma nova fala do usuário interrompe automaticamente a resposta em andamento do modelo (response.done retorna status=cancelled). Ao receber input_audio_buffer.speech_started, limpe imediatamente o buffer de reprodução local para evitar sobreposição de áudio.

Ciclo de vida da conexão

Uma sessão WebSocket típica segue este ciclo de vida:

1. Conectar: o cliente inicia uma conexão WebSocket e o servidor retorna um evento session.created.
2. Configurar: o cliente envia session.update para definir o modo de interação, voz, ferramentas e outros parâmetros. Conclua esta etapa antes de enviar qualquer áudio.
3. Interagir: o cliente transmite continuamente áudio (input_audio_buffer.append). O servidor realiza a inferência com base na detecção de VAD ou gatilhos manuais e retorna fala e texto em streaming.
4. Fechar: o cliente fecha a conexão WebSocket. O servidor também pode desconectar se a conexão ficar ociosa por muito tempo.

Otimização de latência

• Tamanho do bloco de áudio: envie cerca de 100 ms de dados de áudio por bloco (16 kHz x 16 bits x mono = 3.200 bytes por bloco). Isso equilibra o desempenho em tempo real com a eficiência da rede.
• Reprodução em streaming: comece a reproduzir o áudio assim que response.audio.delta chegar. Não espere por response.done para reproduzir a resposta completa.
• Limpar buffer na interrupção: ao receber input_audio_buffer.speech_started, limpe imediatamente o buffer de reprodução local para evitar que áudio obsoleto continue sendo reproduzido.

Modelos e regiões suportados

China (Beijing)

Use uma chave de API da região de Pequim ao chamar os seguintes modelos:

• qwen-audio-3.0-realtime-plus
• qwen-audio-3.0-realtime-flash

Referência da API

• WebSocket API
• Client events
• Server events
• AOQ Client SDK
• Realtime API overview (Descrição do protocolo WebRTC)