diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 41fe75a..ae170b6 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -95,7 +95,6 @@ class BrainRuntime: set_vision_scope: Callable[[dict[str, Any]], None] | None = None vision_function: Any = None set_input_enabled: Callable[[bool], None] | None = None - wait_for_output_stopped: 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 9739e59..46dc6a3 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -1093,16 +1093,6 @@ class WorkflowBrain(BaseBrain): ), ) if result.succeeded: - if ( - completion_policy == MESSAGE_CONFIRMATION - and result.action == "confirmed" - ): - if runtime.wait_for_output_stopped is None: - raise RuntimeError("当前管线不支持等待 Message 确认语音停止") - # ClientToolBroker has already broadcast the interruption - # before resolving the confirmation. Keep the next node behind - # the pipeline and transport playback boundaries. - await runtime.wait_for_output_stopped() await self._emit_trace( "message_completed", nodeId=node_id, diff --git a/backend/services/client_tools.py b/backend/services/client_tools.py index 863c1a0..5f6ca83 100644 --- a/backend/services/client_tools.py +++ b/backend/services/client_tools.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +from collections import deque +from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Any, Literal, Protocol @@ -30,6 +32,14 @@ class _PendingClientToolCall: interrupt_on_result: bool = False +@dataclass(frozen=True) +class _DeferredClientToolResult: + """A successful result waiting for the media interruption barrier.""" + + future: asyncio.Future[dict[str, Any]] + response: dict[str, Any] + + class ClientToolPort(Protocol): async def call( self, @@ -49,8 +59,17 @@ class ClientToolBroker(FrameProcessor): def __init__(self) -> None: super().__init__() self._pending: dict[str, _PendingClientToolCall] = {} + self._deferred_results: deque[_DeferredClientToolResult] = deque() + self._interrupt_handler: Callable[[], Awaitable[None]] | None = None self._closed_message: str | None = None + def set_interrupt_handler( + self, + handler: Callable[[], Awaitable[None]], + ) -> None: + """Use a processor downstream of muted user input as interrupt source.""" + self._interrupt_handler = handler + async def call( self, function_name: str, @@ -142,21 +161,42 @@ class ClientToolBroker(FrameProcessor): logger.debug(f"忽略未知或过期的客户端工具结果: {tool_call_id}") return - status = str(message.get("status") or "error") - # Keep interruption in the input-frame path, matching text input. - # Resolve the tool future only after the interruption has been - # broadcast, so the workflow cannot enqueue its next node first. - if status == "ok" and pending.interrupt_on_result: - await self.broadcast_interruption() - future = pending.future + status = str(message.get("status") or "error") if status == "ok": - future.set_result( - { - "status": "ok", - "data": message.get("data"), - } - ) + response = { + "status": "ok", + "data": message.get("data"), + } + if pending.interrupt_on_result: + # Match text input exactly: broadcasting only starts the + # interruption. The assistant aggregator calls + # on_interruption_processed() after the old response state has + # actually been cleared; only then may the workflow continue. + deferred = _DeferredClientToolResult( + future=future, + response=response, + ) + self._deferred_results.append(deferred) + try: + if self._interrupt_handler is not None: + await self._interrupt_handler() + else: + await self.broadcast_interruption() + except Exception as exc: + try: + self._deferred_results.remove(deferred) + except ValueError: + # The downstream acknowledgement may have won the + # race before an upstream broadcast failure arrived. + pass + if not future.done(): + future.set_exception( + ClientToolError("客户端工具结果中断处理失败") + ) + logger.exception(f"客户端工具结果中断失败: {exc}") + return + future.set_result(response) else: future.set_result( { @@ -166,7 +206,18 @@ class ClientToolBroker(FrameProcessor): } ) + def on_interruption_processed(self) -> None: + """Release one result after the aggregator acknowledges interruption.""" + while self._deferred_results: + deferred = self._deferred_results.popleft() + if deferred.future.done(): + continue + deferred.future.set_result(deferred.response) + logger.debug("客户端工具结果已通过中断处理边界") + return + def _fail_pending(self, message: str) -> None: + self._deferred_results.clear() for pending in self._pending.values(): future = pending.future if not future.done(): diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index b6caa7a..0086a4f 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -8,7 +8,6 @@ import asyncio import base64 -from collections.abc import Awaitable, Callable from io import BytesIO from typing import Any @@ -107,21 +106,6 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = ( ) -async def _wait_for_output_stop( - wait_until_stopped: Callable[[], Awaitable[None]], - *, - timeout_seconds: float = 3.0, -) -> None: - """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: """Keep tool metadata useful without letting it dominate the model context.""" compact = " ".join(value.split()) @@ -437,6 +421,11 @@ async def run_pipeline( should_ignore_update=lambda: call_end.ending, ) client_tools = ClientToolBroker() + # Confirmation nodes mute the user aggregator, which intentionally drops + # InterruptionFrame. Relay confirmation interruptions from the next + # processor instead so LLM, assistant aggregation, TTS, and output all see + # the frame before the workflow continues. + client_tools.set_interrupt_handler(user_turn_router.broadcast_interruption) vision_capture = VisionCaptureProcessor() knowledge_retrieval = KnowledgeRetrievalProcessor( automatic_knowledge_id, @@ -751,11 +740,6 @@ async def run_pipeline( current_enable_interrupt = enable_interrupt current_turn_config = normalized - async def wait_for_output_stopped() -> None: - """Keep workflow continuation behind the interrupted output.""" - if call_end.speaking: - await _wait_for_output_stop(call_end.wait_until_silent) - def set_system_prompt(text: str) -> None: """替换上下文里的系统提示(节点切换时整体替换,而非追加)。""" messages = context.get_messages() @@ -788,7 +772,6 @@ 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), - wait_for_output_stopped=wait_for_output_stopped, apply_turn_config=apply_workflow_turn_config, flow_global_functions=flow_global_functions, ), @@ -877,6 +860,7 @@ async def run_pipeline( vision_enabled=vision_enabled, vision_state=vision_state, submit_user_input=submit_user_input, + client_tools=client_tools, ) runner = WorkerRunner(handle_sigint=False) run_status = "completed" diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py index 5785440..70220c8 100644 --- a/backend/services/pipecat/pipeline_events.py +++ b/backend/services/pipecat/pipeline_events.py @@ -16,6 +16,7 @@ from pipecat.runner.utils import ( maybe_capture_participant_camera, ) from pipecat.utils.time import time_now_iso8601 +from services.client_tools import ClientToolBroker from services.pipecat.processors import UserInput @@ -32,6 +33,7 @@ def bind_cascade_pipeline_events( vision_enabled: bool, vision_state: dict[str, str | None], submit_user_input: Callable[[UserInput], Awaitable[None]] | None = None, + client_tools: ClientToolBroker | None = None, ) -> None: """Connect processors to transport events without owning pipeline assembly.""" @@ -159,6 +161,8 @@ def bind_cascade_pipeline_events( @assistant_aggregator.event_handler("on_interruption_processed") async def on_interruption_processed(_aggregator): + if client_tools is not None: + client_tools.on_interruption_processed() if not pending_user_inputs: return await finish_user_input(pending_user_inputs.pop(0)) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 761e9aa..953d204 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -1541,9 +1541,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ): events.append("message_completed") - async def wait_for_output_stopped(): - events.append("output_stopped") - class OrderedCallEnd(FakeCallEnd): def __init__(self): super().__init__() @@ -1580,7 +1577,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): call_end=call_end, client_tools=client_tools, set_input_enabled=input_states.append, - wait_for_output_stopped=wait_for_output_stopped, ) brain._message_stages.set_client_tools(client_tools) @@ -1606,7 +1602,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(result.action, "confirmed") self.assertFalse(call_end.playback_completion.done()) self.assertEqual(input_states, [False, True]) - self.assertEqual(events[-2:], ["output_stopped", "message_completed"]) + self.assertEqual(events[-1], "message_completed") async def test_speech_only_message_waits_for_transport_playback(self): brain = WorkflowBrain( diff --git a/backend/tests/test_client_tools.py b/backend/tests/test_client_tools.py index b513fd1..46a5ca4 100644 --- a/backend/tests/test_client_tools.py +++ b/backend/tests/test_client_tools.py @@ -138,7 +138,52 @@ class ClientToolExecutorTests(unittest.IsolatedAsyncioTestCase): class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase): - async def test_interrupts_before_resolving_configured_result(self): + async def test_uses_configured_downstream_interrupt_handler(self): + broker = ClientToolBroker() + outbound = [] + interruptions = [] + + async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): + outbound.append((frame, direction)) + + async def interrupt_after_muted_input(): + interruptions.append("downstream") + + broker.push_frame = push_frame + broker.set_interrupt_handler(interrupt_after_muted_input) + call = asyncio.create_task( + broker.call( + "show_message", + {}, + timeout_seconds=1, + response_wait_mode="session", + interrupt_on_result=True, + ) + ) + await asyncio.sleep(0) + message = outbound[0][0].message + + await broker.process_frame( + InputTransportMessageFrame( + message={ + "type": "client-tool-result", + "tool_call_id": message["tool_call_id"], + "status": "ok", + "data": {"action": "confirmed"}, + } + ), + FrameDirection.DOWNSTREAM, + ) + + self.assertEqual(interruptions, ["downstream"]) + self.assertFalse(call.done()) + broker.on_interruption_processed() + self.assertEqual( + await call, + {"status": "ok", "data": {"action": "confirmed"}}, + ) + + async def test_resolves_configured_result_after_interruption_processed(self): broker = ClientToolBroker() outbound = [] observed_future_states = [] @@ -178,6 +223,10 @@ class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(observed_future_states, [[False]]) + self.assertFalse(call.done()) + + broker.on_interruption_processed() + self.assertEqual( await call, {"status": "ok", "data": {"action": "confirmed"}}, diff --git a/backend/tests/test_pipeline_events.py b/backend/tests/test_pipeline_events.py index 158c359..1c39be0 100644 --- a/backend/tests/test_pipeline_events.py +++ b/backend/tests/test_pipeline_events.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -11,7 +10,6 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, ) from services.pipecat.pipeline_events import bind_cascade_pipeline_events -from services.pipecat.pipeline import _wait_for_output_stop class _EventSource: @@ -82,25 +80,37 @@ class _Brain: class PipelineEventTest(unittest.IsolatedAsyncioTestCase): - async def test_interrupted_output_waits_for_transport_stop(self): - events = [] + async def test_interruption_acknowledges_deferred_client_tool_result(self): + transport = _EventSource() + text_input = _EventSource() + user_aggregator = _EventSource() + assistant_aggregator = _EventSource() + worker = _Worker() + brain = _Brain(worker) + acknowledgements = [] + client_tools = SimpleNamespace( + on_interruption_processed=lambda: acknowledgements.append(True) + ) - async def wait_until_stopped(): - events.append("stopped") + bind_cascade_pipeline_events( + transport=transport, + worker=worker, + brain=brain, + context=SimpleNamespace(), + text_input=text_input, + user_aggregator=user_aggregator, + assistant_aggregator=assistant_aggregator, + greeting="", + vision_enabled=False, + vision_state={"client_id": None}, + client_tools=client_tools, + ) - await _wait_for_output_stop(wait_until_stopped) + await assistant_aggregator.handlers["on_interruption_processed"]( + assistant_aggregator + ) - 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, - ) + self.assertEqual(acknowledgements, [True]) async def test_greeting_keeps_playback_timestamp_until_client_ready(self): transport = _EventSource()