From 31c77d8e35b5e1dc6d4574181d80dd4a14f34597 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 20 Jan 2025 14:20:39 -0500 Subject: [PATCH] Update examples for the updated TranscriptProcessor --- .../28a-transcription-processor-openai.py | 67 +++++++++++++++---- .../28b-transcript-processor-anthropic.py | 67 +++++++++++++++---- .../28c-transcription-processor-gemini.py | 67 +++++++++++++++---- 3 files changed, 165 insertions(+), 36 deletions(-) diff --git a/examples/foundational/28a-transcription-processor-openai.py b/examples/foundational/28a-transcription-processor-openai.py index a83f57aa1..8b5b943bf 100644 --- a/examples/foundational/28a-transcription-processor-openai.py +++ b/examples/foundational/28a-transcription-processor-openai.py @@ -7,7 +7,7 @@ import asyncio import os import sys -from typing import List +from typing import List, Optional import aiohttp from dotenv import load_dotenv @@ -15,7 +15,12 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, TranscriptionMessage, TranscriptionUpdateFrame +from pipecat.frames.frames import ( + EndFrame, + StartInterruptionFrame, + TranscriptionMessage, + TranscriptionUpdateFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -33,13 +38,49 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptHandler: - """Simple handler to demonstrate transcript processing. + """Handles real-time transcript processing and output. - Maintains a list of conversation messages and logs them with timestamps. + 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): + 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"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}{message.role}: {message.content}" + + # Always log the message + logger.info(f"Transcript: {line}") + + # Optionally write to file + if self.output_file: + try: + with open(self.output_file, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception as e: + logger.error(f"Error saving transcript message to file: {e}") async def on_transcript_update( self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame @@ -50,13 +91,11 @@ class TranscriptHandler: processor: The TranscriptProcessor that emitted the update frame: TranscriptionUpdateFrame containing new messages """ - self.messages.extend(frame.messages) + logger.debug(f"Received transcript update with {len(frame.messages)} new messages") - # Log the new messages - logger.info("New transcript messages:") for msg in frame.messages: - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - logger.info(f"{timestamp}{msg.role}: {msg.content}") + self.messages.append(msg) + await self.save_message(msg) async def main(): @@ -99,7 +138,8 @@ async def main(): # Create transcript processor and handler transcript = TranscriptProcessor() - transcript_handler = TranscriptHandler() + transcript_handler = TranscriptHandler() # Output to log only + # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log pipeline = Pipeline( [ @@ -110,8 +150,8 @@ async def main(): llm, # LLM tts, # TTS transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses transcript.assistant(), # Assistant transcripts + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -130,6 +170,9 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): + # Interrupt the TTS to stop the TTS generation + await task.queue_frame(StartInterruptionFrame()) + # Stop the gracefully stop the pipeline await task.queue_frame(EndFrame()) runner = PipelineRunner() diff --git a/examples/foundational/28b-transcript-processor-anthropic.py b/examples/foundational/28b-transcript-processor-anthropic.py index 9a75dfb72..18074e430 100644 --- a/examples/foundational/28b-transcript-processor-anthropic.py +++ b/examples/foundational/28b-transcript-processor-anthropic.py @@ -7,7 +7,7 @@ import asyncio import os import sys -from typing import List +from typing import List, Optional import aiohttp from dotenv import load_dotenv @@ -15,7 +15,12 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, TranscriptionMessage, TranscriptionUpdateFrame +from pipecat.frames.frames import ( + EndFrame, + StartInterruptionFrame, + TranscriptionMessage, + TranscriptionUpdateFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -33,13 +38,49 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptHandler: - """Simple handler to demonstrate transcript processing. + """Handles real-time transcript processing and output. - Maintains a list of conversation messages and logs them with timestamps. + 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): + 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"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}{message.role}: {message.content}" + + # Always log the message + logger.info(f"Transcript: {line}") + + # Optionally write to file + if self.output_file: + try: + with open(self.output_file, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception as e: + logger.error(f"Error saving transcript message to file: {e}") async def on_transcript_update( self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame @@ -50,13 +91,11 @@ class TranscriptHandler: processor: The TranscriptProcessor that emitted the update frame: TranscriptionUpdateFrame containing new messages """ - self.messages.extend(frame.messages) + logger.debug(f"Received transcript update with {len(frame.messages)} new messages") - # Log the new messages - logger.info("New transcript messages:") for msg in frame.messages: - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - logger.info(f"{timestamp}{msg.role}: {msg.content}") + self.messages.append(msg) + await self.save_message(msg) async def main(): @@ -99,7 +138,8 @@ async def main(): # Create transcript processor and handler transcript = TranscriptProcessor() - transcript_handler = TranscriptHandler() + transcript_handler = TranscriptHandler() # Output to log only + # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log pipeline = Pipeline( [ @@ -110,8 +150,8 @@ async def main(): llm, # LLM tts, # TTS transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses transcript.assistant(), # Assistant transcripts + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -130,6 +170,9 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): + # Interrupt the TTS to stop the TTS generation + await task.queue_frame(StartInterruptionFrame()) + # Stop the gracefully stop the pipeline await task.queue_frame(EndFrame()) runner = PipelineRunner() diff --git a/examples/foundational/28c-transcription-processor-gemini.py b/examples/foundational/28c-transcription-processor-gemini.py index a04925936..efa400372 100644 --- a/examples/foundational/28c-transcription-processor-gemini.py +++ b/examples/foundational/28c-transcription-processor-gemini.py @@ -7,7 +7,7 @@ import asyncio import os import sys -from typing import List +from typing import List, Optional import aiohttp from dotenv import load_dotenv @@ -15,7 +15,12 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, TranscriptionMessage, TranscriptionUpdateFrame +from pipecat.frames.frames import ( + EndFrame, + StartInterruptionFrame, + TranscriptionMessage, + TranscriptionUpdateFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -34,13 +39,49 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptHandler: - """Simple handler to demonstrate transcript processing. + """Handles real-time transcript processing and output. - Maintains a list of conversation messages and logs them with timestamps. + 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): + 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"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}{message.role}: {message.content}" + + # Always log the message + logger.info(f"Transcript: {line}") + + # Optionally write to file + if self.output_file: + try: + with open(self.output_file, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception as e: + logger.error(f"Error saving transcript message to file: {e}") async def on_transcript_update( self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame @@ -51,13 +92,11 @@ class TranscriptHandler: processor: The TranscriptProcessor that emitted the update frame: TranscriptionUpdateFrame containing new messages """ - self.messages.extend(frame.messages) + logger.debug(f"Received transcript update with {len(frame.messages)} new messages") - # Log the new messages - logger.info("New transcript messages:") for msg in frame.messages: - timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" - logger.info(f"{timestamp}{msg.role}: {msg.content}") + self.messages.append(msg) + await self.save_message(msg) async def main(): @@ -102,7 +141,8 @@ async def main(): # Create transcript processor and handler transcript = TranscriptProcessor() - transcript_handler = TranscriptHandler() + transcript_handler = TranscriptHandler() # Output to log only + # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log pipeline = Pipeline( [ @@ -113,8 +153,8 @@ async def main(): llm, # LLM tts, # TTS transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses transcript.assistant(), # Assistant transcripts + context_aggregator.assistant(), # Assistant spoken responses ] ) @@ -140,6 +180,9 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): + # Interrupt the TTS to stop the TTS generation + await task.queue_frame(StartInterruptionFrame()) + # Stop the gracefully stop the pipeline await task.queue_frame(EndFrame()) runner = PipelineRunner()