diff --git a/backend/README.md b/backend/README.md index 0ac434b..20c43b3 100644 --- a/backend/README.md +++ b/backend/README.md @@ -184,5 +184,6 @@ HTTP 工具可通过“响应变量赋值”把 JSON 路径写回普通会话变 - [ ] 联调 Pipecat 1.3.0 语音链路与各 OpenAI 兼容服务 - [ ] 起本地 SenseVoice / CosyVoice 的 OpenAI 兼容服务 - [x] `realtime` 模式(StepFun StepAudio Realtime) +- [x] `realtime` 模式(Alibaba Cloud Qwen-Audio 3.0 Realtime MVP) - [x] 前端 `DebugVoicePanel` 接 `/ws/voice`(参考 dograh `useWebSocketRTC.tsx`) - [ ] 加 DB 后:助手配置入库(目前随请求内联) diff --git a/backend/db/seed_model_resources.sql b/backend/db/seed_model_resources.sql index 8c1f725..89d7cd6 100644 --- a/backend/db/seed_model_resources.sql +++ b/backend/db/seed_model_resources.sql @@ -50,6 +50,9 @@ VALUES '{"apiKey":"replace-me"}', FALSE, TRUE, FALSE), ('model_016', 'OpenCode 服务', 'Agent', 'opencode', '{"apiUrl":"http://localhost:4096"}', + '{"apiKey":"replace-me"}', FALSE, TRUE, FALSE), + ('model_017', 'Qwen-Audio 3.0 Realtime', 'Realtime', 'qwen-audio-realtime', + '{"modelId":"qwen-audio-3.0-realtime-flash","apiUrl":"wss://replace-with-workspace-id.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime","voice":"longanqian","turnDetection":"server_vad","vadThreshold":0.5,"silenceDurationMs":800,"maxHistoryTurns":20}', '{"apiKey":"replace-me"}', FALSE, TRUE, FALSE) -- Seed defaults must never overwrite resources configured through the UI. ON CONFLICT (id) DO NOTHING; diff --git a/backend/services/interface_catalog.py b/backend/services/interface_catalog.py index 2ac627d..55d28c9 100644 --- a/backend/services/interface_catalog.py +++ b/backend/services/interface_catalog.py @@ -119,6 +119,58 @@ INTERFACE_DEFINITIONS: list[dict] = [ ), ], }, + { + "interface_type": "qwen-audio-realtime", + "name": "Alibaba Cloud Qwen-Audio Realtime", + "capability": "Realtime", + "fields": [ + field( + "modelId", + "Model ID", + type_="select", + required=True, + default="qwen-audio-3.0-realtime-flash", + options=[ + "qwen-audio-3.0-realtime-flash", + "qwen-audio-3.0-realtime-plus", + ], + ), + field( + "apiUrl", + "WebSocket URL (with Workspace ID)", + type_="url", + required=True, + ), + field( + "apiKey", + "API Key", + group="secrets", + type_="password", + required=True, + ), + field("voice", "Voice", default="longanqian"), + field( + "turnDetection", + "Turn Detection", + type_="select", + default="server_vad", + options=["server_vad", "smart_turn"], + ), + field("vadThreshold", "VAD Threshold", type_="number", default=0.5), + field( + "silenceDurationMs", + "VAD Silence Duration (ms)", + type_="number", + default=800, + ), + field( + "maxHistoryTurns", + "Max History Turns", + type_="number", + default=20, + ), + ], + }, { "interface_type": "xfyun-asr", "name": "Xfyun Streaming ASR", diff --git a/backend/services/model_resource_tester.py b/backend/services/model_resource_tester.py index 6b96ea3..6e0bb3c 100644 --- a/backend/services/model_resource_tester.py +++ b/backend/services/model_resource_tester.py @@ -62,7 +62,18 @@ async def test_model_resource( detail="鉴权字段和连接参数完整，请在语音测试页验证签名及音频链路", ) if interface_type == "stepfun-realtime": - return await _test_stepfun_realtime(values, secrets) + return await _test_realtime_websocket( + values, + secrets, + provider="StepFun", + ) + if interface_type == "qwen-audio-realtime": + return await _test_realtime_websocket( + values, + secrets, + provider="Qwen-Audio", + extra_headers={"x-dashscope-dataInspection": "disable"}, + ) if capability == "Realtime": return ModelResourceTestResult( ok=False, @@ -158,8 +169,12 @@ async def test_model_resource( ) -async def _test_stepfun_realtime( - values: dict, secrets: dict +async def _test_realtime_websocket( + values: dict, + secrets: dict, + *, + provider: str, + extra_headers: dict[str, str] | None = None, ) -> ModelResourceTestResult: api_url = str(values.get("apiUrl") or "") model_id = str(values.get("modelId") or "") @@ -171,11 +186,13 @@ async def _test_stepfun_realtime( (parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment) ) started = time.perf_counter() + headers = {"Authorization": f"Bearer {api_key}"} + headers.update(extra_headers or {}) try: async with websocket_connect( url, - additional_headers={"Authorization": f"Bearer {api_key}"}, + additional_headers=headers, open_timeout=TEST_TIMEOUT_SECONDS, close_timeout=2, ) as websocket: @@ -194,7 +211,7 @@ async def _test_stepfun_realtime( ok=True, latency_ms=round((time.perf_counter() - started) * 1000), message="Realtime 连接成功", - detail="StepFun 返回 session.created", + detail=f"{provider} 返回 session.created", ) except TimeoutError: return ModelResourceTestResult( @@ -211,6 +228,6 @@ async def _test_stepfun_realtime( return ModelResourceTestResult( ok=False, latency_ms=round((time.perf_counter() - started) * 1000), - message="无法连接到 StepFun Realtime", + message=f"无法连接到 {provider} Realtime", detail=detail[:300], ) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 3025940..5a397f3 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -26,6 +26,7 @@ from services.pipecat.service_factory import ( create_realtime_service, create_stt, create_tts, + realtime_audio_sample_rates, ) from db.session import SessionLocal from services.knowledge import search as search_knowledge @@ -692,6 +693,7 @@ async def run_realtime_pipeline( cfg, instructions=brain.system_prompt(cfg), ) + input_sample_rate, output_sample_rate = realtime_audio_sample_rates(cfg) text_input = RealtimeTextInputProcessor() dynamic_variables = RealtimeDynamicVariableProcessor(brain, cfg, realtime) greeting = await brain.greeting(cfg) @@ -717,12 +719,8 @@ async def run_realtime_pipeline( pipeline, params=PipelineParams( enable_metrics=False, - audio_in_sample_rate=int( - cfg.realtime_values.get("inputSampleRate") or 24000 - ), - audio_out_sample_rate=int( - cfg.realtime_values.get("outputSampleRate") or 24000 - ), + audio_in_sample_rate=input_sample_rate, + audio_out_sample_rate=output_sample_rate, ), enable_rtvi=False, ) diff --git a/backend/services/pipecat/qwen_audio_realtime.py b/backend/services/pipecat/qwen_audio_realtime.py new file mode 100644 index 0000000..ce6db22 --- /dev/null +++ b/backend/services/pipecat/qwen_audio_realtime.py @@ -0,0 +1,491 @@ +"""Qwen-Audio Realtime speech-to-speech service for Pipecat. + +The adapter intentionally depends only on Pipecat's public frame/service APIs. +Provider-specific WebSocket events stay in this module so the shared pipeline +does not need to know about Qwen-Audio protocol details. +""" + +from __future__ import annotations + +import asyncio +import base64 +import inspect +import json +from collections.abc import Awaitable, Callable +from typing import Any +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit +from uuid import uuid4 + +from loguru import logger +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + InterruptionFrame, + LLMMessagesAppendFrame, + OutputTransportMessageUrgentFrame, + StartFrame, + TTSAudioRawFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_service import AIService +from pipecat.services.settings import ServiceSettings +from pipecat.utils.time import time_now_iso8601 +from websockets.asyncio.client import connect as websocket_connect +from websockets.protocol import State + + +DEFAULT_QWEN_AUDIO_REALTIME_MODEL = "qwen-audio-3.0-realtime-flash" +DEFAULT_QWEN_AUDIO_REALTIME_VOICE = "longanqian" +QWEN_INPUT_SAMPLE_RATE = 16_000 +QWEN_OUTPUT_SAMPLE_RATE = 24_000 +SUPPORTED_TURN_DETECTION_MODES = frozenset({"server_vad", "smart_turn"}) + +ExtraEventHandler = Callable[[dict[str, Any]], Awaitable[None] | None] + + +class QwenAudioRealtimeService(AIService): + """Translate Pipecat frames to Qwen-Audio Realtime WebSocket events. + + ``extra_event_handlers`` is deliberately small: future features such as + Function Calling can subscribe to provider events without changing the + audio, transcript, or interruption paths implemented here. + """ + + def __init__( + self, + *, + api_key: str, + model: str = DEFAULT_QWEN_AUDIO_REALTIME_MODEL, + base_url: str, + instructions: str = "", + voice: str = DEFAULT_QWEN_AUDIO_REALTIME_VOICE, + input_sample_rate: int = QWEN_INPUT_SAMPLE_RATE, + output_sample_rate: int = QWEN_OUTPUT_SAMPLE_RATE, + turn_detection_mode: str = "server_vad", + vad_threshold: float = 0.5, + silence_duration_ms: int = 800, + max_history_turns: int = 20, + extra_event_handlers: dict[str, ExtraEventHandler] | None = None, + **kwargs, + ) -> None: + if turn_detection_mode not in SUPPORTED_TURN_DETECTION_MODES: + supported = ", ".join(sorted(SUPPORTED_TURN_DETECTION_MODES)) + raise ValueError( + f"Unsupported Qwen turn detection mode {turn_detection_mode!r}; " + f"expected one of: {supported}" + ) + if not -1.0 <= vad_threshold <= 1.0: + raise ValueError("Qwen VAD threshold must be between -1.0 and 1.0") + if not 200 <= silence_duration_ms <= 6_000: + raise ValueError( + "Qwen VAD silence duration must be between 200 and 6000 ms" + ) + if not 1 <= max_history_turns <= 50: + raise ValueError("Qwen max history turns must be between 1 and 50") + + super().__init__(settings=ServiceSettings(model=model), **kwargs) + self._api_key = api_key + self._model = model + self._base_url = base_url + self._instructions = instructions + self._voice = voice + self._input_sample_rate = input_sample_rate + self._output_sample_rate = output_sample_rate + self._turn_detection_mode = turn_detection_mode + self._vad_threshold = vad_threshold + self._silence_duration_ms = silence_duration_ms + self._max_history_turns = max_history_turns + self._extra_event_handlers = extra_event_handlers or {} + + self._websocket = None + self._receive_task: asyncio.Task | None = None + self._session_ready = asyncio.Event() + self._pending_events: list[dict[str, Any]] = [] + self._warned_input_sample_rate = False + + self._response_active = False + self._audio_suppressed = False + self._assistant_turn_id: str | None = None + self._assistant_text = "" + self._assistant_timestamp = "" + self._greeting_request_item_id: str | None = None + + async def start(self, frame: StartFrame) -> None: + await super().start(frame) + if not self._api_key or not self._model or not self._base_url: + await self.push_error( + "Qwen-Audio Realtime requires api_key, model, and base_url", + fatal=True, + ) + return + await self._connect() + + async def stop(self, frame: EndFrame) -> None: + await self._disconnect() + await super().stop(frame) + + async def cancel(self, frame: CancelFrame) -> None: + await self._disconnect() + await super().cancel(frame) + + async def cleanup(self) -> None: + await self._disconnect() + await super().cleanup() + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + + if isinstance(frame, InputAudioRawFrame): + if ( + frame.sample_rate != self._input_sample_rate + and not self._warned_input_sample_rate + ): + self._warned_input_sample_rate = True + logger.warning( + "Qwen-Audio Realtime expected {} Hz input, received {} Hz", + self._input_sample_rate, + frame.sample_rate, + ) + await self._send_event( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(frame.audio).decode("ascii"), + } + ) + return + + if isinstance(frame, LLMMessagesAppendFrame): + for message in frame.messages: + text = self._message_text(message) + if text: + await self.send_text( + text, + run_immediately=frame.run_llm is not False, + ) + return + + if isinstance(frame, InterruptionFrame): + await self._cancel_active_response() + + await self.push_frame(frame, direction) + + async def send_text(self, text: str, *, run_immediately: bool = True) -> None: + """Append text to Qwen's conversation and optionally start a response.""" + if not text: + return + await self._send_event( + { + "type": "conversation.item.create", + "item": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": text}], + }, + } + ) + if run_immediately: + await self._send_event({"type": "response.create"}) + + async def interrupt(self) -> None: + """Cancel current model output and notify the rest of the pipeline.""" + await self._cancel_active_response() + await self.broadcast_interruption() + + async def speak(self, text: str) -> None: + """Ask Qwen to speak a fixed greeting, then remove the hidden request. + + Qwen's documented ``response.create`` payload has no per-response + instruction field. A temporary user item keeps this behavior within + the supported protocol; it is deleted after the response completes. + """ + if not text: + return + item_id = f"item_{uuid4().hex}" + self._greeting_request_item_id = item_id + await self._send_event( + { + "type": "conversation.item.create", + "item": { + "id": item_id, + "type": "message", + "role": "user", + "content": [ + { + "type": "input_text", + "text": ( + "请直接朗读下面的开场白，不要增删、解释或添加前后缀：\n" + f"{text}" + ), + } + ], + }, + } + ) + await self._send_event({"type": "response.create"}) + + async def update_instructions(self, instructions: str) -> None: + """Update only instructions after startup. + + Qwen accepts voice and turn detection only in the first session update, + so resending the full initial configuration would break live dynamic + variable updates. + """ + self._instructions = instructions + if self._session_ready.is_set(): + await self._send_event( + { + "type": "session.update", + "session": {"instructions": instructions}, + }, + wait_until_ready=False, + ) + + def _connection_url(self) -> str: + parts = urlsplit(self._base_url) + query = dict(parse_qsl(parts.query)) + query["model"] = self._model + return urlunsplit( + (parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment) + ) + + def _initial_session_config(self) -> dict[str, Any]: + return { + "modalities": ["text", "audio"], + "instructions": self._instructions, + "voice": self._voice, + "input_audio_format": "pcm", + "output_audio_format": "pcm", + "turn_detection": self._turn_detection_config(), + "max_history_turns": self._max_history_turns, + } + + def _turn_detection_config(self) -> dict[str, Any]: + if self._turn_detection_mode == "smart_turn": + return {"type": "smart_turn"} + return { + "type": "server_vad", + "threshold": self._vad_threshold, + "silence_duration_ms": self._silence_duration_ms, + } + + async def _connect(self) -> None: + if self._websocket and self._websocket.state is State.OPEN: + return + try: + self._websocket = await websocket_connect( + self._connection_url(), + additional_headers={ + "Authorization": f"Bearer {self._api_key}", + "x-dashscope-dataInspection": "disable", + }, + max_size=None, + open_timeout=10, + ) + self._receive_task = self.create_task( + self._receive_messages(), name="qwen_audio_realtime_receive" + ) + except Exception as exc: + self._websocket = None + await self.push_error( + f"Qwen-Audio Realtime connection failed: {exc}", + exception=exc, + fatal=True, + ) + + async def _disconnect(self) -> None: + current_task = asyncio.current_task() + task = self._receive_task + self._receive_task = None + if task and task is not current_task: + await self.cancel_task(task) + + websocket = self._websocket + self._websocket = None + self._session_ready.clear() + self._pending_events.clear() + self._response_active = False + if websocket and websocket.state is State.OPEN: + try: + await websocket.close() + except Exception: + pass + + async def _receive_messages(self) -> None: + websocket = self._websocket + if not websocket: + return + try: + async for raw_message in websocket: + await self._handle_server_event(json.loads(raw_message)) + except Exception as exc: + if self._websocket is websocket: + await self.push_error( + f"Qwen-Audio Realtime receive failed: {exc}", exception=exc + ) + finally: + if self._websocket is websocket: + self._websocket = None + self._session_ready.clear() + if self._receive_task is asyncio.current_task(): + self._receive_task = None + + async def _handle_server_event(self, event: dict[str, Any]) -> None: + event_type = str(event.get("type") or "") + if event_type == "session.created": + await self._send_event( + { + "type": "session.update", + "session": self._initial_session_config(), + }, + wait_until_ready=False, + ) + elif event_type == "session.updated": + self._session_ready.set() + pending, self._pending_events = self._pending_events, [] + for payload in pending: + await self._send_event(payload, wait_until_ready=False) + elif event_type == "response.created": + self._response_active = True + self._audio_suppressed = False + elif event_type == "response.audio.delta": + audio = event.get("delta") + if audio and not self._audio_suppressed: + await self.push_frame( + TTSAudioRawFrame( + base64.b64decode(audio), + self._output_sample_rate, + 1, + ) + ) + elif event_type in {"response.audio_transcript.delta", "response.text.delta"}: + if not self._audio_suppressed: + await self._append_assistant_text(str(event.get("delta") or "")) + elif event_type in {"response.audio_transcript.done", "response.text.done"}: + transcript = str(event.get("transcript") or event.get("text") or "") + if transcript and not self._audio_suppressed: + if self._assistant_turn_id: + self._assistant_text = transcript + else: + await self._append_assistant_text(transcript) + elif event_type == "conversation.item.input_audio_transcription.completed": + await self._send_transcript("user", str(event.get("transcript") or "")) + elif event_type == "input_audio_buffer.speech_started": + await self._cancel_active_response() + await self.broadcast_interruption() + elif event_type == "response.done": + response = event.get("response") + status = response.get("status") if isinstance(response, dict) else None + interrupted = status in {"cancelled", "incomplete", "interrupted", "failed"} + self._response_active = False + await self._finish_assistant_text(interrupted=interrupted) + await self._delete_greeting_request() + elif event_type == "error": + error = event.get("error") + message = error.get("message") if isinstance(error, dict) else str(error) + if "cancel" not in str(message).lower(): + await self.push_error(f"Qwen-Audio Realtime error: {message}") + + handler = self._extra_event_handlers.get(event_type) + if handler: + result = handler(event) + if inspect.isawaitable(result): + await result + + async def _cancel_active_response(self) -> None: + if not self._response_active and not self._assistant_turn_id: + return + self._audio_suppressed = True + if self._response_active: + await self._send_event( + {"type": "response.cancel"}, wait_until_ready=False + ) + self._response_active = False + await self._finish_assistant_text(interrupted=True) + + async def _delete_greeting_request(self) -> None: + item_id = self._greeting_request_item_id + self._greeting_request_item_id = None + if item_id: + await self._send_event( + {"type": "conversation.item.delete", "item_id": item_id}, + wait_until_ready=False, + ) + + async def _send_event( + self, payload: dict[str, Any], *, wait_until_ready: bool = True + ) -> None: + if wait_until_ready and not self._session_ready.is_set(): + self._pending_events.append(payload) + return + if not self._websocket or self._websocket.state is not State.OPEN: + return + event = {"event_id": f"event_{uuid4().hex}", **payload} + await self._websocket.send(json.dumps(event, ensure_ascii=False)) + + async def _append_assistant_text(self, delta: str) -> None: + if not delta: + return + if not self._assistant_turn_id: + self._assistant_turn_id = uuid4().hex + self._assistant_timestamp = time_now_iso8601() + await self._send_transport_message( + { + "type": "assistant-text-start", + "turn_id": self._assistant_turn_id, + "timestamp": self._assistant_timestamp, + } + ) + self._assistant_text += delta + await self._send_transport_message( + { + "type": "assistant-text-delta", + "turn_id": self._assistant_turn_id, + "delta": delta, + } + ) + + async def _finish_assistant_text(self, *, interrupted: bool) -> None: + if not self._assistant_turn_id: + return + await self._send_transport_message( + { + "type": "assistant-text-end", + "turn_id": self._assistant_turn_id, + "content": self._assistant_text, + "interrupted": interrupted, + } + ) + self._assistant_turn_id = None + self._assistant_text = "" + self._assistant_timestamp = "" + + async def _send_transcript(self, role: str, content: str) -> None: + if content: + await self._send_transport_message( + { + "type": "transcript", + "role": role, + "content": content, + "timestamp": time_now_iso8601(), + } + ) + + async def _send_transport_message(self, message: dict[str, Any]) -> None: + await self.push_frame(OutputTransportMessageUrgentFrame(message=message)) + + @staticmethod + def _message_text(message: Any) -> str: + if not isinstance(message, dict): + return "" + content = message.get("content") + if isinstance(content, str): + return content.strip() + if isinstance(content, list): + return "\n".join( + str(part.get("text") or "") + for part in content + if isinstance(part, dict) and part.get("type") == "text" + ).strip() + return "" diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index f575be7..c2b6927 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -178,8 +178,24 @@ def create_tts(cfg: AssistantConfig): ) +def realtime_audio_sample_rates(cfg: AssistantConfig) -> tuple[int, int]: + """Resolve provider defaults once for both the service and pipeline. + + Values remain overridable for future model versions with different audio + formats, while today's Qwen-Audio contract defaults to 16 kHz in/24 kHz out. + """ + default_input_rate = ( + 16_000 if cfg.realtime_interface_type == "qwen-audio-realtime" else 24_000 + ) + return ( + int(cfg.realtime_values.get("inputSampleRate") or default_input_rate), + int(cfg.realtime_values.get("outputSampleRate") or 24_000), + ) + + def create_realtime_service(cfg: AssistantConfig, *, instructions: str): """Create a speech-to-speech service that owns STT, LLM, and TTS.""" + input_sample_rate, output_sample_rate = realtime_audio_sample_rates(cfg) if cfg.realtime_interface_type == "stepfun-realtime": from services.pipecat.stepfun_realtime import StepFunRealtimeService @@ -189,12 +205,8 @@ def create_realtime_service(cfg: AssistantConfig, *, instructions: str): base_url=_require(cfg.realtime_base_url, "Realtime apiUrl"), instructions=instructions, voice=str(cfg.realtime_values.get("voice") or "linjiajiejie"), - input_sample_rate=int( - cfg.realtime_values.get("inputSampleRate") or 24000 - ), - output_sample_rate=int( - cfg.realtime_values.get("outputSampleRate") or 24000 - ), + input_sample_rate=input_sample_rate, + output_sample_rate=output_sample_rate, prefix_padding_ms=int( cfg.realtime_values.get("prefixPaddingMs") or 500 ), @@ -205,4 +217,26 @@ def create_realtime_service(cfg: AssistantConfig, *, instructions: str): cfg.realtime_values.get("energyAwakenessThreshold") or 2500 ), ) + if cfg.realtime_interface_type == "qwen-audio-realtime": + from services.pipecat.qwen_audio_realtime import QwenAudioRealtimeService + + return QwenAudioRealtimeService( + api_key=_require(cfg.realtime_api_key, "Realtime apiKey"), + model=_require(cfg.realtimeModel, "Realtime modelId"), + base_url=_require(cfg.realtime_base_url, "Realtime apiUrl"), + instructions=instructions, + voice=str(cfg.realtime_values.get("voice") or "longanqian"), + input_sample_rate=input_sample_rate, + output_sample_rate=output_sample_rate, + turn_detection_mode=str( + cfg.realtime_values.get("turnDetection") or "server_vad" + ), + vad_threshold=float(cfg.realtime_values.get("vadThreshold", 0.5)), + silence_duration_ms=int( + cfg.realtime_values.get("silenceDurationMs") or 800 + ), + max_history_turns=int( + cfg.realtime_values.get("maxHistoryTurns") or 20 + ), + ) raise ValueError(f"不支持的 Realtime 接口类型: {cfg.realtime_interface_type}") diff --git a/backend/tests/test_qwen_audio_realtime.py b/backend/tests/test_qwen_audio_realtime.py new file mode 100644 index 0000000..166ef99 --- /dev/null +++ b/backend/tests/test_qwen_audio_realtime.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import base64 +import json +import unittest +from unittest.mock import AsyncMock, patch + +from models import AssistantConfig +from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSAudioRawFrame +from services import model_resource_tester +from services.interface_catalog import INTERFACE_DEFINITIONS +from services.pipecat.qwen_audio_realtime import QwenAudioRealtimeService +from services.pipecat.service_factory import ( + create_realtime_service, + realtime_audio_sample_rates, +) +from websockets.protocol import State + + +class _OpenWebSocket: + state = State.OPEN + + def __init__(self) -> None: + self.messages: list[dict] = [] + + async def send(self, raw_message: str) -> None: + self.messages.append(json.loads(raw_message)) + + +def _service(**overrides) -> QwenAudioRealtimeService: + settings = { + "api_key": "test-key", + "model": "qwen-audio-3.0-realtime-flash", + "base_url": ( + "wss://workspace.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime" + ), + "instructions": "You are helpful.", + } + settings.update(overrides) + return QwenAudioRealtimeService(**settings) + + +class QwenAudioRealtimeServiceTest(unittest.IsolatedAsyncioTestCase): + def test_provider_defaults_and_session_config(self): + service = _service() + + self.assertEqual( + service._connection_url(), + "wss://workspace.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime" + "?model=qwen-audio-3.0-realtime-flash", + ) + self.assertEqual( + service._initial_session_config(), + { + "modalities": ["text", "audio"], + "instructions": "You are helpful.", + "voice": "longanqian", + "input_audio_format": "pcm", + "output_audio_format": "pcm", + "turn_detection": { + "type": "server_vad", + "threshold": 0.5, + "silence_duration_ms": 800, + }, + "max_history_turns": 20, + }, + ) + + def test_smart_turn_and_value_ranges_are_explicit(self): + service = _service(turn_detection_mode="smart_turn") + self.assertEqual(service._turn_detection_config(), {"type": "smart_turn"}) + + with self.assertRaisesRegex(ValueError, "turn detection mode"): + _service(turn_detection_mode="manual") + with self.assertRaisesRegex(ValueError, "threshold"): + _service(vad_threshold=1.1) + with self.assertRaisesRegex(ValueError, "history"): + _service(max_history_turns=51) + + async def test_session_ready_flushes_queued_events_in_order(self): + service = _service() + websocket = _OpenWebSocket() + service._websocket = websocket + + await service.send_text("hello") + self.assertEqual(websocket.messages, []) + + await service._handle_server_event({"type": "session.created"}) + self.assertEqual(websocket.messages[0]["type"], "session.update") + self.assertEqual( + websocket.messages[0]["session"]["input_audio_format"], "pcm" + ) + + await service._handle_server_event({"type": "session.updated"}) + self.assertEqual( + [message["type"] for message in websocket.messages], + ["session.update", "conversation.item.create", "response.create"], + ) + + async def test_dynamic_instructions_do_not_resend_locked_settings(self): + service = _service() + websocket = _OpenWebSocket() + service._websocket = websocket + service._session_ready.set() + + await service.update_instructions("Call the user Alice.") + + self.assertEqual(websocket.messages[-1]["type"], "session.update") + self.assertEqual( + websocket.messages[-1]["session"], + {"instructions": "Call the user Alice."}, + ) + + async def test_interruption_cancels_and_suppresses_residual_audio(self): + service = _service() + websocket = _OpenWebSocket() + service._websocket = websocket + service._session_ready.set() + service.push_frame = AsyncMock() + service.broadcast_interruption = AsyncMock() + audio = base64.b64encode(b"\x01\x02").decode("ascii") + + await service._handle_server_event({"type": "response.created"}) + await service._handle_server_event( + {"type": "response.audio_transcript.delta", "delta": "Hi"} + ) + await service._handle_server_event( + {"type": "response.audio.delta", "delta": audio} + ) + await service._handle_server_event( + {"type": "input_audio_buffer.speech_started"} + ) + calls_before_residual_audio = service.push_frame.await_count + await service._handle_server_event( + {"type": "response.audio.delta", "delta": audio} + ) + + self.assertEqual(service.push_frame.await_count, calls_before_residual_audio) + self.assertIn("response.cancel", [item["type"] for item in websocket.messages]) + service.broadcast_interruption.assert_awaited_once() + + frames = [call.args[0] for call in service.push_frame.await_args_list] + self.assertTrue(any(isinstance(frame, TTSAudioRawFrame) for frame in frames)) + end_messages = [ + frame.message + for frame in frames + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "assistant-text-end" + ] + self.assertEqual(end_messages[0]["content"], "Hi") + self.assertTrue(end_messages[0]["interrupted"]) + + async def test_greeting_request_is_removed_after_response(self): + service = _service() + websocket = _OpenWebSocket() + service._websocket = websocket + service._session_ready.set() + service.push_frame = AsyncMock() + + await service.speak("Welcome") + greeting_item_id = websocket.messages[0]["item"]["id"] + await service._handle_server_event({"type": "response.created"}) + await service._handle_server_event( + {"type": "response.audio_transcript.done", "transcript": "Welcome"} + ) + await service._handle_server_event( + {"type": "response.done", "response": {"status": "completed"}} + ) + + self.assertEqual(websocket.messages[-1]["type"], "conversation.item.delete") + self.assertEqual(websocket.messages[-1]["item_id"], greeting_item_id) + + def test_pipeline_sample_rates_use_qwen_defaults(self): + cfg = AssistantConfig( + realtime_interface_type="qwen-audio-realtime", + realtime_values={}, + ) + self.assertEqual(realtime_audio_sample_rates(cfg), (16_000, 24_000)) + + def test_factory_and_interface_catalog_register_qwen(self): + definition = next( + item + for item in INTERFACE_DEFINITIONS + if item["interface_type"] == "qwen-audio-realtime" + ) + self.assertEqual(definition["capability"], "Realtime") + self.assertIn( + "smart_turn", + next( + field + for field in definition["fields"] + if field["key"] == "turnDetection" + )["options"], + ) + + cfg = AssistantConfig( + realtime_interface_type="qwen-audio-realtime", + realtimeModel="qwen-audio-3.0-realtime-plus", + realtime_api_key="test-key", + realtime_base_url=( + "wss://workspace.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime" + ), + realtime_values={"turnDetection": "smart_turn"}, + ) + service = create_realtime_service(cfg, instructions="Be concise.") + self.assertIsInstance(service, QwenAudioRealtimeService) + self.assertEqual(service._turn_detection_config(), {"type": "smart_turn"}) + + async def test_resource_tester_routes_qwen_with_dashscope_header(self): + expected = object() + probe = AsyncMock(return_value=expected) + with patch.object( + model_resource_tester, + "_test_realtime_websocket", + probe, + ): + result = await model_resource_tester.test_model_resource( + "qwen-audio-realtime", + "Realtime", + {"apiUrl": "wss://example.test/realtime", "modelId": "model"}, + {"apiKey": "secret"}, + ) + + self.assertIs(result, expected) + probe.assert_awaited_once_with( + {"apiUrl": "wss://example.test/realtime", "modelId": "model"}, + {"apiKey": "secret"}, + provider="Qwen-Audio", + extra_headers={"x-dashscope-dataInspection": "disable"}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/integrations.mdx b/docs/integrations.mdx index 9ca9987..55223a3 100644 --- a/docs/integrations.mdx +++ b/docs/integrations.mdx @@ -13,11 +13,22 @@ icon: plug | 场景 | 推荐接口类型 | | --- | --- | | 通用大模型、转写、合成、向量 | `openai-llm`、`openai-asr`、`openai-tts`、`openai-embedding` | -| 端到端实时语音 | `stepfun-realtime` 或 `openai-realtime` | +| 端到端实时语音 | `stepfun-realtime` 或 `qwen-audio-realtime` | | 讯飞语音 | `xfyun-asr`、`xfyun-tts`、`xfyun-super-tts` | 讯飞接入需要将 `appId`、`apiKey`、`apiSecret` 写入 `secrets`，WebSocket 地址及音色等参数写入 `values`。 +### Qwen-Audio Realtime + +在「组件库 / 模型资源」中新建 `Alibaba Cloud Qwen-Audio Realtime` 资源： + +- WebSocket URL 必须包含百炼业务空间 ID，例如 `wss://.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime`。 +- API Key 填写该业务空间可用的百炼密钥。 +- MVP 支持 `server_vad` 与 `smart_turn`；前者可以调整 VAD 阈值和静音时长。 +- 当前模型固定使用 16 kHz 单声道 PCM 输入、24 kHz 单声道 PCM 输出，管线会自动匹配采样率。 + +保存前可使用「测试连接」验证鉴权、地域、业务空间 ID 和模型是否匹配。 + ## 外部 Agent 创建助手时选择 Dify、FastGPT 或 OpenCode： diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..63f6ada --- /dev/null +++ b/package-lock.json @@ -0,0 +1,6 @@ +{ + "name": "ai-video-fullstack", + "lockfileVersion": 3, + "requires": true, + "packages": {} +}