Qwen-Audio real-time voice model

# 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")
# Singapore (ap-southeast-1). For another region, replace ap-southeast-1 in the URL.
# API keys are region-specific. See https://docs.qwencloud.com/developer-guides/getting-started/regions
API_URL = "wss://ap-southeast-1.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")

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_v2"
    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_v2"
            }
            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(
        # Singapore (ap-southeast-1). For another region, replace ap-southeast-1 in the URL.
        # API keys are region-specific. See https://docs.qwencloud.com/developer-guides/getting-started/regions
        base_url="wss://ap-southeast-1.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.")