diff --git a/changelog/4301.fixed.md b/changelog/4301.fixed.md new file mode 100644 index 000000000..1ee4b092d --- /dev/null +++ b/changelog/4301.fixed.md @@ -0,0 +1 @@ +- Fixed missing tool handlers so unregistered tool calls fail with a normal final tool result instead of leaving tool-call state hanging. diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index b0f0c08c2..59689ba6f 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -737,7 +737,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): logger.warning( f"{self} is calling '{function_call.function_name}', but it's not registered." ) - continue + item = self._build_missing_function_call_registry_item(function_call.function_name) runner_items.append( FunctionCallRunnerItem( @@ -794,12 +794,21 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await self._sequential_runner_queue.put(runner_item) async def _run_function_call(self, runner_item: FunctionCallRunnerItem): + # Re-resolve the registry item at execution time. The function may have + # been unregistered between queuing and execution, in which case we + # fall back to the missing-function handler so the call still terminates + # with a normal tool result. if runner_item.function_name in self._functions.keys(): item = self._functions[runner_item.function_name] elif None in self._functions.keys(): item = self._functions[None] + elif runner_item.registry_item.handler == self._missing_function_call_handler: + item = runner_item.registry_item else: - return + logger.warning( + f"{self} is calling '{runner_item.function_name}', but it was just unregistered." + ) + item = self._build_missing_function_call_registry_item(runner_item.function_name) logger.debug( f"{self} Calling function [{runner_item.function_name}:{runner_item.tool_call_id}] with arguments {runner_item.arguments}" @@ -908,6 +917,20 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): if timeout_task and not timeout_task.done(): await self.cancel_task(timeout_task) + def _build_missing_function_call_registry_item( + self, function_name: str + ) -> FunctionCallRegistryItem: + """Build a registry item that routes to the missing-function handler.""" + return FunctionCallRegistryItem( + function_name=function_name, + handler=self._missing_function_call_handler, + cancel_on_interruption=True, + ) + + async def _missing_function_call_handler(self, params: FunctionCallParams): + """Return a terminal tool result when the LLM calls an unknown function.""" + await params.result_callback(f"Error: function '{params.function_name}' is not registered.") + def _has_async_tools(self) -> bool: """Return True if at least one non-builtin async tool is registered.""" return any( diff --git a/tests/test_llm_service.py b/tests/test_llm_service.py new file mode 100644 index 000000000..c7018d9f0 --- /dev/null +++ b/tests/test_llm_service.py @@ -0,0 +1,173 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest +from unittest.mock import AsyncMock, patch + +from pipecat.frames.frames import ( + FunctionCallFromLLM, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.services.llm_service import LLMService +from pipecat.services.settings import LLMSettings +from pipecat.turns.user_mute.function_call_user_mute_strategy import FunctionCallUserMuteStrategy + + +class MockLLMService(LLMService): + """Minimal LLM service for testing function call execution.""" + + def __init__(self, **kwargs): + settings = LLMSettings( + model="test-model", + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=None, + user_turn_completion_config=None, + ) + super().__init__(settings=settings, **kwargs) + + +class TestLLMService(unittest.IsolatedAsyncioTestCase): + async def _run_function_calls_inline(self, service: MockLLMService): + async def run_inline(runner_items): + for runner_item in runner_items: + await service._run_function_call(runner_item) + + service._run_parallel_function_calls = run_inline + service._run_sequential_function_calls = run_inline + + async def test_missing_function_call_emits_terminal_result(self): + service = MockLLMService() + service._call_event_handler = AsyncMock() + await self._run_function_calls_inline(service) + + recorded_frames = [] + + async def mock_broadcast_frame(frame_cls, **kwargs): + recorded_frames.append(frame_cls(**kwargs)) + + service.broadcast_frame = mock_broadcast_frame + + with patch("pipecat.services.llm_service.logger") as mock_logger: + await service.run_function_calls( + [ + FunctionCallFromLLM( + function_name="missing_tool", + tool_call_id="call_1", + arguments={"query": "weather"}, + context=LLMContext(), + ) + ] + ) + + self.assertEqual( + [type(frame) for frame in recorded_frames], + [ + FunctionCallsStartedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + ], + ) + self.assertEqual(recorded_frames[1].function_name, "missing_tool") + self.assertEqual( + recorded_frames[2].result, + "Error: function 'missing_tool' is not registered.", + ) + + # Only the queue-time warning should fire; the execution-time + # "just unregistered" warning must not double-log. + warnings = [c.args[0] for c in mock_logger.warning.call_args_list] + self.assertTrue(any("not registered" in w for w in warnings)) + self.assertFalse(any("just unregistered" in w for w in warnings)) + + async def test_function_unregistered_between_queue_and_execute(self): + """Function unregistered between queuing and execution still terminates.""" + service = MockLLMService() + service._call_event_handler = AsyncMock() + + async def real_handler(params): + await params.result_callback("should not be called") + + service.register_function("doomed_tool", real_handler) + + recorded_frames = [] + + async def mock_broadcast_frame(frame_cls, **kwargs): + recorded_frames.append(frame_cls(**kwargs)) + + service.broadcast_frame = mock_broadcast_frame + + async def run_inline(runner_items): + # Simulate the function being unregistered after queuing but before execution. + service.unregister_function("doomed_tool") + for runner_item in runner_items: + await service._run_function_call(runner_item) + + service._run_parallel_function_calls = run_inline + service._run_sequential_function_calls = run_inline + + await service.run_function_calls( + [ + FunctionCallFromLLM( + function_name="doomed_tool", + tool_call_id="call_1", + arguments={}, + context=LLMContext(), + ) + ] + ) + + self.assertEqual( + [type(frame) for frame in recorded_frames], + [ + FunctionCallsStartedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + ], + ) + self.assertEqual( + recorded_frames[2].result, + "Error: function 'doomed_tool' is not registered.", + ) + + async def test_missing_function_call_allows_user_mute_cleanup(self): + service = MockLLMService() + service._call_event_handler = AsyncMock() + await self._run_function_calls_inline(service) + + recorded_frames = [] + + async def mock_broadcast_frame(frame_cls, **kwargs): + recorded_frames.append(frame_cls(**kwargs)) + + service.broadcast_frame = mock_broadcast_frame + + await service.run_function_calls( + [ + FunctionCallFromLLM( + function_name="missing_tool", + tool_call_id="call_1", + arguments={}, + context=LLMContext(), + ) + ] + ) + + strategy = FunctionCallUserMuteStrategy() + muted = False + for frame in recorded_frames: + muted = await strategy.process_frame(frame) + + self.assertFalse(muted)