function calling now run in tasks

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- When registering a function call it is now possible to indicate if you want
+  the function call to be cancelled if there's a user interruption via
+  `cancel_on_interruption` (defaults to False). This is now possible because
+  function calls are executed concurrently.
+
 - Added support for detecting idle pipelines. By default, if no activity has
   been detected during 5 minutes, the `PipelineTask` will be automatically
   cancelled. It is possible to override this behavior by passing
@@ -120,6 +125,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Function calls are now executed in tasks. This means that the pipeline will
+  not be blocked while the function call is being executed.
+
 - ⚠️ `PipelineTask` will now be automatically cancelled if no bot activity is
   happening in the pipeline. There are a few settings to configure this
   behavior, see `PipelineTask` documentation for more details.
@@ -140,6 +148,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Deprecated
 
+- Passing a `start_callback` to `LLMService.register_function()` is now
+  deprecated, simply move the code from the start callback to the function call.
+
 - `TTSService` parameter `text_filter` is now deprecated, use `text_filters`
   instead which is now a list. This allows passing multiple filters that will be
   executed in order.
@@ -162,6 +173,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an assistant aggregator issue that could cause assistant text to be
+  split into multiple chunks during function calls.
+
+- Fixed an assistant aggregator issue that was causing assistant text to not be
+  added to the context during function calls. This could lead to duplications.
+
 - Fixed a `SegmentedSTTService` issue that was causing audio to be sent
   prematurely to the STT service. Instead of analyzing the volume in this
   service we rely on VAD events which use both VAD and volume.
@@ -1978,7 +1995,7 @@ async def on_connected(processor):
   completed. If a task is never ran `has_finished()` will return False.
 
 - `PipelineRunner` now supports SIGTERM. If received, the runner will be
-  canceled.
+  cancelled.
 
 ### Fixed
 

src/pipecat/frames/frames.py

@@ -634,6 +634,15 @@ class FunctionCallInProgressFrame(SystemFrame):
     function_name: str
     tool_call_id: str
     arguments: str
+    cancel_on_interruption: bool
+
+
+@dataclass
+class FunctionCallCancelFrame(SystemFrame):
+    """A frame to signal a function call has been cancelled."""
+
+    function_name: str
+    tool_call_id: str
 
 
 @dataclass
@@ -706,6 +715,18 @@ class VisionImageRawFrame(InputImageRawFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
 
 
+@dataclass
+class UserImageMessageFrame(SystemFrame):
+    """An image associated to a user."""
+
+    user_image_raw_frame: UserImageRawFrame
+    text: Optional[str] = None
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, image: {self.user_image_raw_frame}, text: {self.text})"
+
+
 #
 # Control frames
 #

src/pipecat/pipeline/task.py

@@ -409,7 +409,7 @@ class PipelineTask(BaseTask):
     async def _process_push_queue(self):
         """This is the task that runs the pipeline for the first time by sending
         a StartFrame and by pushing any other frames queued by the user. It runs
-        until the tasks is canceled or stopped (e.g. with an EndFrame).
+        until the tasks is cancelled or stopped (e.g. with an EndFrame).
 
         """
         self._clock.start()

src/pipecat/processors/aggregators/llm_response.py

@@ -7,14 +7,20 @@
 import asyncio
 import time
 from abc import abstractmethod
-from typing import List
+from typing import Dict, List
+
+from loguru import logger
 
 from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
     CancelFrame,
     EmulateUserStartedSpeakingFrame,
     EmulateUserStoppedSpeakingFrame,
     EndFrame,
     Frame,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -23,10 +29,12 @@ from pipecat.frames.frames import (
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
     StartInterruptionFrame,
     TextFrame,
     TranscriptionFrame,
+    UserImageMessageFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -35,6 +43,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.time import time_now_iso8601
 
 
 class LLMFullResponseAggregator(FrameProcessor):
@@ -139,68 +148,20 @@ class BaseLLMResponseAggregator(FrameProcessor):
         pass
 
     @abstractmethod
-    async def push_aggregation(self):
+    async def handle_aggregation(self, aggregation: str):
+        """Adds the given aggregation to the aggregator. The aggregator can use
+        a simple list of message or a context. It doesn't not push any frames.
+
+        """
         pass
 
-
+    @abstractmethod
-class LLMResponseAggregator(BaseLLMResponseAggregator):
-    """This is a base LLM aggregator that uses a simple list of messages to
-    store the conversation. It pushes `LLMMessagesFrame` as an aggregation
-    frame.
-
-    """
-
-    def __init__(
-        self,
-        *,
-        messages: List[dict],
-        role: str = "user",
-        **kwargs,
-    ):
-        super().__init__(**kwargs)
-
-        self._messages = messages
-        self._role = role
-
-        self._aggregation = ""
-
-        self.reset()
-
-    @property
-    def messages(self) -> List[dict]:
-        return self._messages
-
-    @property
-    def role(self) -> str:
-        return self._role
-
-    def add_messages(self, messages):
-        self._messages.extend(messages)
-
-    def set_messages(self, messages):
-        self.reset()
-        self._messages.clear()
-        self._messages.extend(messages)
-
-    def set_tools(self, tools):
-        pass
-
-    def reset(self):
-        self._aggregation = ""
-
     async def push_aggregation(self):
-        if len(self._aggregation) > 0:
+        """Pushes the current aggregation. For example, iN the case of context
-            self._messages.append({"role": self._role, "content": self._aggregation})
+        aggregation this might push a new context frame.
 
-            # Reset the aggregation. Reset it before pushing it down, otherwise
+        """
-            # if the tasks gets cancelled we won't be able to clear things up.
+        pass
-            self._aggregation = ""
-
-            frame = LLMMessagesFrame(self._messages)
-            await self.push_frame(frame)
-
-            # Reset our accumulator state.
-            self.reset()
 
 
 class LLMContextResponseAggregator(BaseLLMResponseAggregator):
@@ -247,20 +208,6 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
     def reset(self):
         self._aggregation = ""
 
-    async def push_aggregation(self):
-        if len(self._aggregation) > 0:
-            self._context.add_message({"role": self.role, "content": self._aggregation})
-
-            # Reset the aggregation. Reset it before pushing it down, otherwise
-            # if the tasks gets cancelled we won't be able to clear things up.
-            self._aggregation = ""
-
-            frame = OpenAILLMContextFrame(self._context)
-            await self.push_frame(frame)
-
-            # Reset our accumulator state.
-            self.reset()
-
 
 class LLMUserContextAggregator(LLMContextResponseAggregator):
     """This is a user LLM aggregator that uses an LLM context to store the
@@ -290,12 +237,13 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         self._aggregation_event = asyncio.Event()
         self._aggregation_task = None
 
-        self.reset()
-
     def reset(self):
         super().reset()
         self._seen_interim_results = False
 
+    async def handle_aggregation(self, aggregation: str):
+        self._context.add_message({"role": self.role, "content": self._aggregation})
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -331,6 +279,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         else:
             await self.push_frame(frame, direction)
 
+    async def push_aggregation(self):
+        if len(self._aggregation) > 0:
+            await self.handle_aggregation(self._aggregation)
+
+            # Reset the aggregation. Reset it before pushing it down, otherwise
+            # if the tasks gets cancelled we won't be able to clear things up.
+            self._aggregation = ""
+
+            frame = OpenAILLMContextFrame(self._context)
+            await self.push_frame(frame)
+
+            # Reset our accumulator state.
+            self.reset()
+
     async def _start(self, frame: StartFrame):
         self._create_aggregation_task()
 
@@ -424,17 +386,29 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         super().__init__(context=context, role="assistant", **kwargs)
         self._expect_stripped_words = expect_stripped_words
 
-        self._started = False
+        self._started = 0
+        self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
 
-        self.reset()
+    async def handle_aggregation(self, aggregation: str):
+        self._context.add_message({"role": "assistant", "content": aggregation})
+
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        pass
+
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        pass
+
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
+        pass
+
+    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+        pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, StartInterruptionFrame):
-            await self.push_aggregation()
+            await self._handle_interruptions(frame)
-            # Reset anyways
-            self.reset()
             await self.push_frame(frame, direction)
         elif isinstance(frame, LLMFullResponseStartFrame):
             await self._handle_llm_start(frame)
@@ -448,14 +422,104 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             self.set_messages(frame.messages)
         elif isinstance(frame, LLMSetToolsFrame):
             self.set_tools(frame.tools)
+        elif isinstance(frame, FunctionCallInProgressFrame):
+            await self._handle_function_call_in_progress(frame)
+        elif isinstance(frame, FunctionCallResultFrame):
+            await self._handle_function_call_result(frame)
+        elif isinstance(frame, FunctionCallCancelFrame):
+            await self._handle_function_call_cancel(frame)
+        elif isinstance(frame, UserImageMessageFrame):
+            await self._handle_image_frame_message(frame)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self.push_aggregation()
         else:
             await self.push_frame(frame, direction)
 
+    async def push_aggregation(self):
+        if not self._aggregation:
+            return
+
+        aggregation = self._aggregation.strip()
+        self.reset()
+
+        if aggregation:
+            await self.handle_aggregation(aggregation)
+
+        # Push context frame
+        await self.push_context_frame()
+
+        # Push timestamp frame with current time
+        timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+        await self.push_frame(timestamp_frame)
+
+    async def _handle_interruptions(self, frame: StartInterruptionFrame):
+        await self.push_aggregation()
+        self._started = 0
+        self.reset()
+
+    async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        logger.debug(
+            f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+        await self.handle_function_call_in_progress(frame)
+        self._function_calls_in_progress[frame.tool_call_id] = frame
+
+    async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
+        logger.debug(
+            f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+        if frame.tool_call_id not in self._function_calls_in_progress:
+            logger.warning(
+                f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running"
+            )
+            return
+
+        del self._function_calls_in_progress[frame.tool_call_id]
+
+        properties = frame.properties
+
+        await self.handle_function_call_result(frame)
+
+        # Run inference if the function call result requires it.
+        if frame.result:
+            run_llm = False
+
+            if properties and properties.run_llm is not None:
+                # If the tool call result has a run_llm property, use it
+                run_llm = properties.run_llm
+            else:
+                # Default behavior is to run the LLM if there are no function calls in progress
+                run_llm = not bool(self._function_calls_in_progress)
+
+            if run_llm:
+                await self.push_context_frame(FrameDirection.UPSTREAM)
+
+        # Emit the on_context_updated callback once the function call
+        # result is added to the context
+        if properties and properties.on_context_updated:
+            await properties.on_context_updated()
+
+    async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
+        logger.debug(
+            f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"
+        )
+        if frame.tool_call_id not in self._function_calls_in_progress:
+            return
+
+        if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption:
+            await self.handle_function_call_cancel(frame)
+            del self._function_calls_in_progress[frame.tool_call_id]
+
+    async def _handle_image_frame_message(self, frame: UserImageMessageFrame):
+        await self.handle_image_frame_message(frame)
+        await self.push_aggregation()
+        await self.push_context_frame(FrameDirection.UPSTREAM)
+
     async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
-        self._started = True
+        self._started += 1
 
     async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
-        self._started = False
+        self._started -= 1
         await self.push_aggregation()
 
     async def _handle_text(self, frame: TextFrame):
@@ -474,7 +538,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
 
     async def push_aggregation(self):
         if len(self._aggregation) > 0:
-            self._context.add_message({"role": self.role, "content": self._aggregation})
+            await self.handle_aggregation(self._aggregation)
 
             # Reset the aggregation. Reset it before pushing it down, otherwise
             # if the tasks gets cancelled we won't be able to clear things up.
@@ -493,7 +557,7 @@ class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
 
     async def push_aggregation(self):
         if len(self._aggregation) > 0:
-            self._context.add_message({"role": self.role, "content": self._aggregation})
+            await self.handle_aggregation(self._aggregation)
 
             # Reset the aggregation. Reset it before pushing it down, otherwise
             # if the tasks gets cancelled we won't be able to clear things up.

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -9,9 +9,8 @@ import copy
 import io
 import json
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, List, Optional
+from typing import Any, List, Optional
 
-from loguru import logger
 from openai._types import NOT_GIVEN, NotGiven
 from openai.types.chat import (
     ChatCompletionMessageParam,
@@ -22,12 +21,7 @@ from PIL import Image
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.frames.frames import (
+from pipecat.frames.frames import AudioRawFrame, Frame
-    AudioRawFrame,
-    Frame,
-    FunctionCallInProgressFrame,
-    FunctionCallResultFrame,
-)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 # JSON custom encoder to handle bytes arrays so that we can log contexts
@@ -187,61 +181,6 @@ class OpenAILLMContext:
         # todo: implement for OpenAI models and others
         pass
 
-    async def call_function(
-        self,
-        f: Callable[
-            [str, str, Any, FrameProcessor, "OpenAILLMContext", Callable[[Any], Awaitable[None]]],
-            Awaitable[None],
-        ],
-        *,
-        function_name: str,
-        tool_call_id: str,
-        arguments: str,
-        llm: FrameProcessor,
-        run_llm: bool = True,
-    ) -> None:
-        logger.info(f"Calling function {function_name} with arguments {arguments}")
-        # Push a SystemFrame downstream. This frame will let our assistant context aggregator
-        # know that we are in the middle of a function call. Some contexts/aggregators may
-        # not need this. But some definitely do (Anthropic, for example).
-        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
-        progress_frame_downstream = FunctionCallInProgressFrame(
-            function_name=function_name,
-            tool_call_id=tool_call_id,
-            arguments=arguments,
-        )
-        progress_frame_upstream = FunctionCallInProgressFrame(
-            function_name=function_name,
-            tool_call_id=tool_call_id,
-            arguments=arguments,
-        )
-
-        # Push frame both downstream and upstream
-        await llm.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
-        await llm.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
-
-        # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
-        async def function_call_result_callback(result, *, properties=None):
-            result_frame_downstream = FunctionCallResultFrame(
-                function_name=function_name,
-                tool_call_id=tool_call_id,
-                arguments=arguments,
-                result=result,
-                properties=properties,
-            )
-            result_frame_upstream = FunctionCallResultFrame(
-                function_name=function_name,
-                tool_call_id=tool_call_id,
-                arguments=arguments,
-                result=result,
-                properties=properties,
-            )
-
-            await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
-            await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
-
-        await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback)
-
     def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
         # RIFF chunk descriptor
         header = bytearray()

src/pipecat/services/ai_services.py

@@ -8,7 +8,8 @@ import asyncio
 import io
 import wave
 from abc import abstractmethod
-from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Tuple, Type
+from dataclasses import dataclass
+from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Set, Tuple, Type
 
 from loguru import logger
 
@@ -22,6 +23,9 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     StartFrame,
@@ -138,6 +142,13 @@ class AIService(FrameProcessor):
                     await self.push_frame(f)
 
 
+@dataclass
+class FunctionEntry:
+    function_name: Optional[str]
+    callback: Any  # TODO(aleix): add proper typing.
+    cancel_on_interruption: bool
+
+
 class LLMService(AIService):
     """This class is a no-op but serves as a base class for LLM services."""
 
@@ -147,38 +158,74 @@ class LLMService(AIService):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        self._callbacks = {}
+        self._functions = {}
         self._start_callbacks = {}
         self._adapter = self.adapter_class()
+        self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set()
+
+        self._register_event_handler("on_completion_timeout")
 
     def get_llm_adapter(self) -> BaseLLMAdapter:
         return self._adapter
 
     def create_context_aggregator(
-        self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_kwargs: Mapping[str, Any] = {},
+        assistant_kwargs: Mapping[str, Any] = {},
     ) -> Any:
         pass
 
-        self._register_event_handler("on_completion_timeout")
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
 
-    # TODO-CB: callback function type
+        if isinstance(frame, StartInterruptionFrame):
-    def register_function(self, function_name: Optional[str], callback, start_callback=None):
+            await self._handle_interruptions(frame)
+
+    async def _handle_interruptions(self, frame: StartInterruptionFrame):
+        for function_name, entry in self._functions.items():
+            if entry.cancel_on_interruption:
+                await self._cancel_function_call(function_name)
+
+    def register_function(
+        self,
+        function_name: Optional[str],
+        callback: Any,
+        start_callback=None,
+        *,
+        cancel_on_interruption: bool = False,
+    ):
         # Registering a function with the function_name set to None will run that callback
         # for all functions
-        self._callbacks[function_name] = callback
+        self._functions[function_name] = FunctionEntry(
-        # QUESTION FOR CB: maybe this isn't needed anymore?
+            function_name=function_name,
+            callback=callback,
+            cancel_on_interruption=cancel_on_interruption,
+        )
+
+        # Start callbacks are now deprecated.
         if start_callback:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'start_callback' is deprecated, just put your code on top of the actual function call instead.",
+                    DeprecationWarning,
+                )
+
             self._start_callbacks[function_name] = start_callback
 
     def unregister_function(self, function_name: Optional[str]):
-        del self._callbacks[function_name]
+        del self._functions[function_name]
         if self._start_callbacks[function_name]:
             del self._start_callbacks[function_name]
 
     def has_function(self, function_name: str):
-        if None in self._callbacks.keys():
+        if None in self._functions.keys():
             return True
-        return function_name in self._callbacks.keys()
+        return function_name in self._functions.keys()
 
     async def call_function(
         self,
@@ -188,25 +235,18 @@ class LLMService(AIService):
         function_name: str,
         arguments: str,
         run_llm: bool = True,
-    ) -> None:
+    ):
-        f = None
+        if not function_name in self._functions.keys() and not None in self._functions.keys():
-        if function_name in self._callbacks.keys():
+            return
-            f = self._callbacks[function_name]
+
-        elif None in self._callbacks.keys():
+        task = self.create_task(
-            f = self._callbacks[None]
+            self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
-        else:
-            return None
-        await self.call_start_function(context, function_name)
-        await context.call_function(
-            f,
-            function_name=function_name,
-            tool_call_id=tool_call_id,
-            arguments=arguments,
-            llm=self,
-            run_llm=run_llm,
         )
 
-    # QUESTION FOR CB: maybe this isn't needed anymore?
+        self._function_call_tasks.add((task, tool_call_id, function_name))
+
+        task.add_done_callback(self._function_call_task_finished)
+
     async def call_start_function(self, context: OpenAILLMContext, function_name: str):
         if function_name in self._start_callbacks.keys():
             await self._start_callbacks[function_name](function_name, self, context)
@@ -218,6 +258,106 @@ class LLMService(AIService):
             UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM
         )
 
+    async def _run_function_call(
+        self,
+        context: OpenAILLMContext,
+        tool_call_id: str,
+        function_name: str,
+        arguments: str,
+        run_llm: bool = True,
+    ):
+        if function_name in self._functions.keys():
+            entry = self._functions[function_name]
+        elif None in self._functions.keys():
+            entry = self._functions[None]
+        else:
+            return
+
+        logger.info(f"Calling function {function_name} with arguments {arguments}")
+
+        # NOTE(aleix): This needs to be removed after we remove the deprecation.
+        await self.call_start_function(context, function_name)
+
+        # Push a SystemFrame downstream. This frame will let our assistant context aggregator
+        # know that we are in the middle of a function call. Some contexts/aggregators may
+        # not need this. But some definitely do (Anthropic, for example).
+        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
+        progress_frame_downstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+            cancel_on_interruption=entry.cancel_on_interruption,
+        )
+        progress_frame_upstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+            cancel_on_interruption=entry.cancel_on_interruption,
+        )
+
+        # Push frame both downstream and upstream
+        await self.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
+        await self.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
+
+        # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
+        async def function_call_result_callback(result, *, properties=None):
+            result_frame_downstream = FunctionCallResultFrame(
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                arguments=arguments,
+                result=result,
+                properties=properties,
+            )
+            result_frame_upstream = FunctionCallResultFrame(
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                arguments=arguments,
+                result=result,
+                properties=properties,
+            )
+
+            await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
+            await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
+
+        await entry.callback(
+            function_name, tool_call_id, arguments, self, context, function_call_result_callback
+        )
+
+    async def _cancel_function_call(self, function_name: str):
+        cancelled_tasks = set()
+        for task, tool_call_id, name in self._function_call_tasks:
+            if name == function_name:
+                # We remove the callback because we are going to cancel the task
+                # now, otherwise we will be removing it from the set while we
+                # are iterating.
+                task.remove_done_callback(self._function_call_task_finished)
+
+                logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...")
+
+                await self.cancel_task(task)
+
+                frame = FunctionCallCancelFrame(
+                    function_name=function_name, tool_call_id=tool_call_id
+                )
+                await self.push_frame(frame)
+
+                logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
+
+                cancelled_tasks.add(task)
+
+        # Remove all cancelled tasks from our set.
+        for task in cancelled_tasks:
+            self._function_call_task_finished(task)
+
+    def _function_call_task_finished(self, task: asyncio.Task):
+        tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None)
+        if tuple_to_remove:
+            self._function_call_tasks.discard(tuple_to_remove)
+            # The task is finished so this should exit immediately. We need to
+            # do this because otherwise the task manager would have a dangling
+            # task if we don't remove it.
+            asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())
+
 
 class TTSService(AIService):
     def __init__(
@@ -366,12 +506,14 @@ class TTSService(AIService):
             else:
                 await self.push_frame(frame, direction)
         elif isinstance(frame, TTSSpeakFrame):
+            # Store if we were processing text or not so we can set it back.
+            processing_text = self._processing_text
             await self._push_tts_frames(frame.text)
             # We pause processing incoming frames because we are sending data to
             # the TTS. We pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
             await self.flush_audio()
-            self._processing_text = False
+            self._processing_text = processing_text
         elif isinstance(frame, TTSUpdateSettingsFrame):
             await self._update_settings(frame.settings)
         elif isinstance(frame, BotStoppedSpeakingFrame):

src/pipecat/services/anthropic.py

@@ -21,17 +21,16 @@ from pydantic import BaseModel, Field
 from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
 from pipecat.frames.frames import (
     Frame,
+    FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
-    FunctionCallResultProperties,
     LLMEnablePromptCachingFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
-    OpenAILLMContextAssistantTimestampFrame,
+    UserImageMessageFrame,
-    StartInterruptionFrame,
     UserImageRawFrame,
     UserImageRequestFrame,
     VisionImageRawFrame,
@@ -47,7 +46,6 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import LLMService
-from pipecat.utils.time import time_now_iso8601
 
 try:
     from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
@@ -60,13 +58,6 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-# internal use only -- todo: refactor
-@dataclass
-class AnthropicImageMessageFrame(Frame):
-    user_image_raw_frame: UserImageRawFrame
-    text: Optional[str] = None
-
-
 @dataclass
 class AnthropicContextAggregatorPair:
     _user: "AnthropicUserContextAggregator"
@@ -715,7 +706,7 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
                 text = self._context._user_image_request_context.get(frame.user_id) or ""
                 if text:
                     del self._context._user_image_request_context[frame.user_id]
-                frame = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text)
+                frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text)
                 await self.push_frame(frame)
         except Exception as e:
             logger.error(f"Error processing frame: {e}")
@@ -734,110 +725,61 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
 class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
     def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
         super().__init__(context=context, **kwargs)
-        self._function_call_in_progress = None
-        self._function_call_result = None
-        self._pending_image_frame_message = None
 
-    async def process_frame(self, frame, direction):
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
-        await super().process_frame(frame, direction)
+        assistant_message = {"role": "assistant", "content": []}
-        # See note above about not calling push_frame() here.
+        assistant_message["content"].append(
-        if isinstance(frame, StartInterruptionFrame):
+            {
-            self._function_call_in_progress = None
+                "type": "tool_use",
-            self._function_call_finished = None
+                "id": frame.tool_call_id,
-        elif isinstance(frame, FunctionCallInProgressFrame):
+                "name": frame.function_name,
-            self._function_call_in_progress = frame
+                "input": frame.arguments,
-        elif isinstance(frame, FunctionCallResultFrame):
+            }
-            if (
+        )
-                self._function_call_in_progress
+        self._context.add_message(assistant_message)
-                and self._function_call_in_progress.tool_call_id == frame.tool_call_id
+        self._context.add_message(
-            ):
+            {
-                self._function_call_in_progress = None
+                "role": "user",
-                self._function_call_result = frame
+                "content": [
-                await self.push_aggregation()
+                    {
-            else:
+                        "type": "tool_result",
-                logger.warning(
+                        "tool_use_id": frame.tool_call_id,
-                    "FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id"
+                        "content": "IN_PROGRESS",
-                )
+                    }
-                self._function_call_in_progress = None
+                ],
-                self._function_call_result = None
+            }
-        elif isinstance(frame, AnthropicImageMessageFrame):
+        )
-            self._pending_image_frame_message = frame
-            await self.push_aggregation()
 
-    async def push_aggregation(self):
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
-        if not (
+        if not frame.result:
-            self._aggregation or self._function_call_result or self._pending_image_frame_message
-        ):
             return
 
-        run_llm = False
+        result = json.dumps(frame.result)
-        properties: Optional[FunctionCallResultProperties] = None
 
-        aggregation = self._aggregation.strip()
+        await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
-        self.reset()
 
-        try:
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
-            if aggregation:
+        await self._update_function_call_result(
-                self._context.add_message({"role": "assistant", "content": aggregation})
+            frame.function_name, frame.tool_call_id, "CANCELLED"
+        )
 
-            if self._function_call_result:
+    async def _update_function_call_result(
-                frame = self._function_call_result
+        self, function_name: str, tool_call_id: str, result: str
-                properties = frame.properties
+    ):
-                self._function_call_result = None
+        for message in self._context.messages:
-                if frame.result:
+            if message["role"] == "user":
-                    assistant_message = {"role": "assistant", "content": []}
+                for content in message["content"]:
-                    assistant_message["content"].append(
+                    if (
-                        {
+                        isinstance(content, dict)
-                            "type": "tool_use",
+                        and content["type"] == "tool_result"
-                            "id": frame.tool_call_id,
+                        and content["tool_use_id"] == tool_call_id
-                            "name": frame.function_name,
+                    ):
-                            "input": frame.arguments,
+                        content["content"] = result
-                        }
-                    )
-                    self._context.add_message(assistant_message)
-                    self._context.add_message(
-                        {
-                            "role": "user",
-                            "content": [
-                                {
-                                    "type": "tool_result",
-                                    "tool_use_id": frame.tool_call_id,
-                                    "content": json.dumps(frame.result),
-                                }
-                            ],
-                        }
-                    )
-                    if properties and properties.run_llm is not None:
-                        # If the tool call result has a run_llm property, use it
-                        run_llm = properties.run_llm
-                    else:
-                        # Default behavior
-                        run_llm = True
 
-            if self._pending_image_frame_message:
+    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
-                frame = self._pending_image_frame_message
+        self._context.add_image_frame_message(
-                self._pending_image_frame_message = None
+            format=frame.user_image_raw_frame.format,
-                self._context.add_image_frame_message(
+            size=frame.user_image_raw_frame.size,
-                    format=frame.user_image_raw_frame.format,
+            image=frame.user_image_raw_frame.image,
-                    size=frame.user_image_raw_frame.size,
+            text=frame.text,
-                    image=frame.user_image_raw_frame.image,
+        )
-                    text=frame.text,
-                )
-                run_llm = True
-
-            if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
-
-            # Emit the on_context_updated callback once the function call result is added to the context
-            if properties and properties.on_context_updated is not None:
-                await properties.on_context_updated()
-
-            # Push context frame
-            await self.push_context_frame()
-
-            # Push timestamp frame with current time
-            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
-            await self.push_frame(timestamp_frame)
-
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -39,6 +39,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
     TTSTextFrame,
+    UserImageMessageFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -118,10 +119,10 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    async def push_aggregation(self):
+    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
-        # We don't want to store any images in the context. Revisit this later when the API evolves.
+        # We don't want to store any images in the context. Revisit this later
-        self._pending_image_frame_message = None
+        # when the API evolves.
-        await super().push_aggregation()
+        pass
 
 
 @dataclass

src/pipecat/services/google/google.py

@@ -10,6 +10,7 @@ import io
 import json
 import os
 import time
+import uuid
 
 from google.api_core.exceptions import DeadlineExceeded
 from openai import AsyncStream
@@ -33,20 +34,22 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
-    FunctionCallResultProperties,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
-    OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
     TranscriptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
     URLImageRawFrame,
+    UserImageMessageFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
@@ -565,91 +568,69 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    async def push_aggregation(self):
+    async def handle_aggregation(self, aggregation: str):
-        if not (
+        self._context.add_message(glm.Content(role="model", parts=[glm.Part(text=aggregation)]))
-            self._aggregation or self._function_call_result or self._pending_image_frame_message
+
-        ):
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
+        self._context.add_message(
+            glm.Content(
+                role="model",
+                parts=[
+                    glm.Part(
+                        function_call=glm.FunctionCall(
+                            id=frame.tool_call_id, name=frame.function_name, args=frame.arguments
+                        )
+                    )
+                ],
+            )
+        )
+        self._context.add_message(
+            glm.Content(
+                role="user",
+                parts=[
+                    glm.Part(
+                        function_response=glm.FunctionResponse(
+                            id=frame.tool_call_id,
+                            name=frame.function_name,
+                            response={"response": "IN_PROGRESS"},
+                        )
+                    )
+                ],
+            )
+        )
+
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        if not frame.result:
             return
 
-        run_llm = False
+        if not isinstance(frame.result, str):
-        properties: Optional[FunctionCallResultProperties] = None
+            return
 
-        aggregation = self._aggregation.strip()
+        response = {"response": frame.result}
-        self.reset()
 
-        try:
+        await self._update_function_call_result(frame.function_name, frame.tool_call_id, response)
-            if aggregation:
-                self._context.add_message(
-                    glm.Content(role="model", parts=[glm.Part(text=aggregation)])
-                )
 
-            if self._function_call_result:
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
-                frame = self._function_call_result
+        await self._update_function_call_result(
-                properties = frame.properties
+            frame.function_name, frame.tool_call_id, "CANCELLED"
-                self._function_call_result = None
+        )
-                if frame.result:
-                    logger.debug(f"FunctionCallResultFrame result: {frame.arguments}")
-                    self._context.add_message(
-                        glm.Content(
-                            role="model",
-                            parts=[
-                                glm.Part(
-                                    function_call=glm.FunctionCall(
-                                        name=frame.function_name, args=frame.arguments
-                                    )
-                                )
-                            ],
-                        )
-                    )
-                    response = frame.result
-                    if isinstance(response, str):
-                        response = {"response": response}
-                    self._context.add_message(
-                        glm.Content(
-                            role="user",
-                            parts=[
-                                glm.Part(
-                                    function_response=glm.FunctionResponse(
-                                        name=frame.function_name, response=response
-                                    )
-                                )
-                            ],
-                        )
-                    )
-                    if properties and properties.run_llm is not None:
-                        # If the tool call result has a run_llm property, use it
-                        run_llm = properties.run_llm
-                    else:
-                        # Default behavior is to run the LLM if there are no function calls in progress
-                        run_llm = not bool(self._function_calls_in_progress)
 
-            if self._pending_image_frame_message:
+    async def _update_function_call_result(
-                frame = self._pending_image_frame_message
+        self, function_name: str, tool_call_id: str, result: Any
-                self._pending_image_frame_message = None
+    ):
-                self._context.add_image_frame_message(
+        for message in self._context.messages:
-                    format=frame.user_image_raw_frame.format,
+            if message.role == "user":
-                    size=frame.user_image_raw_frame.size,
+                for part in message.parts:
-                    image=frame.user_image_raw_frame.image,
+                    if part.function_response and part.function_response.id == tool_call_id:
-                    text=frame.text,
+                        part.function_response.response = result
-                )
-                run_llm = True
 
-            if run_llm:
+    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
-                await self.push_context_frame(FrameDirection.UPSTREAM)
+        self._context.add_image_frame_message(
-
+            format=frame.user_image_raw_frame.format,
-            # Emit the on_context_updated callback once the function call result is added to the context
+            size=frame.user_image_raw_frame.size,
-            if properties and properties.on_context_updated is not None:
+            image=frame.user_image_raw_frame.image,
-                await properties.on_context_updated()
+            text=frame.text,
-
+        )
-            # Push context frame
-            await self.push_context_frame()
-
-            # Push timestamp frame with current time
-            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
-            await self.push_frame(timestamp_frame)
-
-        except Exception as e:
-            logger.exception(f"Error processing frame: {e}")
 
 
 @dataclass
@@ -1071,7 +1052,7 @@ class GoogleLLMService(LLMService):
                             args = type(c.function_call).to_dict(c.function_call).get("args", {})
                             await self.call_function(
                                 context=context,
-                                tool_call_id="what_should_this_be",
+                                tool_call_id=str(uuid.uuid4()),
                                 function_name=c.function_call.name,
                                 arguments=args,
                             )

src/pipecat/services/grok.py

@@ -25,94 +25,15 @@ from pipecat.services.openai import (
 )
 
 
-class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    """Custom assistant context aggregator for Grok that handles empty content requirement."""
-
-    async def push_aggregation(self):
-        if not (
-            self._aggregation or self._function_call_result or self._pending_image_frame_message
-        ):
-            return
-
-        run_llm = False
-        properties: Optional[FunctionCallResultProperties] = None
-
-        aggregation = self._aggregation.strip()
-        self.reset()
-
-        try:
-            if aggregation:
-                self._context.add_message({"role": "assistant", "content": aggregation})
-
-            if self._function_call_result:
-                frame = self._function_call_result
-                properties = frame.properties
-                self._function_call_result = None
-                if frame.result:
-                    # Grok requires an empty content field for function calls
-                    self._context.add_message(
-                        {
-                            "role": "assistant",
-                            "content": "",  # Required by Grok
-                            "tool_calls": [
-                                {
-                                    "id": frame.tool_call_id,
-                                    "function": {
-                                        "name": frame.function_name,
-                                        "arguments": json.dumps(frame.arguments),
-                                    },
-                                    "type": "function",
-                                }
-                            ],
-                        }
-                    )
-                    self._context.add_message(
-                        {
-                            "role": "tool",
-                            "content": json.dumps(frame.result),
-                            "tool_call_id": frame.tool_call_id,
-                        }
-                    )
-                    if properties and properties.run_llm is not None:
-                        # If the tool call result has a run_llm property, use it
-                        run_llm = properties.run_llm
-                    else:
-                        # Default behavior is to run the LLM if there are no function calls in progress
-                        run_llm = not bool(self._function_calls_in_progress)
-
-            if self._pending_image_frame_message:
-                frame = self._pending_image_frame_message
-                self._pending_image_frame_message = None
-                self._context.add_image_frame_message(
-                    format=frame.user_image_raw_frame.format,
-                    size=frame.user_image_raw_frame.size,
-                    image=frame.user_image_raw_frame.image,
-                    text=frame.text,
-                )
-                run_llm = True
-
-            if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
-
-            # Emit the on_context_updated callback once the function call result is added to the context
-            if properties and properties.on_context_updated is not None:
-                await properties.on_context_updated()
-
-            await self.push_context_frame()
-
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")
-
-
 @dataclass
 class GrokContextAggregatorPair:
     _user: "OpenAIUserContextAggregator"
-    _assistant: "GrokAssistantContextAggregator"
+    _assistant: "OpenAIAssistantContextAggregator"
 
     def user(self) -> "OpenAIUserContextAggregator":
         return self._user
 
-    def assistant(self) -> "GrokAssistantContextAggregator":
+    def assistant(self) -> "OpenAIAssistantContextAggregator":
         return self._assistant
 
 
@@ -235,5 +156,5 @@ class GrokLLMService(OpenAILLMService):
         context.set_llm_adapter(self.get_llm_adapter())
 
         user = OpenAIUserContextAggregator(context, **user_kwargs)
-        assistant = GrokAssistantContextAggregator(context, **assistant_kwargs)
+        assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs)
         return GrokContextAggregatorPair(_user=user, _assistant=assistant)

src/pipecat/services/openai.py

@@ -27,21 +27,20 @@ from pydantic import BaseModel, Field
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
+    FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
-    FunctionCallResultProperties,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
-    OpenAILLMContextAssistantTimestampFrame,
     StartFrame,
-    StartInterruptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
     URLImageRawFrame,
+    UserImageMessageFrame,
     UserImageRawFrame,
     UserImageRequestFrame,
     VisionImageRawFrame,
@@ -63,7 +62,6 @@ from pipecat.services.ai_services import (
 )
 from pipecat.services.base_whisper import BaseWhisperSTTService, Transcription
 from pipecat.transcriptions.language import Language
-from pipecat.utils.time import time_now_iso8601
 
 ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"]
 
@@ -558,13 +556,6 @@ class OpenAITTSService(TTSService):
             logger.exception(f"{self} error generating TTS: {e}")
 
 
-# internal use only -- todo: refactor
-@dataclass
-class OpenAIImageMessageFrame(Frame):
-    user_image_raw_frame: UserImageRawFrame
-    text: Optional[str] = None
-
-
 class OpenAIUserContextAggregator(LLMUserContextAggregator):
     def __init__(self, context: OpenAILLMContext, **kwargs):
         super().__init__(context=context, **kwargs)
@@ -596,7 +587,7 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator):
                 text = self._context._user_image_request_context.get(frame.user_id) or ""
                 if text:
                     del self._context._user_image_request_context[frame.user_id]
-                frame = OpenAIImageMessageFrame(user_image_raw_frame=frame, text=text)
+                frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text)
                 await self.push_frame(frame)
         except Exception as e:
             logger.error(f"Error processing frame: {e}")
@@ -605,109 +596,59 @@ class OpenAIUserContextAggregator(LLMUserContextAggregator):
 class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
     def __init__(self, context: OpenAILLMContext, **kwargs):
         super().__init__(context=context, **kwargs)
-        self._function_calls_in_progress = {}
-        self._function_call_result = None
-        self._pending_image_frame_message = None
 
-    async def process_frame(self, frame, direction):
+    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
-        await super().process_frame(frame, direction)
+        self._context.add_message(
-        # See note above about not calling push_frame() here.
+            {
-        if isinstance(frame, StartInterruptionFrame):
+                "role": "assistant",
-            self._function_calls_in_progress.clear()
+                "tool_calls": [
-            self._function_call_finished = None
+                    {
-        elif isinstance(frame, FunctionCallInProgressFrame):
+                        "id": frame.tool_call_id,
-            logger.debug(f"FunctionCallInProgressFrame: {frame}")
+                        "function": {
-            self._function_calls_in_progress[frame.tool_call_id] = frame
+                            "name": frame.function_name,
-        elif isinstance(frame, FunctionCallResultFrame):
+                            "arguments": json.dumps(frame.arguments),
-            logger.debug(f"FunctionCallResultFrame: {frame}")
+                        },
-            if frame.tool_call_id in self._function_calls_in_progress:
+                        "type": "function",
-                del self._function_calls_in_progress[frame.tool_call_id]
+                    }
-                self._function_call_result = frame
+                ],
-                # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE
+            }
-                await self.push_aggregation()
+        )
-            else:
+        self._context.add_message(
-                logger.warning(
+            {
-                    "FunctionCallResultFrame tool_call_id does not match any function call in progress"
+                "role": "tool",
-                )
+                "content": "IN_PROGRESS",
-                self._function_call_result = None
+                "tool_call_id": frame.tool_call_id,
-        elif isinstance(frame, OpenAIImageMessageFrame):
+            }
-            self._pending_image_frame_message = frame
+        )
-            await self.push_aggregation()
 
-    async def push_aggregation(self):
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
-        if not (
+        if not frame.result:
-            self._aggregation or self._function_call_result or self._pending_image_frame_message
-        ):
             return
 
-        run_llm = False
+        result = json.dumps(frame.result)
-        properties: Optional[FunctionCallResultProperties] = None
 
-        aggregation = self._aggregation.strip()
+        await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
-        self.reset()
 
-        try:
+    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
-            if aggregation:
+        await self._update_function_call_result(
-                self._context.add_message({"role": "assistant", "content": aggregation})
+            frame.function_name, frame.tool_call_id, "CANCELLED"
+        )
 
-            if self._function_call_result:
+    async def _update_function_call_result(
-                frame = self._function_call_result
+        self, function_name: str, tool_call_id: str, result: str
-                properties = frame.properties
+    ):
-                self._function_call_result = None
+        for message in self._context.messages:
-                if frame.result:
+            if (
-                    self._context.add_message(
+                message["role"] == "tool"
-                        {
+                and message["tool_call_id"]
-                            "role": "assistant",
+                and message["tool_call_id"] == tool_call_id
-                            "tool_calls": [
+            ):
-                                {
+                message["content"] = result
-                                    "id": frame.tool_call_id,
-                                    "function": {
-                                        "name": frame.function_name,
-                                        "arguments": json.dumps(frame.arguments),
-                                    },
-                                    "type": "function",
-                                }
-                            ],
-                        }
-                    )
-                    self._context.add_message(
-                        {
-                            "role": "tool",
-                            "content": json.dumps(frame.result),
-                            "tool_call_id": frame.tool_call_id,
-                        }
-                    )
-                    if properties and properties.run_llm is not None:
-                        # If the tool call result has a run_llm property, use it
-                        run_llm = properties.run_llm
-                    else:
-                        # Default behavior is to run the LLM if there are no function calls in progress
-                        run_llm = not bool(self._function_calls_in_progress)
 
-            if self._pending_image_frame_message:
+    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
-                frame = self._pending_image_frame_message
+        self._context.add_image_frame_message(
-                self._pending_image_frame_message = None
+            format=frame.user_image_raw_frame.format,
-                self._context.add_image_frame_message(
+            size=frame.user_image_raw_frame.size,
-                    format=frame.user_image_raw_frame.format,
+            image=frame.user_image_raw_frame.image,
-                    size=frame.user_image_raw_frame.size,
+            text=frame.text,
-                    image=frame.user_image_raw_frame.image,
+        )
-                    text=frame.text,
-                )
-                run_llm = True
-
-            if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
-
-            # Emit the on_context_updated callback once the function call result is added to the context
-            if properties and properties.on_context_updated is not None:
-                await properties.on_context_updated()
-
-            # Push context frame
-            await self.push_context_frame()
-
-            # Push timestamp frame with current time
-            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
-            await self.push_frame(timestamp_frame)
-
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")

tests/test_context_aggregators.py

@@ -418,7 +418,7 @@ class BaseTestUserContextAggregator:
 class BaseTestAssistantContextAggreagator:
     CONTEXT_CLASS = None  # To be set in subclasses
     AGGREGATOR_CLASS = None  # To be set in subclasses
-    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame]
+    EXPECTED_CONTEXT_FRAMES = None  # To be set in subclasses
 
     def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
         assert context.messages[index]["content"] == content
@@ -577,6 +577,7 @@ class TestLLMAssistantContextAggregator(
 ):
     CONTEXT_CLASS = OpenAILLMContext
     AGGREGATOR_CLASS = LLMAssistantContextAggregator
+    EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
 
 
 #