Conversa de voz em tempo real com Qwen Audio

Qwen-Audio é um modelo de interação de voz em tempo real para conversas de voz de baixa latência. Casos de uso incluem assistentes de voz, serviço ao cliente inteligente, e companheiros de IA.

Visão geral

Qwen-Audio converte áudio em tempo real em voz e texto sobre uma conexão duplex, com entrada de streaming e saída de streaming.

Além do WebSocket, este modelo também suporta os protocolos de QAO e WebRTC. Para integração cliente-side que prioriza latência estável, resiliência em redes fracas, e integração de ruído e cancelamento de eco, o AOQ é recomendado. Para uma comparação de protocolo, veja [Realtime API overview] (/api-reference/realtime-api/overview#supported-models-and-applications).

• Três modos de interação: VAD acústico (servidor_vad), detecção inteligente de turnos semânticos (smart_turn), e controle manual (empurrar-para-falar)
• No modo inteligente, o modelo combina percepção acústica e compreensão semântica para determinar limites de giro, então sons de preenchimento como "uh" ou "hmm" não interrompem a conversação
• Função de suporte para chamadas permite que o modelo decida quando invocar ferramentas externas para informações adicionais.
• O suporte de busca na web permite que o modelo decida quando recuperar informações atualizadas e retornar fontes de resultados de busca (busca na web e Chamada de Função não podem ser ativadas ao mesmo tempo)
• Gestão de contexto de conversação: criar, recuperar e excluir itens de conversação para injetar contexto histórico ou remover itens irrelevantes
• Saída de voz expressiva que ajusta dinamicamente o tom, ritmo e emoção baseado no contexto da conversa
• Suporte para vozes de sistema e vozes clonadas; use a clonagem de voz para criar uma voz de IA personalizada para a saída de voz
• Melhoramento do alto-falante no modo inteligente e de volta: passe áudio pré-gravado de um usuário alvo para que o modelo possa travar no alto-falante durante conversas duplex, bloqueando efetivamente outras vozes e ruído de fundo.

Como funciona

Qwen-Audio usa uma conexão completa com uma arquitetura orientada por 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 as respostas de voz e texto em tempo real. Toda a interação é orientada para eventos: o cliente envia eventos como session.update e input_audio_buffer.appendE o servidor responde com eventos como response.audio.delta e response.done. Não é necessário votação.

Um ciclo de vida típico de conexão é: estabelecer conexão WebSocket, enviar session.update para configurar os parâmetros da sessão, transmitir áudio e receber respostas, então feche a conexão.

Formato de áudio

• Direção * * Format * * Especificação * * | ------------------------- | ---------- | -------------------------------------- | Entrada (cliente para servidor) taxa de amostra de 16 kHz, profundidade de 16 bits, mono Saída (servidor para cliente) taxa de amostra de 24 kHz, profundidade de 16 bits, mono

Linguagens apoiadas

Conversas de voz em tempo real apoiam as seguintes línguas e variedades chinesas:

• Línguas: alemão (de), inglês (en), espanhol (es), francês (fr), indonésio (id), italiano (it), japonês (ja), coreano (ko), português (pt), russo (ru) e chinês (zh).
• Castas chinesas: mandarim, cantonês, Chongqing, nordeste, Gansu, Guizhou, Zhejiang, Hebei, Henan, Hubei, Hunan, Jiangxi, Ningbo, Ningxia, Qingdao, Shaanxi, Shanxi, Shandong, Shanghainese, Sichuan e Yunnan.

Quando se usa qwen-audio-3.1-realtime-plus Você pode usar session.input_audio_transcription.language e session.output_audio.language para especificar a linguagem de origem do áudio de entrada e a linguagem de destino do áudio de saída, respectivamente. Para os códigos de linguagem válidos para cada parâmetro, veja [Eventos clínicos] (/api-reference/qwen-audio-realtime/client-events). Dos dois, session.output_audio.language O modelo faz um esforço para produzir fala na língua alvo, mas isso não é garantido, e a linguagem do texto de resposta não é alterada.

Capacidade de contexto

O modelo mantém o histórico de conversa. Quando o número de voltas ou duração cumulativa de áudio excede os seguintes limites, o histórico anterior é automaticamente descartado. A duração máxima é o limite máximo de áudio cumulativo que o contexto do modelo pode manter.

Ver a tabela estruturada nos quadros de referência desta página ↓

O valor padrão para o máximo de voltas de áudio é 20. Pode aumentar para 50. Para detalhes de configuração, veja [História do controle de turno] (#history-turn-control).

Para orientação sobre escolher entre modelos multimodais, veja [Omni-modal] (/developer-guides/getting-started/vision-models).

Pré-requisitos

• [Obter uma chave API] (/api-reference/preparation/api-key(configurando-o como uma variável de ambiente) (/api-reference/preparation/export-api-key-env).
• Para usar o protocolo AOQ, baixe e integre o cliente AOQ SDK. Para detalhes, veja [SDK visão geral] (/api-reference/realtime-api/aoq-sdk-intro).

-Começo rápido.

Os exemplos básicos de WebSocket abaixo usam qwen-audio-3.1-realtime-plus com a voz do sistema longanqian_v3.1 -Esqueça isso. Para migrar de 3.0 Plus, atualize o model parâmetro de consulta no URL e selecione um correspondente voice [Configuração de voz] (#voice-configuration). O protocolo de eventos WebSocket não mudou.

Siga estes passos para começar uma conversa de voz em tempo real com o Qwen-Audio Modelo.

Para a sequência de interação de eventos WebSocket de cada modo, veja [fluxo de interação de eventos] (/api-reference/qwen-audio-realtime/websocket-api).

O exemplo a seguir demonstra conversas em tempo real sobre microfones sobre uma conexão nativa WebSocket em server_vad Modo. Antes de correr, instale as dependências necessárias:

macOS

```bash
brew install portaudio && pip install pyaudio websockets
```

Debian/Ubuntu

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

Windows

```powershell
pip install pyaudio websockets
```

Guarde o seguinte código como realtime_quickstart.py:

import asyncio
import base64
import json
import os
import pyaudio
import websockets
API_KEY = os.environ["DASHSCOPE_API_KEY"]
URL = "wss://maas.qwencloudapi.com/api-ws/v1/realtime?model=qwen-audio-3.1-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_v3.1",
                "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")

Corra. python realtime_quickstart.py e falar no microfone para ter uma conversa em tempo real com o modelo. O servidor detecta automaticamente o início e o fim da fala e desencadeia respostas.

Exemplo completo

O exemplo a seguir se baseia no fluxo básico de conversa acima, adicionando interrupção de manuseio e supressão de eco. Crie os seguintes dois arquivos no mesmo diretório:

B64PCMPlayer.py

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()

realtime_demo.py

Se o seu `websockets` A versão é menor que 11, mudança. `additional_headers` no código para `extra_headers`, ou atualização: `pip install --upgrade websockets`.

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_v3.1",
            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(
                base_url="wss://maas.qwencloudapi.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.1-realtime-plus",
        voice="longanqian_v3.1",
        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.")

Corra. python realtime_demo.py e falar no microfone para ter uma conversa em tempo real com o modelo. O sistema detecta automaticamente o início e o fim da fala e desencadeia respostas.

O exemplo acima usa o modo servidor_vad (VAD acústico do lado do servidor detecta automaticamente o início e o fim da fala). Para usar o modo inteligente de detecção de turnos semânticos inteligentes ou push-to-talk (controle manual), veja [modos de interação] (#interaction-modes).

Configuração da sessão

Modos de interação

Qwen-Audio suporta três modos de interação: server_vad (VAD acústico detecta automaticamente o início e o fim da fala), smart_turn (Detecção de turnos semânticos inteligentes, combinando sinais acústicos e semânticos) e push-to-talk (cliente controla manualmente a sessão). Para detalhes sobre cada modo e a linha do tempo de interação de eventos (incluindo diagramas de sequência), veja [Modos de interação] (/api-reference/qwen-audio-realtime/websocket-api).

turn_detection só pode ser alterado antes do primeiro áudio ser enviado (estado de IDIE). Mudar de modo de interação após a sessão ter começado requer reconectar.

Mude de modos de interação usando o turn_detection - Não, não, não. session.update evento:

• servidor_vad:

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

• Legendas em Série Sincronia:

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

• Empurra para falar.

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

Exemplo completo de push-to-talk

manual_realtime.py

# pip install websocket-client pyaudio
import json
import os
import base64
import threading
import time
import pyaudio
import websocket
API_KEY = os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://maas.qwencloudapi.com/api-ws/v1/realtime?model=qwen-audio-3.1-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_v3.1",
            "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.

Use o instructions parâmetro para definir a identidade do modelo, estilo de resposta e preferências de comportamento. Configure este parâmetro em session.update- Aplica-se 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."
    }
}

Recomendações:

• Defina uma identidade clara (por exemplo, "Você é um assistente de voz inteligente" ou "Você é um tutor de conversa em inglês"), e opcionalmente inclua detalhes como nome ou gênero.
• Especifique um tom conversacional e estilo de frase, enquanto enfatiza que um tom natural não compromete a completude do conteúdo - detalhes, números, e recomendações específicas ainda devem ser incluídas, apenas expressa de uma forma relaxada, natural.
• Instrua o modelo para explicar todas as restrições de contexto na conversa (como orçamento, preferências, restrições ou acordos anteriores). Quando várias condições se aplicam, dirija-se a cada uma e não omita informações críticas.
• Controle o formato de saída: a menos que o usuário peça o contrário, evite emoji e outros personagens especiais e formatação Markdown. Texto simples de saída para garantir a reprodução natural do TTS.
• Defina estratégia de resposta: manter cumprimentos simples e trocas casuais breves e naturais; para raciocínio, problemas de multi-condições, listas de recomendações ou conselhos de segurança, priorizar a completude – garantir que informações fundamentais (como preços, locais e condições) estejam totalmente presentes, sem preâmbulo desnecessário, repetição ou preenchimento.
• Defina uma estratégia de seguimento: siga o princípio de "responda primeiro a pergunta atual do usuário, e depois, naturalmente, faça um acompanhamento no final para avançar a conversa." Faça apenas uma pergunta de cada vez; não faça várias perguntas seguidas ou confirme repetidamente.

** Configuração padrão**

O seguinte é um recomendado instructions configuração para cenários de conversação de voz em geral. Cobre definição de papéis, estilo conversacional, controle de formato e estratégia de seguimento. Use-o diretamente ou adapte-o à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.

• Exemplos de Persona *

O seguinte: instructions Exemplos cobrem uma variedade de estilos de persona. Escolha um que se encaixe no seu caso de uso ou personalize-o ainda mais:

• • Daisy (Doce e Companheiro Legal)**:

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 (Cool & Sharp-Tongued):

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 (Calm & Carismatic)**:

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 e 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 (Amigos 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 da voz

Use o voice parâmetro para definir a voz TTS para respostas de modelo. O padrão é longanqian_v3.1 Por 3,1 mais e longanqian para 3.0 mais/Flash. As vozes do sistema e as vozes clonadas são suportadas. 3.1 Além disso, mantém as vozes do 3.0 Plus e adiciona oito opções de voz do sistema.

A voz só pode ser definida no primeiro session.update. O campo é ignorado na sequência. session.update Chamadas.

Vozes do sistema: especifique um nome de voz diretamente.

Ver a tabela estruturada nos quadros de referência desta página ↓

O seguinte é um exemplo de configuração 3.1.

{
    "type": "session.update",
    "session": {
        "voice": "longanqian_v3.1"
    }
}

"Criar uma voz clonada usando a API de clonagem de voz"/api-reference/speech-synthesis/voice-cloning/qwen/create-voice(set target_model para qwen-audio-3.1-realtime-plus, qwen-audio-3.0-realtime-plus, ou qwen-audio-3.0-realtime-flash", então passe o devolvido" voice_id como o voice Valor.

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

Modos de saída

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

• ["audio", "text"] Saídas tanto de discurso quanto de texto.
• ["text"] Só sai texto, sem falar. Adequado para depuração, registro ou cenários que só precisam de respostas de texto.

** Session-nível de ajuste**:

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

Use o botão de resposta. response.modalities - Não, não, não. response.create para substituir a configuração da modalidade para uma única resposta.

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

Configuração do VAD

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

• Parameter * * Tipo * * Descrição * * | --------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | threshold - Flutue, sensibilidade VAD. Valores mais baixos aumentam a sensibilidade do DAV, facilitando a detecção de sons fracos (incluindo ruído de fundo) como fala. Valores mais altos diminuem a sensibilidade, requer uma fala mais clara e mais alta para desencadear a detecção. Distância: [-1.0, 1.0]. Padrão: 0,5. | silence_duration_ms Duração mínima do silêncio (em milissegundos) após o fim da fala antes de desencadear uma resposta modelo. Valores mais baixos produzem respostas mais rápidas, mas podem causar falsos gatilhos durante breves pausas. Distância: [200, 6000]. Padrão: 800. Alcance recomendado para conversas: 400-800.

História vira controle

Use o max_history_turns parâmetro para controlar quantas perguntas históricas giram as referências do modelo durante a inferência. Valores mais elevados permitem que o modelo revise mais história de conversa para melhor compreensão do contexto, mas aumentam o consumo de fichas e a latência de inferência.

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

Alcance válido para max_history_turns: 1-50. Predefinição: 20.

Pontas de arremesso:

• Conversas curtas, como perguntas rápidas, definir um valor menor, por exemplo, 5-10 para reduzir a latência.
• Conversas longas (como o serviço multi-turno ao cliente): definir um valor maior (por exemplo, 30-50) para ajudar o modelo a entender o contexto completo.

Recursos avançados

Chamada de Função

Qwen-Audio suporta Chamada de Função, que permite ao modelo decidir quando invocar ferramentas externas baseadas no contexto da conversa.

1. Ferramentas de registro

Configurar tools através 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"]
                }
            }
        }]
    }
}

**2. Receba chamadas de função

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

**3. Execute a ferramenta e retorne os resultados.

Depois de receber response.function_call_arguments.done, execute a ferramenta no cliente e envie o resultado de volta através conversation.item.create:

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

**4. Acione uma resposta de seguimento.

Depois de escrever o resultado da ferramenta, envie response.create para que o modelo gere uma resposta baseada no resultado da ferramenta:

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

Uma única resposta pode conter múltiplos function_call itens, e pode incluir mensagens regulares e chamadas de função. O conteúdo da chamada não é enviado para TTS para reprodução.

Exemplo completo

O exemplo a seguir integra o suporte de Chamadas de Função em cima do realtime_demo.py Desde o começo rápido. Certifique-se. B64PCMPlayer.py está no mesmo diretório antes de correr.

realtime_fc_demo.py

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_v3.1",
            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."""
        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(
                base_url="wss://maas.qwencloudapi.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.1-realtime-plus",
        voice="longanqian_v3.1",
        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.")

Corra. python realtime_fc_demo.py e fale no microfone para tentar conversas em tempo real com Function Calling. Por exemplo, pergunte "Qual é o tempo em Hangzhou?" ou "Quanto custa uma passagem de trem de Pequim para Xangai?" e o modelo automaticamente invoca a ferramenta correspondente e responde com o resultado.

Gestão de contexto de conversação

Qwen-Audio permite gerenciar itens de conversação no contexto através de eventos de clientes. Use isso para injetar contexto histórico, adicionar informações de texto ou remover itens irrelevantes de conversa.

• Crie um item de conversaconversation.item.createIntroduz um item de conversa no contexto. Os três seguintes item.type Os valores são suportados:

• message Uma mensagem de conversa regular. Especifique role (system, user, ou assistant) e a content Array. Use isso para injetar histórico de conversa ou instruções do sistema.

• function_call... uma chamada de função. Especifique call_id, name, e arguments (Corda JSON). Normalmente gerado pelo servidor, mas o cliente também pode usar isso para injetar registros de chamadas de funções históricas.

• function_call_output... um resultado de execução de ferramentas. Especifique call_id e output (Corda JSON). Depois de receber um function_call, execute a ferramenta no cliente e devolva o resultado com este tipo.

O opcional. previous_item_id O parâmetro especifica o item de conversa existente para inserir o novo item. Isso permite inserir conteúdo em qualquer posição no histórico de conversa. Se omitido, o novo item é adicionado ao fim.

• Insira uma mensagem de 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" }
        ]
    }
}

• Retorne o resultado da chamada de função:

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

Se o item.id especificado em conversation.item.create Já existe na conversa, um erro é devolvido.

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

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

• ** Excluir um item de conversa ** (conversation.item.deleteRemove um item específico do contexto de conversa.

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

Transcrição de áudio ambiente

Apenas o modo inteligente de virar. Quando o DAV detecta atividade de fala, mas a análise semântica determina que não é uma volta válida (como ruído ou preenchimento soa como "uh" ou "hmm"), o servidor não desencadeia uma volta de conversa. Em vez disso, envia o resultado da ASR para o cliente como um ambient_audio_transcription evento. Esta transcrição não está escrita no contexto da conversa.

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

Como eventos de transcrição de voz do usuário, transcrição de áudio ambiente inclui delta e completed Fases. Use este evento para implementar monitoramento de áudio ambiente ou consciência de cena de conversação.

- O que é isso?

Apenas o modo inteligente de virar. Passe URLs de áudio pré-gravados do usuário alvo em session.update. O modelo vai travar o alto-falante durante conversas duplex, ignorando as outras vozes e ruído de fundo, permitindo interações duplex fluidas em ambientes abertos.

Configuração: passe URLs de áudio de voz acessíveis publicamente em turn_detection.voiceprint_audio_urls dentro do primeiro 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. Áudio deve ser 16 kHz PCM ou WAV.
• Este parâmetro só faz efeito no primeiro session.update. O campo é ignorado em chamadas subsequentes.

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

• voiceprint_audio_list.in_progress...o registro começou. Enviado antes session.updated, carregando item_id.
• voiceprint_audio_list.completedO registro foi bem sucedido. O item_id combina com o de in_progress.
• voiceprint_audio_list.failedO registro falhou, com um reason campo descrevendo o erro (por exemplo, URL de áudio não é acessível). Falha de registro não bloqueia a conversa em andamento.

Indo ao vivo

- Aplique tolerância à falha.

• Reconexão do cliente: implementar reconexão automática para lidar com o nervosismo da rede. Defina um sinal de reconexão no on_error Retorno e uso de retrocesso exponencial (por exemplo, espera 1s, 2s, 4s) para tentativas de reversão.
• Classificação de erro: erros de cliente.invalid_request_error Não desligue a sessão, registre-os ou ajuste os parâmetros. Erros de servidor (server_error) terminar a conexão e exigir reconexão.
• ** Interrupção de manipulação**: no servidor_vad / smart_turn modos, o novo discurso do usuário interrompe automaticamente a resposta contínua do modelo (response.done retorna status=cancelled). Quando input_audio_buffer.speech_started É recebido, imediatamente, limpar o buffer local de reprodução para evitar sobreposição de áudio.

Ciclo de vida da conexão

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

1. Connect: o cliente inicia uma conexão WebSocket e o servidor retorna uma conexão WebSocket session.created evento.
2. Configurar: o cliente envia session.update para definir o modo de interação, voz, ferramentas e outros parâmetros. Complete este passo antes de enviar qualquer áudio.
3. O cliente transmite continuamente áudioinput_audio_buffer.append). O servidor faz inferências baseadas em detecção de DAV ou gatilhos manuais e retorna fala e texto de forma em streaming.
4. 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**: enviar cerca de 100 ms de dados de áudio por pedaço (16 kHz x 16 bits x mono = 3.200 bytes por pedaço). Isso equilibra o desempenho em tempo real com a eficiência da rede.
• Streaming playback: começar a tocar áudio assim que response.audio.delta Chega. Não esperem. response.done para jogar a resposta completa.
• Quando a interrupção estiver livre input_audio_buffer.speech_started é recebido, imediatamente limpar o buffer local de reprodução para evitar áudio velho de continuar a tocar.

Modelos apoiados.

Use uma região de Singapura [chave API] (https://home.qwencloud.com) quando se chama os seguintes modelos:

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

Referência da API

• [WebSocket API] (/api-reference/qwen-audio-realtime/websocket-api)
• [Acontecimentos do cliente] (/api-reference/qwen-audio-realtime/client-events)
• [Eventos de servidor] (/api-reference/qwen-audio-realtime/server-events)
• [AOQ Cliente SDK] (/api-reference/realtime-api/aoq-sdk-intro)
• [Observação API em tempo real] (/api-reference/realtime-api/overview(Descrição do protocolo WebRTC)