From 79a239489ac823f9d8a1e84dd0fbc289762f7579 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 12 Aug 2026 14:25:24 +0800 Subject: [PATCH] feat: enhance call end coordination with state management - Added support for tracking call end states (pending, speaking, cancelled) in CallEndCoordinator. - Introduced a callback mechanism for state changes to allow external handling of call end states. - Updated pipeline and tests to accommodate new state management features, ensuring proper handling of call end scenarios. --- backend/services/pipecat/call_lifecycle.py | 22 ++++- backend/services/pipecat/pipeline.py | 55 ++++++++++- backend/tests/test_call_lifecycle.py | 13 ++- .../assistant-editor/debug-preview.tsx | 97 +++++++++++++++++-- frontend/src/hooks/use-voice-preview.ts | 59 +++++++++++ 5 files changed, 234 insertions(+), 12 deletions(-) diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index 9aae781..fc39e01 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -63,11 +63,19 @@ def playback_marker_for( return FixedSpeechPlaybackMarkerFrame(completion=completion) +CallEndStateHandler = Callable[[str, bool], Awaitable[None]] + + class CallEndCoordinator: """Coordinate committed and user-cancellable conversation endings.""" - def __init__(self, queue_end: Callable[[str], Awaitable[None]]): + def __init__( + self, + queue_end: Callable[[str], Awaitable[None]], + on_state_change: CallEndStateHandler | None = None, + ): self._queue_end = queue_end + self._on_state_change = on_state_change self._ending = False self._interruptible = False self._armed = False @@ -110,6 +118,7 @@ class CallEndCoordinator: self._armed = False self._finish_after_tracked_speech = False self._reason = reason or "completed" + self._schedule_state("speaking" if self._speaking else "pending") def cancel_pending_end(self) -> bool: """Cancel one pending ending when its tool explicitly allows it.""" @@ -123,8 +132,13 @@ class CallEndCoordinator: # Transport speech state remains authoritative: if audio is currently # playing, the interruption path will still emit BotStoppedSpeakingFrame. logger.info("用户打断结束语,已撤销待挂断状态") + self._schedule_state("cancelled") return True + def _schedule_state(self, state: str) -> None: + if self._on_state_change is not None: + asyncio.create_task(self._on_state_change(state, self.interruptible)) + def begin_response(self) -> None: """Start tracking speech produced by one LLM response.""" self._response_speech_started = False @@ -200,6 +214,8 @@ class CallEndCoordinator: self._speaking = True self._speech_stopped.clear() self._response_speech_started = True + if self._ending: + await self._emit_state("speaking") elif isinstance(frame, BotStoppedSpeakingFrame) and self._speaking: self._speaking = False self._speech_stopped.set() @@ -207,6 +223,10 @@ class CallEndCoordinator: logger.info("结束语播报完毕,挂断通话") await self.finish() + async def _emit_state(self, state: str) -> None: + if self._on_state_change is not None: + await self._on_state_change(state, self.interruptible) + class EndCallAfterSpeechProcessor(FrameProcessor): def __init__(self, coordinator: CallEndCoordinator): diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 043e407..e5596f7 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -201,6 +201,23 @@ def _multimodal_user_input_frame( ) +def _call_end_state_frame( + state: str, + *, + interruptible: bool, + voice_interruptible: bool, +) -> OutputTransportMessageUrgentFrame: + """Build the client event used to render the pending call-end state.""" + return OutputTransportMessageUrgentFrame( + message={ + "type": "call-end-state", + "state": state, + "interruptible": interruptible, + "voice_interruptible": voice_interruptible, + } + ) + + async def run_pipeline( transport, cfg: AssistantConfig, @@ -295,7 +312,24 @@ async def run_pipeline( ) await worker.queue_frame(EndFrame()) - call_end = CallEndCoordinator(queue_call_end) + async def queue_call_end_state(state: str, interruptible: bool) -> None: + worker = worker_holder.get("worker") + if worker is None: + return + await worker.queue_frame( + _call_end_state_frame( + state, + interruptible=interruptible, + voice_interruptible=( + interruptible and current_enable_interrupt + ), + ) + ) + + call_end = CallEndCoordinator( + queue_call_end, + on_state_change=queue_call_end_state, + ) knowledge_config = cfg.knowledge_retrieval_config knowledge_mode = str(knowledge_config.get("mode", "automatic")) @@ -976,7 +1010,24 @@ async def run_realtime_pipeline( ) await worker.queue_frame(EndFrame()) - call_end = CallEndCoordinator(queue_call_end) + async def queue_call_end_state(state: str, interruptible: bool) -> None: + worker = worker_holder.get("worker") + if worker is None: + return + await worker.queue_frame( + _call_end_state_frame( + state, + interruptible=interruptible, + voice_interruptible=( + interruptible and cfg.enableInterrupt + ), + ) + ) + + call_end = CallEndCoordinator( + queue_call_end, + on_state_change=queue_call_end_state, + ) client_tools = ClientToolBroker() async def interrupt_for_client_tool_result() -> None: diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py index 936ca17..89b4ba9 100644 --- a/backend/tests/test_call_lifecycle.py +++ b/backend/tests/test_call_lifecycle.py @@ -19,20 +19,26 @@ from services.pipecat.processors import CallEndingUserMuteStrategy class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.reasons: list[str] = [] + self.states: list[tuple[str, bool]] = [] async def queue_end(reason: str) -> None: self.reasons.append(reason) - self.coordinator = CallEndCoordinator(queue_end) + async def record_state(state: str, interruptible: bool) -> None: + self.states.append((state, interruptible)) + + self.coordinator = CallEndCoordinator(queue_end, record_state) async def test_generated_closing_text_waits_for_audio_stop(self): self.coordinator.begin_response() self.coordinator.begin("prompt_end_call") + await asyncio.sleep(0) await self.coordinator.finish_after_current_speech(has_text=True) self.assertEqual(self.reasons, []) await self.coordinator.observe(BotStartedSpeakingFrame()) self.assertEqual(self.reasons, []) + self.assertEqual(self.states[-1], ("speaking", False)) await self.coordinator.observe(BotStoppedSpeakingFrame()) self.assertEqual(self.reasons, ["prompt_end_call"]) @@ -71,12 +77,17 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): async def test_interruptible_end_keeps_input_open_and_can_be_cancelled(self): self.coordinator.begin_response() self.coordinator.begin("user_goodbye", interruptible=True) + await asyncio.sleep(0) + self.assertEqual(self.states[-1], ("pending", True)) await self.coordinator.finish_after_current_speech(has_text=True) await self.coordinator.observe(BotStartedSpeakingFrame()) + self.assertEqual(self.states[-1], ("speaking", True)) self.assertTrue(self.coordinator.ending) self.assertTrue(self.coordinator.accepts_input) self.assertTrue(self.coordinator.cancel_pending_end()) + await asyncio.sleep(0) + self.assertEqual(self.states[-1], ("cancelled", False)) self.assertFalse(self.coordinator.ending) self.assertTrue(self.coordinator.accepts_input) self.assertTrue(self.coordinator.speaking) diff --git a/frontend/src/components/assistant-editor/debug-preview.tsx b/frontend/src/components/assistant-editor/debug-preview.tsx index a8c25c1..f2ab65d 100644 --- a/frontend/src/components/assistant-editor/debug-preview.tsx +++ b/frontend/src/components/assistant-editor/debug-preview.tsx @@ -58,6 +58,7 @@ import { createUserInputId, useVoicePreview, type ChatMessage, + type CallEndState, type ClientToolDefinition, type UserInputPart, type VoicePreview, @@ -273,6 +274,7 @@ export function DebugDrawer({ )} @@ -724,6 +726,7 @@ function DeviceSelectField({ value, devices, onSelect, + disabled = false, }: { icon: React.ReactNode; ariaLabel: string; @@ -732,9 +735,11 @@ function DeviceSelectField({ value: string; devices: MediaDeviceInfo[]; onSelect: (deviceId: string) => void; + disabled?: boolean; }) { return (