From 8812686b1753c9573275d178bcd23f290044d178 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 12 Jan 2026 16:01:48 -0500 Subject: [PATCH 1/4] Fix parallel function calling with Gemini 3. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gemini expects parallel function calls to be passed in as a single multi-part `Content` block. This is important because only one of the function calls in a batch of parallel function calls gets a thought signature—if they're passed in as separate `Content` blocks, there'd be one or more missing thought signatures, which would result in a Gemini error. --- .../adapters/services/gemini_adapter.py | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index e17d9a508..b00543e95 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -255,6 +255,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): # Apply thought signatures to the corresponding messages self._apply_thought_signatures_to_messages(thought_signature_dicts, messages) + # Merge consecutive tool calls and tool responses into single multi-part messages + messages = self._merge_consecutive_tool_messages(messages) + # Check if we only have function-related messages (no regular text) has_regular_messages = any( len(msg.parts) == 1 @@ -433,6 +436,80 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): tool_call_id_to_name_mapping=tool_call_id_to_name_mapping, ) + def _merge_consecutive_tool_messages(self, messages: List[Content]) -> List[Content]: + """Merge consecutive tool call messages within tool exchange blocks. + + Gemini (and Gemini 3 in particular, where thought signatures are + involved) expects multiple parallel tool calls to be in a single Content + with multiple function_call parts. + + This method detects "tool exchange blocks" (sequences of tool calls and + responses, including alternating patterns like call1, response1, call2, + response2) and merges all tool calls within each block into a single + Content, followed by the individual tool responses. + + Args: + messages: List of Content messages to process. + + Returns: + List of Content messages with tool calls merged within each block. + """ + if not messages: + return messages + + def is_tool_call_message(msg: Content) -> bool: + """Check if message contains only function_call parts.""" + return ( + msg.role == "model" + and msg.parts + and all(getattr(part, "function_call", None) for part in msg.parts) + ) + + def is_tool_response_message(msg: Content) -> bool: + """Check if message contains only function_response parts.""" + return ( + msg.role == "user" + and msg.parts + and all(getattr(part, "function_response", None) for part in msg.parts) + ) + + def is_tool_message(msg: Content) -> bool: + """Check if message is either a tool call or tool response.""" + return is_tool_call_message(msg) or is_tool_response_message(msg) + + merged_messages = [] + i = 0 + + while i < len(messages): + current = messages[i] + + # Check for a tool exchange block (sequence of tool calls and/or responses) + if is_tool_message(current): + tool_call_parts = [] + tool_response_messages = [] + + # Collect all consecutive tool messages (calls and responses) + j = i + while j < len(messages) and is_tool_message(messages[j]): + msg = messages[j] + if is_tool_call_message(msg): + tool_call_parts.extend(msg.parts) + else: # is_tool_response_message + tool_response_messages.append(msg) + j += 1 + + # Output merged tool calls first, then individual tool responses + if tool_call_parts: + merged_messages.append(Content(role="model", parts=tool_call_parts)) + merged_messages.extend(tool_response_messages) + + i = j + else: + merged_messages.append(current) + i += 1 + + return merged_messages + def _apply_thought_signatures_to_messages( self, thought_signature_dicts: List[dict], messages: List[Content] ) -> None: From 6668712f7bea3d52696a57a5f0dcf26e5b628f9b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 12 Jan 2026 17:00:13 -0500 Subject: [PATCH 2/4] Add evals for parallel function calling --- scripts/evals/run-release-evals.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 65902f41a..1cb10146f 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -30,10 +30,15 @@ EVAL_SIMPLE_MATH = EvalConfig( ) EVAL_WEATHER = EvalConfig( - prompt="What's the weather in San Francisco? Temperature should be in fahrenheits.", + prompt="What's the weather in San Francisco? Temperature should be in Fahrenheit.", eval="The user talks about the weather in San Francisco, including the degrees.", ) +EVAL_WEATHER_AND_RESTAURANT = EvalConfig( + prompt="What's the weather in San Francisco, and what's a good restaurant there? Temperature should be in Fahrenheit.", + eval="The user talks about the weather in San Francisco, including the degrees, and provides a restaurant recommendation.", +) + EVAL_ONLINE_SEARCH = EvalConfig( prompt="What's the current date in UTC?", eval=f"Current date in UTC is {datetime.now(timezone.utc).strftime('%A, %B %d, %Y')}.", @@ -145,10 +150,16 @@ TESTS_12 = [ ("12d-describe-image-moondream.py", EVAL_VISION_IMAGE()), ] +# For a few major services, we also test parallel function calling. +# (We don't bother doing this with every single service, as it's expensive and +# most rely on the same OpenAI-compatible implementation.) TESTS_14 = [ ("14-function-calling.py", EVAL_WEATHER), + ("14-function-calling.py", EVAL_WEATHER_AND_RESTAURANT), ("14a-function-calling-anthropic.py", EVAL_WEATHER), + ("14a-function-calling-anthropic.py", EVAL_WEATHER_AND_RESTAURANT), ("14e-function-calling-google.py", EVAL_WEATHER), + ("14e-function-calling-google.py", EVAL_WEATHER_AND_RESTAURANT), ("14f-function-calling-groq.py", EVAL_WEATHER), ("14g-function-calling-grok.py", EVAL_WEATHER), ("14h-function-calling-azure.py", EVAL_WEATHER), @@ -160,6 +171,7 @@ TESTS_14 = [ ("14p-function-calling-gemini-vertex-ai.py", EVAL_WEATHER), ("14q-function-calling-qwen.py", EVAL_WEATHER), ("14r-function-calling-aws.py", EVAL_WEATHER), + ("14r-function-calling-aws.py", EVAL_WEATHER_AND_RESTAURANT), ("14v-function-calling-openai.py", EVAL_WEATHER), ("14w-function-calling-mistral.py", EVAL_WEATHER), ("14x-function-calling-openpipe.py", EVAL_WEATHER), From 30fbcfbf717cf472e763974d0429e598f623297c Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 13 Jan 2026 16:33:59 -0500 Subject: [PATCH 3/4] Rework fix for parallel function calling with Gemini 3 --- .../adapters/services/gemini_adapter.py | 91 ++++++++++--------- 1 file changed, 50 insertions(+), 41 deletions(-) diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index b00543e95..cb6f6d8de 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -255,8 +255,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): # Apply thought signatures to the corresponding messages self._apply_thought_signatures_to_messages(thought_signature_dicts, messages) - # Merge consecutive tool calls and tool responses into single multi-part messages - messages = self._merge_consecutive_tool_messages(messages) + # When thinking is enabled, merge parallel tool calls into single messages + messages = self._merge_parallel_tool_calls_for_thinking(messages) # Check if we only have function-related messages (no regular text) has_regular_messages = any( @@ -436,23 +436,35 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): tool_call_id_to_name_mapping=tool_call_id_to_name_mapping, ) - def _merge_consecutive_tool_messages(self, messages: List[Content]) -> List[Content]: - """Merge consecutive tool call messages within tool exchange blocks. + def _merge_parallel_tool_calls_for_thinking(self, messages: List[Content]) -> List[Content]: + """Merge parallel tool calls into single Content objects when thinking is enabled. - Gemini (and Gemini 3 in particular, where thought signatures are - involved) expects multiple parallel tool calls to be in a single Content - with multiple function_call parts. + Gemini expects parallel tool calls (multiple function calls made + simultaneously) to be in a single Content with multiple function_call + Parts. This method takes a list of Content messages, where parallel + tool calls may be split across multiple messages, and merges them into + single messages. - This method detects "tool exchange blocks" (sequences of tool calls and - responses, including alternating patterns like call1, response1, call2, - response2) and merges all tool calls within each block into a single - Content, followed by the individual tool responses. + This only has an effect when thought_signatures are present (i.e., when + thinking is enabled). When thinking is disabled, merging doesn't matter. + When thinking is enabled, there is a guarantee that the first tool call + (and only the first) in any batch of parallel tool calls will have a + thought_signature. This allows us to distinguish: + + - Parallel tool calls: share a single thought_signature (on the first call) + - Sequential tool calls: each have their own thought_signature + + Algorithm: A tool call message with a thought_signature starts a new + parallel group. Any tool call messages after it without a + thought_signature get merged into that group, regardless of what + messages appear in between. Args: messages: List of Content messages to process. Returns: - List of Content messages with tool calls merged within each block. + List of Content messages with parallel tool calls merged when + thought_signatures are present, otherwise unchanged. """ if not messages: return messages @@ -465,17 +477,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): and all(getattr(part, "function_call", None) for part in msg.parts) ) - def is_tool_response_message(msg: Content) -> bool: - """Check if message contains only function_response parts.""" - return ( - msg.role == "user" - and msg.parts - and all(getattr(part, "function_response", None) for part in msg.parts) - ) - - def is_tool_message(msg: Content) -> bool: - """Check if message is either a tool call or tool response.""" - return is_tool_call_message(msg) or is_tool_response_message(msg) + def message_has_thought_signature(msg: Content) -> bool: + """Check if any part in the message has a thought_signature.""" + return any(getattr(part, "thought_signature", None) for part in msg.parts) merged_messages = [] i = 0 @@ -483,26 +487,31 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): while i < len(messages): current = messages[i] - # Check for a tool exchange block (sequence of tool calls and/or responses) - if is_tool_message(current): - tool_call_parts = [] - tool_response_messages = [] + # If this is a tool call message with a thought signature, start merging + if is_tool_call_message(current) and message_has_thought_signature(current): + merged_parts = list(current.parts) + other_messages = [] + j = i + 1 - # Collect all consecutive tool messages (calls and responses) - j = i - while j < len(messages) and is_tool_message(messages[j]): - msg = messages[j] - if is_tool_call_message(msg): - tool_call_parts.extend(msg.parts) - else: # is_tool_response_message - tool_response_messages.append(msg) - j += 1 - - # Output merged tool calls first, then individual tool responses - if tool_call_parts: - merged_messages.append(Content(role="model", parts=tool_call_parts)) - merged_messages.extend(tool_response_messages) + # Scan forward, merging tool calls without signatures, collecting others + while j < len(messages): + next_msg = messages[j] + if is_tool_call_message(next_msg): + if message_has_thought_signature(next_msg): + # New parallel group starts, stop here + break + else: + # Merge this call into the current group + merged_parts.extend(next_msg.parts) + j += 1 + else: + # Collect non-tool-call message, keep scanning + other_messages.append(next_msg) + j += 1 + # Output merged calls, then collected other messages + merged_messages.append(Content(role="model", parts=merged_parts)) + merged_messages.extend(other_messages) i = j else: merged_messages.append(current) From be49a54856a0f6180715efaec6b45eacf5c07bfa Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 13 Jan 2026 17:32:20 -0500 Subject: [PATCH 4/4] Fast-exit in the fix for parallel function calling with Gemini 3, if we can determine up-front that there's no work to do --- .../adapters/services/gemini_adapter.py | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index cb6f6d8de..2de3742c8 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -256,7 +256,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): self._apply_thought_signatures_to_messages(thought_signature_dicts, messages) # When thinking is enabled, merge parallel tool calls into single messages - messages = self._merge_parallel_tool_calls_for_thinking(messages) + messages = self._merge_parallel_tool_calls_for_thinking(thought_signature_dicts, messages) # Check if we only have function-related messages (no regular text) has_regular_messages = any( @@ -436,7 +436,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): tool_call_id_to_name_mapping=tool_call_id_to_name_mapping, ) - def _merge_parallel_tool_calls_for_thinking(self, messages: List[Content]) -> List[Content]: + def _merge_parallel_tool_calls_for_thinking( + self, thought_signature_dicts: List[dict], messages: List[Content] + ) -> List[Content]: """Merge parallel tool calls into single Content objects when thinking is enabled. Gemini expects parallel tool calls (multiple function calls made @@ -460,6 +462,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): messages appear in between. Args: + thought_signature_dicts: A list of thought signature dicts, used + to determine if the work of merging is necessary. messages: List of Content messages to process. Returns: @@ -469,6 +473,16 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): if not messages: return messages + # Fast-exit if no function-call-related thought signatures + # This is a shortcut for determining both: + # - whether thinking is enabled, and + # - whether there are function calls in the messages + has_function_call_signatures = any( + ts.get("bookmark", {}).get("function_call") for ts in thought_signature_dicts + ) + if not has_function_call_signatures: + return messages + def is_tool_call_message(msg: Content) -> bool: """Check if message contains only function_call parts.""" return (