From e78dc4088a863f83ecb2bd948cc51c09e72c44dd Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Tue, 14 Jul 2026 13:43:42 +0800 Subject: [PATCH] Implement response handling and speech tracking in CallEnd functionality - Add begin_response and finish_after_current_speech methods to CallEndCoordinator for better management of speech events. - Update PromptBrain to utilize new methods, ensuring proper handling of generated closing speech and tool-only calls. - Enhance tests to verify the correct behavior of speech tracking and response handling in various scenarios, including waiting for audio to finish before ending calls. - Introduce a new test suite for CallEndCoordinator to validate the interaction with speech frames. --- backend/services/brains/base.py | 4 ++ backend/services/brains/prompt_brain.py | 26 +++++++++-- backend/services/pipecat/call_lifecycle.py | 31 +++++++++---- backend/tests/test_brains.py | 53 ++++++++++++++++++++++ backend/tests/test_call_lifecycle.py | 49 ++++++++++++++++++++ 5 files changed, 151 insertions(+), 12 deletions(-) create mode 100644 backend/tests/test_call_lifecycle.py diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index ae2db18..deba421 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -51,8 +51,12 @@ class CallEndPort(Protocol): def begin(self, reason: str) -> None: ... + def begin_response(self) -> None: ... + def arm_after_speech(self) -> None: ... + async def finish_after_current_speech(self, *, has_text: bool) -> None: ... + async def finish(self) -> None: ... diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index 7d1bc50..71f17c4 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -35,6 +35,7 @@ class PromptBrain(BaseBrain): self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store) self._runtime: BrainRuntime | None = None + self._waiting_for_generated_end_speech = False async def greeting(self, cfg: AssistantConfig) -> str: return self._store.render(cfg.greeting) if self._dynamic_enabled else cfg.greeting @@ -49,6 +50,7 @@ class PromptBrain(BaseBrain): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: self._runtime = runtime + self._waiting_for_generated_end_speech = False schemas: list[FunctionSchema] = [] for tool in cfg.tools: if tool.type == "end_call": @@ -67,6 +69,10 @@ class PromptBrain(BaseBrain): self._store.record("user", content) self._refresh_prompt() + async def on_assistant_text_start(self, _turn_id: str) -> None: + if self._runtime is not None: + self._runtime.call_end.begin_response() + async def on_assistant_text_end( self, _turn_id: str, @@ -76,6 +82,15 @@ class PromptBrain(BaseBrain): if content and not interrupted: self._store.record("agent", content, completed_agent_turn=True) self._refresh_prompt() + if ( + self._waiting_for_generated_end_speech + and self._runtime is not None + and self._runtime.call_end.ending + ): + self._waiting_for_generated_end_speech = False + await self._runtime.call_end.finish_after_current_speech( + has_text=bool(content.strip()) and not interrupted + ) def _refresh_prompt(self) -> None: if self._dynamic_enabled and self._runtime is not None: @@ -104,8 +119,7 @@ class PromptBrain(BaseBrain): ) return schema, call_http - @staticmethod - def _make_end_call_tool(tool, runtime: BrainRuntime): + def _make_end_call_tool(self, tool, runtime: BrainRuntime): config = (tool.definition or {}).get("config") or {} message_type = str(config.get("message_type") or "none") custom_message = str(config.get("custom_message") or "").strip() @@ -113,14 +127,18 @@ class PromptBrain(BaseBrain): async def end_call(params: FunctionCallParams) -> None: reason = str(params.arguments.get("reason") or "end_call_tool").strip() + uses_custom_message = message_type == "custom" and bool(custom_message) + self._waiting_for_generated_end_speech = not uses_custom_message runtime.call_end.begin(reason) await params.result_callback( {"status": "success", "action": "ending_call"}, properties=FunctionCallResultProperties(run_llm=False), ) - if message_type != "custom" or not custom_message: - await runtime.call_end.finish() + if not uses_custom_message: + # The model may have already streamed a spoken goodbye before + # invoking this tool. Decide at assistant-text-end whether to + # wait for that TTS audio or finish immediately for tool-only calls. return turn_id = uuid4().hex diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index 27eb6c5..6e65872 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -17,6 +17,7 @@ class CallEndCoordinator: self._ending = False self._armed = False self._speaking = False + self._response_speech_started = False self._finished = False self._reason = "completed" @@ -28,7 +29,22 @@ class CallEndCoordinator: self._ending = True self._reason = reason or "completed" + def begin_response(self) -> None: + """Start tracking speech produced by one LLM response.""" + self._response_speech_started = False + def arm_after_speech(self) -> None: + """Wait for the next observed bot speech to finish.""" + self._armed = True + + async def finish_after_current_speech(self, *, has_text: bool) -> None: + """Finish now if speech is absent/done, otherwise wait for its stop.""" + if not has_text: + await self.finish() + return + if self._response_speech_started and not self._speaking: + await self.finish() + return self._armed = True async def finish(self) -> None: @@ -38,15 +54,14 @@ class CallEndCoordinator: await self._queue_end(self._reason) async def observe(self, frame) -> None: - if isinstance(frame, BotStartedSpeakingFrame) and self._armed: + if isinstance(frame, BotStartedSpeakingFrame): self._speaking = True - elif ( - isinstance(frame, BotStoppedSpeakingFrame) - and self._armed - and self._speaking - ): - logger.info("结束语播报完毕,挂断通话") - await self.finish() + self._response_speech_started = True + elif isinstance(frame, BotStoppedSpeakingFrame) and self._speaking: + self._speaking = False + if self._armed: + logger.info("结束语播报完毕,挂断通话") + await self.finish() class EndCallAfterSpeechProcessor(FrameProcessor): diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index a0df965..d4af2f6 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -44,14 +44,26 @@ class FakeCallEnd: self.reason = "" self.armed = False self.finished = False + self.response_started = False + self.waited_for_text: bool | None = None def begin(self, reason: str) -> None: self.ending = True self.reason = reason + def begin_response(self) -> None: + self.response_started = True + def arm_after_speech(self) -> None: self.armed = True + async def finish_after_current_speech(self, *, has_text: bool) -> None: + self.waited_for_text = has_text + if has_text: + self.armed = True + else: + await self.finish() + async def finish(self) -> None: self.finished = True @@ -295,11 +307,52 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(visible_tools[0].name, "end_call") params = FakeFunctionParams({"reason": "用户已完成咨询"}) + await brain.on_assistant_text_start("closing-turn") await llm.functions["end_call"](params) self.assertEqual(call_end.reason, "用户已完成咨询") + self.assertFalse(call_end.finished) + await brain.on_assistant_text_end("closing-turn", "", False) self.assertTrue(call_end.finished) + self.assertFalse(call_end.waited_for_text) + self.assertTrue(call_end.response_started) self.assertEqual(params.result["action"], "ending_call") + async def test_end_call_waits_for_prompt_generated_closing_speech(self): + tool = RuntimeTool( + id="end-call", + name="结束通话", + function_name="end_call", + type="end_call", + definition={"config": {"message_type": "none"}}, + ) + cfg = AssistantConfig(type="prompt", tools=[tool]) + brain = build_brain(cfg) + llm = FakeLLM() + call_end = FakeCallEnd() + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=llm, + queue_frame=lambda _frame: None, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + ), + ) + await brain.on_assistant_text_start("closing-turn") + await llm.functions["end_call"](FakeFunctionParams({})) + await brain.on_assistant_text_end( + "closing-turn", + "感谢您的来电。祝您生活愉快，再见。", + False, + ) + + self.assertFalse(call_end.finished) + self.assertTrue(call_end.armed) + self.assertTrue(call_end.waited_for_text) + async def test_http_tool_renders_secrets_and_updates_prompt_variable(self): requests = [] diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py new file mode 100644 index 0000000..779f7fe --- /dev/null +++ b/backend/tests/test_call_lifecycle.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import unittest + +from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame +from services.pipecat.call_lifecycle import CallEndCoordinator + + +class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.reasons: list[str] = [] + + async def queue_end(reason: str) -> None: + self.reasons.append(reason) + + self.coordinator = CallEndCoordinator(queue_end) + + async def test_generated_closing_text_waits_for_audio_stop(self): + self.coordinator.begin_response() + self.coordinator.begin("prompt_end_call") + await self.coordinator.finish_after_current_speech(has_text=True) + + self.assertEqual(self.reasons, []) + await self.coordinator.observe(BotStartedSpeakingFrame()) + self.assertEqual(self.reasons, []) + await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.assertEqual(self.reasons, ["prompt_end_call"]) + + async def test_already_played_generated_text_finishes_immediately(self): + self.coordinator.begin_response() + await self.coordinator.observe(BotStartedSpeakingFrame()) + await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.coordinator.begin("prompt_end_call") + + await self.coordinator.finish_after_current_speech(has_text=True) + + self.assertEqual(self.reasons, ["prompt_end_call"]) + + async def test_tool_only_end_call_finishes_without_waiting(self): + self.coordinator.begin_response() + self.coordinator.begin("tool_only") + + await self.coordinator.finish_after_current_speech(has_text=False) + + self.assertEqual(self.reasons, ["tool_only"]) + + +if __name__ == "__main__": + unittest.main()