Support for streaming multiple responses via function calls.

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

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_final:
-
+            await self._handle_function_call_finished(frame, in_progress_frame)
-        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,
-                        }
-                    ),
-                }
-            )
         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}]"

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.
+            result: The result of the function call, or an intermediate update.
-            properties: Optional properties for the result.
+            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