From bc98c2e36c897dc0eafaedbeec5600236ec02753 Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Wed, 29 Jan 2025 19:12:15 +0000 Subject: [PATCH] added sqlite storage example --- example.db | Bin 0 -> 8192 bytes ...d-transcription-processor-gemini-sqlite.py | 200 ++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 example.db create mode 100644 examples/foundational/28d-transcription-processor-gemini-sqlite.py diff --git a/example.db b/example.db new file mode 100644 index 0000000000000000000000000000000000000000..8e51d4adfc7589f478c2f61785ade9ce95f72c25 GIT binary patch literal 8192 zcmeI1OK;Oa5XYU?r9ATB5^RP~v~@PIiQ^64b*uDN zi3@xK#24bqiBH0vaUP1Okr3hryNcuNW_R{KznMLZ_UL8X7ZkcI8V~`da$Ct{l&1iS zqU7;ez-QJn_(Pp_@SXI3*Yb*X@h*q6GKKdFeq@sbBmqf45|9KW0ZBj-kOU+FNk9^i z1pZqB7n$WdD=V4HT|rI*I-s1B9_6$5!q%bPbZl^%&)PQ3Mju2hpy1fAoKlaVK?XzE#$`@x--c~_r+M6VVC(qs&^~b9B!fpzbN>ZAOltF7wxATg6h0R2=D+4Y z`#t+9dw_q*CJ9Ial7J*22}lBxfFvLZNCN*RfyX(uWbP8qeJmkDvByF> zf{>ok2>6is0oM%OSYOl2Yevm6^oFtCunf(pmmleRL)TkbwRpT6a~f@t5JoHpL4yE` z7Qn8LRZ4`Vi68c$OI-0JZA#b7dUayz^)0nDm@7H)@OrYt#x`AoHVFjeVg!8>_DIAS z_J=HXJ^U6#u(V9m(saw3*xtOU7B_w^6W+2uO)E1EO|R>d)!iFQrOU;+G{>XB;}F$R z4w28%S?IB#LqqUlx(Y*ro+nQ8zK=8Ts&&jWq~}*vS`Dj)W0lE1jq7T$`jg7@D7mXm zJf(xH8gXq!Tdz&m@@3SxGS}Dw7b74d67qncjR3L5NtKC#>A=Kp$M1GAKv+oj{jkG< zb-|4&5&jvSm6&P4W*s-JO$2XTQ;YSn;BmBQfGDOR>xb+-t=Oz;RkJ=>ytAa1?786Q zu3$q4%3LYhqNTs0FNTM?n(Ng~+1<{nQ=`%^eB0?<8HJx}P^e4P8fV)0& z!ZX5=Y0L#+$_+T2QkS4+REwu(K>8HoAtb>b2@ew>P0uhj%bJF4bK%dD(rOrGt*TGi bS2Akp-dyS)9y^xkh-iA(az(3J(-ZpxU)f&k literal 0 HcmV?d00001 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())