From 11374238f337a6dadf62c049563a6ab7d6d44a43 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 5 Aug 2026 11:25:50 +0800 Subject: [PATCH] fix(realtime): complete workflow message transitions --- backend/services/pipecat/pipeline.py | 17 +++++++- .../services/pipecat/qwen_audio_realtime.py | 39 ++++++++++++++++++- backend/services/pipecat/stepfun_realtime.py | 39 ++++++++++++++++++- backend/services/workflow/realtime.py | 27 +++++++++++-- 4 files changed, 115 insertions(+), 7 deletions(-) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index bbfa09d..79b90c5 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -909,7 +909,22 @@ async def run_realtime_pipeline( call_end = CallEndCoordinator(queue_call_end) client_tools = ClientToolBroker() - client_tools.set_interrupt_handler(realtime.interrupt) + + async def interrupt_for_client_tool_result() -> None: + """Release confirmations after the realtime response is interrupted. + + Fixed Message speech intentionally suppresses the provider transcript, + so the text aggregator may have no assistant turn from which to emit an + ``on_interruption_processed`` event. The realtime adapter owns and + clears that response state itself. Wait for the provider boundary as + well, matching Pipeline's rule that the next node cannot generate until + the interrupted response has fully finished. + """ + await realtime.interrupt() + await realtime.wait_for_response_boundary() + client_tools.on_interruption_processed() + + 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"] diff --git a/backend/services/pipecat/qwen_audio_realtime.py b/backend/services/pipecat/qwen_audio_realtime.py index 465cf81..88cdb1c 100644 --- a/backend/services/pipecat/qwen_audio_realtime.py +++ b/backend/services/pipecat/qwen_audio_realtime.py @@ -27,6 +27,7 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, StartFrame, TTSAudioRawFrame, + TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService @@ -120,6 +121,7 @@ class QwenAudioRealtimeService(AIService): self._assistant_text = "" self._assistant_timestamp = "" self._greeting_request_item_id: str | None = None + self._entry_trigger_item_id: str | None = None self._user_transcript_pending = False self._user_transcript_item_id = "" self._user_transcript_timestamp = "" @@ -215,9 +217,27 @@ class QwenAudioRealtimeService(AIService): await self._cancel_active_response() await self.broadcast_interruption() - async def request_response(self) -> None: + async def request_response(self, *, trigger_text: str | None = None) -> None: + if trigger_text: + item_id = f"item_{uuid4().hex}" + self._entry_trigger_item_id = item_id + await self._send_event( + { + "type": "conversation.item.create", + "item": { + "id": item_id, + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": trigger_text}], + }, + } + ) await self._send_event({"type": "response.create"}) + async def wait_for_response_boundary(self) -> None: + """Wait until Qwen has finished or cancelled the current response.""" + await self._response_done.wait() + def set_speech_started_handler( self, handler: SpeechStartedHandler | None, @@ -387,6 +407,7 @@ class QwenAudioRealtimeService(AIService): self._deferred_assistant_messages.clear() self._tool_session.clear() self._function_names.clear() + self._entry_trigger_item_id = None self._resolve_fixed_speech() if websocket and websocket.state is State.OPEN: try: @@ -412,6 +433,7 @@ class QwenAudioRealtimeService(AIService): self._session_ready.clear() self._response_done.set() self._greeting_request_item_id = None + self._entry_trigger_item_id = None self._resolve_fixed_speech() if self._receive_task is asyncio.current_task(): self._receive_task = None @@ -480,8 +502,10 @@ class QwenAudioRealtimeService(AIService): status = response.get("status") if isinstance(response, dict) else None interrupted = status in {"cancelled", "incomplete", "interrupted", "failed"} self._response_active = False + await self.push_frame(TTSStoppedFrame()) await self._finish_assistant_text(interrupted=interrupted) await self._delete_greeting_request() + await self._delete_entry_trigger() self._resolve_fixed_speech() self._response_done.set() elif event_type == "response.output_item.added": @@ -579,6 +603,15 @@ class QwenAudioRealtimeService(AIService): wait_until_ready=False, ) + async def _delete_entry_trigger(self) -> None: + item_id = self._entry_trigger_item_id + self._entry_trigger_item_id = None + if item_id: + await self._send_event( + {"type": "conversation.item.delete", "item_id": item_id}, + wait_until_ready=False, + ) + async def _start_user_transcript_turn( self, event: dict[str, Any], @@ -643,6 +676,10 @@ class QwenAudioRealtimeService(AIService): async def _send_event( self, payload: dict[str, Any], *, wait_until_ready: bool = True ) -> None: + if payload.get("type") == "response.create": + # Mark the response busy before the server's response.created + # arrives so an immediate confirmation cannot pass the boundary. + self._response_done.clear() if wait_until_ready and not self._session_ready.is_set(): self._pending_events.append(payload) return diff --git a/backend/services/pipecat/stepfun_realtime.py b/backend/services/pipecat/stepfun_realtime.py index 22791bb..d65c58b 100644 --- a/backend/services/pipecat/stepfun_realtime.py +++ b/backend/services/pipecat/stepfun_realtime.py @@ -21,6 +21,7 @@ from pipecat.frames.frames import ( OutputTransportMessageUrgentFrame, StartFrame, TTSAudioRawFrame, + TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService @@ -85,6 +86,7 @@ class StepFunRealtimeService(AIService): ) self._fixed_speech_completion: asyncio.Future[None] | None = None self._fixed_speech_request_item_id: str | None = None + self._entry_trigger_item_id: str | None = None self._suppress_response_transcript = False self._speech_started_handler: SpeechStartedHandler | None = None self._function_names: dict[str, str] = {} @@ -164,9 +166,27 @@ class StepFunRealtimeService(AIService): self._resolve_fixed_speech() await self.broadcast_interruption() - async def request_response(self) -> None: + async def request_response(self, *, trigger_text: str | None = None) -> None: + if trigger_text: + item_id = f"item_{uuid4().hex}" + self._entry_trigger_item_id = item_id + await self._send_event( + { + "type": "conversation.item.create", + "item": { + "id": item_id, + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": trigger_text}], + }, + } + ) await self._send_event({"type": "response.create"}) + async def wait_for_response_boundary(self) -> None: + """Wait until StepFun has finished or cancelled the current response.""" + await self._response_done.wait() + def set_speech_started_handler( self, handler: SpeechStartedHandler | None, @@ -246,6 +266,7 @@ class StepFunRealtimeService(AIService): self._tool_session.clear() self._function_names.clear() self._fixed_speech_request_item_id = None + self._entry_trigger_item_id = None self._resolve_fixed_speech() if websocket and websocket.state is State.OPEN: try: @@ -280,6 +301,7 @@ class StepFunRealtimeService(AIService): self._session_ready.clear() self._response_done.set() self._fixed_speech_request_item_id = None + self._entry_trigger_item_id = None self._resolve_fixed_speech() if self._receive_task is asyncio.current_task(): self._receive_task = None @@ -331,8 +353,10 @@ class StepFunRealtimeService(AIService): "incomplete", "interrupted", } + await self.push_frame(TTSStoppedFrame()) await self._finish_assistant_text(interrupted=interrupted) await self._delete_fixed_speech_request() + await self._delete_entry_trigger() self._resolve_fixed_speech() self._response_done.set() elif event_type == "response.output_item.added": @@ -474,9 +498,22 @@ class StepFunRealtimeService(AIService): wait_until_ready=False, ) + async def _delete_entry_trigger(self) -> None: + item_id = self._entry_trigger_item_id + self._entry_trigger_item_id = None + if item_id: + await self._send_event( + {"type": "conversation.item.delete", "item_id": item_id}, + wait_until_ready=False, + ) + async def _send_event( self, payload: dict[str, Any], *, wait_until_ready: bool = True ) -> None: + if payload.get("type") == "response.create": + # Mark the response busy before the server's response.created + # arrives so an immediate confirmation cannot pass the boundary. + self._response_done.clear() if wait_until_ready and not self._session_ready.is_set(): self._pending_events.append(payload) return diff --git a/backend/services/workflow/realtime.py b/backend/services/workflow/realtime.py index 5d15a3b..f786641 100644 --- a/backend/services/workflow/realtime.py +++ b/backend/services/workflow/realtime.py @@ -35,6 +35,11 @@ from services.workflow_engine import WorkflowEngine MAX_AUTOMATIC_HOPS = 50 +AGENT_ENTRY_TRIGGER = ( + "[工作流内部事件，不是用户消息] 当前 Agent 节点已激活。" + "请严格依据当前 instructions 立即生成进入该节点时的回复，" + "不要提及或复述本事件。" +) ToolHandler = Callable[[dict[str, Any]], Awaitable[RealtimeToolResult]] @@ -80,10 +85,18 @@ class RealtimeWorkflowOutput(WorkflowOutput): **({"nodeId": node_id} if node_id else {}), } ) - return await self._realtime.speak_fixed( + track_speech = getattr(self._runtime.call_end, "track_speech", None) + playback_completion: Awaitable[None] | None = None + if callable(track_speech): + playback_completion = track_speech() + provider_completion = await self._realtime.speak_fixed( content, suppress_transcript=True, ) + # Message playback policy and deterministic continuation must use the + # transport boundary. Provider response.done only means generation + # has finished; audio may still be buffered at the output transport. + return playback_completion or provider_completion class WorkflowRealtimeController: @@ -138,7 +151,7 @@ class WorkflowRealtimeController: allow_visible_actions=True, ) if activation.continue_response: - await self._runtime.realtime.request_response() + await self._request_agent_entry_response() async def on_client_ready(self) -> None: await self._output.mark_client_ready() @@ -160,7 +173,7 @@ class WorkflowRealtimeController: if activation and activation.after_output: await activation.after_output() elif activation and activation.continue_response: - await self._runtime.realtime.request_response() + await self._request_agent_entry_response() return SessionVariableUpdate( changed=changed, dynamic_variables=self._store.public_values(), @@ -321,7 +334,13 @@ class WorkflowRealtimeController: if activation.after_output: await activation.after_output() elif activation.continue_response: - await self._runtime.realtime.request_response() + await self._request_agent_entry_response() + + async def _request_agent_entry_response(self) -> None: + """Give provider inference a temporary turn for Agent entry speech.""" + await self._runtime.realtime.request_response( + trigger_text=AGENT_ENTRY_TRIGGER, + ) async def _follow_edge( self,