import asyncio import base64 import json import logging import time from collections import deque from dataclasses import dataclass from typing import Any, AsyncIterator from inference.core.types import ( AudioChunk, ImageFrame, PluginConfig, ToolCall, ToolResult, VoiceLLMInputEvent, VoiceLLMOutputEvent, VoiceLLMSessionConfig, ) from inference.plugins.qwen_endpoint import dashscope_realtime_ws_url from inference.plugins.voice_llm.base import VoiceCheckError, VoiceLLMPlugin logger = logging.getLogger(__name__) _MAX_IMAGE_BYTES = 500 * 1024 class QwenOmniRealtimePlugin(VoiceLLMPlugin): """DashScope Qwen Omni realtime omni model plugin.""" name = "omni.qwen_omni" def __init__(self) -> None: self.api_key = "" self.model = "qwen3.5-omni-flash-realtime" self.ws_url = "" self.voice = "Tina" self.system_prompt = "" self.input_sample_rate = 16000 self.output_sample_rate = 24000 self.vad_type = "semantic_vad" self.vad_threshold = 0.5 self.vad_silence_duration_ms = 800 self.enable_search: bool | None = None self.search_options: dict[str, Any] | None = None self.temperature: float | None = None self.top_p: float | None = None self.top_k: int | None = None self.max_tokens: int | None = None self._active_ws: Any | None = None async def initialize(self, config: PluginConfig) -> None: params = config.params self.api_key = params.get("api_key", self.api_key) self.model = params.get("model", self.model) self.ws_url = dashscope_realtime_ws_url(self.model, "DASHSCOPE_OMNI_WS_URL") self.voice = params.get("voice", self.voice) self.system_prompt = params.get("system_prompt", self.system_prompt) self.input_sample_rate = int( params.get("input_sample_rate", self.input_sample_rate) ) self.output_sample_rate = int( params.get("output_sample_rate", self.output_sample_rate) ) self.vad_type = params.get("vad_type", self.vad_type) self.vad_threshold = float(params.get("vad_threshold", self.vad_threshold)) self.vad_silence_duration_ms = int( params.get("vad_silence_duration_ms", self.vad_silence_duration_ms) ) self.enable_search = self._optional_bool(params.get("enable_search")) search_options = params.get("search_options") if isinstance(search_options, dict): self.search_options = search_options self.temperature = self._optional_float(params.get("temperature")) self.top_p = self._optional_float(params.get("top_p")) self.top_k = self._optional_int(params.get("top_k")) self.max_tokens = self._optional_int(params.get("max_tokens")) async def check_voice( self, session_config: VoiceLLMSessionConfig | None = None, ) -> None: import websockets ws = await self._connect(websockets) try: await self._configure_session(ws, session_config or VoiceLLMSessionConfig()) except RuntimeError as exc: raise VoiceCheckError(str(exc)) from exc finally: await ws.close() async def converse_stream( self, input_stream: AsyncIterator[VoiceLLMInputEvent], session_config: VoiceLLMSessionConfig | None = None, ) -> AsyncIterator[VoiceLLMOutputEvent]: import websockets config = session_config or VoiceLLMSessionConfig() ws = await self._connect(websockets) self._active_ws = ws response_coordinator = _QwenResponseCoordinator() output_queue: asyncio.Queue[VoiceLLMOutputEvent | Exception | None] = ( asyncio.Queue() ) sender_task: asyncio.Task | None = None receiver_task: asyncio.Task | None = None try: await self._configure_session(ws, config) sender_task = asyncio.create_task( self._send_inputs( ws, input_stream, config.session_id, output_queue, response_coordinator, drain_responses_on_close=True, ) ) receiver_task = asyncio.create_task( self._receive_events( ws, config.session_id, output_queue, response_coordinator, defer_response=config.defer_response, ) ) while True: item = await output_queue.get() if item is None: break if isinstance(item, Exception): raise item yield item finally: for task in (sender_task, receiver_task): if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass if self._active_ws is ws: self._active_ws = None await ws.close() async def interrupt(self) -> None: ws = self._active_ws if ws is None: return for event_type in ("response.cancel", "input_audio_buffer.clear"): try: await self._send_json( ws, { "type": event_type, "event_id": self._event_id("qwen_omni", "interrupt"), }, ) except Exception: logger.debug("Failed to send Qwen Omni interrupt event", exc_info=True) async def _connect(self, websockets: Any): headers = {"Authorization": f"Bearer {self.api_key}"} try: return await websockets.connect( self.ws_url, additional_headers=headers, ) except TypeError: return await websockets.connect( self.ws_url, extra_headers=headers, ) async def _configure_session( self, ws: Any, session_config: VoiceLLMSessionConfig, ) -> None: await self._send_json( ws, { "type": "session.update", "event_id": self._event_id(session_config.session_id, "session"), "session": self._session_payload(session_config), }, ) while True: event = self._decode_message(await ws.recv()) event_type = event.get("type", "") if event_type in {"session.created", "session.updated"}: return if event_type == "error": raise RuntimeError(self._error_message(event)) async def _send_inputs( self, ws: Any, input_stream: AsyncIterator[VoiceLLMInputEvent], session_id: str, output_queue: asyncio.Queue[VoiceLLMOutputEvent | Exception | None], response_coordinator: "_QwenResponseCoordinator | None" = None, drain_responses_on_close: bool = False, ) -> None: response_coordinator = response_coordinator or _QwenResponseCoordinator() response_sender_task = asyncio.create_task( self._send_deferred_responses(ws, response_coordinator, output_queue) ) try: pending_image: ImageFrame | None = None has_sent_audio = False expects_deferred_response = False async for event in input_stream: if event.tool_result: await self._send_tool_result( ws, session_id, event.tool_result, response_coordinator, ) if not event.tool_result.suppress_response: expects_deferred_response = True continue if event.response_instructions is not None: expects_deferred_response = True await self._send_response_instructions( ws, session_id, event.response_instructions, response_coordinator, ) continue if event.text: expects_deferred_response = True await self._send_text(ws, session_id, event.text, response_coordinator) continue if event.audio: has_sent_audio = True await self._send_json( ws, { "type": "input_audio_buffer.append", "event_id": self._event_id(session_id, "audio"), "audio": base64.b64encode(event.audio).decode("ascii"), }, ) # Keep image strictly after an audio append. This avoids provider-side # ordering violations when a new turn starts and an image arrives first. if pending_image is not None: await self._send_image(ws, session_id, pending_image) pending_image = None if event.image is not None: if not self._valid_image(event.image): continue # Always buffer the latest valid frame and flush it only after # the next audio chunk is appended. pending_image = event.image if pending_image is not None and has_sent_audio: # If stream ends after image input, flush once to avoid dropping # the latest frame while still guaranteeing audio-first ordering. await self._send_image(ws, session_id, pending_image) if drain_responses_on_close and expects_deferred_response: await response_coordinator.wait_all_responses_done(timeout=60.0) except Exception as exc: await output_queue.put(exc) finally: await response_coordinator.close() if not response_sender_task.done(): try: await response_sender_task except asyncio.CancelledError: pass try: await ws.close() except Exception: pass async def _send_image(self, ws: Any, session_id: str, image: ImageFrame) -> None: await self._send_json( ws, { "type": "input_image_buffer.append", "event_id": self._event_id(session_id, "image"), "image": base64.b64encode(image.data).decode("ascii"), }, ) async def _send_text( self, ws: Any, session_id: str, text: str, response_coordinator: "_QwenResponseCoordinator", ) -> None: await response_coordinator.enqueue( _QwenDeferredResponse( item_payload={ "type": "conversation.item.create", "event_id": self._event_id(session_id, "text"), "item": { "type": "message", "role": "user", "content": [ { "type": "input_text", "text": text, } ], }, }, response_payload={ "type": "response.create", "event_id": self._event_id(session_id, "text_response"), "response": {"modalities": ["text", "audio"]}, }, ) ) async def _send_response_instructions( self, ws: Any, session_id: str, instructions: str, response_coordinator: "_QwenResponseCoordinator", ) -> None: response: dict[str, Any] = {"modalities": ["text", "audio"]} instructions = str(instructions or "").strip() if instructions: response["instructions"] = instructions await response_coordinator.enqueue( _QwenDeferredResponse( response_payload={ "type": "response.create", "event_id": self._event_id(session_id, "response"), "response": response, } ) ) async def _send_tool_result( self, ws: Any, session_id: str, result: ToolResult, response_coordinator: "_QwenResponseCoordinator", ) -> None: await self._send_json( ws, { "type": "conversation.item.create", "event_id": self._event_id(session_id, "tool_result"), "item": { "type": "function_call_output", "call_id": result.id, "output": json.dumps(result.result, ensure_ascii=False), }, }, ) if result.suppress_response: return await response_coordinator.enqueue( _QwenDeferredResponse( response_payload={ "type": "response.create", "event_id": self._event_id(session_id, "tool_response"), } ) ) async def _send_deferred_responses( self, ws: Any, response_coordinator: "_QwenResponseCoordinator", output_queue: asyncio.Queue[VoiceLLMOutputEvent | Exception | None], ) -> None: try: while True: request = await response_coordinator.next_request() if request is None: return await self._send_deferred_response(ws, response_coordinator, request) except asyncio.CancelledError: raise except Exception as exc: await output_queue.put(exc) async def _send_deferred_response( self, ws: Any, response_coordinator: "_QwenResponseCoordinator", request: "_QwenDeferredResponse", ) -> None: if not await response_coordinator.wait_idle(): return if request.item_payload is not None and not request.item_sent: await self._send_json(ws, request.item_payload) request.item_sent = True await response_coordinator.begin_client_response(request) try: await self._send_json(ws, request.response_payload) except Exception: await response_coordinator.release_client_response(request) raise @staticmethod def _valid_image(image: ImageFrame) -> bool: mime_type = (image.mime_type or "").lower() if mime_type and mime_type not in {"image/jpeg", "image/jpg"}: return False data = image.data or b"" if len(data) == 0 or len(data) > _MAX_IMAGE_BYTES: return False return len(data) >= 3 and data[0] == 0xFF and data[1] == 0xD8 and data[2] == 0xFF async def _receive_events( self, ws: Any, session_id: str, output_queue: asyncio.Queue[VoiceLLMOutputEvent | Exception | None], response_coordinator: "_QwenResponseCoordinator | None" = None, defer_response: bool = False, ) -> None: response_coordinator = response_coordinator or _QwenResponseCoordinator() turn_state = _QwenTurnState(session_id=session_id or "qwen_omni") tool_arg_parts: dict[str, str] = {} emitted_tool_calls: set[str] = set() try: async for message in ws: event = self._decode_message(message) self._log_server_event(session_id, event) event_type = event.get("type", "") if event_type == "error": message = self._error_message(event) if self._is_active_response_error(message): logger.info( "qwen_omni deferred response delayed by active response session=%s", session_id or "qwen_omni", ) await response_coordinator.mark_active_response_error() continue raise RuntimeError(message) if event_type in {"session.created", "session.updated"}: continue if event_type == "response.function_call_arguments.delta": call_id = str(event.get("call_id") or event.get("item_id") or "") if call_id: tool_arg_parts[call_id] = tool_arg_parts.get(call_id, "") + str(event.get("delta", "") or "") continue if event_type == "response.function_call_arguments.done": call = self._tool_call_from_event(event, tool_arg_parts) if call and call.id not in emitted_tool_calls: emitted_tool_calls.add(call.id) await output_queue.put( VoiceLLMOutputEvent( tool_calls=[call], question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "response.output_item.done": item = event.get("item") if isinstance(item, dict) and item.get("type") == "function_call": call = self._tool_call_from_event(item, tool_arg_parts) if call and call.id not in emitted_tool_calls: emitted_tool_calls.add(call.id) await output_queue.put( VoiceLLMOutputEvent( tool_calls=[call], question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "input_audio_buffer.speech_started": if not defer_response: await response_coordinator.mark_response_started() turn_state.start_next_turn() await output_queue.put( VoiceLLMOutputEvent( barge_in=True, question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "response.created": await response_coordinator.mark_response_started() response = event.get("response") if isinstance(response, dict): response_id = str(response.get("id", "") or "") if response_id and not turn_state.question_id: turn_state.start_next_turn() turn_state.reply_id = response_id continue if event_type == "conversation.item.input_audio_transcription.completed": turn_state.ensure_turn() transcript = str(event.get("transcript", "") or "").strip() if transcript: await output_queue.put( VoiceLLMOutputEvent( user_transcript=transcript, question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "response.audio_transcript.delta": turn_state.ensure_turn() delta = str(event.get("delta", "") or "") if delta: turn_state.assistant_text += delta await output_queue.put( VoiceLLMOutputEvent( transcript=delta, question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "response.audio_transcript.done": transcript = str(event.get("transcript", "") or "") if transcript: turn_state.assistant_text = transcript continue if event_type == "response.audio.delta": turn_state.ensure_turn() delta = str(event.get("delta", "") or "") if not delta: continue audio_payload = base64.b64decode(delta) if audio_payload: turn_state.has_audio = True await output_queue.put( VoiceLLMOutputEvent( audio=AudioChunk( data=audio_payload, sample_rate=self.output_sample_rate, channels=1, format="pcm_s16le", ), question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) continue if event_type == "response.done": if turn_state.has_content: await output_queue.put( VoiceLLMOutputEvent( audio=AudioChunk( data=b"", sample_rate=self.output_sample_rate, channels=1, format="pcm_s16le", is_final=True, ) if turn_state.has_audio else None, transcript=turn_state.assistant_text, is_final=True, question_id=turn_state.question_id, reply_id=turn_state.reply_id, ) ) turn_state.reset() await response_coordinator.mark_response_done() continue except Exception as exc: if not getattr(ws, "closed", False): await output_queue.put(exc) finally: await output_queue.put(None) def _session_payload(self, session_config: VoiceLLMSessionConfig) -> dict[str, Any]: payload: dict[str, Any] = { "modalities": ["text", "audio"], "voice": session_config.voice or self.voice, "input_audio_format": "pcm", "output_audio_format": "pcm", "instructions": self._instructions(session_config), "turn_detection": { "type": self.vad_type, "threshold": self.vad_threshold, "silence_duration_ms": self.vad_silence_duration_ms, }, } if session_config.defer_response: payload["turn_detection"]["create_response"] = False has_tools = bool(session_config.tools) optional_values: dict[str, Any] = { "temperature": self.temperature, "top_p": self.top_p, "top_k": self.top_k, "max_tokens": self.max_tokens, } if not has_tools: optional_values["enable_search"] = self.enable_search optional_values["search_options"] = self.search_options for key, value in optional_values.items(): if value is not None: payload[key] = value if has_tools: payload["tools"] = [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.parameters or {"type": "object", "properties": {}}, }, } for tool in session_config.tools ] return payload def _instructions(self, session_config: VoiceLLMSessionConfig) -> str: parts: list[str] = [] if session_config.bot_name: parts.append(f"名字：{session_config.bot_name}") parts.append(session_config.system_prompt or self.system_prompt) if session_config.speaking_style: parts.append(f"说话风格：{session_config.speaking_style}") if session_config.dialog_context: parts.append("以下是最近的对话上下文，请在回答时保持连续性：") for item in session_config.dialog_context: role = "用户" if item.role == "user" else "助手" parts.append(f"{role}：{item.text}") return "\n".join(part for part in parts if part.strip()) @staticmethod def _tool_call_from_event(event: dict[str, Any], arg_parts: dict[str, str]) -> ToolCall | None: call_id = str(event.get("call_id") or event.get("id") or event.get("item_id") or "") name = str(event.get("name") or "") raw_args = event.get("arguments") if raw_args is None and call_id: raw_args = arg_parts.get(call_id, "") if not call_id or not name: return None return ToolCall( id=call_id, name=name, arguments=QwenOmniRealtimePlugin._parse_tool_arguments(raw_args), ) @staticmethod def _parse_tool_arguments(raw: Any) -> dict[str, Any]: if isinstance(raw, dict): return raw if raw is None: return {} try: parsed = json.loads(str(raw)) except json.JSONDecodeError: return {} return parsed if isinstance(parsed, dict) else {} @staticmethod def _clip_text(value: Any, limit: int = 180) -> str: text = str(value or "") if len(text) <= limit: return text return text[:limit] + "..." @classmethod def _server_event_log_fields(cls, event: dict[str, Any]) -> dict[str, Any]: event_type = str(event.get("type") or "") fields: dict[str, Any] = {} for key in ("response_id", "item_id", "call_id", "name", "output_index"): if key in event and event.get(key) not in (None, ""): fields[key] = event.get(key) if event_type == "response.audio.delta": fields["audio_delta_b64_len"] = len(str(event.get("delta") or "")) elif event_type in {"response.audio_transcript.delta", "response.function_call_arguments.delta"}: fields["delta"] = cls._clip_text(event.get("delta")) if "transcript" in event: fields["transcript"] = cls._clip_text(event.get("transcript")) if "arguments" in event: fields["arguments"] = cls._clip_text(event.get("arguments")) item = event.get("item") if isinstance(item, dict): item_fields = { key: item.get(key) for key in ("type", "id", "call_id", "name") if item.get(key) not in (None, "") } if "arguments" in item: item_fields["arguments"] = cls._clip_text(item.get("arguments")) fields["item"] = item_fields response = event.get("response") if isinstance(response, dict): fields["response"] = { key: response.get(key) for key in ("id", "status") if response.get(key) not in (None, "") } error = event.get("error") if error: fields["error"] = cls._clip_text(error) return fields @classmethod def _server_event_log_level(cls, event: dict[str, Any]) -> int: event_type = str(event.get("type") or "") if event_type == "error": return logging.ERROR if event_type in { "session.created", "session.updated", "input_audio_buffer.speech_started", "conversation.item.input_audio_transcription.completed", "response.created", "response.audio_transcript.done", "response.function_call_arguments.done", "response.done", }: return logging.INFO return logging.DEBUG @classmethod def _log_server_event(cls, session_id: str, event: dict[str, Any]) -> None: event_type = str(event.get("type") or "unknown") level = cls._server_event_log_level(event) if not logger.isEnabledFor(level): return fields = cls._server_event_log_fields(event) logger.log( level, "qwen_omni model event session=%s type=%s fields=%s", session_id or "qwen_omni", event_type, json.dumps(fields, ensure_ascii=False, sort_keys=True), ) @staticmethod async def _send_json(ws: Any, payload: dict[str, Any]) -> None: await ws.send(json.dumps(payload, ensure_ascii=False)) @staticmethod def _decode_message(message: str | bytes) -> dict[str, Any]: if isinstance(message, bytes): message = message.decode("utf-8") return json.loads(message) @staticmethod def _event_id(session_id: str, suffix: str) -> str: base = session_id or "qwen_omni" return f"{base}_{suffix}_{int(time.time() * 1000)}" @staticmethod def _error_message(event: dict[str, Any]) -> str: error = event.get("error") if isinstance(error, dict): message = error.get("message") or error.get("msg") or error.get("code") if message: return str(message) if isinstance(error, str): return error return f"Qwen Omni error: {event}" @staticmethod def _is_active_response_error(message: str) -> bool: return "Conversation already has an active response" in message @staticmethod def _optional_bool(value: Any) -> bool | None: if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"true", "1", "yes"}: return True if normalized in {"false", "0", "no"}: return False return None @staticmethod def _optional_float(value: Any) -> float | None: if value is None: return None try: return float(value) except (TypeError, ValueError): return None @staticmethod def _optional_int(value: Any) -> int | None: if value is None: return None try: return int(value) except (TypeError, ValueError): return None async def shutdown(self) -> None: if self._active_ws is not None: await self._active_ws.close() self._active_ws = None @dataclass class _QwenDeferredResponse: response_payload: dict[str, Any] item_payload: dict[str, Any] | None = None item_sent: bool = False class _QwenResponseCoordinator: def __init__(self) -> None: self._pending: deque[_QwenDeferredResponse] = deque() self._pending_condition = asyncio.Condition() self._state_condition = asyncio.Condition() self._idle = True self._closed = False self._current_response: _QwenDeferredResponse | None = None async def enqueue(self, request: _QwenDeferredResponse) -> None: async with self._pending_condition: if self._closed: return self._pending.append(request) self._pending_condition.notify() async def _prepend(self, request: _QwenDeferredResponse) -> None: async with self._pending_condition: if self._closed: return self._pending.appendleft(request) self._pending_condition.notify() async def next_request(self) -> _QwenDeferredResponse | None: async with self._pending_condition: while not self._pending and not self._closed: await self._pending_condition.wait() if self._pending: return self._pending.popleft() return None async def wait_idle(self) -> bool: async with self._state_condition: while not self._idle and not self._closed: await self._state_condition.wait() return self._idle async def begin_client_response(self, request: _QwenDeferredResponse) -> None: async with self._state_condition: self._idle = False self._current_response = request self._state_condition.notify_all() async def release_client_response(self, request: _QwenDeferredResponse) -> None: async with self._state_condition: if self._current_response is request: self._current_response = None self._idle = True self._state_condition.notify_all() async def mark_response_started(self) -> None: async with self._state_condition: self._idle = False self._state_condition.notify_all() async def mark_response_done(self) -> None: async with self._state_condition: self._idle = True self._current_response = None self._state_condition.notify_all() async with self._pending_condition: self._pending_condition.notify_all() async def mark_active_response_error(self) -> None: retry: _QwenDeferredResponse | None = None async with self._state_condition: if self._current_response is not None: retry = self._current_response retry.item_sent = True self._current_response = None self._idle = False self._state_condition.notify_all() if retry is not None: await self._prepend(retry) async def close(self) -> None: async with self._pending_condition: self._closed = True self._pending_condition.notify_all() async with self._state_condition: self._idle = True self._state_condition.notify_all() async def wait_all_responses_done(self, timeout: float) -> None: async def _wait() -> None: while True: async with self._pending_condition: has_pending = bool(self._pending) closed = self._closed async with self._state_condition: idle = self._idle and self._current_response is None if closed or (not has_pending and idle): return await asyncio.sleep(0.01) try: await asyncio.wait_for(_wait(), timeout=timeout) except asyncio.TimeoutError: logger.warning("qwen_omni timed out waiting for deferred response completion") class _QwenTurnState: def __init__(self, session_id: str) -> None: self.session_id = session_id self.turn_index = 0 self.question_id = "" self.reply_id = "" self.assistant_text = "" self.has_audio = False @property def has_content(self) -> bool: return self.has_audio or bool(self.assistant_text) def ensure_turn(self) -> None: if not self.question_id: self.start_next_turn() def start_next_turn(self) -> None: self.turn_index += 1 self.question_id = f"{self.session_id}_q{self.turn_index}" self.reply_id = f"{self.session_id}_r{self.turn_index}" self.assistant_text = "" self.has_audio = False def reset(self) -> None: self.question_id = "" self.reply_id = "" self.assistant_text = "" self.has_audio = False