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

New lint rules and remove unused example file

Browse Source
This commit is contained in:
vipyne
2025-07-01 16:25:59 -05:00
parent a63d0da528
commit 79e51051c7
5 changed files with 125 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py
View File

@@ -1,164 +0,0 @@
import argparse
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
from pipecat.frames.frames import Frame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
from pipecat.services.google.frames import LLMSearchResponseFrame
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
SYSTEM_INSTRUCTION = """
You are a helpful AI assistant that actively uses Google Search to provide up-to-date, accurate information.
IMPORTANT: For ANY question about current events, news, recent developments, real-time information, or anything that might have changed recently, you MUST use the google_search tool to get the latest information.
You should use Google Search for:
- Current news and events
- Recent developments in any field
- Today's weather, stock prices, or other real-time data
- Any question that starts with "what's happening", "latest", "recent", "current", "today", etc.
- When you're not certain about recent information
Always be proactive about using search when the user asks about anything that could benefit from real-time information.
Your output will be converted to audio so don't include special characters in your answers.
Respond to what the user said in a creative and helpful way, always using search for current information.
"""
class GroundingMetadataProcessor(FrameProcessor):
"""Processor to capture and display grounding metadata from Gemini Live API."""
def __init__(self):
super().__init__()
self._grounding_count = 0
async def process_frame(self, frame: Frame, direction: FrameDirection):
# Always call super().process_frame first
await super().process_frame(frame, direction)
# Only log important frame types, not every audio frame
if hasattr(frame, '__class__'):
frame_type = frame.__class__.__name__
if frame_type in ['LLMTextFrame', 'TTSTextFrame', 'LLMFullResponseStartFrame', 'LLMFullResponseEndFrame']:
logger.debug(f"GroundingProcessor received: {frame_type}")
if isinstance(frame, LLMSearchResponseFrame):
self._grounding_count += 1
logger.info(f"\n🔍 GROUNDING METADATA RECEIVED #{self._grounding_count}")
logger.info(f"📝 Search Result Text: {frame.search_result[:200]}...")
if frame.rendered_content:
logger.info(f"🔗 Rendered Content: {frame.rendered_content}")
if frame.origins:
logger.info(f"📍 Number of Origins: {len(frame.origins)}")
for i, origin in enumerate(frame.origins):
logger.info(f" Origin {i+1}: {origin.site_title} - {origin.site_uri}")
if origin.results:
logger.info(f" Results: {len(origin.results)} items")
# Always push the frame downstream
await self.push_frame(frame, direction)
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
logger.info(f"Starting Gemini Live Grounding Test Bot")
# Initialize the SmallWebRTCTransport with the connection
transport = SmallWebRTCTransport(
webrtc_connection=webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=False,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)),
),
)
# Create tools using ToolsSchema with custom tools for Gemini
tools = ToolsSchema(
standard_tools=[], # No standard function declarations needed
custom_tools={
AdapterType.GEMINI: [
{"google_search": {}},
{"code_execution": {}}
]
}
)
llm = GeminiMultimodalLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
system_instruction=SYSTEM_INSTRUCTION,
voice_id="Charon", # Aoede, Charon, Fenrir, Kore, Puck
transcribe_user_audio=True,
tools=tools,
)
# Create a processor to capture grounding metadata
grounding_processor = GroundingMetadataProcessor()
messages = [
{
"role": "user",
"content": 'Please introduce yourself and let me know that you can help with current information by searching the web. Ask me what current information I\'d like to know about.',
},
]
# Set up conversation context and management
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
llm,
grounding_processor, # Add our grounding processor here
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(pipeline)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__":
from run import main
main()

2
src/pipecat/services/gemini_multimodal_live/__init__.py
View File

@@ -1,2 +1,2 @@
from .gemini import GeminiMultimodalLiveLLMService
from .file_api import GeminiFileAPI
from .gemini import GeminiMultimodalLiveLLMService

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

@@ -45,11 +45,12 @@ class ContentPart(BaseModel):
text: Optional[str] = Field(default=None, validate_default=False)
inlineData: Optional[MediaChunk] = Field(default=None, validate_default=False)
fileData: Optional['FileData'] = Field(default=None, validate_default=False)
fileData: Optional["FileData"] = Field(default=None, validate_default=False)
class FileData(BaseModel):
"""Represents a file reference in the Gemini File API."""
mimeType: str
fileUri: str
@@ -255,22 +256,26 @@ class Config(BaseModel):
class SearchEntryPoint(BaseModel):
"""Represents the search entry point with rendered content for search suggestions."""
renderedContent: Optional[str] = None
class WebSource(BaseModel):
"""Represents a web source from grounding chunks."""
uri: Optional[str] = None
title: Optional[str] = None
class GroundingChunk(BaseModel):
"""Represents a grounding chunk containing web source information."""
web: Optional[WebSource] = None
class GroundingSegment(BaseModel):
"""Represents a segment of text that is grounded."""
startIndex: Optional[int] = None
endIndex: Optional[int] = None
text: Optional[str] = None
@@ -278,6 +283,7 @@ class GroundingSegment(BaseModel):
class GroundingSupport(BaseModel):
"""Represents support information for grounded text segments."""
segment: Optional[GroundingSegment] = None
groundingChunkIndices: Optional[List[int]] = None
confidenceScores: Optional[List[float]] = None
@@ -285,6 +291,7 @@ class GroundingSupport(BaseModel):
class GroundingMetadata(BaseModel):
"""Represents grounding metadata from Google Search."""
searchEntryPoint: Optional[SearchEntryPoint] = None
groundingChunks: Optional[List[GroundingChunk]] = None
groundingSupports: Optional[List[GroundingSupport]] = None
@@ -485,15 +492,15 @@ def parse_server_event(message_str):
ServerEvent instance if parsing succeeds, None otherwise.
"""
from loguru import logger # Import logger locally to avoid scoping issues
try:
evt_dict = json.loads(message_str)
# Only log grounding metadata detection if truly needed for debugging
# In production, this could be removed entirely or moved to TRACE level
if 'serverContent' in evt_dict:
server_content = evt_dict['serverContent']
if 'groundingMetadata' in server_content:
if "serverContent" in evt_dict:
server_content = evt_dict["serverContent"]
if "groundingMetadata" in server_content:
# Consider removing this log entirely for production
pass

125
src/pipecat/services/gemini_multimodal_live/file_api.py
View File

@@ -4,26 +4,29 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import aiohttp
import mimetypes
from typing import Dict, Any, Optional
from typing import Any, Dict, Optional
import aiohttp
from loguru import logger
class GeminiFileAPI:
"""Client for the Gemini File API.
This class provides methods for uploading, fetching, listing, and deleting files
through Google's Gemini File API.
Files uploaded through this API remain available for 48 hours and can be referenced
in calls to the Gemini generative models. Maximum file size is 2GB, with total
project storage limited to 20GB.
"""
def __init__(self, api_key: str, base_url: str = "https://generativelanguage.googleapis.com/v1beta/files"):
def __init__(
self, api_key: str, base_url: str = "https://generativelanguage.googleapis.com/v1beta/files"
):
"""Initialize the Gemini File API client.
Args:
api_key: Google AI API key
base_url: Base URL for the Gemini File API (default is the v1beta endpoint)
@@ -32,160 +35,148 @@ class GeminiFileAPI:
self.base_url = base_url
# Upload URL uses the /upload/ path
self.upload_base_url = "https://generativelanguage.googleapis.com/upload/v1beta/files"
async def upload_file(self, file_path: str, display_name: Optional[str] = None) -> Dict[str, Any]:
async def upload_file(
self, file_path: str, display_name: Optional[str] = None
) -> Dict[str, Any]:
"""Upload a file to the Gemini File API using the correct resumable upload protocol.
Args:
file_path: Path to the file to upload
display_name: Optional display name for the file
Returns:
File metadata including uri, name, and display_name
"""
logger.info(f"Uploading file: {file_path}")
async with aiohttp.ClientSession() as session:
# Determine the file's MIME type
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
mime_type = "application/octet-stream"
# Read the file
with open(file_path, "rb") as f:
file_data = f.read()
# Create the metadata payload
metadata = {}
if display_name:
metadata = {"file": {"display_name": display_name}}
# Step 1: Initial resumable request to get upload URL
headers = {
"X-Goog-Upload-Protocol": "resumable",
"X-Goog-Upload-Command": "start",
"X-Goog-Upload-Header-Content-Length": str(len(file_data)),
"X-Goog-Upload-Header-Content-Type": mime_type,
"Content-Type": "application/json"
"Content-Type": "application/json",
}
logger.debug(f"Step 1: Getting upload URL from {self.upload_base_url}")
async with session.post(
f"{self.upload_base_url}?key={self.api_key}",
headers=headers,
json=metadata
f"{self.upload_base_url}?key={self.api_key}", headers=headers, json=metadata
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error initiating file upload: {error_text}")
raise Exception(f"Failed to initiate upload: {response.status} - {error_text}")
# Get the upload URL from the response header
upload_url = response.headers.get("X-Goog-Upload-URL")
if not upload_url:
logger.error(f"Response headers: {dict(response.headers)}")
raise Exception("No upload URL in response headers")
logger.debug(f"Got upload URL: {upload_url}")
# Step 2: Upload the actual file data
upload_headers = {
"Content-Length": str(len(file_data)),
"X-Goog-Upload-Offset": "0",
"X-Goog-Upload-Command": "upload, finalize"
"X-Goog-Upload-Command": "upload, finalize",
}
logger.debug(f"Step 2: Uploading file data to {upload_url}")
async with session.post(
upload_url,
headers=upload_headers,
data=file_data
) as response:
async with session.post(upload_url, headers=upload_headers, data=file_data) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error uploading file data: {error_text}")
raise Exception(f"Failed to upload file: {response.status} - {error_text}")
file_info = await response.json()
logger.info(f"File uploaded successfully: {file_info.get('file', {}).get('name')}")
return file_info
async def get_file(self, name: str) -> Dict[str, Any]:
"""Get metadata for a file.
Args:
name: File name (or full path)
Returns:
File metadata
"""
# Extract just the name part if a full path is provided
if '/' in name:
name = name.split('/')[-1]
if "/" in name:
name = name.split("/")[-1]
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/{name}?key={self.api_key}"
) as response:
async with session.get(f"{self.base_url}/{name}?key={self.api_key}") as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error getting file metadata: {error_text}")
raise Exception(f"Failed to get file metadata: {response.status}")
file_info = await response.json()
return file_info
async def list_files(self, page_size: int = 10, page_token: Optional[str] = None) -> Dict[str, Any]:
async def list_files(
self, page_size: int = 10, page_token: Optional[str] = None
) -> Dict[str, Any]:
"""List uploaded files.
Args:
page_size: Number of files to return per page
page_token: Token for pagination
Returns:
List of files and next page token if available
"""
params = {
"key": self.api_key,
"pageSize": page_size
}
params = {"key": self.api_key, "pageSize": page_size}
if page_token:
params["pageToken"] = page_token
async with aiohttp.ClientSession() as session:
async with session.get(
self.base_url,
params=params
) as response:
async with session.get(self.base_url, params=params) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error listing files: {error_text}")
raise Exception(f"Failed to list files: {response.status}")
result = await response.json()
return result
async def delete_file(self, name: str) -> bool:
"""Delete a file.
Args:
name: File name (or full path)
Returns:
True if deleted successfully
"""
# Extract just the name part if a full path is provided
if '/' in name:
name = name.split('/')[-1]
if "/" in name:
name = name.split("/")[-1]
async with aiohttp.ClientSession() as session:
async with session.delete(
f"{self.base_url}/{name}?key={self.api_key}"
) as response:
async with session.delete(f"{self.base_url}/{name}?key={self.api_key}") as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error deleting file: {error_text}")
raise Exception(f"Failed to delete file: {response.status}")
return True
return True

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

@@ -59,11 +59,8 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
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,
OpenAIUserContextAggregator,
@@ -75,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 .audio_transcriber import AudioTranscriber
from .file_api import GeminiFileAPI
try:
@@ -226,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
@@ -238,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.
@@ -273,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:
@@ -468,7 +468,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Overriding the default adapter to use the Gemini one.
adapter_class = GeminiLLMAdapter
def __init__(
self,
*,
@@ -560,7 +560,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)
@@ -1015,12 +1015,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:
@@ -1167,7 +1169,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Process grounding metadata if we have accumulated any
if self._accumulated_grounding_metadata:
logger.debug("Processing grounding metadata...")
await self._process_grounding_metadata(self._accumulated_grounding_metadata, self._search_result_buffer)
await self._process_grounding_metadata(
self._accumulated_grounding_metadata, self._search_result_buffer
)
else:
logger.debug("No grounding metadata to process")
@@ -1285,17 +1289,23 @@ class GeminiMultimodalLiveLLMService(LLMService):
async def _handle_evt_grounding_metadata(self, evt):
"""Handle dedicated grounding metadata events."""
logger.debug("Received dedicated grounding metadata event.")
if evt.serverContent and evt.serverContent.groundingMetadata:
grounding_metadata = evt.serverContent.groundingMetadata
logger.debug(f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports")
logger.debug(
f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports"
)
# Process the grounding metadata immediately
await self._process_grounding_metadata(grounding_metadata, self._search_result_buffer)
async def _process_grounding_metadata(self, grounding_metadata: events.GroundingMetadata, search_result: str = ""):
async def _process_grounding_metadata(
self, grounding_metadata: events.GroundingMetadata, search_result: str = ""
):
"""Process grounding metadata and emit LLMSearchResponseFrame."""
logger.debug(f"Processing grounding metadata. Search result text length: {len(search_result)}")
logger.debug(
f"Processing grounding metadata. Search result text length: {len(search_result)}"
)
if not grounding_metadata:
logger.warning("No grounding metadata provided to _process_grounding_metadata")
return
@@ -1304,49 +1314,47 @@ class GeminiMultimodalLiveLLMService(LLMService):
# Extract rendered content for search suggestions
rendered_content = None
if grounding_metadata.searchEntryPoint and grounding_metadata.searchEntryPoint.renderedContent:
if (
grounding_metadata.searchEntryPoint
and grounding_metadata.searchEntryPoint.renderedContent
):
rendered_content = grounding_metadata.searchEntryPoint.renderedContent
# Convert grounding chunks and supports to LLMSearchOrigin format
origins = []
if grounding_metadata.groundingChunks and grounding_metadata.groundingSupports:
# Create a mapping of chunk indices to origins
chunk_to_origin = {}
for index, chunk in enumerate(grounding_metadata.groundingChunks):
if chunk.web:
origin = LLMSearchOrigin(
site_uri=chunk.web.uri,
site_title=chunk.web.title,
results=[]
site_uri=chunk.web.uri, site_title=chunk.web.title, results=[]
)
chunk_to_origin[index] = origin
origins.append(origin)
# Add grounding support results to the appropriate origins
for support in grounding_metadata.groundingSupports:
if support.segment and support.groundingChunkIndices:
text = support.segment.text or ""
confidence_scores = support.confidenceScores or []
# Add this result to all origins referenced by this support
for chunk_index in support.groundingChunkIndices:
if chunk_index in chunk_to_origin:
result = LLMSearchResult(
text=text,
confidence=confidence_scores
)
result = LLMSearchResult(text=text, confidence=confidence_scores)
chunk_to_origin[chunk_index].results.append(result)
# Create and push the search response frame
search_frame = LLMSearchResponseFrame(
search_result=search_result,
origins=origins,
rendered_content=rendered_content
search_result=search_result, origins=origins, rendered_content=rendered_content
)
logger.debug(
f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}"
)
logger.debug(f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}")
await self.push_frame(search_frame)
def create_context_aggregator(