diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 035f1b2f2..82c2e7303 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -601,15 +601,7 @@ class LLMService(AIService): cancel_on_interruption=item.cancel_on_interruption, ) - # Start a timeout task for deferred function calls - async def timeout_handler(): - await asyncio.sleep(self._function_call_timeout_secs) - logger.warning( - f"{self} Function call [{runner_item.function_name}:{runner_item.tool_call_id}] timed out after {self._function_call_timeout_secs} seconds" - ) - await function_call_result_callback(None) - - timeout_task = self.create_task(timeout_handler()) + timeout_task: Optional[asyncio.Task] = None # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream. async def function_call_result_callback( @@ -631,6 +623,24 @@ class LLMService(AIService): properties=properties, ) + # Start a timeout task for deferred function calls + async def timeout_handler(): + try: + await asyncio.sleep(self._function_call_timeout_secs) + logger.warning( + f"{self} Function call [{runner_item.function_name}:{runner_item.tool_call_id}] timed out after {self._function_call_timeout_secs} seconds" + ) + await function_call_result_callback(None) + except asyncio.CancelledError: + raise + + timeout_task = self.create_task(timeout_handler()) + + # Yield to the event loop so the timeout task coroutine gets entered + # before it could be cancelled. Without this, cancelling the task before + # it starts would leave the coroutine in a "never awaited" state. + await asyncio.sleep(0) + try: if isinstance(item.handler, DirectFunctionWrapper): # Handler is a DirectFunctionWrapper @@ -667,6 +677,9 @@ class LLMService(AIService): ) await item.handler(params) except Exception as e: + # Cancel timeout task if it exists + if timeout_task and not timeout_task.done(): + await self.cancel_task(timeout_task) error_message = f"Error executing function call [{runner_item.function_name}]: {e}" logger.error(f"{self} {error_message}") await self.push_error(error_msg=error_message, exception=e, fatal=False)