import asyncio
import base64
import json
import logging
import os
import time
from typing import Any, Callable, Dict, Optional
import numpy as np
import soxr
import websockets
from dotenv import load_dotenv
from agentduet import (
BufferFullError,
Call,
CallAudioConfig,
CallClosedError,
IncomingCallNotification,
SessionManager,
SessionManagerConfig,
new_session_id,
)
load_dotenv()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
REGION = os.getenv("DASHSCOPE_REGION", "intl")
BASE_DOMAIN = (
"dashscope-intl.aliyuncs.com" if REGION == "intl" else "dashscope.aliyuncs.com"
)
QWEN_WS_URL = f"wss://{BASE_DOMAIN}/api-ws/v1/realtime"
MODEL = "qwen3.5-omni-flash-realtime"
class QwenRealtimeClient:
"""Minimal Qwen-Omni Realtime WebSocket client."""
def __init__(
self,
url: str,
api_key: str,
model: str,
voice: str = "Jennifer",
instructions: str = "You are a helpful and respectful AI assistant.",
on_audio_delta: Optional[Callable[[bytes], None]] = None,
on_interruption: Optional[Callable[[], Any]] = None,
):
self.url = f"{url}?model={model}"
self.api_key = api_key
self.voice = voice
self.instructions = instructions
self.on_audio_delta = on_audio_delta
self.on_interruption = on_interruption
self.ws = None
self._is_responding = False
async def connect(self):
headers = {"Authorization": f"Bearer {self.api_key}"}
self.ws = await websockets.connect(self.url, additional_headers=headers)
logger.info("Connected to Qwen-Omni WebSocket")
await self.send_event(
{
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": self.voice,
"instructions": self.instructions,
"input_audio_format": "pcm16", # 16-bit 16 kHz mono
"output_audio_format": "pcm24", # 16-bit 24 kHz mono
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500,
},
"input_audio_transcription": {"model": "gummy-realtime-v1"},
},
}
)
async def send_event(self, event: Dict[str, Any]):
if "event_id" not in event:
event["event_id"] = f"evt_{int(time.time() * 1000)}"
await self.ws.send(json.dumps(event))
async def stream_audio(self, audio_chunk: bytes):
await self.send_event(
{
"type": "input_audio_buffer.append",
"audio": base64.b64encode(audio_chunk).decode(),
}
)
async def cancel_response(self):
if self._is_responding:
logger.info("Sending response.cancel to Qwen")
await self.send_event({"type": "response.cancel"})
self._is_responding = False
async def receive_loop(self):
try:
async for message in self.ws:
event = json.loads(message)
event_type = event.get("type")
if event_type == "response.audio.delta":
audio_bytes = base64.b64decode(event["delta"])
if self.on_audio_delta:
self.on_audio_delta(audio_bytes)
elif event_type == "input_audio_buffer.speech_started":
logger.debug("Speech start - possible interruption")
if self.on_interruption:
await self.on_interruption()
elif event_type == "response.created":
self._is_responding = True
elif event_type == "response.done":
self._is_responding = False
elif event_type == "error":
logger.error("Qwen error: %s", event.get("error"))
elif event_type == "conversation.item.input_audio_transcription.completed":
logger.info("User: %s", event.get("transcript"))
elif event_type == "response.audio_transcript.done":
logger.info("AI: %s", event.get("transcript"))
except websockets.exceptions.ConnectionClosed:
logger.info("Qwen WebSocket closed")
except Exception:
logger.exception("Error in Qwen receive loop")
async def close(self):
if self.ws:
await self.ws.close()
class QwenRealtimeIntegration:
def __init__(self, call: Call, qwen_client: QwenRealtimeClient):
self._call = call
self._qwen_client = qwen_client
self._stream_task: Optional[asyncio.Task] = None
self._receive_task: Optional[asyncio.Task] = None
self._terminated = False
async def _on_hangup(self, evt):
logger.info("Call terminated - cleaning up Qwen")
self._terminated = True
await self._qwen_client.close()
if self._stream_task:
self._stream_task.cancel()
if self._receive_task:
self._receive_task.cancel()
async def handle_interruption(self):
logger.info("Interruption - cancel Qwen response and clear send buffer")
await self._qwen_client.cancel_response()
await self._call.clear_send_audio_buffer()
async def run(self):
self._call.on_hangup(self._on_hangup)
await self._qwen_client.connect()
self._stream_task = asyncio.create_task(self.stream_to_qwen())
self._receive_task = asyncio.create_task(self._qwen_client.receive_loop())
await asyncio.gather(
self._stream_task, self._receive_task, return_exceptions=True
)
async def stream_to_qwen(self):
try:
async for audio_chunk in self._call.caller.audio_stream():
if self._terminated:
break
resampled = self.downsample_24to16(audio_chunk)
await self._qwen_client.stream_audio(resampled)
except CallClosedError:
logger.debug("Call closed; stopping stream to Qwen")
except Exception:
logger.exception("Error in stream to Qwen")
def downsample_24to16(self, audio_data: bytes) -> bytes:
if not audio_data:
return b""
samples = np.frombuffer(audio_data, dtype=np.int16)
resampled = soxr.resample(samples, 24000, 16000)
return resampled.astype(np.int16).tobytes()
def on_qwen_audio(self, audio_bytes: bytes):
if self._terminated:
return
asyncio.create_task(self._send_audio_to_call(audio_bytes))
async def _send_audio_to_call(self, audio_bytes: bytes):
try:
await self._call.send_audio(audio_bytes)
except BufferFullError:
logger.warning("Call audio buffer full, dropping chunk")
except CallClosedError:
logger.debug("Call closed; dropping Qwen audio")
except Exception:
logger.exception("Error sending audio to call")
async def main():
config = SessionManagerConfig.create(
api_key=os.getenv("AGENTDUET_API_KEY"),
connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
call_audio=CallAudioConfig(
sample_rate=24000,
buffer_size=8 * 1024 * 1024,
),
)
async with SessionManager(config) as sm:
logger.info("Connected. Waiting for calls...")
@sm.on_incoming_call
async def on_call(noti: IncomingCallNotification):
session = await sm.open_session(new_session_id(), noti.subscriber)
call = await session.process_call(noti)
logger.info("Incoming call: %s", call.id)
try:
integration: Optional[QwenRealtimeIntegration] = None
async def on_interruption_handler():
if integration:
await integration.handle_interruption()
qwen_client = QwenRealtimeClient(
url=QWEN_WS_URL,
api_key=os.getenv("DASHSCOPE_API_KEY"),
model=MODEL,
on_audio_delta=lambda data: (
integration.on_qwen_audio(data) if integration else None
),
on_interruption=on_interruption_handler,
)
integration = QwenRealtimeIntegration(call, qwen_client)
result = await call.answer()
if not result:
logger.error(
"Answer failed %s: %s (%s)",
call.id,
result.error_message,
result.error_code,
)
return
await integration.run()
except Exception:
logger.exception("Error in Qwen bridge")
await call.close()
await sm.run_forever()
if __name__ == "__main__":
asyncio.run(main())
