Qwen Omni

1
import asyncio
2
import base64
3
import json
4
import logging
5
import os
6
import time
7
from typing import Any, Callable, Dict, Optional
8
9
import numpy as np
10
import soxr
11
import websockets
12
from dotenv import load_dotenv
13
14
from agentduet import (
15
    BufferFullError,
16
    Call,
17
    CallAudioConfig,
18
    CallClosedError,
19
    IncomingCallNotification,
20
    SessionManager,
21
    SessionManagerConfig,
22
    new_session_id,
23
)
24
25
load_dotenv()
26
27
logging.basicConfig(
28
    level=logging.INFO,
29
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
30
)
31
logger = logging.getLogger(__name__)
32
33
REGION = os.getenv("DASHSCOPE_REGION", "intl")
34
BASE_DOMAIN = (
35
    "dashscope-intl.aliyuncs.com" if REGION == "intl" else "dashscope.aliyuncs.com"
36
)
37
QWEN_WS_URL = f"wss://{BASE_DOMAIN}/api-ws/v1/realtime"
38
MODEL = "qwen3.5-omni-flash-realtime"
39
40
41
class QwenRealtimeClient:
42
    """Minimal Qwen-Omni Realtime WebSocket client."""
43
44
    def __init__(
45
        self,
46
        url: str,
47
        api_key: str,
48
        model: str,
49
        voice: str = "Jennifer",
50
        instructions: str = "You are a helpful and respectful AI assistant.",
51
        on_audio_delta: Optional[Callable[[bytes], None]] = None,
52
        on_interruption: Optional[Callable[[], Any]] = None,
53
    ):
54
        self.url = f"{url}?model={model}"
55
        self.api_key = api_key
56
        self.voice = voice
57
        self.instructions = instructions
58
        self.on_audio_delta = on_audio_delta
59
        self.on_interruption = on_interruption
60
        self.ws = None
61
        self._is_responding = False
62
63
    async def connect(self):
64
        headers = {"Authorization": f"Bearer {self.api_key}"}
65
        self.ws = await websockets.connect(self.url, additional_headers=headers)
66
        logger.info("Connected to Qwen-Omni WebSocket")
67
        await self.send_event(
68
            {
69
                "type": "session.update",
70
                "session": {
71
                    "modalities": ["text", "audio"],
72
                    "voice": self.voice,
73
                    "instructions": self.instructions,
74
                    "input_audio_format": "pcm16",  # 16-bit 16 kHz mono
75
                    "output_audio_format": "pcm24",  # 16-bit 24 kHz mono
76
                    "turn_detection": {
77
                        "type": "server_vad",
78
                        "threshold": 0.5,
79
                        "prefix_padding_ms": 300,
80
                        "silence_duration_ms": 500,
81
                    },
82
                    "input_audio_transcription": {"model": "gummy-realtime-v1"},
83
                },
84
            }
85
        )
86
87
    async def send_event(self, event: Dict[str, Any]):
88
        if "event_id" not in event:
89
            event["event_id"] = f"evt_{int(time.time() * 1000)}"
90
        await self.ws.send(json.dumps(event))
91
92
    async def stream_audio(self, audio_chunk: bytes):
93
        await self.send_event(
94
            {
95
                "type": "input_audio_buffer.append",
96
                "audio": base64.b64encode(audio_chunk).decode(),
97
            }
98
        )
99
100
    async def cancel_response(self):
101
        if self._is_responding:
102
            logger.info("Sending response.cancel to Qwen")
103
            await self.send_event({"type": "response.cancel"})
104
            self._is_responding = False
105
106
    async def receive_loop(self):
107
        try:
108
            async for message in self.ws:
109
                event = json.loads(message)
110
                event_type = event.get("type")
111
112
                if event_type == "response.audio.delta":
113
                    audio_bytes = base64.b64decode(event["delta"])
114
                    if self.on_audio_delta:
115
                        self.on_audio_delta(audio_bytes)
116
117
                elif event_type == "input_audio_buffer.speech_started":
118
                    logger.debug("Speech start - possible interruption")
119
                    if self.on_interruption:
120
                        await self.on_interruption()
121
122
                elif event_type == "response.created":
123
                    self._is_responding = True
124
125
                elif event_type == "response.done":
126
                    self._is_responding = False
127
128
                elif event_type == "error":
129
                    logger.error("Qwen error: %s", event.get("error"))
130
131
                elif event_type == "conversation.item.input_audio_transcription.completed":
132
                    logger.info("User: %s", event.get("transcript"))
133
134
                elif event_type == "response.audio_transcript.done":
135
                    logger.info("AI: %s", event.get("transcript"))
136
137
        except websockets.exceptions.ConnectionClosed:
138
            logger.info("Qwen WebSocket closed")
139
        except Exception:
140
            logger.exception("Error in Qwen receive loop")
141
142
    async def close(self):
143
        if self.ws:
144
            await self.ws.close()
145
146
147
class QwenRealtimeIntegration:
148
    def __init__(self, call: Call, qwen_client: QwenRealtimeClient):
149
        self._call = call
150
        self._qwen_client = qwen_client
151
        self._stream_task: Optional[asyncio.Task] = None
152
        self._receive_task: Optional[asyncio.Task] = None
153
        self._terminated = False
154
155
    async def _on_hangup(self, evt):
156
        logger.info("Call terminated - cleaning up Qwen")
157
        self._terminated = True
158
        await self._qwen_client.close()
159
        if self._stream_task:
160
            self._stream_task.cancel()
161
        if self._receive_task:
162
            self._receive_task.cancel()
163
164
    async def handle_interruption(self):
165
        logger.info("Interruption - cancel Qwen response and clear send buffer")
166
        await self._qwen_client.cancel_response()
167
        await self._call.clear_send_audio_buffer()
168
169
    async def run(self):
170
        self._call.on_hangup(self._on_hangup)
171
        await self._qwen_client.connect()
172
        self._stream_task = asyncio.create_task(self.stream_to_qwen())
173
        self._receive_task = asyncio.create_task(self._qwen_client.receive_loop())
174
        await asyncio.gather(
175
            self._stream_task, self._receive_task, return_exceptions=True
176
        )
177
178
    async def stream_to_qwen(self):
179
        try:
180
            async for audio_chunk in self._call.caller.audio_stream():
181
                if self._terminated:
182
                    break
183
                resampled = self.downsample_24to16(audio_chunk)
184
                await self._qwen_client.stream_audio(resampled)
185
        except CallClosedError:
186
            logger.debug("Call closed; stopping stream to Qwen")
187
        except Exception:
188
            logger.exception("Error in stream to Qwen")
189
190
    def downsample_24to16(self, audio_data: bytes) -> bytes:
191
        if not audio_data:
192
            return b""
193
        samples = np.frombuffer(audio_data, dtype=np.int16)
194
        resampled = soxr.resample(samples, 24000, 16000)
195
        return resampled.astype(np.int16).tobytes()
196
197
    def on_qwen_audio(self, audio_bytes: bytes):
198
        if self._terminated:
199
            return
200
        asyncio.create_task(self._send_audio_to_call(audio_bytes))
201
202
    async def _send_audio_to_call(self, audio_bytes: bytes):
203
        try:
204
            await self._call.send_audio(audio_bytes)
205
        except BufferFullError:
206
            logger.warning("Call audio buffer full, dropping chunk")
207
        except CallClosedError:
208
            logger.debug("Call closed; dropping Qwen audio")
209
        except Exception:
210
            logger.exception("Error sending audio to call")
211
212
213
async def main():
214
    config = SessionManagerConfig.create(
215
        api_key=os.getenv("AGENTDUET_API_KEY"),
216
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
217
        call_audio=CallAudioConfig(
218
            sample_rate=24000,
219
            buffer_size=8 * 1024 * 1024,
220
        ),
221
    )
222
223
    async with SessionManager(config) as sm:
224
        logger.info("Connected. Waiting for calls...")
225
226
        @sm.on_incoming_call
227
        async def on_call(noti: IncomingCallNotification):
228
            session = await sm.open_session(new_session_id(), noti.subscriber)
229
            call = await session.process_call(noti)
230
            logger.info("Incoming call: %s", call.id)
231
            try:
232
                integration: Optional[QwenRealtimeIntegration] = None
233
234
                async def on_interruption_handler():
235
                    if integration:
236
                        await integration.handle_interruption()
237
238
                qwen_client = QwenRealtimeClient(
239
                    url=QWEN_WS_URL,
240
                    api_key=os.getenv("DASHSCOPE_API_KEY"),
241
                    model=MODEL,
242
                    on_audio_delta=lambda data: (
243
                        integration.on_qwen_audio(data) if integration else None
244
                    ),
245
                    on_interruption=on_interruption_handler,
246
                )
247
                integration = QwenRealtimeIntegration(call, qwen_client)
248
249
                result = await call.answer()
250
                if not result:
251
                    logger.error(
252
                        "Answer failed %s: %s (%s)",
253
                        call.id,
254
                        result.error_message,
255
                        result.error_code,
256
                    )
257
                    return
258
                await integration.run()
259
            except Exception:
260
                logger.exception("Error in Qwen bridge")
261
                await call.close()
262
263
        await sm.run_forever()
264
265
266
if __name__ == "__main__":
267
    asyncio.run(main())