fix(realtime): unblock fixed speech after tool calls

backend/services/pipecat/qwen_audio_realtime.py

@@ -109,6 +109,8 @@ class QwenAudioRealtimeService(AIService):
         self._websocket = None
         self._receive_task: asyncio.Task | None = None
         self._session_ready = asyncio.Event()
+        self._response_done = asyncio.Event()
+        self._response_done.set()
         self._pending_events: list[dict[str, Any]] = []
         self._warned_input_sample_rate = False
 
@@ -123,7 +125,10 @@ class QwenAudioRealtimeService(AIService):
         self._user_transcript_timestamp = ""
         self._deferred_assistant_messages: list[dict[str, Any]] = []
         self._tools: list[RealtimeTool] = []
-        self._tool_session = RealtimeToolSession(self._send_tool_event)
+        self._tool_session = RealtimeToolSession(
+            self._send_tool_event,
+            wait_for_response_boundary=self._response_done.wait,
+        )
         self._fixed_speech_completion: asyncio.Future[None] | None = None
         self._suppress_response_transcript = False
         self._speech_started_handler: SpeechStartedHandler | None = None
@@ -373,6 +378,7 @@ class QwenAudioRealtimeService(AIService):
         websocket = self._websocket
         self._websocket = None
         self._session_ready.clear()
+        self._response_done.set()
         self._pending_events.clear()
         self._response_active = False
         self._user_transcript_pending = False
@@ -404,6 +410,9 @@ class QwenAudioRealtimeService(AIService):
             if self._websocket is websocket:
                 self._websocket = None
                 self._session_ready.clear()
+            self._response_done.set()
+            self._greeting_request_item_id = None
+            self._resolve_fixed_speech()
             if self._receive_task is asyncio.current_task():
                 self._receive_task = None
 
@@ -424,6 +433,7 @@ class QwenAudioRealtimeService(AIService):
                 await self._send_event(payload, wait_until_ready=False)
         elif event_type == "response.created":
             self._response_active = True
+            self._response_done.clear()
             self._audio_suppressed = False
         elif event_type == "response.audio.delta":
             audio = event.get("delta")
@@ -473,13 +483,17 @@ class QwenAudioRealtimeService(AIService):
             await self._finish_assistant_text(interrupted=interrupted)
             await self._delete_greeting_request()
             self._resolve_fixed_speech()
+            self._response_done.set()
         elif event_type == "response.output_item.added":
             self._remember_function_call(event)
         elif event_type in {
             "response.function_call_arguments.done",
             "response.output_item.done",
         }:
-            await self._handle_function_call_event(event)
+            self.create_task(
+                self._handle_function_call_event(dict(event)),
+                name="qwen_audio_realtime_tool_call",
+            )
         elif event_type == "error":
             error = event.get("error")
             message = error.get("message") if isinstance(error, dict) else str(error)
@@ -527,11 +541,17 @@ class QwenAudioRealtimeService(AIService):
         )
         if not name:
             return
+        try:
             await self._tool_session.handle_call(
                 name=name,
                 call_id=call_id,
                 arguments=source.get("arguments", event.get("arguments")),
             )
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001 - background provider task
+            logger.exception(f"Qwen-Audio Realtime 工具调用处理失败:{exc}")
+        finally:
             self._function_names.pop(call_id, None)
 
     def _remember_function_call(self, event: dict[str, Any]) -> None:

backend/services/pipecat/realtime_tools.py

@@ -46,13 +46,20 @@ RealtimeToolDispatcher = Callable[
     [str, dict[str, Any], str], Awaitable[RealtimeToolResult]
 ]
 SendProviderEvent = Callable[[dict[str, Any]], Awaitable[None]]
+WaitForResponseBoundary = Callable[[], Awaitable[Any]]
 
 
 class RealtimeToolSession:
     """Serialize provider calls and answer every call id at most once."""
 
-    def __init__(self, send_event: SendProviderEvent) -> None:
+    def __init__(
+        self,
+        send_event: SendProviderEvent,
+        *,
+        wait_for_response_boundary: WaitForResponseBoundary | None = None,
+    ) -> None:
         self._send_event = send_event
+        self._wait_for_response_boundary = wait_for_response_boundary
         self._dispatcher: RealtimeToolDispatcher | None = None
         self._handled_call_ids: set[str] = set()
         self._lock = asyncio.Lock()
@@ -105,6 +112,12 @@ class RealtimeToolSession:
                     },
                 }
             )
+            # Function-call argument events arrive before the provider's
+            # response.done boundary. Starting another response sooner is
+            # rejected by both supported realtime providers. The provider
+            # receive loop stays free while this background task waits.
+            if self._wait_for_response_boundary is not None:
+                await self._wait_for_response_boundary()
             if result.after_output is not None:
                 await result.after_output()
             if result.continue_response:

backend/services/pipecat/stepfun_realtime.py

@@ -72,13 +72,19 @@ class StepFunRealtimeService(AIService):
         self._websocket = None
         self._receive_task: asyncio.Task | None = None
         self._session_ready = asyncio.Event()
+        self._response_done = asyncio.Event()
+        self._response_done.set()
         self._pending_events: list[dict[str, Any]] = []
         self._assistant_turn_id: str | None = None
         self._assistant_text = ""
         self._assistant_timestamp = ""
         self._tools: list[RealtimeTool] = []
-        self._tool_session = RealtimeToolSession(self._send_tool_event)
+        self._tool_session = RealtimeToolSession(
+            self._send_tool_event,
+            wait_for_response_boundary=self._response_done.wait,
+        )
         self._fixed_speech_completion: asyncio.Future[None] | None = None
+        self._fixed_speech_request_item_id: str | None = None
         self._suppress_response_transcript = False
         self._speech_started_handler: SpeechStartedHandler | None = None
         self._function_names: dict[str, str] = {}
@@ -184,14 +190,25 @@ class StepFunRealtimeService(AIService):
         self._resolve_fixed_speech()
         self._fixed_speech_completion = completion
         self._suppress_response_transcript = suppress_transcript
+        item_id = f"item_{uuid4().hex}"
+        self._fixed_speech_request_item_id = item_id
         await self._send_event(
             {
-                "type": "response.create",
+                "type": "conversation.item.create",
-                "session": {
+                "item": {
-                    "instructions": f"请原样无修改地输出下面的话：\n{text}",
+                    "id": item_id,
+                    "type": "message",
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "input_text",
+                            "text": f"请原样无修改地朗读下面的话：\n{text}",
+                        }
+                    ],
                 },
             }
         )
+        await self._send_event({"type": "response.create"})
         return completion
 
     async def _connect(self) -> None:
@@ -225,8 +242,10 @@ class StepFunRealtimeService(AIService):
         websocket = self._websocket
         self._websocket = None
         self._session_ready.clear()
+        self._response_done.set()
         self._tool_session.clear()
         self._function_names.clear()
+        self._fixed_speech_request_item_id = None
         self._resolve_fixed_speech()
         if websocket and websocket.state is State.OPEN:
             try:
@@ -259,6 +278,9 @@ class StepFunRealtimeService(AIService):
             if self._websocket is websocket:
                 self._websocket = None
                 self._session_ready.clear()
+            self._response_done.set()
+            self._fixed_speech_request_item_id = None
+            self._resolve_fixed_speech()
             if self._receive_task is asyncio.current_task():
                 self._receive_task = None
 
@@ -271,6 +293,8 @@ class StepFunRealtimeService(AIService):
             pending, self._pending_events = self._pending_events, []
             for payload in pending:
                 await self._send_event(payload, wait_until_ready=False)
+        elif event_type == "response.created":
+            self._response_done.clear()
         elif event_type == "response.audio.delta":
             audio = event.get("delta")
             if audio:
@@ -308,14 +332,19 @@ class StepFunRealtimeService(AIService):
                 "interrupted",
             }
             await self._finish_assistant_text(interrupted=interrupted)
+            await self._delete_fixed_speech_request()
             self._resolve_fixed_speech()
+            self._response_done.set()
         elif event_type == "response.output_item.added":
             self._remember_function_call(event)
         elif event_type in {
             "response.function_call_arguments.done",
             "response.output_item.done",
         }:
-            await self._handle_function_call_event(event)
+            self.create_task(
+                self._handle_function_call_event(dict(event)),
+                name="stepfun_realtime_tool_call",
+            )
         elif event_type == "error":
             error = event.get("error")
             message = error.get("message") if isinstance(error, dict) else str(error)
@@ -407,11 +436,17 @@ class StepFunRealtimeService(AIService):
         )
         if not name:
             return
+        try:
             await self._tool_session.handle_call(
                 name=name,
                 call_id=call_id,
                 arguments=source.get("arguments", event.get("arguments")),
             )
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001 - background provider task
+            logger.exception(f"StepFun Realtime 工具调用处理失败:{exc}")
+        finally:
             self._function_names.pop(call_id, None)
 
     def _remember_function_call(self, event: dict[str, Any]) -> None:
@@ -430,6 +465,15 @@ class StepFunRealtimeService(AIService):
         if completion is not None and not completion.done():
             completion.set_result(None)
 
+    async def _delete_fixed_speech_request(self) -> None:
+        item_id = self._fixed_speech_request_item_id
+        self._fixed_speech_request_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: