wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update Gemini service to include function name when sending function responses in context

Browse Source
This commit is contained in:
Paul Kompfner
2025-10-24 10:38:17 -04:00
parent 99d94fc625
commit 4ad15f9a01
3 changed files with 81 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/adapters/services/anthropic_adapter.py
View File

@@ -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:

2
src/pipecat/adapters/services/bedrock_adapter.py
View File

@@ -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:

106
src/pipecat/adapters/services/gemini_adapter.py
View File

@@ -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 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::
@@ -248,26 +280,36 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
"""
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"]),
)
)
@@ -284,9 +326,16 @@ 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.from_function_response(
name="tool_call_result", # seems to work to hard-code the same name every time
name=function_name,
response=response_dict,
)
)
@@ -310,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,
)