From dfdb92958b7e9affed4b20198d4230a64fb4cc12 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Tue, 31 Mar 2026 17:26:06 -0300 Subject: [PATCH] Fix async tool handling for compatibility with all LLMs. --- src/pipecat/frames/frames.py | 4 ++++ .../aggregators/llm_response_universal.py | 19 ++++++++++++++----- src/pipecat/services/llm_service.py | 17 +++++++++++++++++ .../context/llm_context_summarization.py | 14 ++++++++++---- 4 files changed, 45 insertions(+), 9 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 29369f03a..bb7b3d68b 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1918,12 +1918,16 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame): tool_call_id: Unique identifier for this function call. arguments: Arguments passed to the function. cancel_on_interruption: Whether to cancel this call if interrupted. + is_async: Whether this function call runs asynchronously. When True, + the LLM continues the conversation immediately without waiting for + the result. The result is injected later via a developer message. """ function_name: str tool_call_id: str arguments: Any cancel_on_interruption: bool = False + is_async: bool = False @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 605db31f6..9d98750fe 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -1067,16 +1067,25 @@ class LLMAssistantAggregator(LLMContextAggregator): ) return + in_progress_frame = self._function_calls_in_progress[frame.tool_call_id] + is_async = in_progress_frame.is_async if in_progress_frame else False del self._function_calls_in_progress[frame.tool_call_id] properties = frame.properties - # Update context with the function call result - if frame.result: - result = json.dumps(frame.result, ensure_ascii=False) - self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED" + + if is_async: + # For async function calls instead of updating the existing IN_PROGRESS tool message we inject + # a new developer message so the LLM is notified of the completed result. + self._context.add_message( + { + "role": "developer", + "content": f"Async function with id '{frame.tool_call_id}' completed with result: {result}", + } + ) else: - self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED") + self._update_function_call_result(frame.function_name, frame.tool_call_id, result) run_llm = False diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index f2e247de9..cbfbcc88c 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -124,6 +124,9 @@ class FunctionCallRegistryItem: function_name: The name of the function (None for catch-all handler). handler: The handler for processing function call parameters. cancel_on_interruption: Whether to cancel the call on interruption. + is_async: Whether this function call runs asynchronously. When True, + the LLM continues the conversation immediately without waiting for + the result. The result is injected later via a developer message. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. """ @@ -131,6 +134,7 @@ class FunctionCallRegistryItem: function_name: Optional[str] handler: FunctionCallHandler | "DirectFunctionWrapper" cancel_on_interruption: bool + is_async: bool = False timeout_secs: Optional[float] = None @@ -578,6 +582,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): handler: Any, *, cancel_on_interruption: bool = True, + is_async: bool = False, timeout_secs: Optional[float] = None, ): """Register a function handler for LLM function calls. @@ -589,6 +594,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): parameter. cancel_on_interruption: Whether to cancel this function call when an interruption occurs. Defaults to True. + is_async: Whether this function call runs asynchronously. When True, + the LLM continues the conversation immediately without waiting for + the result. The result is injected later via a developer message. + Defaults to False. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. Defaults to None, which uses the global timeout. @@ -599,6 +608,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, + is_async=is_async, timeout_secs=timeout_secs, ) @@ -607,6 +617,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): handler: DirectFunction, *, cancel_on_interruption: bool = True, + is_async: bool = False, timeout_secs: Optional[float] = None, ): """Register a direct function handler for LLM function calls. @@ -619,6 +630,10 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): handler: The direct function to register. Must follow DirectFunction protocol. cancel_on_interruption: Whether to cancel this function call when an interruption occurs. Defaults to True. + is_async: Whether this function call runs asynchronously. When True, + the LLM continues the conversation immediately without waiting for + the result. The result is injected later via a developer message. + Defaults to False. timeout_secs: Optional per-tool timeout in seconds. Overrides the global ``function_call_timeout_secs`` for this specific function. Defaults to None, which uses the global timeout. @@ -628,6 +643,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): function_name=wrapper.name, handler=wrapper, cancel_on_interruption=cancel_on_interruption, + is_async=is_async, timeout_secs=timeout_secs, ) @@ -766,6 +782,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, cancel_on_interruption=item.cancel_on_interruption, + is_async=item.is_async, ) timeout_task: Optional[asyncio.Task] = None diff --git a/src/pipecat/utils/context/llm_context_summarization.py b/src/pipecat/utils/context/llm_context_summarization.py index 259d0bae5..39fa46644 100644 --- a/src/pipecat/utils/context/llm_context_summarization.py +++ b/src/pipecat/utils/context/llm_context_summarization.py @@ -389,9 +389,13 @@ class LLMContextSummarizationUtil: Scans messages from ``start_idx`` up to (but not including) ``summary_end`` to identify tool calls whose responses either don't - exist yet or fall in the kept portion of the context (>= summary_end). + exist yet, fall in the kept portion of the context (>= summary_end), + or are still marked as ``IN_PROGRESS`` (async calls whose results have + not yet arrived). + This prevents summarizing tool call requests when their responses would - remain in the kept context as orphans, which the OpenAI API rejects. + remain in the kept context as orphans, which the OpenAI API rejects, + and avoids summarizing async function calls before their results arrive. Args: messages: List of messages to check. @@ -428,11 +432,13 @@ class LLMContextSummarizationUtil: if tool_call_id: pending_tool_calls[tool_call_id] = i - # Check for tool results + # Check for tool results — treat IN_PROGRESS as still pending + # (async function calls whose results have not yet arrived). if role == "tool": tool_call_id = msg.get("tool_call_id") if tool_call_id and tool_call_id in pending_tool_calls: - pending_tool_calls.pop(tool_call_id) + if msg.get("content") != "IN_PROGRESS": + pending_tool_calls.pop(tool_call_id) # If we have pending tool calls, return the earliest index if pending_tool_calls: