diff --git a/example.db b/example.db deleted file mode 100644 index 8e51d4adf..000000000 Binary files a/example.db and /dev/null differ diff --git a/examples/foundational/28c-transcription-processor-gemini.py b/examples/foundational/28c-transcription-processor-gemini.py index 377274ae1..c93dfb302 100644 --- a/examples/foundational/28c-transcription-processor-gemini.py +++ b/examples/foundational/28c-transcription-processor-gemini.py @@ -6,6 +6,7 @@ import asyncio import os +import sqlite3 import sys from typing import List, Optional @@ -48,22 +49,33 @@ class TranscriptHandler: 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. + def __init__(self, output_file: Optional[str] = None, output_db: Optional[str] = None): + """Initialize handler with optional file or database 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 + self.output_db: Optional[str] = output_db + + if self.output_db: + self.con = sqlite3.connect("example.db") + self.db = self.con.cursor() + + table = self.db.execute("SELECT name FROM sqlite_master WHERE name='messages'") + if not (table.fetchone()): + self.db.execute( + "CREATE TABLE messages(role TEXT, content TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP )" + ) logger.debug( - f"TranscriptHandler initialized {'with output_file=' + output_file if output_file else 'with log output only'}" + f"TranscriptHandler initialized; output file: {output_file}, output DB: {output_db}" ) async def save_message(self, message: TranscriptionMessage): """Save a single transcript message. - Outputs the message to the log and optionally to a file. + Outputs the message to the log and optionally to a SQLite database or file. Args: message: The message to save @@ -82,6 +94,14 @@ class TranscriptHandler: except Exception as e: logger.error(f"Error saving transcript message to file: {e}") + # and/or to a SQLite database + if self.output_db: + self.db.execute( + "INSERT INTO messages VALUES (?, ?, ?)", + (message.role, message.content, message.timestamp), + ) + self.con.commit() + async def on_transcript_update( self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame ): @@ -140,7 +160,7 @@ async def main(): # Create transcript processor and handler transcript = TranscriptProcessor() - transcript_handler = TranscriptHandler() # Output to log only + transcript_handler = TranscriptHandler(output_db="example.db") # Output to log only # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log pipeline = Pipeline( diff --git a/examples/foundational/28d-transcription-processor-gemini-sqlite.py b/examples/foundational/28d-transcription-processor-gemini-sqlite.py deleted file mode 100644 index f25c3cbe1..000000000 --- a/examples/foundational/28d-transcription-processor-gemini-sqlite.py +++ /dev/null @@ -1,200 +0,0 @@ -# -# 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())