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

Update examples/foundational/26f-gemini-multimodal-live-files-api.py

Browse Source
This commit is contained in:
vipyne
2025-07-01 16:26:52 -05:00
parent 79e51051c7
commit f1c9f5040b
1 changed files with 95 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
examples/foundational/26f-gemini-multimodal-live-files-api.py
View File

@@ -18,67 +18,87 @@ from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.gemini_multimodal_live.gemini import ( from pipecat.services.gemini_multimodal_live.gemini import (
GeminiMultimodalLiveLLMService,
GeminiMultimodalLiveContext, GeminiMultimodalLiveContext,
GeminiMultimodalLiveLLMService,
) )
from pipecat.transports.base_transport import TransportParams from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.services.daily import DailyParams
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True) load_dotenv(override=True)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=False,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=False,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
video_in_enabled=False,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)),
),
}
sample_file_path = ""
async def create_sample_file(): async def create_sample_file():
"""Create a sample text file for testing the File API.""" if sample_file_path:
content = """# Sample Document for Gemini File API Test return sample_file_path
else:
"""Create a sample text file for testing the File API."""
content = """# Sample Document for Gemini File API Test
This is a test document to demonstrate the Gemini File API functionality. This is a test document to demonstrate the Gemini File API functionality.
## Key Information: ## Key Information:
- This document was created for testing purposes - This document was created for testing purposes
- It contains information about AI assistants - It contains information about AI assistants
- The document should be analyzed by Gemini - The document should be analyzed by Gemini
- The secret phrase for the test is "Pineapple Pizza" - The secret phrase for the test is "Pineapple Pizza"
## AI Assistant Capabilities: ## AI Assistant Capabilities:
1. Natural language processing 1. Natural language processing
2. File analysis and understanding 2. File analysis and understanding
3. Context-aware conversations 3. Context-aware conversations
4. Multi-modal interactions 4. Multi-modal interactions
## Conclusion: ## Conclusion:
This document serves as a test case for the Gemini File API integration with Pipecat. This document serves as a test case for the Gemini File API integration with Pipecat.
The AI should be able to reference and discuss the contents of this file. The AI should be able to reference and discuss the contents of this file.
""" """
# Create a temporary file # Create a temporary file
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write(content) f.write(content)
return f.name return f.name
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
logger.info(f"Starting File API bot") logger.info(f"Starting File API bot")
# Create a sample file to upload # Create a sample file to upload
sample_file_path = await create_sample_file() sample_file_path = await create_sample_file()
logger.info(f"Created sample file: {sample_file_path}") logger.info(f"Created sample file: {sample_file_path}")
# 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)),
),
)
system_instruction = """ system_instruction = """
You are a helpful AI assistant with access to a document that has been uploaded for analysis. You are a helpful AI assistant with access to a document that has been uploaded for analysis.
The document contains test information including a secret phrase. You should be able to: The document contains test information.
You should be able to:
- Reference and discuss the contents of the uploaded document - Reference and discuss the contents of the uploaded document
- Answer questions about what's in the document - Answer questions about what's in the document
- Use the information from the document in our conversation - Use the information from the document in our conversation
@@ -100,15 +120,14 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
file_info = None file_info = None
try: try:
file_info = await llm.file_api.upload_file( file_info = await llm.file_api.upload_file(
sample_file_path, sample_file_path, display_name="Sample Test Document"
display_name="Sample Test Document"
) )
logger.info(f"File uploaded successfully: {file_info['file']['name']}") logger.info(f"File uploaded successfully: {file_info['file']['name']}")
# Get file URI and mime type # Get file URI and mime type
file_uri = file_info["file"]["uri"] file_uri = file_info["file"]["uri"]
mime_type = "text/plain" mime_type = "text/plain"
# Create context with file reference # Create context with file reference
context = OpenAILLMContext( context = OpenAILLMContext(
[ [
@@ -117,22 +136,19 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
"content": [ "content": [
{ {
"type": "text", "type": "text",
"text": "Greet the user and let them know you have access to a document they can ask you about. Mention that you can discuss its contents." "text": "Greet the user and let them know you have access to a document they can ask you about. Mention that you can discuss its contents.",
}, },
{ {
"type": "file_data", "type": "file_data",
"file_data": { "file_data": {"mime_type": mime_type, "file_uri": file_uri},
"mime_type": mime_type, },
"file_uri": file_uri ],
}
}
]
} }
] ]
) )
logger.info("File reference added to conversation context") logger.info("File reference added to conversation context")
except Exception as e: except Exception as e:
logger.error(f"Error uploading file: {e}") logger.error(f"Error uploading file: {e}")
# Continue with a basic context if file upload fails # Continue with a basic context if file upload fails
@@ -140,7 +156,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
[ [
{ {
"role": "user", "role": "user",
"content": "Greet the user and explain that there was an issue with file upload, but you're ready to help with other tasks." "content": "Greet the user and explain that there was an issue with file upload, but you're ready to help with other tasks.",
} }
] ]
) )
@@ -149,13 +165,15 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
context_aggregator = llm.create_context_aggregator(context) context_aggregator = llm.create_context_aggregator(context)
# Build the pipeline # Build the pipeline
pipeline = Pipeline([ pipeline = Pipeline(
transport.input(), [
context_aggregator.user(), transport.input(),
llm, context_aggregator.user(),
transport.output(), llm,
context_aggregator.assistant(), transport.output(),
]) context_aggregator.assistant(),
]
)
# Configure the pipeline task # Configure the pipeline task
task = PipelineTask( task = PipelineTask(
@@ -195,7 +213,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
logger.info("Cleaned up uploaded file from Gemini") logger.info("Cleaned up uploaded file from Gemini")
except Exception as e: except Exception as e:
logger.error(f"Error cleaning up file: {e}") logger.error(f"Error cleaning up file: {e}")
# Remove temporary file # Remove temporary file
try: try:
os.unlink(sample_file_path) os.unlink(sample_file_path)
@@ -205,6 +223,20 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
if __name__ == "__main__": if __name__ == "__main__":
from run import main from pipecat.examples.run import main
main() upload_example_file = input("""
Please pass in a TEXT filepath to test upload.
NOTE: Files are stored on Google's servers for 48 hours.
Press Enter to use a default test file.
text filepath : """)
if upload_example_file:
print(f"Uploading file: {upload_example_file}")
sample_file_path = upload_example_file.strip()
else:
print(f"Using default file")
main(run_example, transport_params=transport_params)