From e6301af36a401f8e257268dd6e64e78edcae821c Mon Sep 17 00:00:00 2001 From: Abhishek Kumar Date: Wed, 15 Jul 2026 18:36:19 +0530 Subject: [PATCH] fix: fix ultravox node transitions --- .../pipecat/realtime/ultravox_realtime.py | 37 ++++++++++++++ api/tests/test_ultravox_realtime_wrapper.py | 49 +++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/api/services/pipecat/realtime/ultravox_realtime.py b/api/services/pipecat/realtime/ultravox_realtime.py index 11d00f0d..0ca8359e 100644 --- a/api/services/pipecat/realtime/ultravox_realtime.py +++ b/api/services/pipecat/realtime/ultravox_realtime.py @@ -11,6 +11,7 @@ the Dograh engine contract by: the existing call keeps its complete audio-native history - updating the next stage's system prompt and selected tools without a disconnect/reconnect cycle +- deferring workflow-control tools until any active Ultravox response ends - handling Dograh-only frames such as user mute and idle append prompts - tagging user transcripts with ``finalized=True`` for downstream parity """ @@ -67,6 +68,12 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService): # the context aggregator. Unlike Gemini, this ID is part of the wire # protocol needed to update the existing call without reconnecting. self._pending_node_transition_tool_call_ids: set[str] = set() + # A stage result can replace the active prompt and tools immediately. + # Hold transition invocations separately so ordinary tools can still + # run during speech while workflow control waits for response end. + self._deferred_node_transition_tool_invocations: list[ + tuple[str, str, dict[str, Any]] + ] = [] self._pending_user_text_messages: list[str] = [] async def start(self, frame): @@ -124,6 +131,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService): self._call_started = False self._started_placeholder_sent = set() self._pending_node_transition_tool_call_ids = set() + self._deferred_node_transition_tool_invocations = [] self._disconnecting = False async def _send_user_audio(self, frame): @@ -206,8 +214,37 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService): ): if self._function_is_node_transition(tool_name): self._pending_node_transition_tool_call_ids.add(invocation_id) + if self._bot_responding: + self._deferred_node_transition_tool_invocations.append( + (tool_name, invocation_id, parameters) + ) + logger.debug( + f"{self}: deferring workflow-control call {tool_name} " + "until bot turn ends" + ) + return await super()._handle_tool_invocation(tool_name, invocation_id, parameters) + async def _handle_response_end(self): + """Close the current response before applying queued workflow control.""" + await super()._handle_response_end() + await self._run_deferred_node_transition_tool_invocations() + + async def _run_deferred_node_transition_tool_invocations(self): + if not self._deferred_node_transition_tool_invocations: + return + + invocations = self._deferred_node_transition_tool_invocations + self._deferred_node_transition_tool_invocations = [] + logger.debug( + f"{self}: executing {len(invocations)} deferred workflow-control " + "call(s) after bot turn ended" + ) + for tool_name, invocation_id, parameters in invocations: + await super()._handle_tool_invocation( + tool_name, invocation_id, parameters + ) + async def _send_tool_result(self, tool_call_id: str, result: str): is_node_transition = tool_call_id in self._pending_node_transition_tool_call_ids try: diff --git a/api/tests/test_ultravox_realtime_wrapper.py b/api/tests/test_ultravox_realtime_wrapper.py index 27bc4c28..75f6d0ae 100644 --- a/api/tests/test_ultravox_realtime_wrapper.py +++ b/api/tests/test_ultravox_realtime_wrapper.py @@ -204,6 +204,55 @@ async def test_only_registered_node_transition_invocations_are_tracked(): assert service.run_function_calls.await_count == 2 +@pytest.mark.asyncio +async def test_node_transition_invocation_waits_for_response_end(): + service = _make_service() + service.run_function_calls = AsyncMock() + service.stop_processing_metrics = AsyncMock() + service.push_frame = AsyncMock() + service.register_function( + "transition_to_next_node", + AsyncMock(), + is_node_transition=True, + ) + service._bot_responding = "voice" + + await service._handle_tool_invocation( + "transition_to_next_node", "call-transition", {"reason": "pricing"} + ) + + service.run_function_calls.assert_not_awaited() + assert service._deferred_node_transition_tool_invocations == [ + ( + "transition_to_next_node", + "call-transition", + {"reason": "pricing"}, + ) + ] + + await service._handle_response_end() + + service.run_function_calls.assert_awaited_once() + function_call = service.run_function_calls.await_args.args[0][0] + assert function_call.function_name == "transition_to_next_node" + assert function_call.tool_call_id == "call-transition" + assert function_call.arguments == {"reason": "pricing"} + assert service._deferred_node_transition_tool_invocations == [] + assert service._bot_responding is None + + +@pytest.mark.asyncio +async def test_ordinary_tool_invocation_runs_while_response_is_active(): + service = _make_service() + service.run_function_calls = AsyncMock() + service._bot_responding = "voice" + + await service._handle_tool_invocation("lookup_price", "call-lookup", {}) + + service.run_function_calls.assert_awaited_once() + assert service._deferred_node_transition_tool_invocations == [] + + def test_ultravox_requires_transition_context_aggregation(): service = _make_service()