diff --git a/backend/db/seed_model_resources.sql b/backend/db/seed_model_resources.sql index 243ca42..2292e4a 100644 --- a/backend/db/seed_model_resources.sql +++ b/backend/db/seed_model_resources.sql @@ -25,7 +25,7 @@ VALUES '{"apiUrl":"https://tts-api.xfyun.cn/v2/tts","voice":"xiaoyan","speed":50,"volume":50,"pitch":50,"sourceSampleRate":16000}', '{"appId":"replace-me","apiKey":"replace-me","apiSecret":"replace-me"}', FALSE, TRUE, TRUE), ('model_008', 'CosyVoice 合成', 'TTS', 'dashscope-tts', - '{"modelId":"cosyvoice-v1","apiUrl":"https://dashscope.aliyuncs.com/api/v1/services/audio/tts","voice":"longxiaochun"}', + '{"modelId":"cosyvoice-v1","apiUrl":"wss://dashscope.aliyuncs.com/api-ws/v1/inference","voice":"longxiaochun"}', '{"apiKey":"replace-me"}', FALSE, TRUE, FALSE), ('model_009', 'GPT Realtime', 'Realtime', 'openai-realtime', '{"modelId":"gpt-4o-realtime-preview","apiUrl":"https://api.openai.com/v1/realtime"}', diff --git a/backend/services/interface_catalog.py b/backend/services/interface_catalog.py index 0eb0219..844c7ce 100644 --- a/backend/services/interface_catalog.py +++ b/backend/services/interface_catalog.py @@ -219,9 +219,14 @@ INTERFACE_DEFINITIONS: list[dict] = [ }, { "interface_type": "dashscope-tts", - "name": "DashScope TTS", + "name": "Alibaba Cloud Streaming TTS", "capability": "TTS", - "fields": OPENAI_COMMON + [field("voice", "Voice")], + "fields": [ + field("modelId", "Model ID", required=True), + field("apiUrl", "WebSocket URL (/api-ws/v1/inference)", type_="url", required=True), + field("apiKey", "API Key", group="secrets", type_="password", required=True), + field("voice", "Voice", required=True), + ], }, { "interface_type": "gemini-realtime", diff --git a/backend/services/model_resource_tester.py b/backend/services/model_resource_tester.py index 6e0bb3c..6577e83 100644 --- a/backend/services/model_resource_tester.py +++ b/backend/services/model_resource_tester.py @@ -55,6 +55,8 @@ async def test_model_resource( values: dict, secrets: dict, ) -> ModelResourceTestResult: + if interface_type == "dashscope-tts": + return await _test_dashscope_tts(values, secrets) if interface_type.startswith("xfyun-"): return ModelResourceTestResult( ok=True, @@ -231,3 +233,35 @@ async def _test_realtime_websocket( message=f"无法连接到 {provider} Realtime", detail=detail[:300], ) + + +async def _test_dashscope_tts(values: dict, secrets: dict) -> ModelResourceTestResult: + from services.pipecat.dashscope_tts import SynthesisSession + + started = time.perf_counter() + session = None + try: + session = SynthesisSession( + str(values.get("apiUrl") or ""), str(secrets.get("apiKey") or ""), + str(values.get("modelId") or ""), str(values.get("voice") or ""), + ) + async with asyncio.timeout(TEST_TIMEOUT_SECONDS): + await session.start() + await session.text("你好，这是语音合成测试。") + await session.finish() + async for _ in session.audio(): + pass + return ModelResourceTestResult( + ok=True, message="百炼 TTS 合成成功", + detail="已收到 PCM 音频及 task-finished", + latency_ms=round((time.perf_counter() - started) * 1000), + ) + except Exception as exc: + detail = "合成测试超时" if isinstance(exc, TimeoutError) else str(exc) + for secret in secrets.values(): + if secret: + detail = detail.replace(str(secret), "***") + return ModelResourceTestResult(ok=False, message="百炼 TTS 合成失败", detail=detail[:300]) + finally: + if session: + await session.close() diff --git a/backend/services/pipecat/dashscope_tts.py b/backend/services/pipecat/dashscope_tts.py new file mode 100644 index 0000000..85cfcb6 --- /dev/null +++ b/backend/services/pipecat/dashscope_tts.py @@ -0,0 +1,219 @@ +"""百炼 CosyVoice / Qwen-Audio-TTS 双向流式适配。 + +每个 Pipecat 音频上下文对应一个任务和连接；文本增量发送，音频独立接收。 +不复用连接，避免取消后没有 task_id 的二进制音频串入下一轮。 +""" +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from urllib.parse import urlsplit +from uuid import uuid4 + +from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, TTSAudioRawFrame, TTSStoppedFrame +from pipecat.services.settings import TTSSettings +from pipecat.services.tts_service import TextAggregationMode, TTSService +from websockets.asyncio.client import connect + +SOURCE_SAMPLE_RATE = 24000 +TIMEOUT_SECONDS = 30 +SUPPORTED_MODELS = ( + "qwen-audio-3.1-tts-plus", "qwen-audio-3.1-tts-flash", + "qwen-audio-3.0-tts-plus", "qwen-audio-3.0-tts-flash", + "cosyvoice-v3.5-plus", "cosyvoice-v3.5-flash", + "cosyvoice-v3-plus", "cosyvoice-v3-flash", "cosyvoice-v2", "cosyvoice-v1", +) + + +def validate_config(url: str, model: str, voice: str, api_key: str) -> None: + parts = urlsplit(url) + if parts.scheme != "wss" or not parts.hostname or parts.path != "/api-ws/v1/inference": + raise ValueError("百炼 TTS API URL 必须为 wss://<域名>/api-ws/v1/inference，请更新旧 HTTP 配置") + if model not in SUPPORTED_MODELS: + raise ValueError(f"百炼流式 TTS 不支持模型: {model}") + if not voice or not api_key: + raise ValueError("百炼 TTS 需要 voice 和 apiKey") + + +class SynthesisSession: + """一个百炼任务。连接测试和正式合成使用相同的协议实现。""" + + def __init__(self, url: str, api_key: str, model: str, voice: str): + validate_config(url, model, voice, api_key) + self.url, self.api_key = url, api_key + self.model, self.voice = model, voice + self.task_id = uuid4().hex + self.socket = None + self.finished_input = False + + async def send(self, action: str, input_: dict, **payload) -> None: + await self.socket.send(json.dumps({ + "header": {"action": action, "task_id": self.task_id, "streaming": "duplex"}, + "payload": {"input": input_, **payload}, + }, ensure_ascii=False)) + + def event(self, raw: str) -> str: + header = json.loads(raw).get("header", {}) + if header.get("task_id") != self.task_id: + raise RuntimeError("百炼 TTS 返回了不匹配的 task_id") + if header.get("event") == "task-failed": + raise RuntimeError(f"百炼 TTS: {header.get('error_code')}: {header.get('error_message')}") + return header.get("event", "") + + async def start(self) -> None: + self.socket = await connect( + self.url, additional_headers={"Authorization": f"Bearer {self.api_key}"}, + open_timeout=TIMEOUT_SECONDS, close_timeout=2, max_size=None, + ) + await self.send("run-task", {}, task_group="audio", task="tts", + function="SpeechSynthesizer", model=self.model, + parameters={"text_type": "PlainText", "voice": self.voice, + "format": "pcm", "sample_rate": SOURCE_SAMPLE_RATE}) + raw = await asyncio.wait_for(self.socket.recv(), TIMEOUT_SECONDS) + if not isinstance(raw, str) or self.event(raw) != "task-started": + raise RuntimeError("百炼 TTS 未返回 task-started") + + async def text(self, text: str) -> None: + if self.finished_input: + raise RuntimeError("百炼 TTS 任务已结束文本输入") + # 单次最多 20000 字符；保留 token 间的空格。 + for offset in range(0, len(text), 20000): + await self.send("continue-task", {"text": text[offset:offset + 20000]}) + + async def finish(self) -> None: + if not self.finished_input: + await self.send("finish-task", {}) + self.finished_input = True + + async def audio(self) -> AsyncGenerator[bytes, None]: + received_audio = False + while True: + raw = await asyncio.wait_for(self.socket.recv(), TIMEOUT_SECONDS) + if isinstance(raw, bytes): + if raw: + received_audio = True + yield raw + elif self.event(raw) == "task-finished": + if not received_audio: + raise RuntimeError("百炼 TTS 合成结束但没有返回音频") + return + + async def close(self) -> None: + if self.socket: + await self.socket.close() + self.socket = None + + +@dataclass +class _Stream: + session: SynthesisSession + receiver: asyncio.Task | None = None + + +class DashScopeTTSService(TTSService): + """核心参数只有模型、音色、地址和密钥；输出采样率由流水线决定。""" + + def __init__(self, *, api_key: str, model: str, voice: str, url: str, **kwargs): + validate_config(url, model, voice, api_key) + super().__init__( + text_aggregation_mode=TextAggregationMode.TOKEN, + push_text_frames=True, push_start_frame=True, push_stop_frames=False, + pause_frame_processing=False, + settings=TTSSettings(model=model, voice=voice), **kwargs, + ) + self._url, self._api_key = url, api_key + self._streams: dict[str, _Stream] = {} + # A failed turn must not silently restart and speak only its remaining tokens. + self._failed_contexts: set[str] = set() + + @property + def supports_processing_metrics(self) -> bool: + return False # 音频在独立接收任务中生成。 + + async def stop(self, frame: EndFrame) -> None: + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame) -> None: + await super().cancel(frame) + await self._disconnect() + + async def cleanup(self) -> None: + await self._disconnect() + await super().cleanup() + + def _safe_error(self, exc: Exception) -> str: + return str(exc).replace(self._api_key, "***")[:300] + + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]: + if not text or context_id in self._failed_contexts: + return + try: + stream = self._streams.get(context_id) + if stream is None: + session = SynthesisSession(self._url, self._api_key, + self._settings.model, self._settings.voice) + stream = _Stream(session) + self._streams[context_id] = stream + await session.start() + stream.receiver = self.create_task(self._receive(context_id, session)) + await stream.session.text(text) + except Exception as exc: + self._failed_contexts.add(context_id) + await self._close_stream(context_id) + yield ErrorFrame(error=self._safe_error(exc)) + await self._finish_context(context_id) + yield None + + async def flush_audio(self, context_id: str | None = None) -> None: + stream = self._streams.get(context_id) + if stream: + try: + await stream.session.finish() + except Exception as exc: + self._failed_contexts.add(context_id) + await self.push_error(error_msg=self._safe_error(exc)) + await self._close_stream(context_id) + await self._finish_context(context_id) + + async def _receive(self, context_id: str, session: SynthesisSession) -> None: + try: + async for audio in session.audio(): + if self.audio_context_available(context_id): + if self.sample_rate != SOURCE_SAMPLE_RATE: + audio = await self._resampler.resample(audio, SOURCE_SAMPLE_RATE, self.sample_rate) + await self.append_to_audio_context( + context_id, TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=context_id)) + except Exception as exc: + self._failed_contexts.add(context_id) + await self.push_error(error_msg=self._safe_error(exc)) + finally: + await session.close() + self._streams.pop(context_id, None) + await self._finish_context(context_id) + + async def _finish_context(self, context_id: str) -> None: + if self.audio_context_available(context_id): + await self.append_to_audio_context(context_id, TTSStoppedFrame(context_id=context_id)) + await self.remove_audio_context(context_id) + + async def _close_stream(self, context_id: str) -> None: + stream = self._streams.pop(context_id, None) + if stream: + if stream.receiver: + await self.cancel_task(stream.receiver) + await stream.session.close() + + async def on_audio_context_interrupted(self, context_id: str) -> None: + await self._close_stream(context_id) + self._failed_contexts.discard(context_id) + + async def on_audio_context_completed(self, context_id: str) -> None: + self._failed_contexts.discard(context_id) + + async def _disconnect(self) -> None: + for context_id in list(self._streams): + await self._close_stream(context_id) + self._failed_contexts.clear() diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index 13af449..76f6561 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -137,7 +137,7 @@ def create_llm(cfg: AssistantConfig): def create_tts(cfg: AssistantConfig): - """CosyVoice 等,走 OpenAI 兼容的 /v1/audio/speech。""" + """根据资源的协议创建独立 TTS 服务。""" voice = _require(cfg.voice, "TTS voice") if cfg.tts_interface_type == "xfyun-super-tts": return XfyunSuperTTSService( @@ -171,7 +171,16 @@ def create_tts(cfg: AssistantConfig): push_stop_frames=True, stop_frame_timeout_s=WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S, ) - if cfg.tts_interface_type not in {"openai-tts", "dashscope-tts"}: + if cfg.tts_interface_type == "dashscope-tts": + from services.pipecat.dashscope_tts import DashScopeTTSService + + return DashScopeTTSService( + api_key=_require(cfg.tts_api_key, "TTS apiKey"), + model=_require(cfg.tts_model, "TTS modelId"), + voice=voice, + url=_require(cfg.tts_base_url, "TTS apiUrl"), + ) + if cfg.tts_interface_type != "openai-tts": raise ValueError(f"不支持的 TTS 接口类型: {cfg.tts_interface_type}") # Pipecat 默认只接受 OpenAI 官方音色。OpenAI 兼容服务常使用自定义 voice id, diff --git a/backend/tests/test_dashscope_tts.py b/backend/tests/test_dashscope_tts.py new file mode 100644 index 0000000..cf72378 --- /dev/null +++ b/backend/tests/test_dashscope_tts.py @@ -0,0 +1,130 @@ +import asyncio +import json +import unittest +from unittest.mock import AsyncMock, patch + +from pipecat.frames.frames import TTSAudioRawFrame, TTSStoppedFrame +from services.pipecat.dashscope_tts import DashScopeTTSService, SynthesisSession, SUPPORTED_MODELS +from services.model_resource_tester import test_model_resource + +URL = 'wss://workspace.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference' + + +class Socket: + def __init__(self): + self.sent = [] + self.queue = asyncio.Queue() + self.close = AsyncMock() + + async def send(self, raw): + msg = json.loads(raw) + self.sent.append(msg) + if msg['header']['action'] == 'run-task': + self.task_id = msg['header']['task_id'] + self.event('task-started') + + def event(self, name): + self.queue.put_nowait(json.dumps({'header': {'event': name, 'task_id': self.task_id}})) + + async def recv(self): + return await self.queue.get() + + +class DashScopeTests(unittest.IsolatedAsyncioTestCase): + async def test_protocol_streams_before_finish_and_preserves_spaces(self): + socket = Socket() + session = SynthesisSession(URL, 'secret', 'cosyvoice-v1', 'longxiaochun') + with patch('services.pipecat.dashscope_tts.connect', AsyncMock(return_value=socket)): + await session.start() + await session.text('Hello') + await session.text(' world') + socket.queue.put_nowait(b'\x01\x00' * 20) + audio = session.audio() + self.assertEqual(await anext(audio), b'\x01\x00' * 20) + self.assertFalse(session.finished_input) + await session.finish() + await session.finish() + socket.event('task-finished') + with self.assertRaises(StopAsyncIteration): + await anext(audio) + await session.close() + self.assertEqual([m['header']['action'] for m in socket.sent], + ['run-task', 'continue-task', 'continue-task', 'finish-task']) + self.assertEqual(socket.sent[2]['payload']['input']['text'], ' world') + self.assertEqual(socket.sent[0]['payload']['parameters']['format'], 'pcm') + self.assertEqual(len({m['header']['task_id'] for m in socket.sent}), 1) + + async def test_empty_audio_and_provider_failure(self): + for failure in (False, True): + socket = Socket() + session = SynthesisSession(URL, 'secret', 'cosyvoice-v2', 'voice') + with patch('services.pipecat.dashscope_tts.connect', AsyncMock(return_value=socket)): + await session.start() + socket.event('task-failed' if failure else 'task-finished') + with self.assertRaises(RuntimeError): + await anext(session.audio()) + await session.close() + + async def test_pipecat_audio_completion_and_interruption(self): + socket = Socket() + service = DashScopeTTSService(api_key='secret', model='qwen-audio-3.1-tts-flash', + voice='voice', url=URL) + service._sample_rate = 24000 + service.create_task = asyncio.create_task + service.cancel_task = AsyncMock(side_effect=self.cancel_task) + service.audio_context_available = lambda _: True + service.append_to_audio_context = AsyncMock() + service.remove_audio_context = AsyncMock() + with patch('services.pipecat.dashscope_tts.connect', AsyncMock(return_value=socket)): + async for _ in service.run_tts('你好', 'turn1'): + pass + receiver = service._streams['turn1'].receiver + socket.queue.put_nowait(b'\0\0' * 20) + await service.flush_audio('turn1') + socket.event('task-finished') + await receiver + frames = [call.args[1] for call in service.append_to_audio_context.call_args_list] + self.assertIsInstance(frames[0], TTSAudioRawFrame) + self.assertIsInstance(frames[-1], TTSStoppedFrame) + async for _ in service.run_tts('下一轮', 'turn2'): + pass + await service.on_audio_context_interrupted('turn2') + self.assertFalse(service._streams) + self.assertGreaterEqual(socket.close.await_count, 2) + + async def cancel_task(self, task): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def test_tester_requires_audio_and_completion(self): + with patch.object(SynthesisSession, 'start', AsyncMock()), \ + patch.object(SynthesisSession, 'text', AsyncMock()), \ + patch.object(SynthesisSession, 'finish', AsyncMock()), \ + patch.object(SynthesisSession, 'close', AsyncMock()): + async def no_audio(self): + raise RuntimeError('no audio secret') + yield b'' + with patch.object(SynthesisSession, 'audio', no_audio): + result = await test_model_resource('dashscope-tts', 'TTS', + {'apiUrl': URL, 'modelId': 'cosyvoice-v2', 'voice': 'voice'}, {'apiKey': 'secret'}) + self.assertFalse(result.ok) + self.assertNotIn('secret', result.detail) + + def test_factory_routes_dashscope_to_token_streaming(self): + from models import AssistantConfig + from services.pipecat.service_factory import create_tts + service = create_tts(AssistantConfig( + tts_interface_type="dashscope-tts", tts_model="cosyvoice-v1", + voice="longxiaochun", tts_api_key="secret", tts_base_url=URL)) + self.assertIsInstance(service, DashScopeTTSService) + self.assertTrue(service._is_streaming_tokens) + + def test_models_and_old_http_url(self): + for model in SUPPORTED_MODELS: + SynthesisSession(URL, 'secret', model, 'voice') + with self.assertRaisesRegex(ValueError, '旧 HTTP'): + SynthesisSession('https://dashscope.aliyuncs.com/api/v1/services/audio/tts', + 'secret', 'cosyvoice-v1', 'voice') diff --git a/docs-developer/integrations.mdx b/docs-developer/integrations.mdx index cfcba58..4d36221 100644 --- a/docs-developer/integrations.mdx +++ b/docs-developer/integrations.mdx @@ -18,6 +18,24 @@ icon: plug 讯飞接入需要将 `appId`、`apiKey`、`apiSecret` 写入 `secrets`，WebSocket 地址及音色等参数写入 `values`。 +### 百炼双向流式 TTS + +在模型资源中选择 `Alibaba Cloud Streaming TTS`（`dashscope-tts`），填写四个字段： + +- `modelId`：Qwen-Audio-TTS 3.1/3.0 的 plus、flash，或 CosyVoice v3.5/v3 的 plus、flash、v2。 +- `apiUrl`：北京为 `wss://.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference`，其他地域使用对应业务空间域名。 +- `apiKey`：对应地域和业务空间的 API Key。 +- `voice`：与模型匹配的系统音色或已创建成功的音色 ID。CosyVoice v3.5 需要复刻或设计音色。 + +保留 `cosyvoice-v1` 配置兼容，但服务是否仍可用以账号实际调用为准。 +已有资源若填写旧的 HTTP `/services/audio/tts` 地址，必须手动更新为上述 WebSocket 地址；种子文件更新不会覆盖已有数据库资源。 + +适配器逐段发送 LLM 文本，同时接收音频，每轮回复使用独立连接。固定请求 24 kHz 单声道 PCM，由服务适配到流水线采样率；不开放语速、情感、SSML 等高级参数。打断会关闭该轮连接。 +“测试连接”会实际合成一句短文本，收到音频和完成事件才返回成功，会产生少量合成用量。 + +此接口不支持 Qwen3-TTS-Realtime、Sambert 或 MiniMax，它们需要不同协议适配。 +协议参考：[百炼 WebSocket API](https://help.aliyun.com/zh/model-studio/cosyvoice-websocket-api)。 + ### Qwen-Audio Realtime 在「组件库 / 模型资源」中新建 `Alibaba Cloud Qwen-Audio Realtime` 资源：