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

Update GeminiMultimodalLiveLLMService docstrings

Browse Source
This commit is contained in:
Mark Backman
2025-06-26 11:47:30 -04:00
parent 9e518cf2ba
commit d123cd4b2b
2 changed files with 411 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
src/pipecat/services/gemini_multimodal_live/events.py
View File

@@ -3,7 +3,8 @@
#
# SPDX-License-Identifier: BSD 2-Clause License
#
#
"""Event models and utilities for Google Gemini Multimodal Live API."""
import base64
import io
@@ -22,16 +23,37 @@ from pipecat.frames.frames import ImageRawFrame
class MediaChunk(BaseModel):
"""Represents a chunk of media data for transmission.
Parameters:
mimeType: MIME type of the media content.
data: Base64-encoded media data.
"""
mimeType: str
data: str
class ContentPart(BaseModel):
"""Represents a part of content that can contain text or media.
Parameters:
text: Text content. Defaults to None.
inlineData: Inline media data. Defaults to None.
"""
text: Optional[str] = Field(default=None, validate_default=False)
inlineData: Optional[MediaChunk] = Field(default=None, validate_default=False)
class Turn(BaseModel):
"""Represents a conversational turn in the dialogue.
Parameters:
role: The role of the speaker, either "user" or "model". Defaults to "user".
parts: List of content parts that make up the turn.
"""
role: Literal["user", "model"] = "user"
parts: List[ContentPart]
@@ -53,7 +75,15 @@ class EndSensitivity(str, Enum):
class AutomaticActivityDetection(BaseModel):
"""Configures automatic detection of activity."""
"""Configures automatic detection of voice activity.
Parameters:
disabled: Whether automatic activity detection is disabled. Defaults to None.
start_of_speech_sensitivity: Sensitivity for detecting speech start. Defaults to None.
prefix_padding_ms: Padding before speech start in milliseconds. Defaults to None.
end_of_speech_sensitivity: Sensitivity for detecting speech end. Defaults to None.
silence_duration_ms: Duration of silence to detect speech end. Defaults to None.
"""
disabled: Optional[bool] = None
start_of_speech_sensitivity: Optional[StartSensitivity] = None
@@ -63,25 +93,57 @@ class AutomaticActivityDetection(BaseModel):
class RealtimeInputConfig(BaseModel):
"""Configures the realtime input behavior."""
"""Configures the realtime input behavior.
Parameters:
automatic_activity_detection: Voice activity detection configuration. Defaults to None.
"""
automatic_activity_detection: Optional[AutomaticActivityDetection] = None
class RealtimeInput(BaseModel):
"""Contains realtime input media chunks.
Parameters:
mediaChunks: List of media chunks for realtime processing.
"""
mediaChunks: List[MediaChunk]
class ClientContent(BaseModel):
"""Content sent from client to the Gemini Live API.
Parameters:
turns: List of conversation turns. Defaults to None.
turnComplete: Whether the client's turn is complete. Defaults to False.
"""
turns: Optional[List[Turn]] = None
turnComplete: bool = False
class AudioInputMessage(BaseModel):
"""Message containing audio input data.
Parameters:
realtimeInput: Realtime input containing audio chunks.
"""
realtimeInput: RealtimeInput
@classmethod
def from_raw_audio(cls, raw_audio: bytes, sample_rate: int) -> "AudioInputMessage":
"""Create an audio input message from raw audio data.
Args:
raw_audio: Raw audio bytes.
sample_rate: Audio sample rate in Hz.
Returns:
AudioInputMessage instance with encoded audio data.
"""
data = base64.b64encode(raw_audio).decode("utf-8")
return cls(
realtimeInput=RealtimeInput(
@@ -91,10 +153,24 @@ class AudioInputMessage(BaseModel):
class VideoInputMessage(BaseModel):
"""Message containing video/image input data.
Parameters:
realtimeInput: Realtime input containing video/image chunks.
"""
realtimeInput: RealtimeInput
@classmethod
def from_image_frame(cls, frame: ImageRawFrame) -> "VideoInputMessage":
"""Create a video input message from an image frame.
Args:
frame: Image frame to encode.
Returns:
VideoInputMessage instance with encoded image data.
"""
buffer = io.BytesIO()
Image.frombytes(frame.format, frame.size, frame.image).save(buffer, format="JPEG")
data = base64.b64encode(buffer.getvalue()).decode("utf-8")
@@ -104,18 +180,44 @@ class VideoInputMessage(BaseModel):
class ClientContentMessage(BaseModel):
"""Message containing client content for the API.
Parameters:
clientContent: The client content to send.
"""
clientContent: ClientContent
class SystemInstruction(BaseModel):
"""System instruction for the model.
Parameters:
parts: List of content parts that make up the system instruction.
"""
parts: List[ContentPart]
class AudioTranscriptionConfig(BaseModel):
"""Configuration for audio transcription."""
pass
class Setup(BaseModel):
"""Setup configuration for the Gemini Live session.
Parameters:
model: Model identifier to use.
system_instruction: System instruction for the model. Defaults to None.
tools: List of available tools/functions. Defaults to None.
generation_config: Generation configuration parameters. Defaults to None.
input_audio_transcription: Input audio transcription config. Defaults to None.
output_audio_transcription: Output audio transcription config. Defaults to None.
realtime_input_config: Realtime input configuration. Defaults to None.
"""
model: str
system_instruction: Optional[SystemInstruction] = None
tools: Optional[List[dict]] = None
@@ -126,6 +228,12 @@ class Setup(BaseModel):
class Config(BaseModel):
"""Configuration message for session setup.
Parameters:
setup: Setup configuration for the session.
"""
setup: Setup
@@ -135,36 +243,86 @@ class Config(BaseModel):
class SetupComplete(BaseModel):
"""Indicates that session setup is complete."""
pass
class InlineData(BaseModel):
"""Inline data embedded in server responses.
Parameters:
mimeType: MIME type of the data.
data: Base64-encoded data content.
"""
mimeType: str
data: str
class Part(BaseModel):
"""Part of a server response containing data or text.
Parameters:
inlineData: Inline binary data. Defaults to None.
text: Text content. Defaults to None.
"""
inlineData: Optional[InlineData] = None
text: Optional[str] = None
class ModelTurn(BaseModel):
"""Represents a turn from the model in the conversation.
Parameters:
parts: List of content parts in the model's response.
"""
parts: List[Part]
class ServerContentInterrupted(BaseModel):
"""Indicates server content was interrupted.
Parameters:
interrupted: Whether the content was interrupted.
"""
interrupted: bool
class ServerContentTurnComplete(BaseModel):
"""Indicates the server's turn is complete.
Parameters:
turnComplete: Whether the turn is complete.
"""
turnComplete: bool
class BidiGenerateContentTranscription(BaseModel):
"""Transcription data from bidirectional content generation.
Parameters:
text: The transcribed text content.
"""
text: str
class ServerContent(BaseModel):
"""Content sent from server to client.
Parameters:
modelTurn: Model's conversational turn. Defaults to None.
interrupted: Whether content was interrupted. Defaults to None.
turnComplete: Whether the turn is complete. Defaults to None.
inputTranscription: Transcription of input audio. Defaults to None.
outputTranscription: Transcription of output audio. Defaults to None.
"""
modelTurn: Optional[ModelTurn] = None
interrupted: Optional[bool] = None
turnComplete: Optional[bool] = None
@@ -173,12 +331,26 @@ class ServerContent(BaseModel):
class FunctionCall(BaseModel):
"""Represents a function call from the model.
Parameters:
id: Unique identifier for the function call.
name: Name of the function to call.
args: Arguments to pass to the function.
"""
id: str
name: str
args: dict
class ToolCall(BaseModel):
"""Contains one or more function calls.
Parameters:
functionCalls: List of function calls to execute.
"""
functionCalls: List[FunctionCall]
@@ -193,14 +365,32 @@ class Modality(str, Enum):
class ModalityTokenCount(BaseModel):
"""Token count for a specific modality."""
"""Token count for a specific modality.
Parameters:
modality: The modality type.
tokenCount: Number of tokens for this modality.
"""
modality: Modality
tokenCount: int
class UsageMetadata(BaseModel):
"""Usage metadata about the response."""
"""Usage metadata about the API response.
Parameters:
promptTokenCount: Number of tokens in the prompt. Defaults to None.
cachedContentTokenCount: Number of cached content tokens. Defaults to None.
responseTokenCount: Number of tokens in the response. Defaults to None.
toolUsePromptTokenCount: Number of tokens for tool use prompts. Defaults to None.
thoughtsTokenCount: Number of tokens for model thoughts. Defaults to None.
totalTokenCount: Total number of tokens used. Defaults to None.
promptTokensDetails: Detailed breakdown of prompt tokens by modality. Defaults to None.
cacheTokensDetails: Detailed breakdown of cache tokens by modality. Defaults to None.
responseTokensDetails: Detailed breakdown of response tokens by modality. Defaults to None.
toolUsePromptTokensDetails: Detailed breakdown of tool use tokens by modality. Defaults to None.
"""
promptTokenCount: Optional[int] = None
cachedContentTokenCount: Optional[int] = None
@@ -215,6 +405,15 @@ class UsageMetadata(BaseModel):
class ServerEvent(BaseModel):
"""Server event received from the Gemini Live API.
Parameters:
setupComplete: Setup completion notification. Defaults to None.
serverContent: Content from the server. Defaults to None.
toolCall: Tool/function call request. Defaults to None.
usageMetadata: Token usage metadata. Defaults to None.
"""
setupComplete: Optional[SetupComplete] = None
serverContent: Optional[ServerContent] = None
toolCall: Optional[ToolCall] = None
@@ -222,6 +421,14 @@ class ServerEvent(BaseModel):
def parse_server_event(str):
"""Parse a server event from JSON string.
Args:
str: JSON string containing the server event.
Returns:
ServerEvent instance if parsing succeeds, None otherwise.
"""
try:
evt = json.loads(str)
return ServerEvent.model_validate(evt)
@@ -231,7 +438,12 @@ def parse_server_event(str):
class ContextWindowCompressionConfig(BaseModel):
"""Configuration for context window compression."""
"""Configuration for context window compression.
Parameters:
sliding_window: Whether to use sliding window compression. Defaults to True.
trigger_tokens: Token count threshold to trigger compression. Defaults to None.
"""
sliding_window: Optional[bool] = Field(default=True)
trigger_tokens: Optional[int] = Field(default=None)

227
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -4,6 +4,13 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Google Gemini Multimodal Live API service implementation.
This module provides real-time conversational AI capabilities using Google's
Gemini Multimodal Live API, supporting both text and audio modalities with
voice transcription, streaming responses, and tool usage.
"""
import base64
import json
import time
@@ -79,7 +86,11 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
Source:
https://ai.google.dev/api/generate-content#MediaResolution
Returns None if the language is not supported by Gemini Live.
Args:
language: The language enum value to convert.
Returns:
The Gemini language code string, or None if the language is not supported.
"""
language_map = {
# Arabic
@@ -166,8 +177,22 @@ def language_to_gemini_language(language: Language) -> Optional[str]:
class GeminiMultimodalLiveContext(OpenAILLMContext):
"""Extended OpenAI context for Gemini Multimodal Live API.
Provides Gemini-specific context management including system instruction
extraction and message format conversion for the Live API.
"""
@staticmethod
def upgrade(obj: OpenAILLMContext) -> "GeminiMultimodalLiveContext":
"""Upgrade an OpenAI context to Gemini context.
Args:
obj: The OpenAI context to upgrade.
Returns:
The upgraded Gemini context instance.
"""
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, GeminiMultimodalLiveContext):
logger.debug(f"Upgrading to Gemini Multimodal Live Context: {obj}")
obj.__class__ = GeminiMultimodalLiveContext
@@ -178,6 +203,11 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
pass
def extract_system_instructions(self):
"""Extract system instructions from context messages.
Returns:
Combined system instruction text from all system messages.
"""
system_instruction = ""
for item in self.messages:
if item.get("role") == "system":
@@ -189,6 +219,11 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
return system_instruction
def get_messages_for_initializing_history(self):
"""Get messages formatted for Gemini history initialization.
Returns:
List of messages in Gemini format for conversation history.
"""
messages = []
for item in self.messages:
role = item.get("role")
@@ -216,7 +251,19 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
"""User context aggregator for Gemini Multimodal Live.
Extends OpenAI user aggregator to handle Gemini-specific message passing
while maintaining compatibility with the standard aggregation pipeline.
"""
async def process_frame(self, frame, direction):
"""Process incoming frames for user context aggregation.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
await super().process_frame(frame, direction)
# kind of a hack just to pass the LLMMessagesAppendFrame through, but it's fine for now
if isinstance(frame, LLMMessagesAppendFrame):
@@ -224,15 +271,33 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
# The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
# but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
# need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
# are process. This ensures that the context gets only one set of messages.
"""Assistant context aggregator for Gemini Multimodal Live.
Handles assistant response aggregation while filtering out LLMTextFrames
to prevent duplicate context entries, as Gemini Live pushes both
LLMTextFrames and TTSTextFrames.
"""
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames for assistant context aggregation.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
# The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
# but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
# need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
# are process. This ensures that the context gets only one set of messages.
if not isinstance(frame, LLMTextFrame):
await super().process_frame(frame, direction)
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle user image frames.
Args:
frame: The user image frame to handle.
"""
# We don't want to store any images in the context. Revisit this later
# when the API evolves.
pass
@@ -240,17 +305,36 @@ class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggre
@dataclass
class GeminiMultimodalLiveContextAggregatorPair:
"""Pair of user and assistant context aggregators for Gemini Multimodal Live.
Parameters:
_user: The user context aggregator instance.
_assistant: The assistant context aggregator instance.
"""
_user: GeminiMultimodalLiveUserContextAggregator
_assistant: GeminiMultimodalLiveAssistantContextAggregator
def user(self) -> GeminiMultimodalLiveUserContextAggregator:
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> GeminiMultimodalLiveAssistantContextAggregator:
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class GeminiMultimodalModalities(Enum):
"""Supported modalities for Gemini Multimodal Live."""
TEXT = "TEXT"
AUDIO = "AUDIO"
@@ -265,7 +349,15 @@ class GeminiMediaResolution(str, Enum):
class GeminiVADParams(BaseModel):
"""Voice Activity Detection parameters."""
"""Voice Activity Detection parameters for Gemini Live.
Parameters:
disabled: Whether to disable VAD. Defaults to None.
start_sensitivity: Sensitivity for speech start detection. Defaults to None.
end_sensitivity: Sensitivity for speech end detection. Defaults to None.
prefix_padding_ms: Prefix padding in milliseconds. Defaults to None.
silence_duration_ms: Silence duration threshold in milliseconds. Defaults to None.
"""
disabled: Optional[bool] = Field(default=None)
start_sensitivity: Optional[events.StartSensitivity] = Field(default=None)
@@ -275,7 +367,12 @@ class GeminiVADParams(BaseModel):
class ContextWindowCompressionParams(BaseModel):
"""Parameters for context window compression."""
"""Parameters for context window compression in Gemini Live.
Parameters:
enabled: Whether compression is enabled. Defaults to False.
trigger_tokens: Token count to trigger compression. None uses 80% of context window.
"""
enabled: bool = Field(default=False)
trigger_tokens: Optional[int] = Field(
@@ -284,6 +381,23 @@ class ContextWindowCompressionParams(BaseModel):
class InputParams(BaseModel):
"""Input parameters for Gemini Multimodal Live generation.
Parameters:
frequency_penalty: Frequency penalty for generation (0.0-2.0). Defaults to None.
max_tokens: Maximum tokens to generate. Must be >= 1. Defaults to 4096.
presence_penalty: Presence penalty for generation (0.0-2.0). Defaults to None.
temperature: Sampling temperature (0.0-2.0). Defaults to None.
top_k: Top-k sampling parameter. Must be >= 0. Defaults to None.
top_p: Top-p sampling parameter (0.0-1.0). Defaults to None.
modalities: Response modalities. Defaults to AUDIO.
language: Language for generation. Defaults to EN_US.
media_resolution: Media resolution setting. Defaults to UNSPECIFIED.
vad: Voice activity detection parameters. Defaults to None.
context_window_compression: Context compression settings. Defaults to None.
extra: Additional parameters. Defaults to empty dict.
"""
frequency_penalty: Optional[float] = Field(default=None, ge=0.0, le=2.0)
max_tokens: Optional[int] = Field(default=4096, ge=1)
presence_penalty: Optional[float] = Field(default=None, ge=0.0, le=2.0)
@@ -310,23 +424,18 @@ class GeminiMultimodalLiveLLMService(LLMService):
responses, and tool usage.
Args:
api_key (str): Google AI API key
base_url (str, optional): API endpoint base URL. Defaults to
"generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent".
model (str, optional): Model identifier to use. Defaults to
"models/gemini-2.0-flash-live-001".
voice_id (str, optional): TTS voice identifier. Defaults to "Charon".
start_audio_paused (bool, optional): Whether to start with audio input paused.
Defaults to False.
start_video_paused (bool, optional): Whether to start with video input paused.
Defaults to False.
system_instruction (str, optional): System prompt for the model. Defaults to None.
tools (Union[List[dict], ToolsSchema], optional): Tools/functions available to the model.
Defaults to None.
params (InputParams, optional): Configuration parameters for the model.
Defaults to InputParams().
inference_on_context_initialization (bool, optional): Whether to generate a response
when context is first set. Defaults to True.
api_key: Google AI API key for authentication.
base_url: API endpoint base URL. Defaults to the official Gemini Live endpoint.
model: Model identifier to use. Defaults to "models/gemini-2.0-flash-live-001".
voice_id: TTS voice identifier. Defaults to "Charon".
start_audio_paused: Whether to start with audio input paused. Defaults to False.
start_video_paused: Whether to start with video input paused. Defaults to False.
system_instruction: System prompt for the model. Defaults to None.
tools: Tools/functions available to the model. Defaults to None.
params: Configuration parameters for the model. Defaults to InputParams().
inference_on_context_initialization: Whether to generate a response when context
is first set. Defaults to True.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the Gemini one.
@@ -408,19 +517,43 @@ class GeminiMultimodalLiveLLMService(LLMService):
}
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True as Gemini Live supports token usage metrics.
"""
return True
def set_audio_input_paused(self, paused: bool):
"""Set the audio input pause state.
Args:
paused: Whether to pause audio input.
"""
self._audio_input_paused = paused
def set_video_input_paused(self, paused: bool):
"""Set the video input pause state.
Args:
paused: Whether to pause video input.
"""
self._video_input_paused = paused
def set_model_modalities(self, modalities: GeminiMultimodalModalities):
"""Set the model response modalities.
Args:
modalities: The modalities to use for responses.
"""
self._settings["modalities"] = modalities
def set_language(self, language: Language):
"""Set the language for generation."""
"""Set the language for generation.
Args:
language: The language to use for generation.
"""
self._language = language
self._language_code = language_to_gemini_language(language) or "en-US"
self._settings["language"] = self._language_code
@@ -433,6 +566,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
way to trigger the pipeline. This sends the history to the server. The `inference_on_context_initialization`
flag controls whether to set the turnComplete flag when we do this. Without that flag, the model will
not respond. This is often what we want when setting the context at the beginning of a conversation.
Args:
context: The OpenAI LLM context to set.
"""
if self._context:
logger.error(
@@ -447,14 +583,29 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def start(self, frame: StartFrame):
"""Start the service and establish websocket connection.
Args:
frame: The start frame.
"""
await super().start(frame)
await self._connect()
async def stop(self, frame: EndFrame):
"""Stop the service and close connections.
Args:
frame: The end frame.
"""
await super().stop(frame)
await self._disconnect()
async def cancel(self, frame: CancelFrame):
"""Cancel the service and close connections.
Args:
frame: The cancel frame.
"""
await super().cancel(frame)
await self._disconnect()
@@ -489,6 +640,12 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames for the Gemini Live service.
Args:
frame: The frame to process.
direction: The frame processing direction.
"""
await super().process_frame(frame, direction)
if isinstance(frame, TranscriptionFrame):
@@ -544,6 +701,11 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def send_client_event(self, event):
"""Send a client event to the Gemini Live API.
Args:
event: The event to send.
"""
await self._ws_send(event.model_dump(exclude_none=True))
async def _connect(self):
@@ -1033,22 +1195,19 @@ class GeminiMultimodalLiveLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GeminiMultimodalLiveContextAggregatorPair:
"""Create an instance of GeminiMultimodalLiveContextAggregatorPair from
an OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create an instance of GeminiMultimodalLiveContextAggregatorPair from an OpenAILLMContext.
Constructor keyword arguments for both the user and assistant aggregators can be provided.
Args:
context (OpenAILLMContext): The LLM context.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
context: The LLM context to use.
user_params: User aggregator parameters. Defaults to LLMUserAggregatorParams().
assistant_params: Assistant aggregator parameters. Defaults to LLMAssistantAggregatorParams().
Returns:
GeminiMultimodalLiveContextAggregatorPair: A pair of context
aggregators, one for the user and one for the assistant,
encapsulated in an GeminiMultimodalLiveContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())