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

Progress on LLM failover support

Browse Source
This commit is contained in:
Paul Kompfner
2025-07-25 14:56:38 -04:00
parent 602724b984
commit ee2aade12c
6 changed files with 418 additions and 668 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src/pipecat/adapters/base_llm_adapter.py
View File

@@ -11,22 +11,26 @@ adapters that handle tool format conversion and standardization.
"""
from abc import ABC, abstractmethod
from typing import Any, List, Union, cast
from typing import Any, Generic, List, TypedDict, TypeVar, Union, cast
from loguru import logger
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
# Should be a TypedDict
TLLMInvocationParams = TypeVar("TLLMInvocationParams", bound=dict[str, Any])
class BaseLLMAdapter(ABC):
# TODO: fix everywhere we subclass BaseLLMAdapter...
class BaseLLMAdapter(ABC, Generic[TLLMInvocationParams]):
"""Abstract base class for LLM provider adapters.
Provides a standard interface for converting to provider-specific formats.
Handles:
- Converting universal LLM context to provider-specific parameters for LLM
invocation.
- Extracting provider-specific parameters for LLM invocation from a
universal LLM context
- Converting standardized tools schema to provider-specific tool formats.
- Extracting messages from the LLM context for the purposes of logging
about the specific provider.
@@ -35,7 +39,7 @@ class BaseLLMAdapter(ABC):
"""
@abstractmethod
def get_llm_invocation_params(self, context: LLMContext) -> dict[str, Any]:
def get_llm_invocation_params(self, context: LLMContext) -> TLLMInvocationParams:
"""Get provider-specific LLM invocation parameters from a universal LLM context.
Args:
@@ -71,6 +75,7 @@ class BaseLLMAdapter(ABC):
"""
pass
# TODO: should this also be able to return NotGiven?
def from_standard_tools(self, tools: Any) -> List[Any]:
"""Convert tools from standard format to provider format.
@@ -87,4 +92,38 @@ class BaseLLMAdapter(ABC):
# Fallback to return the same tools in case they are not in a standard format
return tools
def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
"""Create a WAV file header for audio data.
Args:
sample_rate: Audio sample rate in Hz.
num_channels: Number of audio channels.
bits_per_sample: Bits per audio sample.
data_size: Size of audio data in bytes.
Returns:
WAV header as a bytearray.
"""
# RIFF chunk descriptor
header = bytearray()
header.extend(b"RIFF") # ChunkID
header.extend((data_size + 36).to_bytes(4, "little")) # ChunkSize: total size - 8
header.extend(b"WAVE") # Format
# "fmt " sub-chunk
header.extend(b"fmt ") # Subchunk1ID
header.extend((16).to_bytes(4, "little")) # Subchunk1Size (16 for PCM)
header.extend((1).to_bytes(2, "little")) # AudioFormat (1 for PCM)
header.extend(num_channels.to_bytes(2, "little")) # NumChannels
header.extend(sample_rate.to_bytes(4, "little")) # SampleRate
# Calculate byte rate and block align
byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
block_align = num_channels * (bits_per_sample // 8)
header.extend(byte_rate.to_bytes(4, "little")) # ByteRate
header.extend(block_align.to_bytes(2, "little")) # BlockAlign
header.extend(bits_per_sample.to_bytes(2, "little")) # BitsPerSample
# "data" sub-chunk
header.extend(b"data") # Subchunk2ID
header.extend(data_size.to_bytes(4, "little")) # Subchunk2Size
return header
# TODO: we can move the logic to also handle the Messages here

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

@@ -6,21 +6,68 @@
"""Gemini LLM adapter for Pipecat."""
from typing import Any, Dict, List, Union
import base64
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, TypedDict, Union
from loguru import logger
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
try:
from google.genai.types import (
Blob,
Content,
FunctionCall,
FunctionResponse,
Part,
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
raise Exception(f"Missing module: {e}")
class GeminiLLMAdapter(BaseLLMAdapter):
"""LLM adapter for Google's Gemini service.
class GeminiLLMInvocationParams(TypedDict):
"""Context-based parameters for invoking Gemini LLM."""
Provides tool schema conversion functionality to transform standard tool
definitions into Gemini's specific function-calling format for use with
Gemini LLM models.
system_instruction: Optional[str]
messages: List[Content]
tools: List[Any]
class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
"""Gemini-specific adapter for Pipecat.
Handles:
- Extracting parameters for Gemini's API from a universal
LLM context
- Converting Pipecat's standardized tools schema to Gemini's function-calling format.
- Extracting and sanitizing messages from the LLM context for logging with Gemini.
"""
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
def get_llm_invocation_params(self, context: LLMContext) -> GeminiLLMInvocationParams:
"""Get Gemini-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for Gemini's API.
"""
# TODO: remove when done testing
print(f"[pk] {self}: Getting LLM invocation params...")
messages = self._from_standard_messages(context.messages)
return {
"system_instruction": messages.system_instruction,
"messages": messages.messages,
"tools": self.from_standard_tools(context.tools),
}
def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
"""Convert tool schemas to Gemini's function-calling format.
Args:
@@ -39,3 +86,217 @@ class GeminiLLMAdapter(BaseLLMAdapter):
custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])
return formatted_standard_tools + custom_gemini_tools
def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
"""Get messages from the LLM context in a format ready for logging about Gemini.
Removes or truncates sensitive data like image content for safe logging.
Args:
context: The LLM context containing messages.
Returns:
List of messages in a format ready for logging about Gemini.
"""
# Get messages in Gemini's format
messages = self._from_standard_messages(context.messages).messages
# Sanitize messages for logging
messages_for_logging = []
for message in messages:
obj = message.to_json_dict()
try:
if "parts" in obj:
for part in obj["parts"]:
if "inline_data" in part:
part["inline_data"]["data"] = "..."
except Exception as e:
logger.debug(f"Error: {e}")
messages_for_logging.append(obj)
return messages_for_logging
@dataclass
class ConvertedMessages:
"""Container for converted messages.
Holds the converted messages in a format suitable for Gemini's API.
"""
messages: List[Content]
system_instruction: Optional[str] = None
def _from_standard_messages(
self, standard_messages: List[LLMContextMessage]
) -> ConvertedMessages:
"""Restructures messages to ensure proper Google format and message ordering.
This method handles conversion of OpenAI-formatted messages to Google format,
with special handling for function calls, function responses, and system messages.
System messages are added back to the context as user messages when needed.
The final message order is preserved as:
1. Function calls (from model)
2. Function responses (from user)
3. Text messages (converted from system messages)
Note:
System messages are only added back when there are no regular text
messages in the context, ensuring proper conversation continuity
after function calls.
"""
system_instruction = None
messages = []
# Process each message, preserving Google-formatted messages and converting others
for message in standard_messages:
if isinstance(message, Content):
# Keep existing Google-formatted messages (e.g., function calls/responses)
# TODO: this branch is probably not needed anymore, since LLMContext contains a universal format
messages.append(message)
continue
# Convert standard format to Google format
converted = self._from_standard_message(message)
if isinstance(converted, Content):
# Regular (non-system) message
messages.append(converted)
else:
# System instruction
system_instruction = converted
# Check if we only have function-related messages (no regular text)
has_regular_messages = any(
len(msg.parts) == 1
and getattr(msg.parts[0], "text", None)
and not getattr(msg.parts[0], "function_call", None)
and not getattr(msg.parts[0], "function_response", None)
for msg in self._messages
)
# Add system instruction back as a user message if we only have function messages
if system_instruction and not has_regular_messages:
messages.append(Content(role="user", parts=[Part(text=system_instruction)]))
# Remove any empty messages
messages = [m for m in messages if m.parts]
return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
def _from_standard_message(self, message: LLMContextMessage) -> Content | str:
"""Convert standard format message to Google Content object.
Handles conversion of text, images, and function calls to Google's
format.
System instructions are returned as a plain string.
Args:
message: Message in standard format.
Returns:
Content object with role and parts, or a plain string for system
messages.
Examples:
Standard text message::
{
"role": "user",
"content": "Hello there"
}
Converts to Google Content with::
Content(
role="user",
parts=[Part(text="Hello there")]
)
Standard function call message::
{
"role": "assistant",
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "test"}'
}
}
]
}
Converts to Google Content with::
Content(
role="model",
parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
)
"""
role = message["role"]
content = message.get("content", [])
if role == "system":
# System instructions are returned as plain text
# TODO: here we've always assumed that system instructions are plain text...is that a safe assumption?
return content
elif role == "assistant":
role = "model"
parts = []
if message.get("tool_calls"):
for tc in message["tool_calls"]:
parts.append(
Part(
function_call=FunctionCall(
name=tc["function"]["name"],
args=json.loads(tc["function"]["arguments"]),
)
)
)
elif role == "tool":
role = "model"
parts.append(
Part(
function_response=FunctionResponse(
name="tool_call_result", # seems to work to hard-code the same name every time
response=json.loads(message["content"]),
)
)
)
elif isinstance(content, str):
parts.append(Part(text=content))
elif isinstance(content, list):
for c in content:
if c["type"] == "text":
parts.append(Part(text=c["text"]))
elif c["type"] == "image_url":
parts.append(
Part(
inline_data=Blob(
mime_type="image/jpeg",
data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
)
)
)
elif c["type"] == "input_audio":
input_audio = c["input_audio"]
parts.append(
Part(
inline_data=Blob(
mime_type="audio/wav",
data=(
bytes(
self.create_wav_header(
input_audio["sample_rate"],
input_audio["num_channels"],
16,
len(input_audio["data"]),
)
+ input_audio["data"]
)
),
)
)
)
message = Content(role=role, parts=parts)
return message

55
src/pipecat/adapters/services/open_ai_adapter.py
View File

@@ -8,34 +8,55 @@
import copy
import json
from typing import Any, List
from typing import Any, List, TypedDict
from openai.types.chat import ChatCompletionToolParam
from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import (
ChatCompletionMessageParam,
ChatCompletionToolChoiceOptionParam,
ChatCompletionToolParam,
)
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMContextMessage,
LLMContextToolChoice,
NotGiven,
)
class OpenAILLMAdapter(BaseLLMAdapter):
"""Adapter for converting tool schemas to OpenAI's format.
class OpenAILLMInvocationParams(TypedDict):
"""Context-based parameters for invoking OpenAI ChatCompletion API."""
Provides conversion utilities for transforming Pipecat's standard tool
schemas into the format expected by OpenAI's ChatCompletion API for
function calling capabilities.
messages: List[ChatCompletionMessageParam]
tools: List[ChatCompletionToolParam] | OpenAINotGiven
tool_choice: ChatCompletionToolChoiceOptionParam | OpenAINotGiven
class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
"""OpenAI-specific adapter for Pipecat.
Handles:
- Extracting parameters for OpenAI's ChatCompletion API from a universal
LLM context
- Converting Pipecat's standardized tools schema to OpenAI's function-calling format.
- Extracting and sanitizing messages from the LLM context for logging with OpenAI.
"""
def get_llm_invocation_params(self, context: LLMContext) -> dict[str, Any]:
def get_llm_invocation_params(self, context: LLMContext) -> OpenAILLMInvocationParams:
"""Get OpenAI-specific LLM invocation parameters from a universal LLM context.
Args:
context: The LLM context containing messages, tools, etc.
Returns:
Dictionary of parameters for OpenAI's chat completion API.
Dictionary of parameters for OpenAI's ChatCompletion API.
"""
return {
"messages": context.messages,
"messages": self._from_standard_messages(context.messages),
# TODO: doesn't seem right that we may or may not need to convert tools here; they should already be guaranteed to exist in a universal format in the LLMContext, right?
"tools": self.from_standard_tools(context.tools),
"tool_choice": context.tool_choice,
@@ -81,3 +102,15 @@ class OpenAILLMAdapter(BaseLLMAdapter):
msg["data"] = "..."
msgs.append(msg)
return json.dumps(msgs, ensure_ascii=False)
def _from_standard_messages(
self, messages: List[LLMContextMessage]
) -> List[ChatCompletionMessageParam]:
# Just a pass-through: messages is already the right type
return messages
def _from_standard_tool_choice(
self, tool_choice: LLMContextToolChoice | NotGiven
) -> ChatCompletionToolChoiceOptionParam | OpenAINotGiven:
# Just a pass-through: tool_choice is already the right type
return tool_choice

37
src/pipecat/processors/aggregators/llm_context.py
View File

@@ -32,6 +32,9 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import AudioRawFrame, Frame
# "Re-export" types from OpenAI that we're using as universal context types.
# NOTE: this is just for convenience, for now. As soon as the universal types
# diverge from OpenAI's, we should ditch this. In fact, audio frames already
# diverge from OpenAI's standard format...we really ought to do this.
LLMContextMessage = ChatCompletionMessageParam
LLMContextTool = ChatCompletionToolParam
LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
@@ -148,6 +151,7 @@ class LLMContext:
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
# TODO: we might not want the universal format to be base64 encoded, since encoding is not needed by all LLM services; today, te Gemini adapter has to decode from base64, which is less than ideal.
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
content = []
@@ -158,18 +162,39 @@ class LLMContext:
)
self.add_message({"role": "user", "content": content})
def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None):
# NOTE: today we've only built support for audio frames with the Google
# LLM, so this "universal" representation skews towards that.
# When we add support for other LLMs, we may need to adjust this.
def add_audio_frames_message(
self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
):
"""Add a message containing audio frames.
Args:
audio_frames: List of audio frame objects to include.
text: Optional text to include with the audio.
Note:
This method is currently a placeholder for future implementation.
"""
# TODO: implement storing universal representation of audio frames in context (only used by Google for now)
pass
if not audio_frames:
return
sample_rate = audio_frames[0].sample_rate
num_channels = audio_frames[0].num_channels
content = []
content.append({"type": "text", "text": text})
data = b"".join(frame.audio for frame in audio_frames)
# TODO: filter this out in OpenAI adapter, since it doesn't support audio frames
content.append(
{
"type": "input_audio",
"input_audio": {
"data": data,
"sample_rate": sample_rate,
"num_channels": num_channels,
},
}
)
self.add_message({"role": "user", "content": content})
@dataclass

665
src/pipecat/services/google/llm.py
View File

@@ -10,49 +10,29 @@ This module provides Google Gemini integration for the Pipecat framework,
including LLM services, context management, and message aggregation.
"""
import base64
import io
import json
import os
import uuid
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter, GeminiLLMInvocationParams
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
UserImageRawFrame,
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextFrame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchResponseFrame
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
)
from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
from pipecat.utils.tracing.service_decorators import traced_llm
@@ -63,13 +43,8 @@ try:
from google import genai
from google.api_core.exceptions import DeadlineExceeded
from google.genai.types import (
Blob,
Content,
FunctionCall,
FunctionResponse,
GenerateContentConfig,
HttpOptions,
Part,
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -77,577 +52,12 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
class GoogleUserContextAggregator(OpenAIUserContextAggregator):
"""Google-specific user context aggregator.
Extends OpenAI user context aggregator to handle Google AI's specific
Content and Part message format for user messages.
"""
async def push_aggregation(self):
"""Push aggregated user text as a Google Content message."""
if len(self._aggregation) > 0:
self._context.add_message(Content(role="user", parts=[Part(text=self._aggregation)]))
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
# Push context frame
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Reset our accumulator state.
await self.reset()
class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Google-specific assistant context aggregator.
Extends OpenAI assistant context aggregator to handle Google AI's specific
Content and Part message format for assistant responses and function calls.
"""
async def handle_aggregation(self, aggregation: str):
"""Handle aggregated assistant text response.
Args:
aggregation: The aggregated text response from the assistant.
"""
self._context.add_message(Content(role="model", parts=[Part(text=aggregation)]))
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle function call in progress frame.
Args:
frame: Frame containing function call details.
"""
self._context.add_message(
Content(
role="model",
parts=[
Part(
function_call=FunctionCall(
id=frame.tool_call_id, name=frame.function_name, args=frame.arguments
)
)
],
)
)
self._context.add_message(
Content(
role="user",
parts=[
Part(
function_response=FunctionResponse(
id=frame.tool_call_id,
name=frame.function_name,
response={"response": "IN_PROGRESS"},
)
)
],
)
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle function call result frame.
Args:
frame: Frame containing function call result.
"""
if frame.result:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, frame.result
)
else:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "COMPLETED"
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle function call cancellation frame.
Args:
frame: Frame containing function call cancellation details.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
async def _update_function_call_result(
self, function_name: str, tool_call_id: str, result: Any
):
for message in self._context.messages:
if message.role == "user":
for part in message.parts:
if part.function_response and part.function_response.id == tool_call_id:
part.function_response.response = {"value": json.dumps(result)}
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle user image frame.
Args:
frame: Frame containing user image data and request context.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
self._context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.request.context,
)
@dataclass
class GoogleContextAggregatorPair:
"""Pair of Google context aggregators for user and assistant messages.
Parameters:
_user: User context aggregator for handling user messages.
_assistant: Assistant context aggregator for handling assistant responses.
"""
_user: GoogleUserContextAggregator
_assistant: GoogleAssistantContextAggregator
def user(self) -> GoogleUserContextAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> GoogleAssistantContextAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class GoogleLLMContext(OpenAILLMContext):
"""Google AI LLM context that extends OpenAI context for Google-specific formatting.
This class handles conversion between OpenAI-style messages and Google AI's
Content/Part format, including system messages, function calls, and media.
"""
def __init__(
self,
messages: Optional[List[dict]] = None,
tools: Optional[List[dict]] = None,
tool_choice: Optional[dict] = None,
):
"""Initialize GoogleLLMContext.
Args:
messages: Initial messages in OpenAI format.
tools: Available tools/functions for the model.
tool_choice: Tool choice configuration.
"""
super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
self.system_message = None
@staticmethod
def upgrade_to_google(obj: OpenAILLMContext) -> "GoogleLLMContext":
"""Upgrade an OpenAI context to a Google context.
Args:
obj: OpenAI LLM context to upgrade.
Returns:
GoogleLLMContext instance with converted messages.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GoogleLLMContext):
logger.debug(f"Upgrading to Google: {obj}")
obj.__class__ = GoogleLLMContext
obj._restructure_from_openai_messages()
return obj
def set_messages(self, messages: List):
"""Set messages and restructure them for Google format.
Args:
messages: List of messages to set.
"""
self._messages[:] = messages
self._restructure_from_openai_messages()
def add_messages(self, messages: List):
"""Add messages to the context, converting to Google format as needed.
Args:
messages: List of messages to add (can be mixed formats).
"""
# Convert each message individually
converted_messages = []
for msg in messages:
if isinstance(msg, Content):
# Already in Gemini format
converted_messages.append(msg)
else:
# Convert from standard format to Gemini format
converted = self.from_standard_message(msg)
if converted is not None:
converted_messages.append(converted)
# Add the converted messages to our existing messages
self._messages.extend(converted_messages)
def get_messages_for_logging(self):
"""Get messages formatted for logging with sensitive data redacted.
Returns:
List of message dictionaries with inline data redacted.
"""
msgs = []
for message in self.messages:
obj = message.to_json_dict()
try:
if "parts" in obj:
for part in obj["parts"]:
if "inline_data" in part:
part["inline_data"]["data"] = "..."
except Exception as e:
logger.debug(f"Error: {e}")
msgs.append(obj)
return msgs
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add an image message to the context.
Args:
format: Image format (e.g., 'RGB', 'RGBA').
size: Image dimensions as (width, height).
image: Raw image bytes.
text: Optional text to accompany the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
parts = []
if text:
parts.append(Part(text=text))
parts.append(Part(inline_data=Blob(mime_type="image/jpeg", data=buffer.getvalue())))
self.add_message(Content(role="user", parts=parts))
def add_audio_frames_message(
self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
):
"""Add audio frames as a message to the context.
Args:
audio_frames: List of audio frames to add.
text: Text description of the audio content.
"""
if not audio_frames:
return
sample_rate = audio_frames[0].sample_rate
num_channels = audio_frames[0].num_channels
parts = []
data = b"".join(frame.audio for frame in audio_frames)
# NOTE(aleix): According to the docs only text or inline_data should be needed.
# (see https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/inference)
parts.append(Part(text=text))
parts.append(
Part(
inline_data=Blob(
mime_type="audio/wav",
data=(
bytes(
self.create_wav_header(sample_rate, num_channels, 16, len(data)) + data
)
),
)
),
)
self.add_message(Content(role="user", parts=parts))
# message = {"mime_type": "audio/mp3", "data": bytes(data + create_wav_header(sample_rate, num_channels, 16, len(data)))}
# self.add_message(message)
def from_standard_message(self, message):
"""Convert standard format message to Google Content object.
Handles conversion of text, images, and function calls to Google's format.
System messages are stored separately and return None.
Args:
message: Message in standard format.
Returns:
Content object with role and parts, or None for system messages.
Examples:
Standard text message::
{
"role": "user",
"content": "Hello there"
}
Converts to Google Content with::
Content(
role="user",
parts=[Part(text="Hello there")]
)
Standard function call message::
{
"role": "assistant",
"tool_calls": [
{
"function": {
"name": "search",
"arguments": '{"query": "test"}'
}
}
]
}
Converts to Google Content with::
Content(
role="model",
parts=[Part(function_call=FunctionCall(name="search", args={"query": "test"}))]
)
System message returns None and stores content in self.system_message.
"""
role = message["role"]
content = message.get("content", [])
if role == "system":
self.system_message = content
return None
elif role == "assistant":
role = "model"
parts = []
if message.get("tool_calls"):
for tc in message["tool_calls"]:
parts.append(
Part(
function_call=FunctionCall(
name=tc["function"]["name"],
args=json.loads(tc["function"]["arguments"]),
)
)
)
elif role == "tool":
role = "model"
parts.append(
Part(
function_response=FunctionResponse(
name="tool_call_result", # seems to work to hard-code the same name every time
response=json.loads(message["content"]),
)
)
)
elif isinstance(content, str):
parts.append(Part(text=content))
elif isinstance(content, list):
for c in content:
if c["type"] == "text":
parts.append(Part(text=c["text"]))
elif c["type"] == "image_url":
parts.append(
Part(
inline_data=Blob(
mime_type="image/jpeg",
data=base64.b64decode(c["image_url"]["url"].split(",")[1]),
)
)
)
message = Content(role=role, parts=parts)
return message
def to_standard_messages(self, obj) -> list:
"""Convert Google Content object to standard structured format.
Handles text, images, and function calls from Google's Content/Part objects.
Args:
obj: Google Content object with role and parts.
Returns:
List containing a single message in standard format.
Examples:
Google Content with text::
Content(
role="user",
parts=[Part(text="Hello")]
)
Converts to::
[
{
"role": "user",
"content": [{"type": "text", "text": "Hello"}]
}
]
Google Content with function call::
Content(
role="model",
parts=[Part(function_call=FunctionCall(name="search", args={"q": "test"}))]
)
Converts to::
[
{
"role": "assistant",
"tool_calls": [
{
"id": "search",
"type": "function",
"function": {
"name": "search",
"arguments": '{"q": "test"}'
}
}
]
}
]
Google Content with image::
Content(
role="user",
parts=[Part(inline_data=Blob(mime_type="image/jpeg", data=bytes_data))]
)
Converts to::
[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,<encoded_data>"}
}
]
}
]
"""
msg = {"role": obj.role, "content": []}
if msg["role"] == "model":
msg["role"] = "assistant"
for part in obj.parts:
if part.text:
msg["content"].append({"type": "text", "text": part.text})
elif part.inline_data:
encoded = base64.b64encode(part.inline_data.data).decode("utf-8")
msg["content"].append(
{
"type": "image_url",
"image_url": {"url": f"data:{part.inline_data.mime_type};base64,{encoded}"},
}
)
elif part.function_call:
args = part.function_call.args if hasattr(part.function_call, "args") else {}
msg["tool_calls"] = [
{
"id": part.function_call.name,
"type": "function",
"function": {
"name": part.function_call.name,
"arguments": json.dumps(args),
},
}
]
elif part.function_response:
msg["role"] = "tool"
resp = (
part.function_response.response
if hasattr(part.function_response, "response")
else {}
)
msg["tool_call_id"] = part.function_response.name
msg["content"] = json.dumps(resp)
# there might be no content parts for tool_calls messages
if not msg["content"]:
del msg["content"]
return [msg]
def _restructure_from_openai_messages(self):
"""Restructures messages to ensure proper Google format and message ordering.
This method handles conversion of OpenAI-formatted messages to Google format,
with special handling for function calls, function responses, and system messages.
System messages are added back to the context as user messages when needed.
The final message order is preserved as:
1. Function calls (from model)
2. Function responses (from user)
3. Text messages (converted from system messages)
Note:
System messages are only added back when there are no regular text
messages in the context, ensuring proper conversation continuity
after function calls.
"""
self.system_message = None
converted_messages = []
# Process each message, preserving Google-formatted messages and converting others
for message in self._messages:
if isinstance(message, Content):
# Keep existing Google-formatted messages (e.g., function calls/responses)
converted_messages.append(message)
continue
# Convert OpenAI format to Google format, system messages return None
converted = self.from_standard_message(message)
if converted is not None:
converted_messages.append(converted)
# Update message list
self._messages[:] = converted_messages
# Check if we only have function-related messages (no regular text)
has_regular_messages = any(
len(msg.parts) == 1
and getattr(msg.parts[0], "text", None)
and not getattr(msg.parts[0], "function_call", None)
and not getattr(msg.parts[0], "function_response", None)
for msg in self._messages
)
# Add system message back as a user message if we only have function messages
if self.system_message and not has_regular_messages:
self._messages.append(Content(role="user", parts=[Part(text=self.system_message)]))
# Remove any empty messages
self._messages = [m for m in self._messages if m.parts]
class GoogleLLMService(LLMService):
"""Google AI (Gemini) LLM service implementation.
This class implements inference with Google's AI models, translating internally
from OpenAILLMContext to the messages format expected by the Google AI model.
We use OpenAILLMContext as a lingua franca for all LLM services to enable
easy switching between different LLMs.
from the universal LLMContext to the message format expected by the Google
AI model.
"""
# Overriding the default adapter to use the Gemini one.
@@ -750,7 +160,7 @@ class GoogleLLMService(LLMService):
logger.exception(f"Failed to unset thinking budget: {e}")
@traced_llm
async def _process_context(self, context: OpenAILLMContext):
async def _process_context(self, context: LLMContext):
await self.push_frame(LLMFullResponseStartFrame())
prompt_tokens = 0
@@ -763,19 +173,31 @@ class GoogleLLMService(LLMService):
search_result = ""
try:
adapter = self.get_llm_adapter()
llm_invocation_params: GeminiLLMInvocationParams = adapter.get_llm_invocation_params(
context
)
logger.debug(
# f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
f"{self}: Generating chat [{context.get_messages_for_logging()}]"
# TODO: figure out a nice way to also log system instruction
# f"{self}: Generating chat [{self._system_instruction}] | [{adapter.get_messages_for_logging(context)}]"
f"{self}: Generating chat [{adapter.get_messages_for_logging(context)}]"
)
messages = context.messages
if context.system_message and self._system_instruction != context.system_message:
logger.debug(f"System instruction changed: {context.system_message}")
self._system_instruction = context.system_message
if (
llm_invocation_params.get("system_instruction")
and self._system_instruction != llm_invocation_params["system_instruction"]
):
logger.debug(
f"System instruction changed: {llm_invocation_params['system_instruction']}"
)
self._system_instruction = llm_invocation_params["system_instruction"]
# TODO: test what happens when there are no tools
tools = []
if context.tools:
tools = context.tools
if llm_invocation_params.get("tools"):
tools = llm_invocation_params["tools"]
elif self._tools:
tools = self._tools
tool_config = None
@@ -922,12 +344,12 @@ class GoogleLLMService(LLMService):
context = None
if isinstance(frame, OpenAILLMContextFrame):
context = GoogleLLMContext.upgrade_to_google(frame.context)
if isinstance(frame, LLMContextFrame):
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
context = GoogleLLMContext(frame.messages)
context = LLMContext(messages=frame.messages)
elif isinstance(frame, VisionImageRawFrame):
context = GoogleLLMContext()
context = LLMContext()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text
)
@@ -938,34 +360,3 @@ class GoogleLLMService(LLMService):
if context:
await self._process_context(context)
def create_context_aggregator(
self,
context: OpenAILLMContext,
*,
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GoogleContextAggregatorPair:
"""Create Google-specific context aggregators.
Creates a pair of context aggregators optimized for Google's message format,
including support for function calls, tool usage, and image handling.
Args:
context: The LLM context to create aggregators for.
user_params: Parameters for user message aggregation.
assistant_params: Parameters for assistant message aggregation.
Returns:
GoogleContextAggregatorPair: A pair of context aggregators, one for
the user and one for the assistant, encapsulated in an
GoogleContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())
if isinstance(context, OpenAILLMContext):
context = GoogleLLMContext.upgrade_to_google(context)
user = GoogleUserContextAggregator(context, params=user_params)
assistant = GoogleAssistantContextAggregator(context, params=assistant_params)
return GoogleContextAggregatorPair(_user=user, _assistant=assistant)

5
src/pipecat/services/openai/base_llm.py
View File

@@ -21,6 +21,7 @@ from openai import (
from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
from pydantic import BaseModel, Field
from pipecat.adapters.services.open_ai_adapter import OpenAILLMInvocationParams
from pipecat.frames.frames import (
Frame,
LLMFullResponseEndFrame,
@@ -173,7 +174,7 @@ class BaseOpenAILLMService(LLMService):
return True
async def get_chat_completions(
self, params_from_context: dict[str, Any]
self, params_from_context: OpenAILLMInvocationParams
) -> AsyncStream[ChatCompletionChunk]:
"""Get streaming chat completions from OpenAI API.
@@ -211,7 +212,7 @@ class BaseOpenAILLMService(LLMService):
adapter = self.get_llm_adapter()
logger.debug(f"{self}: Generating chat [{adapter.get_messages_for_logging(context)}]")
params = adapter.get_llm_invocation_params(context)
params: OpenAILLMInvocationParams = adapter.get_llm_invocation_params(context)
chunks = await self.get_chat_completions(params)