diff --git a/example.db b/example.db new file mode 100644 index 000000000..8e51d4adf Binary files /dev/null and b/example.db differ diff --git a/examples/foundational/28d-transcription-processor-gemini-sqlite.py b/examples/foundational/28d-transcription-processor-gemini-sqlite.py new file mode 100644 index 000000000..f25c3cbe1 --- /dev/null +++ b/examples/foundational/28d-transcription-processor-gemini-sqlite.py @@ -0,0 +1,200 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sqlite3 +import sys +from typing import List, Optional + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import ( + CancelFrame, + TranscriptionMessage, + TranscriptionUpdateFrame, +) +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.transcript_processor import TranscriptProcessor +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.google import GoogleLLMService +from pipecat.services.openai import OpenAILLMContext +from pipecat.transports.services.daily import DailyParams, DailyTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +con = sqlite3.connect("example.db") +db = con.cursor() + +table = db.execute("SELECT name FROM sqlite_master WHERE name='messages'") +if not (table.fetchone()): + db.execute( + "CREATE TABLE messages(role TEXT, content TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP )" + ) + + +class TranscriptHandler: + """Handles real-time transcript processing and output. + + Maintains a list of conversation messages and outputs them either to a log + or to a file as they are received. Each message includes its timestamp and role. + + Attributes: + messages: List of all processed transcript messages + output_file: Optional path to file where transcript is saved. If None, outputs to log only. + """ + + def __init__(self, output_file: Optional[str] = None): + """Initialize handler with optional file output. + + Args: + output_file: Path to output file. If None, outputs to log only. + """ + self.messages: List[TranscriptionMessage] = [] + self.output_file: Optional[str] = output_file + logger.debug( + f"TranscriptHandler initialized {'with output_file=' + output_file if output_file else 'with log output only'}" + ) + + async def save_message(self, message: TranscriptionMessage): + """Save a single transcript message. + + Outputs the message to the log and optionally to a file. + + Args: + message: The message to save + """ + timestamp_f = f"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp_f}{message.role}: {message.content}" + + # Always log the message + logger.info(f"Transcript: {line}") + + # Write to database + db.execute( + "INSERT INTO messages VALUES (?, ?, ?)", + (message.role, message.content, message.timestamp), + ) + con.commit() + + async def on_transcript_update( + self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame + ): + """Handle new transcript messages. + + Args: + processor: The TranscriptProcessor that emitted the update + frame: TranscriptionUpdateFrame containing new messages + """ + logger.debug(f"Received transcript update with {len(frame.messages)} new messages") + + for msg in frame.messages: + self.messages.append(msg) + await self.save_message(msg) + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + None, + "Respond bot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = GoogleLLMService( + model="models/gemini-2.0-flash-exp", + # model="gemini-exp-1114", + api_key=os.getenv("GOOGLE_API_KEY"), + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. 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, helpful, and brief way.", + }, + {"role": "user", "content": "Say hello."}, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + # Create transcript processor and handler + transcript = TranscriptProcessor() + transcript_handler = TranscriptHandler() # Output to log only + # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + transcript.user(), # User transcripts + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + transcript.assistant(), # Assistant transcripts + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + # Register event handler for transcript updates + @transcript.event_handler("on_transcript_update") + async def on_transcript_update(processor, frame): + await transcript_handler.on_transcript_update(processor, frame) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + # Stop the pipeline immediately when the participant leaves + await task.queue_frame(CancelFrame()) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main())