From efa5f133d747783c43c11ff5b1aad929a1f01484 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Thu, 20 Mar 2025 11:14:59 -0700
Subject: [PATCH] openai_realtime: fix and update function calling

---
 .../services/openai_realtime_beta/context.py  | 72 +++----------------
 .../services/openai_realtime_beta/openai.py   |  4 +-
 2 files changed, 11 insertions(+), 65 deletions(-)

diff --git a/src/pipecat/services/openai_realtime_beta/context.py b/src/pipecat/services/openai_realtime_beta/context.py
index 31639dc6b..c8381976f 100644
--- a/src/pipecat/services/openai_realtime_beta/context.py
+++ b/src/pipecat/services/openai_realtime_beta/context.py
@@ -12,6 +12,7 @@ from loguru import logger
 
 from pipecat.frames.frames import (
     Frame,
+    FunctionCallResultFrame,
     FunctionCallResultProperties,
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
@@ -174,67 +175,12 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    async def push_aggregation(self):
-        # the only thing we implement here is function calling. in all other cases, messages
-        # are added to the context when we receive openai realtime api events
-        if not self._function_call_result:
-            return
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        await super().handle_function_call_result(frame)
 
-        properties: Optional[FunctionCallResultProperties] = None
-
-        self.reset()
-        try:
-            run_llm = True
-            frame = self._function_call_result
-            properties = frame.properties
-            self._function_call_result = None
-            if frame.result:
-                # The "tool_call" message from the LLM that triggered the function call
-                self._context.add_message(
-                    {
-                        "role": "assistant",
-                        "tool_calls": [
-                            {
-                                "id": frame.tool_call_id,
-                                "function": {
-                                    "name": frame.function_name,
-                                    "arguments": json.dumps(frame.arguments),
-                                },
-                                "type": "function",
-                            }
-                        ],
-                    }
-                )
-                # The result of the function call. Need to add this both to our context here and to
-                # the openai realtime api context.
-                result_message = {
-                    "role": "tool",
-                    "content": json.dumps(frame.result),
-                    "tool_call_id": frame.tool_call_id,
-                }
-
-                self._context.add_message(result_message)
-                # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,
-                # so we didn't have a chance to add the result to the openai realtime api context. Let's push a
-                # special frame to do that.
-                await self.push_frame(
-                    RealtimeFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
-                )
-                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 is not None:
-                await properties.on_context_updated()
-
-            await self.push_context_frame()
-
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")
+        # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself,
+        # so we didn't have a chance to add the result to the openai realtime api context. Let's push a
+        # special frame to do that.
+        await self.push_frame(
+            RealtimeFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
+        )
diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py
index c8f1f597a..a1f4bc731 100644
--- a/src/pipecat/services/openai_realtime_beta/openai.py
+++ b/src/pipecat/services/openai_realtime_beta/openai.py
@@ -579,7 +579,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             arguments = json.loads(item.arguments)
             if self.has_function(function_name):
                 run_llm = index == total_items - 1
-                if function_name in self._callbacks.keys():
+                if function_name in self._functions.keys():
                     await self.call_function(
                         context=self._context,
                         tool_call_id=tool_id,
@@ -587,7 +587,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
                         arguments=arguments,
                         run_llm=run_llm,
                     )
-                elif None in self._callbacks.keys():
+                elif None in self._functions.keys():
                     await self.call_function(
                         context=self._context,
                         tool_call_id=tool_id,