diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index ae170b6..8b8e146 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -95,6 +95,7 @@ class BrainRuntime: set_vision_scope: Callable[[dict[str, Any]], None] | None = None vision_function: Any = None set_input_enabled: Callable[[bool], None] | None = None + interrupt_output: Callable[[], Awaitable[None]] | None = None apply_turn_config: ( Callable[[bool, dict[str, Any]], Awaitable[None]] | None ) = None diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index c1333c6..ae1dd91 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -19,7 +19,6 @@ from pipecat.flows import ( NodeConfig, ) from pipecat.frames.frames import ( - InterruptionFrame, LLMRunFrame, LLMUpdateSettingsFrame, OutputTransportMessageUrgentFrame, @@ -1098,10 +1097,12 @@ class WorkflowBrain(BaseBrain): completion_policy == MESSAGE_CONFIRMATION and result.action == "confirmed" ): - # The client-tool result closes the confirmation gate, while - # InterruptionFrame closes any speech still owned by this - # Message before the next node can enqueue new output. - await runtime.queue_frame(InterruptionFrame()) + if runtime.interrupt_output is None: + raise RuntimeError("当前管线不支持中断 Message 确认语音") + # Keep the client-tool result as the confirmation gate, then + # wait for the pipeline-owned interruption boundary before the + # next node is allowed to enqueue new output. + await runtime.interrupt_output() await self._emit_trace( "message_completed", nodeId=node_id, diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index d156056..fe4eb5d 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -104,6 +104,17 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = ( "先调用 search_knowledge_base 检索；回答资料事实时只根据检索内容，" "资料不足要明确说明。" ) + + +async def _interrupt_pipeline_output( + source: FrameProcessor, + worker: PipelineWorker, +) -> None: + """Broadcast an interruption and wait until it crosses the media pipeline.""" + await source.broadcast_interruption() + await worker.flush_pipeline(timeout=1.0) + + def _compact_knowledge_metadata(value: str, max_length: int) -> str: """Keep tool metadata useful without letting it dominate the model context.""" compact = " ".join(value.split()) @@ -733,6 +744,10 @@ async def run_pipeline( current_enable_interrupt = enable_interrupt current_turn_config = normalized + async def interrupt_output() -> None: + """Stop active output through the same boundary as text/voice input.""" + await _interrupt_pipeline_output(user_input, worker) + def set_system_prompt(text: str) -> None: """替换上下文里的系统提示(节点切换时整体替换,而非追加)。""" @@ -766,6 +781,7 @@ async def run_pipeline( set_vision_scope=lambda scope: workflow_vision_scope.update(scope), vision_function=workflow_vision_function, set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), + interrupt_output=interrupt_output, apply_turn_config=apply_workflow_turn_config, flow_global_functions=flow_global_functions, ), diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 4f77e25..8b1d601 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, patch from models import AssistantConfig, RuntimeTool from pipecat.flows import FlowManager from pipecat.frames.frames import ( - InterruptionFrame, LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -1536,14 +1535,15 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def queue_frame(frame): if isinstance(frame, TTSSpeakFrame): events.append(("speech", frame.text)) - elif isinstance(frame, InterruptionFrame): - events.append("interrupted") elif ( isinstance(frame, OutputTransportMessageUrgentFrame) and frame.message.get("event") == "message_completed" ): events.append("message_completed") + async def interrupt_output(): + events.append("interrupted") + class OrderedCallEnd(FakeCallEnd): def __init__(self): super().__init__() @@ -1580,6 +1580,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): call_end=call_end, client_tools=client_tools, set_input_enabled=input_states.append, + interrupt_output=interrupt_output, ) brain._message_stages.set_client_tools(client_tools) diff --git a/backend/tests/test_pipeline_events.py b/backend/tests/test_pipeline_events.py index e7450a1..cf1872c 100644 --- a/backend/tests/test_pipeline_events.py +++ b/backend/tests/test_pipeline_events.py @@ -2,7 +2,7 @@ from __future__ import annotations import unittest from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, patch from pipecat.frames.frames import ( BotStartedSpeakingFrame, @@ -10,6 +10,7 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, ) from services.pipecat.pipeline_events import bind_cascade_pipeline_events +from services.pipecat.pipeline import _interrupt_pipeline_output class _EventSource: @@ -80,6 +81,25 @@ class _Brain: class PipelineEventTest(unittest.IsolatedAsyncioTestCase): + async def test_output_interruption_broadcasts_before_flush_barrier(self): + events = [] + source = SimpleNamespace( + broadcast_interruption=AsyncMock( + side_effect=lambda: events.append("broadcast") + ) + ) + worker = SimpleNamespace( + flush_pipeline=AsyncMock( + side_effect=lambda **_kwargs: events.append("flush") + ) + ) + + await _interrupt_pipeline_output(source, worker) + + self.assertEqual(events, ["broadcast", "flush"]) + source.broadcast_interruption.assert_awaited_once_with() + worker.flush_pipeline.assert_awaited_once_with(timeout=1.0) + async def test_greeting_keeps_playback_timestamp_until_client_ready(self): transport = _EventSource() text_input = _EventSource()