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

Merge branch 'main' into groundingMetadata

Browse Source
This commit is contained in:
Pete
2025-07-03 18:47:47 -04:00
committed by GitHub
parent 14c22234bb dbc76389d8
commit c7e758fc36
200 changed files with 10261 additions and 1831 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -59,10 +59,9 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
@@ -75,6 +74,7 @@ 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
from .audio_transcriber import AudioTranscriber
from .file_api import GeminiFileAPI
@@ -242,16 +242,18 @@ 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):
def get_messages_for_initializing_history(self):
"""Get messages formatted for Gemini history initialization.
Returns:
@@ -277,12 +279,16 @@ 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:
@@ -386,7 +392,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
@@ -503,6 +516,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)
@@ -572,6 +586,12 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._search_result_buffer = ""
self._accumulated_grounding_metadata = None
# Initialize the File API client
self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url)
# Initialize the File API client
self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url)
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
@@ -580,6 +600,17 @@ class GeminiMultimodalLiveLLMService(LLMService):
"""
return True
def needs_mcp_alternate_schema(self) -> bool:
"""Check if this LLM service requires alternate MCP schema.
Google/Gemini has stricter JSON schema validation and requires
certain properties to be removed or modified for compatibility.
Returns:
True for Google/Gemini services.
"""
return True
def set_audio_input_paused(self, paused: bool):
"""Set the audio input pause state.
@@ -765,6 +796,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.
@@ -869,6 +901,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
@@ -885,6 +918,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:
@@ -905,6 +939,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]}")
@@ -940,6 +975,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
@@ -956,6 +992,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
@@ -969,6 +1006,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
@@ -994,7 +1032,10 @@ class GeminiMultimodalLiveLLMService(LLMService):
self._needs_turn_complete_message = True
async def _create_single_response(self, messages_list):
# Refactor to combine this logic with same logic in GeminiMultimodalLiveContext
"""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:
role = item.get("role")
@@ -1015,12 +1056,15 @@ 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:
@@ -1044,6 +1088,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")
@@ -1069,6 +1114,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.
@@ -1077,6 +1123,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
@@ -1123,6 +1170,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
@@ -1143,6 +1191,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
@@ -1232,6 +1281,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
)
async def _handle_evt_output_transcription(self, evt):
"""Handle the output transcription event."""
if not evt.serverContent.outputTranscription:
return
@@ -1314,6 +1364,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self.push_frame(search_frame)
async def _handle_evt_usage_metadata(self, evt):
"""Handle the usage metadata event."""
if not evt.usageMetadata:
return