diff --git a/backend/db/session.py b/backend/db/session.py index 7efdce6..4cbdca0 100644 --- a/backend/db/session.py +++ b/backend/db/session.py @@ -59,6 +59,7 @@ async def sync_default_tools() -> None: "message_type": "none", "custom_message": "", "capture_reason": True, + "allow_end_interruption": False, } ) default_system_tools.append( diff --git a/backend/schemas.py b/backend/schemas.py index 1aca5cd..cc31a69 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -300,6 +300,7 @@ class SystemToolConfig(CamelModel): message_type: Literal["none", "custom"] = "none" custom_message: str = "" capture_reason: bool = True + allow_end_interruption: bool = False class HttpToolConfig(CamelModel): diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 55b4bbe..1181463 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -58,7 +58,12 @@ class CallEndPort(Protocol): @property def ending(self) -> bool: ... - def begin(self, reason: str) -> None: ... + @property + def accepts_input(self) -> bool: ... + + def begin(self, reason: str, *, interruptible: bool = False) -> None: ... + + def cancel_pending_end(self) -> bool: ... def begin_response(self) -> None: ... diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index e1d9a1a..b21c0f4 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -488,15 +488,12 @@ 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 - ): + if self._waiting_for_generated_end_speech: self._waiting_for_generated_end_speech = False - await self._runtime.call_end.finish_after_current_speech( - has_text=bool(content.strip()) and not interrupted - ) + if self._runtime is not None and self._runtime.call_end.ending: + 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: @@ -594,12 +591,18 @@ class PromptBrain(BaseBrain): message_type = str(config.get("message_type") or "none") custom_message = str(config.get("custom_message") or "").strip() capture_reason = bool(config.get("capture_reason", True)) + allow_end_interruption = bool( + config.get("allow_end_interruption", False) + ) 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) + runtime.call_end.begin( + reason, + interruptible=allow_end_interruption, + ) await params.result_callback( {"status": "success", "action": "ending_call"}, properties=FunctionCallResultProperties(run_llm=False), @@ -610,6 +613,8 @@ class PromptBrain(BaseBrain): # invoking this tool. Decide at assistant-text-end whether to # wait for that TTS audio or finish immediately for tool-only calls. return + if not runtime.call_end.ending: + return turn_id = uuid4().hex timestamp = time_now_iso8601() @@ -634,6 +639,8 @@ class PromptBrain(BaseBrain): await runtime.queue_frame( OutputTransportMessageUrgentFrame(message=message) ) + if not runtime.call_end.ending: + return runtime.call_end.arm_after_speech() await runtime.queue_frame( TTSSpeakFrame(custom_message, append_to_context=False) diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 07b4eb9..517e939 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -603,15 +603,12 @@ class WorkflowBrain(BaseBrain): self._state.consume_user_turn() if self._engine.node_type(self._state.current_node_id) == "agent": self._state.status = WorkflowStatus.WAITING_USER - if ( - self._waiting_for_generated_end_speech - and self._runtime is not None - and self._runtime.call_end.ending - ): + if self._waiting_for_generated_end_speech: self._waiting_for_generated_end_speech = False - await self._runtime.call_end.finish_after_current_speech( - has_text=bool(content.strip()) and not interrupted - ) + if self._runtime is not None and self._runtime.call_end.ending: + await self._runtime.call_end.finish_after_current_speech( + has_text=bool(content.strip()) and not interrupted + ) async def on_realtime_user_speech_started(self) -> None: if self._realtime_controller is not None: @@ -1090,13 +1087,19 @@ class WorkflowBrain(BaseBrain): message_type = str(config.get("message_type") or "none") custom_message = str(config.get("custom_message") or "").strip() capture_reason = bool(config.get("capture_reason", True)) + allow_end_interruption = bool( + config.get("allow_end_interruption", False) + ) async def handler(args, _flow_manager): reason = str((args or {}).get("reason") or "end_conversation").strip() uses_custom_message = message_type == "custom" and bool(custom_message) self._waiting_for_generated_end_speech = not uses_custom_message runtime = self._require_runtime() - runtime.call_end.begin(reason) + runtime.call_end.begin( + reason, + interruptible=allow_end_interruption, + ) if uses_custom_message: await self._queue_visible_speech( custom_message, diff --git a/backend/services/pipecat/call_lifecycle.py b/backend/services/pipecat/call_lifecycle.py index 391b970..9aae781 100644 --- a/backend/services/pipecat/call_lifecycle.py +++ b/backend/services/pipecat/call_lifecycle.py @@ -64,11 +64,12 @@ def playback_marker_for( class CallEndCoordinator: - """End immediately or after the currently armed closing speech finishes.""" + """Coordinate committed and user-cancellable conversation endings.""" def __init__(self, queue_end: Callable[[str], Awaitable[None]]): self._queue_end = queue_end self._ending = False + self._interruptible = False self._armed = False self._speaking = False self._speech_stopped = asyncio.Event() @@ -83,6 +84,15 @@ class CallEndCoordinator: def ending(self) -> bool: return self._ending + @property + def accepts_input(self) -> bool: + """Keep input open while an interruptible ending is still pending.""" + return not self._finished and (not self._ending or self._interruptible) + + @property + def interruptible(self) -> bool: + return self._ending and self._interruptible and not self._finished + @property def speaking(self) -> bool: """Whether transport output is currently producing bot speech.""" @@ -92,16 +102,37 @@ class CallEndCoordinator: """Wait for the transport-owned bot speech boundary.""" await self._speech_stopped.wait() - def begin(self, reason: str) -> None: + def begin(self, reason: str, *, interruptible: bool = False) -> None: + if self._finished: + return self._ending = True + self._interruptible = interruptible + self._armed = False + self._finish_after_tracked_speech = False self._reason = reason or "completed" + def cancel_pending_end(self) -> bool: + """Cancel one pending ending when its tool explicitly allows it.""" + if not self.interruptible: + return False + self._ending = False + self._interruptible = False + self._armed = False + self._finish_after_tracked_speech = False + self._response_speech_started = False + # Transport speech state remains authoritative: if audio is currently + # playing, the interruption path will still emit BotStoppedSpeakingFrame. + logger.info("用户打断结束语,已撤销待挂断状态") + return True + 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.""" + if not self._ending: + return self._armed = True def track_speech(self) -> SpeechPlaybackCompletion: @@ -128,12 +159,16 @@ class CallEndCoordinator: async def arm_after_tracked_speech(self) -> None: """Finish after every already queued fixed utterance has played.""" + if not self._ending: + return self._finish_after_tracked_speech = True if not self._tracked_speech_completions: await self.finish() 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 self._ending: + return if not has_text: await self.finish() return @@ -143,9 +178,10 @@ class CallEndCoordinator: self._armed = True async def finish(self) -> None: - if self._finished: + if self._finished or not self._ending: return self._finished = True + self._interruptible = False await self._queue_end(self._reason) async def observe(self, frame) -> None: diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 71f4d41..043e407 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -68,6 +68,7 @@ from services.pipecat.turn_config import ( create_user_turn_strategies, create_vad_analyzer, create_vad_params, + is_configured_user_turn_start, ) from services.pipecat.processors import ( KNOWLEDGE_CONTEXT_MARKER, @@ -330,6 +331,16 @@ async def run_pipeline( ) ) input_state = {"enabled": True} + current_enable_interrupt = cfg.enableInterrupt + current_turn_config = dict(cfg.turnConfig) + + def is_voice_end_interruption(frame) -> bool: + return ( + input_state["enabled"] + and current_enable_interrupt + and is_configured_user_turn_start(frame, current_turn_config) + ) + tool_interruption_strategy = ToolInterruptionUserMuteStrategy( { tool.function_name: policy_for_tool(tool).execution_mode @@ -355,7 +366,11 @@ async def run_pipeline( vad_analyzer=create_vad_analyzer(cfg.turnConfig), user_mute_strategies=[ CallEndingUserMuteStrategy( - lambda: call_end.ending or not input_state["enabled"] + lambda: ( + not call_end.accepts_input or not input_state["enabled"] + ), + is_user_interruption=is_voice_end_interruption, + cancel_pending_end=call_end.cancel_pending_end, ), tool_interruption_strategy, ], @@ -369,10 +384,11 @@ async def run_pipeline( assistant_aggregator = PassthroughLLMAssistantAggregator(context) user_input = UserInputProcessor( should_ignore_input=lambda: ( - call_end.ending + not call_end.accepts_input or not input_state["enabled"] or tool_interruption_strategy.is_muted - ) + ), + cancel_pending_end=call_end.cancel_pending_end, ) session_update = SessionUpdateProcessor( brain.on_session_update, @@ -723,9 +739,6 @@ async def run_pipeline( **current_voice_services, }, ) - current_enable_interrupt = cfg.enableInterrupt - current_turn_config = dict(cfg.turnConfig) - async def apply_workflow_turn_config( enable_interrupt: bool, turn_config: dict[str, Any], @@ -983,11 +996,12 @@ async def run_realtime_pipeline( client_tools.set_interrupt_handler(interrupt_for_client_tool_result) user_input = RealtimeUserInputProcessor( should_ignore_input=lambda: ( - call_end.ending or not input_state["enabled"] - ) + not call_end.accepts_input or not input_state["enabled"] + ), + cancel_pending_end=call_end.cancel_pending_end, ) input_gate = RealtimeInputAudioGateProcessor( - lambda: not call_end.ending and input_state["enabled"] + lambda: call_end.accepts_input and input_state["enabled"] ) dynamic_variables = RealtimeDynamicVariableProcessor(brain, cfg, realtime) diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index 702cf0d..42233bd 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -230,9 +230,14 @@ def message_text(message: dict) -> str: class UserInputProcessor(FrameProcessor): """Validate app inputs and coordinate their interruption boundary.""" - def __init__(self, should_ignore_input: Callable[[], bool] | None = None): + def __init__( + self, + should_ignore_input: Callable[[], bool] | None = None, + cancel_pending_end: Callable[[], bool] | None = None, + ): super().__init__() self._should_ignore_input = should_ignore_input or (lambda: False) + self._cancel_pending_end = cancel_pending_end or (lambda: False) self._register_event_handler("on_user_input") self._register_event_handler("on_client_ready") @@ -256,6 +261,9 @@ class UserInputProcessor(FrameProcessor): await self.push_frame(frame, direction) return + if user_input.run_immediately and user_input.interrupt: + self._cancel_pending_end() + if self._should_ignore_input(): logger.debug("当前会话输入已暂停,忽略用户输入") await self._emit_result( @@ -371,15 +379,25 @@ class SessionUpdateProcessor(FrameProcessor): class CallEndingUserMuteStrategy(BaseUserMuteStrategy): - """Keep user media muted after an end-call tool starts terminating a call.""" + """Mute committed endings and cancel interruptible ones on user activity.""" - def __init__(self, is_call_ending: Callable[[], bool]): + def __init__( + self, + should_mute: Callable[[], bool], + *, + is_user_interruption: Callable[[object], bool] | None = None, + cancel_pending_end: Callable[[], bool] | None = None, + ): super().__init__() - self._is_call_ending = is_call_ending + self._should_mute = should_mute + self._is_user_interruption = is_user_interruption or (lambda _frame: False) + self._cancel_pending_end = cancel_pending_end or (lambda: False) async def process_frame(self, frame) -> bool: await super().process_frame(frame) - return self._is_call_ending() + if self._is_user_interruption(frame): + self._cancel_pending_end() + return self._should_mute() @dataclass @@ -570,9 +588,14 @@ class RealtimeInputAudioGateProcessor(FrameProcessor): class RealtimeUserInputProcessor(FrameProcessor): """Route text-only user-input messages to a realtime service.""" - def __init__(self, should_ignore_input: Callable[[], bool] | None = None): + def __init__( + self, + should_ignore_input: Callable[[], bool] | None = None, + cancel_pending_end: Callable[[], bool] | None = None, + ): super().__init__() self._should_ignore_input = should_ignore_input or (lambda: False) + self._cancel_pending_end = cancel_pending_end or (lambda: False) self._register_event_handler("on_user_input") async def process_frame(self, frame, direction: FrameDirection): @@ -590,6 +613,8 @@ class RealtimeUserInputProcessor(FrameProcessor): if user_input is None: await self.push_frame(frame, direction) return + if user_input.run_immediately and user_input.interrupt: + self._cancel_pending_end() if self._should_ignore_input(): await self._emit_error( user_input.input_id, diff --git a/backend/services/pipecat/turn_config.py b/backend/services/pipecat/turn_config.py index 0d97a1a..81724ed 100644 --- a/backend/services/pipecat/turn_config.py +++ b/backend/services/pipecat/turn_config.py @@ -7,6 +7,7 @@ from typing import Any from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import TranscriptionFrame, VADUserStartedSpeakingFrame from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( LLMUserAggregator, @@ -100,6 +101,14 @@ def create_user_turn_strategies( return UserTurnStrategies(start=start, stop=stop) +def is_configured_user_turn_start(frame, turn_config: dict[str, Any]) -> bool: + """Match the frame used by the configured barge-in start strategy.""" + barge_in = _section(turn_config, "barge_in", "bargeIn") + if barge_in.get("strategy", "vad") == "transcription": + return isinstance(frame, TranscriptionFrame) and bool(frame.text.strip()) + return isinstance(frame, VADUserStartedSpeakingFrame) + + class ConfigurableLLMUserAggregator(LLMUserAggregator): """LLM user aggregator with one stable project-level runtime update API. diff --git a/backend/services/workflow/realtime.py b/backend/services/workflow/realtime.py index dccc39c..fe50051 100644 --- a/backend/services/workflow/realtime.py +++ b/backend/services/workflow/realtime.py @@ -235,6 +235,7 @@ class WorkflowRealtimeController: async def on_user_speech_started(self) -> None: """Let an interruptible Message finish its deterministic continuation.""" + self._runtime.call_end.cancel_pending_end() current = self._state.current_node_id if ( self._engine.node_type(current) != "message" @@ -263,11 +264,12 @@ class WorkflowRealtimeController: self._state.consume_user_turn() if self._engine.node_type(self._state.current_node_id) == "agent": self._state.status = WorkflowStatus.WAITING_USER - if self._waiting_generated_end and self._runtime.call_end.ending: + if self._waiting_generated_end: self._waiting_generated_end = False - await self._runtime.call_end.finish_after_current_speech( - has_text=bool(content.strip()) and not interrupted - ) + if self._runtime.call_end.ending: + await self._runtime.call_end.finish_after_current_speech( + has_text=bool(content.strip()) and not interrupted + ) async def dispatch_tool( self, @@ -759,10 +761,18 @@ class WorkflowRealtimeController: request_handoff, ) if kind == "end_conversation": + config = (tool.definition or {}).get("config") or {} + allow_end_interruption = bool( + config.get("allow_end_interruption", False) + ) + async def end_conversation(arguments: dict[str, Any]) -> RealtimeToolResult: reason = str(arguments.get("reason") or "end_conversation") self._waiting_generated_end = True - self._runtime.call_end.begin(reason) + self._runtime.call_end.begin( + reason, + interruptible=allow_end_interruption, + ) return RealtimeToolResult( {"status": "success", "action": "ending_call"} ) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 5dd970a..0d89aa7 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -60,10 +60,24 @@ class FakeCallEnd: self.response_started = False self.waited_for_text: bool | None = None self.tracked_speeches = 0 + self.interruptible = False - def begin(self, reason: str) -> None: + @property + def accepts_input(self) -> bool: + return not self.ending or self.interruptible + + def begin(self, reason: str, *, interruptible: bool = False) -> None: self.ending = True self.reason = reason + self.interruptible = interruptible + + def cancel_pending_end(self) -> bool: + if not self.ending or not self.interruptible: + return False + self.ending = False + self.interruptible = False + self.armed = False + return True def begin_response(self) -> None: self.response_started = True @@ -105,14 +119,14 @@ async def noop_queue_frame(_frame): return None -def system_runtime_tool(kind: str) -> RuntimeTool: +def system_runtime_tool(kind: str, **config) -> RuntimeTool: return RuntimeTool( id=f"tool_{kind}", name=kind, function_name=kind, type="system", description=f"{kind} system tool", - definition={"type": "system", "config": {"kind": kind}}, + definition={"type": "system", "config": {"kind": kind, **config}}, ) @@ -175,6 +189,12 @@ class BrainRegistryTests(unittest.TestCase): ): self.assertEqual(SystemToolConfig(kind=kind).kind, kind) + def test_end_interruption_is_opt_in(self): + self.assertFalse(SystemToolConfig().allow_end_interruption) + self.assertTrue( + SystemToolConfig(allowEndInterruption=True).allow_end_interruption + ) + def test_system_tool_config_rejects_unknown_kind(self): with self.assertRaises(ValueError): SystemToolConfig(kind="magic") @@ -1018,6 +1038,36 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): self.assertFalse(call_end.finished) self.assertTrue(call_end.armed) + async def test_end_conversation_passes_interruptible_tool_setting(self): + cfg = AssistantConfig( + type="prompt", + tools=[ + system_runtime_tool( + "end_conversation", + allow_end_interruption=True, + ) + ], + ) + 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 llm.functions["end_conversation"](FakeFunctionParams({})) + + self.assertTrue(call_end.ending) + self.assertTrue(call_end.interruptible) + async def test_end_conversation_finishes_when_no_speech(self): cfg = AssistantConfig( type="prompt", @@ -1481,7 +1531,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): call_end=call_end, ) tool = brain._workflow_end_conversation_tool( - system_runtime_tool("end_conversation"), + system_runtime_tool( + "end_conversation", + allow_end_interruption=True, + ), "agent", ) @@ -1489,6 +1542,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): result = await tool.handler({"reason": "用户告别"}, None) self.assertEqual(result["action"], "ending_call") self.assertTrue(call_end.ending) + self.assertTrue(call_end.interruptible) self.assertEqual(call_end.reason, "用户告别") self.assertTrue(getattr(tool.handler, "_suppress_followup_llm")) diff --git a/backend/tests/test_call_lifecycle.py b/backend/tests/test_call_lifecycle.py index 43b6699..936ca17 100644 --- a/backend/tests/test_call_lifecycle.py +++ b/backend/tests/test_call_lifecycle.py @@ -7,11 +7,13 @@ from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, InterruptionFrame, + VADUserStartedSpeakingFrame, ) from services.pipecat.call_lifecycle import ( CallEndCoordinator, playback_marker_for, ) +from services.pipecat.processors import CallEndingUserMuteStrategy class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): @@ -66,6 +68,59 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(self.reasons, ["tool_only"]) + 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 self.coordinator.finish_after_current_speech(has_text=True) + await self.coordinator.observe(BotStartedSpeakingFrame()) + + self.assertTrue(self.coordinator.ending) + self.assertTrue(self.coordinator.accepts_input) + self.assertTrue(self.coordinator.cancel_pending_end()) + self.assertFalse(self.coordinator.ending) + self.assertTrue(self.coordinator.accepts_input) + self.assertTrue(self.coordinator.speaking) + + await self.coordinator.observe(BotStoppedSpeakingFrame()) + self.assertFalse(self.coordinator.speaking) + self.assertEqual(self.reasons, []) + + async def test_committed_end_rejects_cancellation(self): + self.coordinator.begin("required_shutdown") + + self.assertFalse(self.coordinator.accepts_input) + self.assertFalse(self.coordinator.cancel_pending_end()) + self.assertTrue(self.coordinator.ending) + + async def test_late_arm_and_finish_cannot_restore_cancelled_end(self): + self.coordinator.begin("user_goodbye", interruptible=True) + self.assertTrue(self.coordinator.cancel_pending_end()) + + self.coordinator.arm_after_speech() + await self.coordinator.arm_after_tracked_speech() + await self.coordinator.finish_after_current_speech(has_text=False) + await self.coordinator.finish() + await self.coordinator.observe(BotStartedSpeakingFrame()) + await self.coordinator.observe(BotStoppedSpeakingFrame()) + + self.assertEqual(self.reasons, []) + + async def test_voice_turn_start_cancels_interruptible_end_before_muting(self): + strategy = CallEndingUserMuteStrategy( + lambda: not self.coordinator.accepts_input, + is_user_interruption=lambda frame: isinstance( + frame, VADUserStartedSpeakingFrame + ), + cancel_pending_end=self.coordinator.cancel_pending_end, + ) + self.coordinator.begin("user_goodbye", interruptible=True) + + muted = await strategy.process_frame(VADUserStartedSpeakingFrame()) + + self.assertFalse(muted) + self.assertFalse(self.coordinator.ending) + self.assertEqual(self.reasons, []) + async def test_workflow_end_waits_for_every_queued_fixed_speech(self): first_completion = self.coordinator.track_speech() second_completion = self.coordinator.track_speech() @@ -134,6 +189,19 @@ class CallEndCoordinatorTest(unittest.IsolatedAsyncioTestCase): await marker.completion.mark_played() self.assertTrue(completion.done()) + async def test_interrupted_fixed_end_does_not_finish_after_cancellation(self): + completion = self.coordinator.track_speech() + marker = playback_marker_for(completion) + marker.completion.mark_queued() + self.coordinator.begin("user_goodbye", interruptible=True) + await self.coordinator.arm_after_tracked_speech() + + self.assertTrue(self.coordinator.cancel_pending_end()) + await self.coordinator.observe(InterruptionFrame()) + + self.assertTrue(completion.done()) + self.assertEqual(self.reasons, []) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_user_input.py b/backend/tests/test_user_input.py index 9df12dd..cedfe99 100644 --- a/backend/tests/test_user_input.py +++ b/backend/tests/test_user_input.py @@ -1,8 +1,19 @@ import unittest +from unittest.mock import AsyncMock -from pipecat.frames.frames import LLMMessagesAppendFrame, UserImageRawFrame +from pipecat.frames.frames import ( + InputTransportMessageFrame, + LLMMessagesAppendFrame, + UserImageRawFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from services.pipecat.call_lifecycle import CallEndCoordinator from services.pipecat.pipeline import _multimodal_user_input_frame -from services.pipecat.processors import UserInputError, parse_user_input +from services.pipecat.processors import ( + UserInputError, + UserInputProcessor, + parse_user_input, +) class UserInputParserTests(unittest.TestCase): @@ -127,5 +138,40 @@ class UserInputParserTests(unittest.TestCase): ) +class UserInputProcessorTests(unittest.IsolatedAsyncioTestCase): + async def test_interrupting_text_cancels_pending_end_before_mute_check(self): + async def queue_end(_reason: str) -> None: + self.fail("撤销后不应挂断") + + call_end = CallEndCoordinator(queue_end) + call_end.begin("user_goodbye", interruptible=True) + processor = UserInputProcessor( + should_ignore_input=lambda: not call_end.accepts_input, + cancel_pending_end=call_end.cancel_pending_end, + ) + processor._call_event_handler = AsyncMock() + processor.broadcast_interruption = AsyncMock() + + await processor.process_frame( + InputTransportMessageFrame( + message={ + "type": "user-input", + "schema_version": 1, + "input_id": "input_interrupt_end", + "parts": [{"type": "input_text", "text": "等等"}], + "options": { + "run_immediately": True, + "interrupt": True, + }, + } + ), + FrameDirection.DOWNSTREAM, + ) + + self.assertFalse(call_end.ending) + processor._call_event_handler.assert_awaited_once() + processor.broadcast_interruption.assert_awaited_once() + + if __name__ == "__main__": unittest.main() diff --git a/frontend/src/components/pages/ComponentsToolsPage.tsx b/frontend/src/components/pages/ComponentsToolsPage.tsx index 18fe86b..9a93a2b 100644 --- a/frontend/src/components/pages/ComponentsToolsPage.tsx +++ b/frontend/src/components/pages/ComponentsToolsPage.tsx @@ -95,6 +95,7 @@ type ToolForm = { messageType: "none" | "custom"; customMessage: string; captureReason: boolean; + allowEndInterruption: boolean; method: HttpMethod; url: string; timeoutSeconds: string; @@ -125,6 +126,7 @@ function blankForm(): ToolForm { messageType: "none", customMessage: "", captureReason: true, + allowEndInterruption: false, method: "GET", url: "", timeoutSeconds: "15", @@ -159,6 +161,8 @@ function formFromTool(tool: Tool): ToolForm { base.messageType = tool.definition.config.messageType; base.customMessage = tool.definition.config.customMessage; base.captureReason = tool.definition.config.captureReason; + base.allowEndInterruption = + tool.definition.config.allowEndInterruption ?? false; return base; } if (tool.definition.type === "mcp") return base; @@ -262,6 +266,7 @@ function payloadFromForm(form: ToolForm): ToolUpsert { ? form.customMessage : "", captureReason: form.captureReason, + allowEndInterruption: form.allowEndInterruption, }, }, }; @@ -958,6 +963,23 @@ function SystemToolFields({ } /> +
+
+
允许用户撤销结束
+
+ 结束语期间收到有效输入时撤销结束；语音遵循 Agent 的打断配置 +
+
+ + setForm((current) => ({ + ...current, + allowEndInterruption, + })) + } + /> +
)} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index e74b98b..65a31dc 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -513,6 +513,7 @@ export type SystemToolDefinition = { messageType: "none" | "custom"; customMessage: string; captureReason: boolean; + allowEndInterruption?: boolean; }; };