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

Qwen-Audio est un modèle d'interaction vocale en temps réel de bout en bout qui utilise le protocole de streaming WebSocket pour des conversations vocales à faible latence. Les cas d'usage incluent les assistants vocaux, le service client intelligent et les compagnons IA.

Présentation

Qwen-Audio convertit l'audio en temps réel en parole et en texte via une connexion full-duplex, avec une entrée et une sortie en streaming.

Outre WebSocket, ce modèle prend également en charge les protocoles AOQ et WebRTC. Pour une intégration côté client privilégiant une latence stable, une résilience sur les réseaux faibles, ainsi qu'une suppression du bruit et une annulation de l'écho full-duplex intégrées, le protocole AOQ est recommandé. Pour comparer les protocoles, consultez Présentation de l'API Realtime.

• Trois modes d'interaction : VAD acoustique (server_vad), détection intelligente des tours de parole sémantiques (smart_turn) et contrôle manuel (push-to-talk)
• En mode smart_turn, le modèle combine la perception acoustique et la compréhension sémantique pour déterminer les limites des tours de parole, afin que les sons d'hésitation comme « euh » ou « hmm » n'interrompent pas la conversation
• La prise en charge du Function Calling permet au modèle de décider quand invoquer des outils externes pour obtenir des informations supplémentaires
• Gestion du contexte de conversation : créez, récupérez et supprimez des éléments de conversation pour injecter un historique ou retirer des éléments non pertinents
• Sortie vocale expressive ajustant dynamiquement le ton, le rythme et l'émotion selon le contexte de la conversation
• Prise en charge des voix système et des voix clonées ; utilisez Voice Cloning pour créer une voix IA personnalisée pour la synthèse vocale
• Amélioration du locuteur en mode smart_turn : transmettez un audio préenregistré d'un utilisateur cible afin que le modèle puisse se verrouiller sur ce locuteur lors de conversations duplex, bloquant ainsi efficacement les autres voix et le bruit ambiant

Fonctionnement

Qwen-Audio utilise une connexion full-duplex WebSocket reposant sur une architecture orientée événements. Le client et le serveur échangent des données simultanément via une connexion persistante : le client diffuse en continu l'audio du microphone tandis que le serveur renvoie des réponses vocales et textuelles en temps réel. L'ensemble de l'interaction est piloté par des événements : le client envoie des événements tels que session.update et input_audio_buffer.append, et le serveur répond par des événements tels que response.audio.delta et response.done. Aucune interrogation (polling) n'est nécessaire.

Le cycle de vie typique d'une connexion comprend les étapes suivantes : établissement de la connexion WebSocket, envoi de session.update pour configurer les paramètres de session, streaming audio et réception des réponses, puis fermeture de la connexion.

Format audio

Capacité du contexte

Le modèle conserve l'historique de la conversation. Lorsque le nombre de tours ou la durée cumulée de l'audio dépasse les limites suivantes, l'historique le plus ancien est automatiquement supprimé. La durée maximale correspond à la limite supérieure de l'audio cumulé que le contexte du modèle peut conserver.

La valeur par défaut du nombre maximal de tours audio est de 20. Vous pouvez l'augmenter jusqu'à 50. Pour plus de détails sur la configuration, consultez Contrôle des tours de l'historique.

Pour obtenir des conseils sur le choix entre les modèles multimodaux, consultez Omni-modal.

Prérequis

• Obtenez une clé API et configurez-la en tant que variable d'environnement.
• Pour utiliser le protocole AOQ, téléchargez et intégrez le SDK client AOQ. Pour plus de détails, consultez Présentation du SDK.

Démarrage rapide

Suivez ces étapes pour démarrer une conversation vocale en temps réel avec le modèle 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.")

Configuration de la session

Modes d'interaction

Qwen-Audio prend en charge trois modes d'interaction : server_vad (VAD acoustique pour la détection automatique de la parole), smart_turn (détection intelligente des tours de parole sémantiques, combinant analyse acoustique et sémantique), et push-to-talk (contrôle manuel par le client). Pour des descriptions détaillées et des diagrammes de flux d'interaction des événements, consultez Modes d'interaction.

Pour changer de mode d'interaction, définissez le champ turn_detection dans un événement 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
      }
  }
  

Exemple complet 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")

Instructions système

Utilisez le paramètre instructions pour définir le rôle du modèle, son style de réponse et ses préférences comportementales. Configurez ce paramètre dans session.update pour l'appliquer à l'ensemble de la session.

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

Conseils :

• Définissez une identité de rôle claire (par exemple, « Vous êtes un assistant vocal intelligent » ou « Vous êtes un tuteur de conversation en anglais ») et incluez éventuellement des détails tels que le nom ou le genre.
• Spécifiez un ton conversationnel et un style de formulation, tout en précisant qu'un ton naturel ne doit pas compromettre l'exhaustivité du contenu : les détails, les chiffres et les recommandations spécifiques doivent toujours être inclus, mais exprimés de manière détendue et naturelle.
• Demandez au modèle de prendre en compte toutes les contraintes contextuelles de la conversation (telles que le budget, les préférences, les restrictions ou les accords préalables). Lorsque plusieurs conditions s'appliquent, traitez chacune d'elles sans omettre d'information critique.
• Contrôlez le format de sortie : sauf demande contraire de l'utilisateur, évitez les émojis, les caractères spéciaux et le formatage Markdown. Privilégiez le texte brut pour garantir une lecture TTS naturelle.
• Définissez une stratégie de réponse : gardez les salutations simples et les échanges informels brefs et naturels ; pour les raisonnements, les problèmes à conditions multiples, les listes de recommandations ou les conseils de sécurité, privilégiez l'exhaustivité. Assurez-vous que les informations clés (prix, lieux, conditions) sont entièrement présentes, sans préambule inutile, répétition ou remplissage.
• Établissez une stratégie de relance : suivez le principe « répondre d'abord pleinement à la question actuelle de l'utilisateur, puis poser naturellement une question de suivi à la fin pour faire avancer la conversation ». Ne posez qu'une seule question à la fois ; évitez les questions en série ou les confirmations répétées.

Voici une configuration instructions recommandée pour les scénarios généraux de conversation vocale. Elle couvre la définition du rôle, le style conversationnel, le contrôle du format et la stratégie de relance. Utilisez-la directement ou adaptez-la à vos besoins :

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.

Les exemples instructions suivants couvrent une variété de styles de personas. Choisissez celui qui correspond à votre cas d'usage ou personnalisez-le davantage :

• Daisy (Compagne douce & cool) :

  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 & langue acérée) :

  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 (Calme & charismatique) :

  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 (Élégant & incisif) :

  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 (Pote du Nord-Est) :

  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.
  

Configuration de la voix

Utilisez le paramètre voice pour définir la voix TTS des réponses du modèle. La valeur par défaut est longanqian. Deux types de voix sont pris en charge.

Voix système : spécifiez directement un nom de voix. Valeurs disponibles : longanqian, longanlingxin, longanlingxi, longanxiaoxin, longanlufeng.

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

Voix clonées : créez une voix clonée à l'aide de l'API Voice Cloning (définissez target_model sur qwen-audio-3.0-realtime-plus ou qwen-audio-3.0-realtime-flash), puis transmettez la valeur voice_id retournée comme valeur du paramètre voice.

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

Modalités de sortie

Utilisez le paramètre modalities pour contrôler les types de sortie du modèle :

• ["audio", "text"] (par défaut) : génère à la fois la parole et le texte.
• ["text"] : génère uniquement du texte, sans parole. Adapté pour le débogage, la journalisation ou les scénarios ne nécessitant que des réponses textuelles.

Paramétrage au niveau de la session :

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

Remplacement par réponse : utilisez le champ response.modalities dans response.create pour remplacer le paramètre de modalité pour une seule réponse.

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

Configuration du VAD

En mode server_vad, configurez les paramètres suivants dans l'objet session.turn_detection pour ajuster le comportement du VAD (ces paramètres n'ont aucun effet en mode smart_turn) :

Contrôle des tours de l'historique

Utilisez le paramètre max_history_turns pour contrôler le nombre de tours de questions-réponses historiques auxquels le modèle se réfère lors de l'inférence. Des valeurs plus élevées permettent au modèle d'examiner un historique de conversation plus large pour une meilleure compréhension du contexte, mais augmentent la consommation de tokens et la latence d'inférence.

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

Plage valide pour max_history_turns : 1-50. Par défaut : 20.

Conseils d'ajustement :

• Conversations courtes (comme des Q&R rapides) : définissez une valeur plus basse (par exemple, 5-10) pour réduire la latence.
• Conversations longues (comme un service client multi-tours) : définissez une valeur plus élevée (par exemple, 30-50) pour aider le modèle à comprendre l'ensemble du contexte.

Fonctionnalités avancées

Function Calling

Qwen-Audio prend en charge le Function Calling, ce qui permet au modèle de décider quand invoquer des outils externes en fonction du contexte de la conversation.

Configurez les tools via 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"]
                }
            }
        }]
    }
}

Lorsque le modèle décide d'appeler un outil, le serveur envoie la séquence d'événements suivante :

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

Après avoir reçu response.function_call_arguments.done, exécutez l'outil côté client et renvoyez le résultat via conversation.item.create :

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

Après avoir renvoyé le résultat de l'outil, envoyez response.create pour que le modèle génère une réponse basée sur ce résultat :

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

L'exemple suivant intègre la prise en charge du Function Calling en plus du script realtime_demo.py du démarrage rapide. Assurez-vous que B64PCMPlayer.py se trouve dans le même répertoire avant l'exécution.

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

Exécutez python realtime_fc_demo.py et parlez dans votre microphone pour tester des conversations en temps réel avec Function Calling. Par exemple, demandez « Quel temps fait-il à Hangzhou ? » ou « Combien coûte un billet de train de Pékin à Shanghai ? », et le modèle invoquera automatiquement l'outil correspondant pour répondre avec le résultat.

Gestion du contexte de conversation

Qwen-Audio vous permet de gérer les éléments de conversation dans le contexte via des événements client. Utilisez cette fonctionnalité pour injecter un historique, ajouter des informations textuelles ou supprimer des éléments de conversation non pertinents.

• Créer un élément de conversation (conversation.item.create) : insère un élément de conversation dans le contexte. Trois valeurs item.type sont prises en charge :

  • message : un message de conversation standard. Spécifiez le role (system, user ou assistant) et un tableau content. Utilisez ceci pour injecter un historique de conversation ou des instructions système.
  • function_call : une demande d'appel de fonction. Spécifiez call_id, name et arguments (chaîne JSON). Généralement créé par le serveur, mais le client peut aussi l'utiliser pour injecter des historiques d'appels de fonction.
  • function_call_output : un résultat d'exécution d'outil. Spécifiez call_id et output (chaîne JSON). Après avoir reçu un function_call, exécutez l'outil côté client et renvoyez le résultat avec ce type.

  Le paramètre optionnel previous_item_id spécifie l'élément de conversation existant après lequel insérer le nouvel élément. Cela permet d'insérer du contenu à n'importe quelle position dans l'historique de la conversation. S'il est omis, le nouvel élément est ajouté à la fin.

  • Insérer un message utilisateur à une position spécifique :

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

  • Renvoyer un résultat de Function Calling :

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

  RemarqueSi l'identifiant item.id spécifié dans conversation.item.create existe déjà dans la conversation, une erreur est renvoyée.

• Récupérer un élément de conversation (conversation.item.retrieve) : interroge un élément de conversation stocké sur le serveur. Pour le contenu de type audio, seul le texte de transcription est renvoyé, et non les données audio brutes.

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

• Supprimer un élément de conversation (conversation.item.delete) : retire un élément spécifique du contexte de conversation.

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

Transcription audio ambiante

Mode smart_turn uniquement. Lorsque le VAD détecte une activité vocale mais que l'analyse sémantique détermine qu'il ne s'agit pas d'un tour de parole valide (comme du bruit ou des sons d'hésitation tels que « euh » ou « hmm »), le serveur ne déclenche pas de tour de conversation. À la place, il envoie le résultat ASR au client sous forme d'événement ambient_audio_transcription. Cette transcription n'est pas écrite dans le contexte de la conversation.

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

Comme les événements de transcription de la parole utilisateur, la transcription audio ambiante comprend des phases delta et completed. Utilisez cet événement pour mettre en œuvre une surveillance audio ambiante ou une conscience de la scène de conversation.

Amélioration du locuteur

Mode smart_turn uniquement. Transmettez des URL d'audio préenregistré de l'utilisateur cible dans session.update. Le modèle se verrouillera sur ce locuteur lors des conversations duplex, ignorant efficacement les autres voix et le bruit ambiant, ce qui permet des interactions duplex fluides dans des environnements ouverts.

Configuration : transmettez des URL d'audio d'empreinte vocale accessibles publiquement dans turn_detection.voiceprint_audio_urls lors du premier appel session.update.

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

Exigences relatives aux paramètres :

• Jusqu'à 5 URL. L'audio doit être au format PCM 16 kHz ou WAV.
• Ce paramètre ne prend effet que lors du premier appel session.update. Le champ est ignoré dans les appels suivants.

Événements d'enregistrement : après réception de la configuration, le serveur effectue de manière asynchrone l'enregistrement de l'empreinte vocale et notifie le résultat via les événements suivants :

• voiceprint_audio_list.in_progress : l'enregistrement a commencé. Envoyé avant session.updated, portant l'identifiant item_id.
• voiceprint_audio_list.completed : l'enregistrement a réussi. L'identifiant item_id correspond à celui de in_progress.
• voiceprint_audio_list.failed : l'enregistrement a échoué, avec un champ reason décrivant l'erreur (par exemple, l'URL audio n'est pas accessible). Un échec d'enregistrement ne bloque pas la conversation en cours.

Mise en production

Configurer la tolérance aux pannes

• Reconnexion client : mettez en œuvre une reconnexion automatique pour gérer l'instabilité du réseau. Définissez un signal de reconnexion dans le callback on_error et utilisez un backoff exponentiel (par exemple, attendre 1s, 2s, 4s) pour les nouvelles tentatives.
• Classification des erreurs : les erreurs client (invalid_request_error) ne déconnectent pas la session ; journalisez-les ou ajustez les paramètres. Les erreurs serveur (server_error) interrompent la connexion et nécessitent une reconnexion.
• Gestion des interruptions : en mode server_vad / smart_turn, toute nouvelle parole de l'utilisateur interrompt automatiquement la réponse en cours du modèle (response.done renvoie status=cancelled). Lors de la réception de input_audio_buffer.speech_started, videz immédiatement le tampon de lecture local pour éviter le chevauchement audio.

Cycle de vie de la connexion

Une session WebSocket typique suit le cycle de vie suivant :

1. Connexion : le client initie une connexion WebSocket et le serveur renvoie un événement session.created.
2. Configuration : le client envoie session.update pour définir le mode d'interaction, la voix, les outils et d'autres paramètres. Effectuez cette étape avant d'envoyer tout flux audio.
3. Interaction : le client diffuse en continu l'audio (input_audio_buffer.append). Le serveur effectue l'inférence en fonction de la détection VAD ou des déclenchements manuels, et renvoie la parole et le texte en streaming.
4. Fermeture : le client ferme la connexion WebSocket. Le serveur peut également se déconnecter si la connexion reste inactive trop longtemps.

Optimisation de la latence

• Taille des chunks audio : envoyez environ 100 ms de données audio par chunk (16 kHz x 16 bits x mono = 3 200 octets par chunk). Cela permet d'équilibrer les performances en temps réel et l'efficacité réseau.
• Lecture en streaming : commencez à lire l'audio dès que response.audio.delta arrive. N'attendez pas response.done pour lire la réponse complète.
• Vider le tampon lors d'une interruption : lorsque input_audio_buffer.speech_started est reçu, videz immédiatement le tampon de lecture local pour empêcher la poursuite de la lecture d'un audio obsolète.

Modèles et régions pris en charge

Chine (Pékin)

Utilisez une clé API de la région de Pékin lors de l'appel des modèles suivants :

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

Référence API

• API WebSocket
• Événements client
• Événements serveur
• SDK Client AOQ
• Présentation de l'API Realtime (description du protocole WebRTC)