diff --git a/backend/requirements.txt b/backend/requirements.txt index b0e9716..494ee8f 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -2,7 +2,7 @@ # webrtc -> SmallWebRTCTransport / SmallWebRTCConnection + aiortc # silero -> 本地 VAD(判断用户说话起止),语音必备 # openai -> OpenAI 兼容的 LLM/STT/TTS 客户端(DeepSeek、SenseVoice、CosyVoice 都走它) -pipecat-ai[webrtc,websocket,silero,openai,mcp]==1.5.0 +pipecat-ai[webrtc,websocket,silero,openai,mcp]==1.7.0 Pillow>=11.1.0,<13 # FastGPT 类型助手:本地 SDK(包 /api/v1/chat/completions 流式 + chatId 会话) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index a22d1ce..b6caa7a 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -107,19 +107,19 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = ( ) -async def _wait_for_interrupted_output( - worker: PipelineWorker, +async def _wait_for_output_stop( + wait_until_stopped: Callable[[], Awaitable[None]], *, - wait_until_stopped: Callable[[], Awaitable[None]] | None = None, + timeout_seconds: float = 3.0, ) -> None: - """Wait until an in-band interruption crosses pipeline and playback.""" - if not await worker.flush_pipeline(timeout=2.0): - raise RuntimeError("输出中断帧未能及时穿过媒体管线") - if wait_until_stopped is not None: - try: - await asyncio.wait_for(wait_until_stopped(), timeout=2.0) - except TimeoutError as exc: - raise RuntimeError("等待客户端语音停止超时") from exc + """Wait for the transport's real stop event after an interruption.""" + try: + await asyncio.wait_for( + wait_until_stopped(), + timeout=timeout_seconds, + ) + except TimeoutError as exc: + raise RuntimeError("等待客户端语音停止超时") from exc def _compact_knowledge_metadata(value: str, max_length: int) -> str: @@ -753,14 +753,8 @@ async def run_pipeline( async def wait_for_output_stopped() -> None: """Keep workflow continuation behind the interrupted output.""" - wait_until_stopped = ( - call_end.wait_until_silent if call_end.speaking else None - ) - await _wait_for_interrupted_output( - worker, - wait_until_stopped=wait_until_stopped, - ) - + if call_end.speaking: + await _wait_for_output_stop(call_end.wait_until_silent) def set_system_prompt(text: str) -> None: """替换上下文里的系统提示(节点切换时整体替换,而非追加)。""" diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index c2b6927..92389a5 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -20,10 +20,12 @@ from services.pipecat.xfyun_super_tts import ( ) from services.pipecat.xfyun_tts import DEFAULT_XFYUN_TTS_URL, XfyunTTSService -# TTS「说完」判定的空闲时长:默认 3.0s 过长(导致工作流结束节点说完后还要等约 3s -# 才挂断,也拖慢日常轮次的交还)。设 1.0s 既能让结束语文字/音频送达,又更跟手。 -# 流式 TTS 句间音频间隔通常远小于 1s,不会把一段多句回复误判为结束。 -TTS_STOP_FRAME_TIMEOUT_S = 1.0 +# HTTP TTS may pause between response chunks. Keep Pipecat's wider default so +# a long utterance is not marked stopped and routed onward before late chunks +# arrive. WebSocket TTS has an explicit completion event and can use the short +# idle fallback without delaying the end of a call. +HTTP_TTS_STOP_FRAME_TIMEOUT_S = 3.0 +WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S = 1.0 def config_with_resource( @@ -158,7 +160,7 @@ def create_tts(cfg: AssistantConfig): volume=int(cfg.tts_values.get("volume") or 50), pitch=int(cfg.tts_values.get("pitch") or 50), push_stop_frames=True, - stop_frame_timeout_s=TTS_STOP_FRAME_TIMEOUT_S, + stop_frame_timeout_s=WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S, ) if cfg.tts_interface_type not in {"openai-tts", "dashscope-tts"}: raise ValueError(f"不支持的 TTS 接口类型: {cfg.tts_interface_type}") @@ -169,7 +171,7 @@ def create_tts(cfg: AssistantConfig): return OpenAITTSService( api_key=_require(cfg.tts_api_key, "TTS apiKey"), base_url=_require(cfg.tts_base_url, "TTS apiUrl"), - stop_frame_timeout_s=TTS_STOP_FRAME_TIMEOUT_S, + stop_frame_timeout_s=HTTP_TTS_STOP_FRAME_TIMEOUT_S, settings=OpenAITTSService.Settings( model=_require(cfg.tts_model, "TTS modelId"), voice=voice, diff --git a/backend/tests/test_pipeline_events.py b/backend/tests/test_pipeline_events.py index 6e6b576..158c359 100644 --- a/backend/tests/test_pipeline_events.py +++ b/backend/tests/test_pipeline_events.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -10,7 +11,7 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, ) from services.pipecat.pipeline_events import bind_cascade_pipeline_events -from services.pipecat.pipeline import _wait_for_interrupted_output +from services.pipecat.pipeline import _wait_for_output_stop class _EventSource: @@ -81,31 +82,25 @@ class _Brain: class PipelineEventTest(unittest.IsolatedAsyncioTestCase): - async def test_interrupted_output_waits_for_flush_and_stop(self): + async def test_interrupted_output_waits_for_transport_stop(self): events = [] async def wait_until_stopped(): events.append("stopped") - worker = SimpleNamespace( - flush_pipeline=AsyncMock( - side_effect=lambda **_kwargs: events.append("flush") or True + await _wait_for_output_stop(wait_until_stopped) + + self.assertEqual(events, ["stopped"]) + + async def test_interrupted_output_rejects_stop_timeout(self): + async def wait_forever(): + await asyncio.Event().wait() + + with self.assertRaisesRegex(RuntimeError, "语音停止超时"): + await _wait_for_output_stop( + wait_forever, + timeout_seconds=0.001, ) - ) - - await _wait_for_interrupted_output( - worker, - wait_until_stopped=wait_until_stopped, - ) - - self.assertEqual(events, ["flush", "stopped"]) - worker.flush_pipeline.assert_awaited_once_with(timeout=2.0) - - async def test_interrupted_output_rejects_flush_timeout(self): - worker = SimpleNamespace(flush_pipeline=AsyncMock(return_value=False)) - - with self.assertRaisesRegex(RuntimeError, "中断帧"): - await _wait_for_interrupted_output(worker) async def test_greeting_keeps_playback_timestamp_until_client_ready(self): transport = _EventSource() diff --git a/backend/tests/test_service_factory.py b/backend/tests/test_service_factory.py new file mode 100644 index 0000000..b4862ae --- /dev/null +++ b/backend/tests/test_service_factory.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from models import AssistantConfig +from services.pipecat.service_factory import ( + HTTP_TTS_STOP_FRAME_TIMEOUT_S, + WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S, + create_tts, +) + + +class TTSServiceFactoryTest(unittest.TestCase): + def test_http_tts_keeps_wider_audio_chunk_timeout(self): + config = AssistantConfig( + tts_interface_type="openai-tts", + tts_model="test-model", + voice="test-voice", + tts_api_key="test-key", + tts_base_url="https://tts.example.test/v1", + ) + + with patch( + "services.pipecat.service_factory.OpenAITTSService" + ) as service_type: + create_tts(config) + + self.assertEqual(HTTP_TTS_STOP_FRAME_TIMEOUT_S, 3.0) + self.assertEqual( + service_type.call_args.kwargs["stop_frame_timeout_s"], + HTTP_TTS_STOP_FRAME_TIMEOUT_S, + ) + + def test_websocket_tts_keeps_short_completion_fallback(self): + config = AssistantConfig( + tts_interface_type="xfyun-tts", + voice="test-voice", + tts_secrets={ + "appId": "test-app", + "apiKey": "test-key", + "apiSecret": "test-secret", + }, + ) + + with patch( + "services.pipecat.service_factory.XfyunTTSService" + ) as service_type: + create_tts(config) + + self.assertEqual(WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S, 1.0) + self.assertEqual( + service_type.call_args.kwargs["stop_frame_timeout_s"], + WEBSOCKET_TTS_STOP_FRAME_TIMEOUT_S, + ) + + +if __name__ == "__main__": + unittest.main()