Merge pull request #3224 from pipecat-ai/pk/simplify-gemini-thinking
Clean up logic related to applying Gemini thought signatures to conte…
This commit is contained in:
kompfner
@@ -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
|
||||
|
||||
3
changelog/3224.fixed.2.md
Normal file
3
changelog/3224.fixed.2.md
Normal file
@@ -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.
|
||||
3
changelog/3224.fixed.md
Normal file
3
changelog/3224.fixed.md
Normal file
@@ -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.
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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(
|
||||
# {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -281,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"):
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -209,7 +211,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 +220,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 +252,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(
|
||||
@@ -416,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]),
|
||||
)
|
||||
)
|
||||
)
|
||||
@@ -447,136 +433,138 @@ 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
|
||||
# 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
|
||||
|
||||
return False
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -1470,6 +1466,23 @@ 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 an image generated by the assistant.
|
||||
|
||||
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_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_data: Optional[bytes] = None
|
||||
original_mime_type: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class InputDTMFFrame(DTMFFrame, SystemFrame):
|
||||
"""DTMF keypress input frame from transport."""
|
||||
@@ -1745,16 +1758,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
|
||||
|
||||
@@ -150,21 +150,29 @@ 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():
|
||||
buffer = io.BytesIO()
|
||||
Image.frombytes(format, size, image).save(buffer, format="JPEG")
|
||||
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
|
||||
if image_already_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)
|
||||
|
||||
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)
|
||||
|
||||
@@ -334,18 +342,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 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.
|
||||
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)
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -740,10 +743,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):
|
||||
@@ -831,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_data and frame.original_mime_type:
|
||||
await self._context.add_image_frame_message(
|
||||
format=frame.original_mime_type,
|
||||
size=frame.size, # Technically doesn't matter, since already encoded
|
||||
image=frame.original_data,
|
||||
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
|
||||
|
||||
|
||||
@@ -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,
|
||||
@@ -478,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]),
|
||||
)
|
||||
)
|
||||
)
|
||||
@@ -932,7 +938,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 +949,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 +971,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,46 +981,43 @@ 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:
|
||||
# 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"
|
||||
await self.push_frame(
|
||||
AssistantImageRawFrame(
|
||||
image=image.tobytes(),
|
||||
size=image.size,
|
||||
format="RGB",
|
||||
original_data=part.inline_data.data,
|
||||
original_mime_type=part.inline_data.mime_type,
|
||||
)
|
||||
)
|
||||
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 +1025,37 @@ 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:
|
||||
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 +1104,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"],
|
||||
)
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user