Merge pull request #2914 from pipecat-ai/pk/gemini-function-calling-fixes

Gemini function calling fixes

src/pipecat/adapters/services/anthropic_adapter.py

@@ -110,7 +110,7 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
         system = NOT_GIVEN
         messages = []
 
-        # first, map messages using self._from_universal_context_message(m)
+        # First, map messages using self._from_universal_context_message(m)
         try:
             messages = [self._from_universal_context_message(m) for m in universal_context_messages]
         except Exception as e:

src/pipecat/adapters/services/bedrock_adapter.py

@@ -107,7 +107,7 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
         system = None
         messages = []
 
-        # first, map messages using self._from_universal_context_message(m)
+        # First, map messages using self._from_universal_context_message(m)
         try:
             messages = [self._from_universal_context_message(m) for m in universal_context_messages]
         except Exception as e:

src/pipecat/adapters/services/gemini_adapter.py

@@ -8,8 +8,8 @@
 
 import base64
 import json
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, TypedDict
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple, TypedDict
 
 from loguru import logger
 from openai import NotGiven
@@ -133,6 +133,28 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
         messages: List[Content]
         system_instruction: Optional[str] = None
 
+    @dataclass
+    class MessageConversionResult:
+        """Result of converting a single universal context message to Google format.
+
+        Either content (a Google Content object) or a system instruction string
+        is guaranteed to be set.
+
+        Also returns a tool call ID to name mapping for any tool calls
+        discovered in the message.
+        """
+
+        content: Optional[Content] = None
+        system_instruction: Optional[str] = None
+        tool_call_id_to_name_mapping: Dict[str, str] = field(default_factory=dict)
+
+    @dataclass
+    class MessageConversionParams:
+        """Parameters for converting a single universal context message to Google format."""
+
+        already_have_system_instruction: bool
+        tool_call_id_to_name_mapping: Dict[str, str]
+
     def _from_universal_context_messages(
         self, universal_context_messages: List[LLMContextMessage]
     ) -> ConvertedMessages:
@@ -156,24 +178,26 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
         """
         system_instruction = None
         messages = []
+        tool_call_id_to_name_mapping = {}
 
         # Process each message, preserving Google-formatted messages and converting others
         for message in universal_context_messages:
-            if isinstance(message, LLMSpecificMessage):
-                # Assume that LLMSpecificMessage wraps a message in Google format
-                messages.append(message.message)
-                continue
-
-            # Convert standard format to Google format
-            converted = self._from_standard_message(
-                message, already_have_system_instruction=bool(system_instruction)
+            result = self._from_universal_context_message(
+                message,
+                params=self.MessageConversionParams(
+                    already_have_system_instruction=bool(system_instruction),
+                    tool_call_id_to_name_mapping=tool_call_id_to_name_mapping,
+                ),
             )
-            if isinstance(converted, Content):
-                # Regular (non-system) message
-                messages.append(converted)
-            else:
-                # System instruction
-                system_instruction = converted
+            # Each result is either a Content or a system instruction
+            if result.content:
+                messages.append(result.content)
+            elif result.system_instruction:
+                system_instruction = result.system_instruction
+
+            # Merge tool call ID to name mapping
+            if result.tool_call_id_to_name_mapping:
+                tool_call_id_to_name_mapping.update(result.tool_call_id_to_name_mapping)
 
         # Check if we only have function-related messages (no regular text)
         has_regular_messages = any(
@@ -193,9 +217,16 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
 
         return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
 
+    def _from_universal_context_message(
+        self, message: LLMContextMessage, *, params: MessageConversionParams
+    ) -> MessageConversionResult:
+        if isinstance(message, LLMSpecificMessage):
+            return self.MessageConversionResult(content=message.message)
+        return self._from_standard_message(message, params=params)
+
     def _from_standard_message(
-        self, message: LLMStandardMessage, already_have_system_instruction: bool
-    ) -> Content | str:
+        self, message: LLMStandardMessage, *, params: MessageConversionParams
+    ) -> MessageConversionResult:
         """Convert standard universal context message to Google Content object.
 
         Handles conversion of text, images, and function calls to Google's
@@ -205,10 +236,11 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
         Args:
             message: Message in standard universal context format.
             already_have_system_instruction: Whether we already have a system instruction
+            params: Parameters for conversion.
 
         Returns:
-            Content object with role and parts, or a plain string for system
-            messages.
+            MessageConversionResult containing either a Content object or a
+            system instruction string.
 
         Examples:
             Standard text message::
@@ -242,38 +274,48 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
             Converts to Google Content with::
 
                 Content(
-                    role="model",
+                    role="user",
                     parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
                 )
         """
         role = message["role"]
         content = message.get("content", [])
+
         if role == "system":
-            if already_have_system_instruction:
+            if params.already_have_system_instruction:
                 role = "user"  # Convert system message to user role if we already have a system instruction
             else:
-                # System instructions are returned as plain text
+                system_instruction: str = None
                 if isinstance(content, str):
-                    return content
+                    system_instruction = content
                 elif isinstance(content, list):
                     # If content is a list, we assume it's a list of text parts, per the standard
-                    return " ".join(part["text"] for part in content if part.get("type") == "text")
+                    system_instruction = " ".join(
+                        part["text"] for part in content if part.get("type") == "text"
+                    )
+                if system_instruction:
+                    return self.MessageConversionResult(system_instruction=system_instruction)
         elif role == "assistant":
             role = "model"
 
         parts = []
+        tool_call_id_to_name_mapping = {}
+
         if message.get("tool_calls"):
             for tc in message["tool_calls"]:
+                id = tc["id"]
+                name = tc["function"]["name"]
+                tool_call_id_to_name_mapping[id] = name
                 parts.append(
                     Part(
                         function_call=FunctionCall(
-                            name=tc["function"]["name"],
+                            name=name,
                             args=json.loads(tc["function"]["arguments"]),
                         )
                     )
                 )
         elif role == "tool":
-            role = "model"
+            role = "user"
             try:
                 response = json.loads(message["content"])
                 if isinstance(response, dict):
@@ -284,12 +326,17 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                 # Response might not be JSON-deserializable.
                 # This occurs with a UserImageFrame, for example, where we get a plain "COMPLETED" string.
                 response_dict = {"value": message["content"]}
+
+            # Get function name from mapping using tool_call_id, or fallback
+            tool_call_id = message.get("tool_call_id")
+            function_name = "tool_call_result"  # Default fallback
+            if tool_call_id and tool_call_id in params.tool_call_id_to_name_mapping:
+                function_name = params.tool_call_id_to_name_mapping[tool_call_id]
+
             parts.append(
-                Part(
-                    function_response=FunctionResponse(
-                        name="tool_call_result",  # seems to work to hard-code the same name every time
-                        response=response_dict,
-                    )
+                Part.from_function_response(
+                    name=function_name,
+                    response=response_dict,
                 )
             )
         elif isinstance(content, str):
@@ -312,4 +359,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                     audio_bytes = base64.b64decode(input_audio["data"])
                     parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
 
-        return Content(role=role, parts=parts)
+        return self.MessageConversionResult(
+            content=Content(role=role, parts=parts),
+            tool_call_id_to_name_mapping=tool_call_id_to_name_mapping,
+        )

src/pipecat/services/google/llm.py

@@ -1034,6 +1034,23 @@ class GoogleLLMService(LLMService):
         if context:
             await self._process_context(context)
 
+    async def stop(self, frame):
+        """Override stop to gracefully close the client."""
+        await super().stop(frame)
+        await self._close_client()
+
+    async def cancel(self, frame):
+        """Override cancel to gracefully close the client."""
+        await super().cancel(frame)
+        await self._close_client()
+
+    async def _close_client(self):
+        try:
+            await self._client.aio.aclose()
+        except Exception:
+            # Do nothing - we're shutting down anyway
+            pass
+
     def create_context_aggregator(
         self,
         context: OpenAILLMContext,