From 64471d65f8222fdf19f90c450d62b8d7633deb01 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 12 Dec 2025 10:50:16 -0500 Subject: [PATCH 1/4] Clean up logic related to applying Gemini thought signatures to context messages --- changelog/3175.added.md | 24 +- examples/foundational/49b-thinking-google.py | 5 +- .../49d-thinking-functions-google.py | 4 +- .../adapters/services/gemini_adapter.py | 240 ++++++++---------- src/pipecat/frames/frames.py | 8 - .../aggregators/llm_response_universal.py | 4 - src/pipecat/services/google/llm.py | 97 +++---- src/pipecat/services/llm_service.py | 6 - 8 files changed, 170 insertions(+), 218 deletions(-) diff --git a/changelog/3175.added.md b/changelog/3175.added.md index 16c946068..60557f84e 100644 --- a/changelog/3175.added.md +++ b/changelog/3175.added.md @@ -10,31 +10,19 @@ - `LLMThoughtStartFrame` - `LLMThoughtTextFrame` - `LLMThoughtEndFrame` - 3. A mechanism for appending arbitrary context messages after a function call - message, used specifically to support Google's function-call-related - "thought signatures", which are necessary to ensure thinking continuity - between function calls in a chain (where the model thinks, makes a function - call, thinks some more, etc.). See: - - `append_extra_context_messages` field in `FunctionInProgressFrame` and - helper types - - `GoogleLLMService` leveraging the new mechanism to add a Google-specific - `"fn_thought_signature"` message - - `LLMAssistantAggregator` handling of `append_extra_context_messages` - - `GeminiLLMAdapter` handling of `"fn_thought_signature"` messages - 4. A generic mechanism for recording LLM thoughts to context, used + 3. A generic mechanism for recording LLM thoughts to context, used specifically to support Anthropic, whose thought signatures are expected to appear alongside the text of the thoughts within assistant context messages. See: - `LLMThoughtEndFrame.signature` - `LLMAssistantAggregator` handling of the above field - `AnthropicLLMAdapter` handling of `"thought"` context messages - 5. Google-specific logic for inserting non-function-call-related thought - signatures into the context, to help maintain thinking continuity in a - chain of LLM calls. See: + 4. Google-specific logic for inserting thought signatures into the context, + to help maintain thinking continuity in a chain of LLM calls. See: - `GoogleLLMService` sending `LLMMessagesAppendFrame`s to add LLM-specific - `"non_fn_thought_signature"` messages to context - - `GeminiLLMAdapter` handling of `"non_fn_thought_signature"` messages - 6. An expansion of `TranscriptProcessor` to process LLM thoughts in addition + `"thought_signature"` messages to context + - `GeminiLLMAdapter` handling of `"thought_signature"` messages + 5. An expansion of `TranscriptProcessor` to process LLM thoughts in addition to user and assistant utterances. See: - `TranscriptProcessor(process_thoughts=True)` (defaults to `False`) - `ThoughtTranscriptionMessage`, which is now also emitted with the diff --git a/examples/foundational/49b-thinking-google.py b/examples/foundational/49b-thinking-google.py index 947ab39c9..ca0d6d34d 100644 --- a/examples/foundational/49b-thinking-google.py +++ b/examples/foundational/49b-thinking-google.py @@ -123,8 +123,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): "content": "Say hello briefly.", } ) - # Here are some example prompts conducive to demonstrating - # thinking (picked from Google and Anthropic docs). + # Replace the above with one of these example prompts to demonstrate + # thinking. + # These examples come from Gemini and Anthropic docs. # messages.append( # { # "role": "user", diff --git a/examples/foundational/49d-thinking-functions-google.py b/examples/foundational/49d-thinking-functions-google.py index cdf4621b1..e75e70baa 100644 --- a/examples/foundational/49d-thinking-functions-google.py +++ b/examples/foundational/49d-thinking-functions-google.py @@ -149,8 +149,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): "content": "Say hello briefly.", } ) - # Here is an example prompt conducive to demonstrating thinking and - # function calling. + # Replace the above with one of these example prompts to demonstrate + # thinking and function calling. # This example comes from Gemini docs. # messages.append( # { diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 5a7387aca..b504967de 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -209,7 +209,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): system_instruction = None messages = [] tool_call_id_to_name_mapping = {} - non_fn_thought_signatures = [] + thought_signature_dicts = [] # Process each message, converting to Google format as needed for message in universal_context_messages: @@ -218,29 +218,11 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): # special way, or a message already in Google format that we can # use directly if isinstance(message, LLMSpecificMessage): - # Special handling for function-call-related thought signature - # messages if ( isinstance(message.message, dict) - and message.message.get("type") == "fn_thought_signature" - and (thought_signature := message.message.get("signature")) + and message.message.get("type") == "thought_signature" ): - self._apply_function_thought_signature_to_messages( - thought_signature, message.message.get("tool_call_id"), messages - ) - continue - - # Special handling for non-function-call-related thought- - # signature-containing messages - if ( - isinstance(message.message, dict) - and message.message.get("type") == "non_fn_thought_signature" - and (thought_signature := message.message.get("signature")) - and (bookmark := message.message.get("bookmark")) - ): - non_fn_thought_signatures.append( - {"signature": thought_signature, "bookmark": bookmark} - ) + thought_signature_dicts.append(message.message) continue # Fall back to assuming that the message is already in Google @@ -268,9 +250,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): if result.tool_call_id_to_name_mapping: tool_call_id_to_name_mapping.update(result.tool_call_id_to_name_mapping) - # Apply non-function-call-related thought signatures to the appropriate - # messages - self._apply_non_function_thought_signatures_to_messages(non_fn_thought_signatures, messages) + # Apply thought signatures to the corresponding messages + self._apply_thought_signatures_to_messages(thought_signature_dicts, messages) # Check if we only have function-related messages (no regular text) has_regular_messages = any( @@ -447,136 +428,135 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): tool_call_id_to_name_mapping=tool_call_id_to_name_mapping, ) - def _apply_function_thought_signature_to_messages( - self, thought_signature: bytes, tool_call_id: str, messages: List[Content] + def _apply_thought_signatures_to_messages( + self, thought_signature_dicts: List[dict], messages: List[Content] ) -> None: - """Apply a function-related thought signature to the corresponding function call message. + """Apply thought signatures to corresponding assistant messages. + + See GoogleLLMService for more details about thought signatures. Args: - thought_signature: The thought signature bytes to apply. - tool_call_id: ID of the tool call message to find and modify. - messages: List of messages to search through. - """ - # Search backwards through messages to find the matching function call - for message in reversed(messages): - if not isinstance(message, Content) or not message.parts: - continue - # Find the specific part with the matching function call - for part in message.parts: - if ( - hasattr(part, "function_call") - and part.function_call - and part.function_call.id == tool_call_id - ): - part.thought_signature = thought_signature - break - else: - # Continue outer loop if inner loop didn't break - continue - # Break outer loop if inner loop broke (found match) - break - - def _apply_non_function_thought_signatures_to_messages( - self, thought_signatures: List[dict], messages: List[Content] - ) -> None: - """Apply (optional, but recommended) non-function-call-related thought signatures to the last part of corresponding non-function-call assistant messages. - - Gemini 3 Pro (and, somewhat surprisingly, other models, too, when - functions are involved in the conversation) outputs thought signatures - at the end of assistant responses. - - Args: - thought_signatures: A list of dicts containing: + thought_signature_dicts: A list of dicts containing: - "signature": a thought signature - "bookmark": a bookmark to identify the message part to apply the signature to. - The bookmark may contain either: - - "text" - - "inline_data" - messages: List of messages to search through. + The bookmark may contain one of: + - "function_call" (a function call ID string) + - "text" (a text string) + - "inline_data" (a Blob) + The list of thought signature dicts is in order. + messages: List of messages to apply the thought signatures to. """ - if not thought_signatures: + if not thought_signature_dicts: return # For debugging, print out thought signatures and their bookmarks - logger.trace(f"Thought signatures to apply: {len(thought_signatures)}") - for ts in thought_signatures: + logger.debug(f"Thought signatures to apply: {len(thought_signature_dicts)}") + for ts in thought_signature_dicts: bookmark = ts.get("bookmark") - if bookmark.get("text"): + if bookmark.get("function_call"): + logger.trace(f" - To function call: {bookmark['function_call']}") + elif bookmark.get("text"): text = bookmark["text"] log_display_text = f"{text[:50]}..." if len(text) > 50 else text - logger.trace(f" - At text: {log_display_text}") + logger.trace(f" - To text: {log_display_text}") elif bookmark.get("inline_data"): - logger.trace(f" - At inline data") + logger.trace(f" - To inline data") - # Find all assistant (model) messages that aren't function calls - non_fn_assistant_messages = [] - for message in messages: - if not isinstance(message, Content) or not message.parts: - continue - # Check if this is a model message without function calls - if message.role == "model": - has_function_call = any( - hasattr(part, "function_call") and part.function_call for part in message.parts - ) - if not has_function_call: - non_fn_assistant_messages.append(message) + # Get all assistant messages + assistant_messages = [ + message + for message in messages + if isinstance(message, Content) and message.role == "model" + ] - # Apply thought signatures to the corresponding assistant messages - # Match them using content heuristics, maintaining order (messages without signatures are skipped) - message_start_index = 0 # Track where to start searching for the next match - for thought_signature_dict in thought_signatures: + # Apply thought signatures to the corresponding assistant messages. + # Thought signatures are already in message order. + thought_signatures_applied = 0 + message_start_index = 0 # Track where to start searching for the next matching message. + for thought_signature_dict in thought_signature_dicts: signature = thought_signature_dict.get("signature") bookmark = thought_signature_dict.get("bookmark") - if not signature: + if not signature or not bookmark: continue - # Search through remaining non-function assistant messages for a match - for i in range(message_start_index, len(non_fn_assistant_messages)): - message = non_fn_assistant_messages[i] + # Search through remaining assistant messages for a match + for i in range(message_start_index, len(assistant_messages)): + message = assistant_messages[i] if not message.parts: continue + # We're assuming that the thought signature always applies to the last part last_part = message.parts[-1] - matched = False - # If it's a text bookmark, check that the last message part text has the same text or - # - is a prefix of that text (in case spoken text was truncated due to interruption) - # - is prefixed by that text (in case bookmark represents just first chunk of multi-chunk text) - if bookmark_text := bookmark.get("text"): - if hasattr(last_part, "text") and last_part.text: - # Normalize whitespace for comparison - signed_text = " ".join(bookmark_text.split()) - last_text = " ".join(last_part.text.split()) - if ( - last_text == signed_text - or signed_text.startswith(last_text) - or last_text.startswith(signed_text) - ): - log_display_text = ( - f"{last_part.text[:50]}..." - if len(last_part.text) > 50 - else last_part.text - ) - logger.trace( - f"Applying thought signature to part with matching text: {log_display_text}" - ) - last_part.thought_signature = signature - matched = True + # If the bookmark matches the part... + if self._thought_signature_bookmark_matches_part(bookmark, last_part): + # Apply the thought signature + last_part.thought_signature = signature + thought_signatures_applied += 1 - # Check if signed part has inline_data and last message part has matching inline_data - elif inline_data := bookmark.get("inline_data"): - if ( - hasattr(last_part, "inline_data") - and last_part.inline_data - and last_part.inline_data.data == inline_data.data - ): - logger.trace( - f"Applying thought signature to part with matching inline_data" - ) - last_part.thought_signature = signature - matched = True - - # If we found a match, update start index and stop searching for this signed part - if matched: + # Update the start index and stop searching for a match message_start_index = i + 1 break + + # For debugging, print out how many thought signatures were applied + logger.debug(f"Applied {thought_signatures_applied} thought signatures.") + + def _thought_signature_bookmark_matches_part(self, bookmark: dict, part: Part) -> bool: + if function_call_bookmark := bookmark.get("function_call"): + return self._thought_signature_function_call_bookmark_matches_part( + function_call_bookmark, part + ) + elif text_bookmark := bookmark.get("text"): + return self._thought_signature_text_bookmark_matches_part(text_bookmark, part) + elif inline_data := bookmark.get("inline_data"): + return self._thought_signature_inline_data_bookmark_matches_part(inline_data, part) + else: + logger.warning(f"Unknown thought signature bookmark type: {bookmark}") + + return False + + def _thought_signature_function_call_bookmark_matches_part( + self, bookmark_function_call_id: str, part: Part + ) -> bool: + if ( + hasattr(part, "function_call") + and part.function_call + and part.function_call.id == bookmark_function_call_id + ): + logger.trace(f"Thought signature function call match: {bookmark_function_call_id}") + return True + + return False + + def _thought_signature_text_bookmark_matches_part(self, bookmark_text: str, part: Part) -> bool: + if hasattr(part, "text") and part.text: + # Normalize whitespace for comparison + bookmark_text = " ".join(bookmark_text.split()) + part_text = " ".join(part.text.split()) + # Check that either: + # - the part text is the same as the bookmark text + # - a prefix of the bookmark text (in case the part text was truncated due to interruption) + # - the bookmark text is a prefix of the part text (in case the bookmark represents just first chunk of multi-chunk text) + if ( + part_text == bookmark_text + or bookmark_text.startswith(part_text) + or part_text.startswith(bookmark_text) + ): + log_display_text = f"{part.text[:50]}..." if len(part.text) > 50 else part.text + logger.trace(f"Thought signature text match: {log_display_text}") + return True + + return False + + def _thought_signature_inline_data_bookmark_matches_part( + self, bookmark_inline_data: Blob, part: Part + ) -> bool: + if ( + hasattr(part, "inline_data") + and part.inline_data + and part.inline_data.data == bookmark_inline_data.data + ): + logger.trace(f"Thought signature inline data match") + return True + + return False diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 542f21fc0..56a2b18c8 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1197,16 +1197,12 @@ class FunctionCallFromLLM: tool_call_id: A unique identifier for the function call. arguments: The arguments to pass to the function. context: The LLM context when the function call was made. - append_extra_context_messages: Optional extra messages to append to the - context after the function call message. Used to add Google - function-call-related thought signatures to the context. """ function_name: str tool_call_id: str arguments: Mapping[str, Any] context: Any - append_extra_context_messages: Optional[List["LLMContextMessage"]] = None @dataclass @@ -1745,16 +1741,12 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame): tool_call_id: Unique identifier for this function call. arguments: Arguments passed to the function. cancel_on_interruption: Whether to cancel this call if interrupted. - append_extra_context_messages: Optional extra messages to append to the - context after the function call message. Used to add Google - function-call-related thought signatures to the context. """ function_name: str tool_call_id: str arguments: Any cancel_on_interruption: bool = False - append_extra_context_messages: Optional[List["LLMContextMessage"]] = None @dataclass diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index debfefb07..686285443 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -740,10 +740,6 @@ class LLMAssistantAggregator(LLMContextAggregator): } ) - # Append to context any specified extra context messages - if frame.append_extra_context_messages: - self._context.add_messages(frame.append_extra_context_messages) - self._function_calls_in_progress[frame.tool_call_id] = frame async def _handle_function_call_result(self, frame: FunctionCallResultFrame): diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 4fd291868..7b3c2e831 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -932,7 +932,7 @@ class GoogleLLMService(LLMService): reasoning_tokens = 0 grounding_metadata = None - search_result = "" + accumulated_text = "" try: # Generate content using either OpenAILLMContext or universal LLMContext @@ -943,7 +943,6 @@ class GoogleLLMService(LLMService): ) function_calls = [] - previous_part = None async for chunk in response: # Stop TTFB metrics after the first chunk await self.stop_ttfb_metrics() @@ -966,6 +965,7 @@ class GoogleLLMService(LLMService): for candidate in chunk.candidates: if candidate.content and candidate.content.parts: for part in candidate.content.parts: + function_call_id = None if part.text: if part.thought: # Gemini emits fully-formed thoughts rather @@ -975,29 +975,20 @@ class GoogleLLMService(LLMService): await self.push_frame(LLMThoughtTextFrame(part.text)) await self.push_frame(LLMThoughtEndFrame()) else: - search_result += part.text + accumulated_text += part.text await self.push_frame(LLMTextFrame(part.text)) elif part.function_call: function_call = part.function_call - id = function_call.id or str(uuid.uuid4()) - logger.debug(f"Function call: {function_call.name}:{id}") + function_call_id = function_call.id or str(uuid.uuid4()) + logger.debug( + f"Function call: {function_call.name}:{function_call_id}" + ) function_calls.append( FunctionCallFromLLM( context=context, - tool_call_id=id, + tool_call_id=function_call_id, function_name=function_call.name, arguments=function_call.args or {}, - append_extra_context_messages=[ - self.get_llm_adapter().create_llm_specific_message( - { - "type": "fn_thought_signature", - "signature": part.thought_signature, - "tool_call_id": id, - } - ) - ] - if part.thought_signature - else None, ) ) elif part.inline_data and part.inline_data.data: @@ -1007,14 +998,14 @@ class GoogleLLMService(LLMService): ) await self.push_frame(frame) - # With Gemini 3 Pro (and, contrary to Google's - # docs, other models models, too, especially when - # functions are involved in the conversation), - # thought signatures can be associated with any - # kind of Part, not just function calls. + # Handle Gemini thought signatures. # - # They should always be included in the last - # response Part. (*) + # - Gemini 2.5: they appear on function_call Parts, + # and then (surprisingly) on the last(*) Part of + # model responses following the first function_call + # in a conversation. + # - Gemini 3 Pro: they appear on the last(*) Part + # of model responses, regardless of Part type. # # (*) Since we're using the streaming API, though, # where text Parts may be split across multiple @@ -1022,34 +1013,44 @@ class GoogleLLMService(LLMService): # signatures may actually appear with the first # chunk (Gemini 2.5) or in a trailing empty-text # chunk (Gemini 3 Pro). - if part.thought_signature and not part.function_call: + if part.thought_signature: # Save a "bookmark" for the signature, so we - # can later stick it in the right place in - # context when sending it back to the LLM to - # continue the conversation. + # can later be sure we've put it in the right + # place in context when sending the context + # back to the LLM to continue the conversation. bookmark = {} - if part.inline_data and part.inline_data.data: - bookmark["inline_data"] = {"inline_data": part.inline_data} + if part.function_call: + bookmark["function_call"] = function_call_id + elif part.inline_data and part.inline_data.data: + # NOTE: missing feature: we don't store + # inline_data messages (like generated + # images) in context today, so this thought + # signature is not fully supported yet. + # (A conversation with + # "gemini-3-pro-image-preview" doesn't work + # today due to the missing context.) + bookmark["inline_data"] = part.inline_data elif part.text is not None: # Account for Gemini 3 Pro trailing - # empty-text chunk by using search_result, - # which accumulates all text so far. - bookmark["text"] = search_result - await self.push_frame( - LLMMessagesAppendFrame( - [ - self.get_llm_adapter().create_llm_specific_message( - { - "type": "non_fn_thought_signature", - "signature": part.thought_signature, - "bookmark": bookmark, - } - ) - ] + # empty-text chunk by using all the text + # seen so far in this response's chunks. + bookmark["text"] = accumulated_text + else: + logger.warning("Thought signature found on unhandled Part type") + if bookmark: + await self.push_frame( + LLMMessagesAppendFrame( + [ + self.get_llm_adapter().create_llm_specific_message( + { + "type": "thought_signature", + "signature": part.thought_signature, + "bookmark": bookmark, + } + ) + ] + ) ) - ) - - previous_part = part if ( candidate.grounding_metadata @@ -1098,7 +1099,7 @@ class GoogleLLMService(LLMService): finally: if grounding_metadata and isinstance(grounding_metadata, dict): llm_search_frame = LLMSearchResponseFrame( - search_result=search_result, + search_result=accumulated_text, origins=grounding_metadata["origins"], rendered_content=grounding_metadata["rendered_content"], ) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 91358dcf3..ea0da01f6 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -132,9 +132,6 @@ class FunctionCallRunnerItem: tool_call_id: A unique identifier for the function call. arguments: The arguments for the function. context: The LLM context. - append_extra_context_messages: Optional extra messages to append to the - context after the function call message. Used to add Google - function-call-related thought signatures to the context. run_llm: Optional flag to control LLM execution after function call. """ @@ -143,7 +140,6 @@ class FunctionCallRunnerItem: tool_call_id: str arguments: Mapping[str, Any] context: OpenAILLMContext | LLMContext - append_extra_context_messages: Optional[List[LLMContextMessage]] = None run_llm: Optional[bool] = None @@ -465,7 +461,6 @@ class LLMService(AIService): tool_call_id=function_call.tool_call_id, arguments=function_call.arguments, context=function_call.context, - append_extra_context_messages=function_call.append_extra_context_messages, ) ) @@ -590,7 +585,6 @@ class LLMService(AIService): function_name=runner_item.function_name, tool_call_id=runner_item.tool_call_id, arguments=runner_item.arguments, - append_extra_context_messages=runner_item.append_extra_context_messages, cancel_on_interruption=item.cancel_on_interruption, ) From abc2ad8cbc422d50eabca60893d24367bf8a13a8 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 12 Dec 2025 13:01:45 -0500 Subject: [PATCH 2/4] Avoid printing out entire thought signatures in logs --- src/pipecat/adapters/services/anthropic_adapter.py | 2 ++ src/pipecat/adapters/services/gemini_adapter.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index 4cecaf810..2429957e4 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -94,6 +94,8 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): for item in msg["content"]: if item["type"] == "image": item["source"]["data"] = "..." + if item["type"] == "thinking" and item.get("signature"): + item["signature"] = "..." messages_for_logging.append(msg) return messages_for_logging diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index b504967de..ba9dffd87 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -151,6 +151,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): for part in obj["parts"]: if "inline_data" in part: part["inline_data"]["data"] = "..." + if "thought_signature" in part: + part["thought_signature"] = "..." except Exception as e: logger.debug(f"Error: {e}") messages_for_logging.append(obj) From e604e9b490e5caa22bd13bc3186570e7af34cd59 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Fri, 12 Dec 2025 14:38:07 -0500 Subject: [PATCH 3/4] Support conversations with Gemini 3 Pro Image (model "gemini-3-pro-image-preview"). Prior to this change, after the model generated an image the conversation would not be able to progress. It would stall out because we were never storing the image in context, so the model would never realize it already did the work of generating an image. We didn't run into issues with Gemini 2.5 Flash Image, because that model always followed up an image with a text message. --- changelog/3224.fixed.2.md | 3 ++ changelog/3224.fixed.md | 3 ++ .../07n-interruptible-gemini-image.py | 1 + .../adapters/services/gemini_adapter.py | 5 ++- src/pipecat/frames/frames.py | 14 ++++++++ .../processors/aggregators/llm_context.py | 26 ++++++++++---- .../aggregators/llm_response_universal.py | 21 +++++++++++ src/pipecat/services/google/llm.py | 36 +++++++++++++------ src/pipecat/transports/base_output.py | 7 +++- 9 files changed, 97 insertions(+), 19 deletions(-) create mode 100644 changelog/3224.fixed.2.md create mode 100644 changelog/3224.fixed.md diff --git a/changelog/3224.fixed.2.md b/changelog/3224.fixed.2.md new file mode 100644 index 000000000..ad54a9572 --- /dev/null +++ b/changelog/3224.fixed.2.md @@ -0,0 +1,3 @@ +- Better support conversation history with Gemini 2.5 Flash Image (model + "gemini-2.5-flash-image"). Prior to this fix, the model had no memory of + previous images it had generated, so it wouldn't be able to iterate on them. diff --git a/changelog/3224.fixed.md b/changelog/3224.fixed.md new file mode 100644 index 000000000..ddae072cf --- /dev/null +++ b/changelog/3224.fixed.md @@ -0,0 +1,3 @@ +- Support conversations with Gemini 3 Pro Image (model + "gemini-3-pro-image-preview"). Prior to this fix, after the model generated + an image the conversation would not be able to progress. diff --git a/examples/foundational/07n-interruptible-gemini-image.py b/examples/foundational/07n-interruptible-gemini-image.py index ef37d1fb6..62af2cf46 100644 --- a/examples/foundational/07n-interruptible-gemini-image.py +++ b/examples/foundational/07n-interruptible-gemini-image.py @@ -89,6 +89,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = GoogleLLMService( api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.5-flash-image", + # model="gemini-3-pro-image-preview", # A more powerful model, but slower ) messages = [ diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index ba9dffd87..92600a16e 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -556,7 +556,10 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): if ( hasattr(part, "inline_data") and part.inline_data - and part.inline_data.data == bookmark_inline_data.data + # Comparing length should be good enough for matching inline data, + # especially since we're already matching thought signatures in + # strict message order. Comparing actual data is expensive. + and len(part.inline_data.data) == len(bookmark_inline_data.data) ): logger.trace(f"Thought signature inline data match") return True diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 56a2b18c8..867015f9f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1466,6 +1466,20 @@ class UserImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, text: {self.text}, append_to_context: {self.append_to_context})" +@dataclass +class AssistantImageRawFrame(OutputImageRawFrame): + """Frame containing image generated by the assistant. + + An image generated by the assistant. Gets appended to the LLM context. + + Parameters: + original_jpeg: The already-JPEG-encoded image bytes, which may be + appended directly to the LLM context without further encoding. + """ + + original_jpeg: Optional[bytes] = None + + @dataclass class InputDTMFFrame(DTMFFrame, SystemFrame): """DTMF keypress input frame from transport.""" diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 99b9aeaa9..614b6789e 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -157,9 +157,15 @@ class LLMContext: """ def encode_image(): - buffer = io.BytesIO() - Image.frombytes(format, size, image).save(buffer, format="JPEG") - encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") + if format == "JPEG": + # Already JPEG-encoded + bytes = image + else: + # Encode to JPEG + buffer = io.BytesIO() + Image.frombytes(format, size, image).save(buffer, format="JPEG") + bytes = buffer.getvalue() + encoded_image = base64.b64encode(bytes).decode("utf-8") return encoded_image encoded_image = await asyncio.to_thread(encode_image) @@ -334,18 +340,26 @@ class LLMContext: self._tool_choice = tool_choice async def add_image_frame_message( - self, *, format: str, size: tuple[int, int], image: bytes, text: Optional[str] = None + self, + *, + format: str, + size: tuple[int, int], + image: bytes, + text: Optional[str] = None, + role: str = "user", ): """Add a message containing an image frame. Args: - format: Image format (e.g., 'RGB', 'RGBA'). + format: Image format (e.g., 'RGB', 'RGBA', or, if already + JPEG-encoded, "JPEG"). size: Image dimensions as (width, height) tuple. image: Raw image bytes. text: Optional text to include with the image. + role: The role of this message (defaults to "user"). """ message = await LLMContext.create_image_message( - format=format, size=size, image=image, text=text + role=role, format=format, size=size, image=image, text=text ) self.add_message(message) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 686285443..913209818 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -24,6 +24,7 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( + AssistantImageRawFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -663,6 +664,8 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._handle_function_call_cancel(frame) elif isinstance(frame, UserImageRawFrame): await self._handle_user_image_frame(frame) + elif isinstance(frame, AssistantImageRawFrame): + await self._handle_assistant_image_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): await self.push_aggregation() await self.push_frame(frame, direction) @@ -827,6 +830,24 @@ class LLMAssistantAggregator(LLMContextAggregator): await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) + async def _handle_assistant_image_frame(self, frame: AssistantImageRawFrame): + logger.debug(f"{self} Appending AssistantImageRawFrame to LLM context (size: {frame.size})") + + if frame.original_jpeg: + await self._context.add_image_frame_message( + format="JPEG", + size=frame.size, # Technically doesn't matter, since already encoded + image=frame.original_jpeg, + role="assistant", + ) + else: + await self._context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + role="assistant", + ) + async def _handle_llm_start(self, _: LLMFullResponseStartFrame): self._started += 1 diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 7b3c2e831..7734d456c 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -24,6 +24,7 @@ from pydantic import BaseModel, Field from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams from pipecat.frames.frames import ( + AssistantImageRawFrame, AudioRawFrame, Frame, FunctionCallCancelFrame, @@ -43,7 +44,7 @@ from pipecat.frames.frames import ( UserImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage -from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, @@ -992,11 +993,25 @@ class GoogleLLMService(LLMService): ) ) elif part.inline_data and part.inline_data.data: + # Here we assume that inline_data is an image. image = Image.open(io.BytesIO(part.inline_data.data)) - frame = OutputImageRawFrame( - image=image.tobytes(), size=image.size, format="RGB" + # NOTE: Gemini 3 Pro Image seems to always give + # JPEGs. It expects us to send back the + # original JPEG data in the context, along with + # the corresponding thought signature. JPEG + # happens to be the format our universal + # context uses for images, so we can just pass + # it through as-is. + await self.push_frame( + AssistantImageRawFrame( + image=image.tobytes(), + size=image.size, + format="RGB", + original_jpeg=part.inline_data.data + if part.inline_data.mime_type == "image/jpeg" + else None, + ) ) - await self.push_frame(frame) # Handle Gemini thought signatures. # @@ -1022,13 +1037,12 @@ class GoogleLLMService(LLMService): if part.function_call: bookmark["function_call"] = function_call_id elif part.inline_data and part.inline_data.data: - # NOTE: missing feature: we don't store - # inline_data messages (like generated - # images) in context today, so this thought - # signature is not fully supported yet. - # (A conversation with - # "gemini-3-pro-image-preview" doesn't work - # today due to the missing context.) + # With Gemini 3 Pro (where sending the + # thought signature is required for images) + # this is the JPEG-encoded image data that + # we sent to be written to the context + # as-is, so it is usable as a bookmark (it + # will match the context data). bookmark["inline_data"] = part.inline_data elif part.text is not None: # Account for Gemini 3 Pro trailing diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index ace1e2edf..b2815b2c6 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -23,6 +23,7 @@ from pipecat.audio.dtmf.utils import load_dtmf_audio from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer from pipecat.audio.utils import create_stream_resampler, is_silence from pipecat.frames.frames import ( + AssistantImageRawFrame, BotSpeakingFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, @@ -335,6 +336,10 @@ class BaseOutputTransport(FrameProcessor): await sender.handle_audio_frame(frame) elif isinstance(frame, (OutputImageRawFrame, SpriteFrame)): await sender.handle_image_frame(frame) + if isinstance(frame, AssistantImageRawFrame): + # This will push it further, to be handled by the assistant + # aggregator, say + await sender.handle_sync_frame(frame) elif isinstance(frame, MixerControlFrame): await sender.handle_mixer_control_frame(frame) elif frame.pts: @@ -753,7 +758,7 @@ class BaseOutputTransport(FrameProcessor): await self._handle_frame(frame) # If we are not able to write to the transport we shouldn't - # pushb downstream. + # push downstream. push_downstream = True # Try to send audio to the transport. From 54926f390daf72bce1d0ce2729d8a5a4e3029531 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 15 Dec 2025 10:34:33 -0500 Subject: [PATCH 4/4] Make image writing to and reading from `LLMContext` more robust; let's allow storing in context image types other than JPEG, meaning not lossily and unnecessarily re-encoding non-JPEG images as JPEG. --- .../adapters/services/anthropic_adapter.py | 7 +++-- .../adapters/services/bedrock_adapter.py | 13 ++++----- .../adapters/services/gemini_adapter.py | 7 +++-- src/pipecat/frames/frames.py | 15 ++++++----- .../processors/aggregators/llm_context.py | 14 +++++----- .../aggregators/llm_response_universal.py | 6 ++--- src/pipecat/services/google/llm.py | 27 +++++++------------ 7 files changed, 46 insertions(+), 43 deletions(-) diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index 2429957e4..5dd2e1276 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/src/pipecat/adapters/services/anthropic_adapter.py @@ -283,11 +283,14 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]): # handle image_url -> image conversion if item["type"] == "image_url": if item["image_url"]["url"].startswith("data:"): + # Extract MIME type from data URL (format: "data:image/jpeg;base64,...") + url = item["image_url"]["url"] + mime_type = url.split(":")[1].split(";")[0] item["type"] = "image" item["source"] = { "type": "base64", - "media_type": "image/jpeg", - "data": item["image_url"]["url"].split(",")[1], + "media_type": mime_type, + "data": url.split(",")[1], } del item["image_url"] elif item["image_url"]["url"].startswith("http"): diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index 213e3b01c..6b67a118b 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/src/pipecat/adapters/services/bedrock_adapter.py @@ -257,14 +257,15 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]): # handle image_url -> image conversion if item["type"] == "image_url": if item["image_url"]["url"].startswith("data:"): + # Extract format from data URL (format: "data:image/jpeg;base64,...") + url = item["image_url"]["url"] + mime_type = url.split(":")[1].split(";")[0] + # Bedrock expects format like "jpeg", "png" etc., not "image/jpeg" + image_format = mime_type.split("/")[1] new_item = { "image": { - "format": "jpeg", - "source": { - "bytes": base64.b64decode( - item["image_url"]["url"].split(",")[1] - ) - }, + "format": image_format, + "source": {"bytes": base64.b64decode(url.split(",")[1])}, } } new_content.append(new_item) diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 92600a16e..725713887 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -399,11 +399,14 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): if c["type"] == "text": parts.append(Part(text=c["text"])) elif c["type"] == "image_url" and c["image_url"]["url"].startswith("data:"): + # Extract MIME type from data URL (format: "data:image/jpeg;base64,...") + url = c["image_url"]["url"] + mime_type = url.split(":")[1].split(";")[0] parts.append( Part( inline_data=Blob( - mime_type="image/jpeg", - data=base64.b64decode(c["image_url"]["url"].split(",")[1]), + mime_type=mime_type, + data=base64.b64decode(url.split(",")[1]), ) ) ) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 867015f9f..9653f1611 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -227,7 +227,7 @@ class ImageRawFrame: Parameters: image: Raw image bytes. size: Image dimensions as (width, height) tuple. - format: Image format (e.g., 'JPEG', 'PNG'). + format: Image format (e.g., 'RGB', 'RGBA'). """ image: bytes @@ -1468,16 +1468,19 @@ class UserImageRawFrame(InputImageRawFrame): @dataclass class AssistantImageRawFrame(OutputImageRawFrame): - """Frame containing image generated by the assistant. + """Frame containing an image generated by the assistant. - An image generated by the assistant. Gets appended to the LLM context. + Contains both the raw frame for display (superclass functionality) as well + as the original image, which can get used directly in LLM contexts. Parameters: - original_jpeg: The already-JPEG-encoded image bytes, which may be - appended directly to the LLM context without further encoding. + original_data: The original image data, which can get used directly in + an LLM context message without further encoding. + original_mime_type: The MIME type of the original image data. """ - original_jpeg: Optional[bytes] = None + original_data: Optional[bytes] = None + original_mime_type: Optional[str] = None @dataclass diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 614b6789e..a6c1b6df1 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -150,15 +150,17 @@ class LLMContext: Args: role: The role of this message (defaults to "user"). - format: Image format (e.g., 'RGB', 'RGBA'). + format: Image format (e.g., 'RGB', 'RGBA', or, if already encoded, + the MIME type like 'image/jpeg'). size: Image dimensions as (width, height) tuple. image: Raw image bytes. text: Optional text to include with the image. """ + # Format is a mime type: image is already encoded + image_already_encoded = format.startswith("image/") def encode_image(): - if format == "JPEG": - # Already JPEG-encoded + if image_already_encoded: bytes = image else: # Encode to JPEG @@ -170,7 +172,7 @@ class LLMContext: encoded_image = await asyncio.to_thread(encode_image) - url = f"data:image/jpeg;base64,{encoded_image}" + url = f"data:{format if image_already_encoded else 'image/jpeg'};base64,{encoded_image}" return LLMContext.create_image_url_message(role=role, url=url, text=text) @@ -351,8 +353,8 @@ class LLMContext: """Add a message containing an image frame. Args: - format: Image format (e.g., 'RGB', 'RGBA', or, if already - JPEG-encoded, "JPEG"). + format: Image format (e.g., 'RGB', 'RGBA', or, if already encoded, + the MIME type like 'image/jpeg'). size: Image dimensions as (width, height) tuple. image: Raw image bytes. text: Optional text to include with the image. diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 913209818..810280729 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -833,11 +833,11 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_assistant_image_frame(self, frame: AssistantImageRawFrame): logger.debug(f"{self} Appending AssistantImageRawFrame to LLM context (size: {frame.size})") - if frame.original_jpeg: + if frame.original_data and frame.original_mime_type: await self._context.add_image_frame_message( - format="JPEG", + format=frame.original_mime_type, size=frame.size, # Technically doesn't matter, since already encoded - image=frame.original_jpeg, + image=frame.original_data, role="assistant", ) else: diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 7734d456c..66af5bd10 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -479,11 +479,16 @@ class GoogleLLMContext(OpenAILLMContext): if c["type"] == "text": parts.append(Part(text=c["text"])) elif c["type"] == "image_url": + # Extract MIME type from data URL (format: "data:image/jpeg;base64,...") + url = c["image_url"]["url"] + mime_type = ( + url.split(":")[1].split(";")[0] if url.startswith("data:") else "image/jpeg" + ) parts.append( Part( inline_data=Blob( - mime_type="image/jpeg", - data=base64.b64decode(c["image_url"]["url"].split(",")[1]), + mime_type=mime_type, + data=base64.b64decode(url.split(",")[1]), ) ) ) @@ -995,21 +1000,13 @@ class GoogleLLMService(LLMService): elif part.inline_data and part.inline_data.data: # Here we assume that inline_data is an image. image = Image.open(io.BytesIO(part.inline_data.data)) - # NOTE: Gemini 3 Pro Image seems to always give - # JPEGs. It expects us to send back the - # original JPEG data in the context, along with - # the corresponding thought signature. JPEG - # happens to be the format our universal - # context uses for images, so we can just pass - # it through as-is. await self.push_frame( AssistantImageRawFrame( image=image.tobytes(), size=image.size, format="RGB", - original_jpeg=part.inline_data.data - if part.inline_data.mime_type == "image/jpeg" - else None, + original_data=part.inline_data.data, + original_mime_type=part.inline_data.mime_type, ) ) @@ -1037,12 +1034,6 @@ class GoogleLLMService(LLMService): if part.function_call: bookmark["function_call"] = function_call_id elif part.inline_data and part.inline_data.data: - # With Gemini 3 Pro (where sending the - # thought signature is required for images) - # this is the JPEG-encoded image data that - # we sent to be written to the context - # as-is, so it is usable as a bookmark (it - # will match the context data). bookmark["inline_data"] = part.inline_data elif part.text is not None: # Account for Gemini 3 Pro trailing