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

Gemini Live fixes, plus additional docstrings

Browse Source
This commit is contained in:
Mark Backman
2025-07-02 07:41:15 -07:00
parent e71cb3ba68
commit f590a476e7
2 changed files with 55 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,7 +72,6 @@ from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt
from . import events
from .file_api import GeminiFileAPI
try:
@@ -223,9 +222,9 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
def add_file_reference(self, file_uri: str, mime_type: str, text: Optional[str] = None):
"""Add a file reference to the context.
This adds a user message with a file reference that will be sent during context initialization.
Args:
file_uri: URI of the uploaded file
mime_type: MIME type of the file
@@ -235,15 +234,17 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
parts = []
if text:
parts.append({"type": "text", "text": text})
# Add file reference part
parts.append({"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}})
parts.append(
{"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}}
)
# Add to messages
message = {"role": "user", "content": parts}
self.messages.append(message)
logger.info(f"Added file reference to context: {file_uri}")
def get_messages_for_initializing_history(self):
"""Get messages formatted for Gemini history initialization.
@@ -270,12 +271,14 @@ class GeminiMultimodalLiveContext(OpenAILLMContext):
parts.append({"text": part.get("text")})
elif part.get("type") == "file_data":
file_data = part.get("file_data", {})
parts.append({
"fileData": {
"mimeType": file_data.get("mime_type"),
"fileUri": file_data.get("file_uri")
parts.append(
{
"fileData": {
"mimeType": file_data.get("mime_type"),
"fileUri": file_data.get("file_uri"),
}
}
})
)
else:
logger.warning(f"Unsupported content type: {str(part)[:80]}")
else:
@@ -379,7 +382,14 @@ class GeminiMultimodalModalities(Enum):
class GeminiMediaResolution(str, Enum):
"""Media resolution options for Gemini Multimodal Live."""
"""Media resolution options for Gemini Multimodal Live.
Parameters:
UNSPECIFIED: Use default resolution setting.
LOW: Low resolution with 64 tokens.
MEDIUM: Medium resolution with 256 tokens.
HIGH: High resolution with zoomed reframing and 256 tokens.
"""
UNSPECIFIED = "MEDIA_RESOLUTION_UNSPECIFIED" # Use default
LOW = "MEDIA_RESOLUTION_LOW" # 64 tokens
@@ -465,7 +475,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Overriding the default adapter to use the Gemini one.
adapter_class = GeminiLLMAdapter
def __init__(
self,
*,
@@ -496,6 +506,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
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.
file_api_base_url: Base URL for the Gemini File API. Defaults to the official endpoint.
**kwargs: Additional arguments passed to parent LLMService.
"""
super().__init__(base_url=base_url, **kwargs)
@@ -557,7 +568,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
else {},
"extra": params.extra if isinstance(params.extra, dict) else {},
}
# Initialize the File API client
self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url)
@@ -757,6 +768,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self._ws_send(event.model_dump(exclude_none=True))
async def _connect(self):
"""Establish WebSocket connection to Gemini Live API."""
if self._websocket:
# Here we assume that if we have a websocket, we are connected. We
# handle disconnections in the send/recv code paths.
@@ -861,6 +873,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._websocket = None
async def _disconnect(self):
"""Disconnect from Gemini Live API and clean up resources."""
logger.info("Disconnecting from Gemini service")
try:
self._disconnecting = True
@@ -877,6 +890,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
logger.error(f"{self} error disconnecting: {e}")
async def _ws_send(self, message):
"""Send a message to the WebSocket connection."""
# logger.debug(f"Sending message to websocket: {message}")
try:
if self._websocket:
@@ -897,6 +911,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def _receive_task_handler(self):
"""Handle incoming messages from the WebSocket connection."""
async for message in WatchdogAsyncIterator(self._websocket, manager=self.task_manager):
evt = events.parse_server_event(message)
# logger.debug(f"Received event: {message[:500]}")
@@ -925,6 +940,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
#
async def _send_user_audio(self, frame):
"""Send user audio frame to Gemini Live API."""
if self._audio_input_paused:
return
# Send all audio to Gemini
@@ -941,6 +957,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._user_audio_buffer = self._user_audio_buffer[-length:]
async def _send_user_video(self, frame):
"""Send user video frame to Gemini Live API."""
if self._video_input_paused:
return
@@ -954,6 +971,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self.send_client_event(evt)
async def _create_initial_response(self):
"""Create initial response based on context history."""
if not self._api_session_ready:
self._run_llm_when_api_session_ready = True
return
@@ -979,6 +997,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._needs_turn_complete_message = True
async def _create_single_response(self, messages_list):
"""Create a single response from a list of messages."""
# Refactor to combine this logic with same logic in GeminiMultimodalLiveContext
messages = []
for item in messages_list:
@@ -1000,12 +1019,14 @@ class GeminiMultimodalLiveLLMService(LLMService):
parts.append({"text": part.get("text")})
elif part.get("type") == "file_data":
file_data = part.get("file_data", {})
parts.append({
"fileData": {
"mimeType": file_data.get("mime_type"),
"fileUri": file_data.get("file_uri")
parts.append(
{
"fileData": {
"mimeType": file_data.get("mime_type"),
"fileUri": file_data.get("file_uri"),
}
}
})
)
else:
logger.warning(f"Unsupported content type: {str(part)[:80]}")
else:
@@ -1029,6 +1050,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
@traced_gemini_live(operation="llm_tool_result")
async def _tool_result(self, tool_result_message):
"""Send tool result back to the API."""
# For now we're shoving the name into the tool_call_id field, so this
# will work until we revisit that.
id = tool_result_message.get("tool_call_id")
@@ -1054,6 +1076,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
@traced_gemini_live(operation="llm_setup")
async def _handle_evt_setup_complete(self, evt):
"""Handle the setup complete event."""
# If this is our first context frame, run the LLM
self._api_session_ready = True
# Now that we've configured the session, we can run the LLM if we need to.
@@ -1062,6 +1085,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self._create_initial_response()
async def _handle_evt_model_turn(self, evt):
"""Handle the model turn event."""
part = evt.serverContent.modelTurn.parts[0]
if not part:
return
@@ -1103,6 +1127,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
@traced_gemini_live(operation="llm_tool_call")
async def _handle_evt_tool_call(self, evt):
"""Handle tool call events."""
function_calls = evt.toolCall.functionCalls
if not function_calls:
return
@@ -1123,6 +1148,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
@traced_gemini_live(operation="llm_response")
async def _handle_evt_turn_complete(self, evt):
"""Handle the turn complete event."""
self._bot_is_speaking = False
text = self._bot_text_buffer
@@ -1206,6 +1232,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
async def _handle_evt_output_transcription(self, evt):
"""Handle the output transcription event."""
if not evt.serverContent.outputTranscription:
return
@@ -1224,6 +1251,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self.push_frame(TTSTextFrame(text=text))
async def _handle_evt_usage_metadata(self, evt):
"""Handle the usage metadata event."""
if not evt.usageMetadata:
return