diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b73f94d26..00f38cab8 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -663,10 +663,14 @@ class FunctionCallResultProperties: Parameters: run_llm: Whether to run the LLM after receiving this result. on_context_updated: Callback to execute when context is updated. + is_final: Whether this is the final result for the function call. When + ``False`` the result is treated as an intermediate update. Defaults to ``True``. + Only meaningful for async function calls (``cancel_on_interruption=False``). """ run_llm: Optional[bool] = None on_context_updated: Optional[Callable[[], Awaitable[None]]] = None + is_final: bool = True @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 44e17faef..616636494 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -25,6 +25,8 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.audio.vad.vad_controller import VADController from pipecat.frames.frames import ( AssistantImageRawFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, @@ -832,6 +834,13 @@ class LLMAssistantAggregator(LLMContextAggregator): self._context_updated_tasks: Set[asyncio.Task] = set() self._user_speaking: bool = False + self._bot_speaking: bool = False + + # When a function call result arrives while the bot is speaking, we defer the LLM + # re-invocation until the bot stops speaking. This flag is set to True in that case + # so that `BotStoppedSpeakingFrame` knows to push a context frame. Multiple results + # arriving in the same speaking window are bundled into a single deferred push. + self._push_context_on_bot_stopped_speaking: bool = False self._assistant_turn_start_timestamp = "" @@ -872,6 +881,7 @@ class LLMAssistantAggregator(LLMContextAggregator): """Reset the aggregation state.""" await super().reset() await self._reset_thought_aggregation() # Just to be safe + self._push_context_on_bot_stopped_speaking = False async def _reset_thought_aggregation(self): """Reset the thought aggregation state.""" @@ -943,6 +953,15 @@ class LLMAssistantAggregator(LLMContextAggregator): elif isinstance(frame, UserStoppedSpeakingFrame): self._user_speaking = False await self.push_frame(frame, direction) + elif isinstance(frame, BotStartedSpeakingFrame): + self._bot_speaking = True + await self.push_frame(frame, direction) + elif isinstance(frame, BotStoppedSpeakingFrame): + self._bot_speaking = False + await self.push_frame(frame, direction) + if self._push_context_on_bot_stopped_speaking and not self._user_speaking: + logger.debug(f"{self}: Bot stopped speaking — pushing deferred context frame!") + await self.push_context_frame(FrameDirection.UPSTREAM) else: await self.push_frame(frame, direction) @@ -973,6 +992,15 @@ class LLMAssistantAggregator(LLMContextAggregator): return aggregation + async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a context frame in the specified direction. + + Args: + direction: The direction to push the frame (upstream or downstream). + """ + await super().push_context_frame(direction) + self._push_context_on_bot_stopped_speaking = False + async def _handle_llm_run(self, frame: LLMRunFrame): await self.push_context_frame(FrameDirection.UPSTREAM) @@ -1036,9 +1064,12 @@ class LLMAssistantAggregator(LLMContextAggregator): "content": json.dumps( { "type": "async_tool", - "status": "started", + "status": "running", "tool_call_id": frame.tool_call_id, - "description": "The tool associated with this tool_call_id is still in progress, and the result is not yet available. It will be provided in a subsequent message with the same tool_call_id.", + "description": "An asynchronous task associated with this tool_call_id has started running. " + + "Expect results to arrive later as developer messages that look roughly like this one (with 'type=async_tool' and a matching tool_call_id) but with a 'result' field. " + + "Note that there *may* be more than one result (i.e., a stream of results), but there doesn't have to be (there may be only one). " + + "The last result will come in a message with 'status=finished'.", } ), "tool_call_id": frame.tool_call_id, @@ -1066,33 +1097,14 @@ class LLMAssistantAggregator(LLMContextAggregator): return in_progress_frame = self._function_calls_in_progress[frame.tool_call_id] - is_async = not in_progress_frame.cancel_on_interruption if in_progress_frame else False group_id = in_progress_frame.group_id if in_progress_frame else None - - del self._function_calls_in_progress[frame.tool_call_id] - properties = frame.properties + is_final = frame.properties.is_final if frame.properties else True - 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": json.dumps( - { - "type": "async_tool", - "tool_call_id": frame.tool_call_id, - "status": "finished", - "result": result, - } - ), - } - ) + if is_final: + await self._handle_function_call_finished(frame, in_progress_frame) else: - self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + await self._handle_function_call_intermediate_result(frame, in_progress_frame) run_llm = False @@ -1119,14 +1131,38 @@ class LLMAssistantAggregator(LLMContextAggregator): # otherwise always execute as soon as we receive the result. if group_id: run_llm = not any( - f is not None and f.group_id == group_id + f is not None + and f.group_id == group_id + # We are now able to receive "updates", so the current + # frame can still be in the in progress list, and we need to + # ignore it. + and f.tool_call_id != frame.tool_call_id for f in self._function_calls_in_progress.values() ) else: run_llm = True if run_llm and not self._user_speaking: - await self.push_context_frame(FrameDirection.UPSTREAM) + if self.has_queued_frame(FunctionCallResultFrame): + # Another FunctionCallResultFrame is already queued. Defer the context push + # to bundle all results into a single LLM call instead of triggering one + # inference pass per result. The context will be pushed once the last + # function call in the queue is processed. + logger.debug( + f"{self}: More FunctionCallResultFrames queued — deferring context frame push." + ) + elif self._bot_speaking: + # Defer the context frame push until the bot finishes speaking. If multiple + # function call results arrive while the bot is speaking, they all accumulate + # in the context and a single push is performed once speaking stops, preventing + # the LLM from running multiple times and producing duplicated responses. + # This should be an edge case, since it would require a FunctionCallResultFrame + # being queued between an LLM response start and end frame. + logger.debug(f"{self}: Bot is speaking — deferring context frame push.") + self._push_context_on_bot_stopped_speaking = True + else: + logger.debug(f"{self}: Pushing context frame!") + await self.push_context_frame(FrameDirection.UPSTREAM) # Call the `on_context_updated` callback once the function call result # is added to the context. Also, run this in a separate task to make @@ -1137,6 +1173,70 @@ class LLMAssistantAggregator(LLMContextAggregator): self._context_updated_tasks.add(task) task.add_done_callback(self._context_updated_task_finished) + async def _handle_function_call_intermediate_result( + self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame + ): + """Handle an intermediate result for an async function call. + + Injects an intermediate developer message into the context without + removing the call from the in-progress map. + """ + if not frame.result: + logger.warning(f"{self} result_callback called with is_final=False but no result!") + return + + result = json.dumps(frame.result, ensure_ascii=False) + self._context.add_message( + { + "role": "developer", + "content": json.dumps( + { + "type": "async_tool", + "tool_call_id": frame.tool_call_id, + "status": "running", + "description": "This is an intermediate result for the asynchronous task associated with this tool_call_id. " + + "The task is still running. More intermediate results may follow, or the next result may be the final one with 'status=finished'.", + "result": result, + } + ), + } + ) + + async def _handle_function_call_finished( + self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame + ): + """Handle the final result of a function call. + + Removes the call from the in-progress map, updates the context, and + triggers LLM inference when appropriate. + """ + is_async = not in_progress_frame.cancel_on_interruption + del self._function_calls_in_progress[frame.tool_call_id] + + result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED" + + if is_async: + # For async function calls inject a developer message so the LLM is + # notified of the completed result instead of updating the IN_PROGRESS + # tool message. + self._context.add_message( + { + "role": "developer", + "content": json.dumps( + { + "type": "async_tool", + "tool_call_id": frame.tool_call_id, + "status": "finished", + "description": "This is the final result for the asynchronous task associated with this tool_call_id. " + + "The task has completed. No further results will arrive for this tool_call_id.", + "result": result, + } + ), + } + ) + else: + self._update_function_call_result(frame.function_name, frame.tool_call_id, result) + async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame): logger.debug( f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]" diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 1430aceac..03fcb115f 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -73,7 +73,10 @@ FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]] class FunctionCallResultCallback(Protocol): """Protocol for function call result callbacks. - Handles the result of an LLM function call execution. + Used for both final results and intermediate updates. Pass + ``properties=FunctionCallResultProperties(is_final=False)`` to send an + intermediate update (only valid for async function calls registered with + ``cancel_on_interruption=False``). """ async def __call__( @@ -82,8 +85,9 @@ class FunctionCallResultCallback(Protocol): """Call the result callback. Args: - result: The result of the function call. - properties: Optional properties for the result. + result: The result of the function call, or an intermediate update. + properties: Optional properties. Set ``is_final=False`` to send an + intermediate update instead of the final result. """ ... @@ -98,7 +102,10 @@ class FunctionCallParams: arguments: The arguments for the function. llm: The LLMService instance being used. context: The LLM context. - result_callback: Callback to handle the result of the function call. + result_callback: Callback to deliver the result of the function call. + For async function calls (``cancel_on_interruption=False``), call + it with ``properties=FunctionCallResultProperties(is_final=False)`` + to push intermediate updates before the final result. """ function_name: str @@ -756,10 +763,21 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): timeout_task: Optional[asyncio.Task] = None - # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream. + # Single callback for both intermediate updates and final results. + # Pass properties=FunctionCallResultProperties(is_final=False) for updates. async def function_call_result_callback( result: Any, *, properties: Optional[FunctionCallResultProperties] = None ): + is_final = properties.is_final if properties else True + if not is_final and item.cancel_on_interruption: + logger.warning( + f"{self} result_callback called with is_final=False on sync function call" + f" [{runner_item.function_name}:{runner_item.tool_call_id}]." + " Intermediate updates are only valid for async function calls" + " (cancel_on_interruption=False)." + ) + return + nonlocal timeout_task # Cancel timeout task if it exists