diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index f9fd2377a..cf3d891b2 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -7,26 +7,35 @@ import asyncio import os import sys +from typing import List import aiohttp from dotenv import load_dotenv from loguru import logger from runner import configure -from pipecat.frames.frames import Frame, LLMMessagesFrame, TextFrame +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import ( + EndFrame, + Frame, + LLMMessagesFrame, + TranscriptionFrame, + TranscriptionMessage, + TranscriptionUpdateFrame, +) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.aggregators.llm_response import LLMFullResponseAggregator -from pipecat.processors.aggregators.sentence import SentenceAggregator +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.azure import AzureTTSService +from pipecat.processors.transcript_processor import TranscriptProcessor +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import ( DailyParams, - DailyTranscriptionSettings, DailyTransport, - DailyTransportMessageFrame, ) load_dotenv(override=True) @@ -44,18 +53,34 @@ It also isn't saving what the user or bot says into the context object for use i # We need to use a custom service here to yield LLM frames without saving # any context class TranslationProcessor(FrameProcessor): - def __init__(self, language): + """A processor that translates text frames from a source language to a target language.""" + + def __init__(self, in_language, out_language): + """Initialize the TranslationProcessor with source and target languages. + + Args: + in_language (str): The language of the input text. + out_language (str): The language to translate the text into. + """ super().__init__() - self._language = language + self._out_language = out_language + self._in_language = in_language async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process a frame and translate text frames. + + Args: + frame (Frame): The frame to process. + direction (FrameDirection): The direction of the frame. + """ await super().process_frame(frame, direction) - if isinstance(frame, TextFrame): + if isinstance(frame, TranscriptionFrame): + logger.debug(f"Translating {self._in_language}: {frame.text} to {self._out_language}") context = [ { "role": "system", - "content": f"You will be provided with a sentence in English, and your task is to translate it into {self._language}.", + "content": f"You will be provided with a sentence in {self._in_language}, and your task is to only translate it into {self._out_language}.", }, {"role": "user", "content": frame.text}, ] @@ -64,28 +89,45 @@ class TranslationProcessor(FrameProcessor): await self.push_frame(frame) -class TranslationSubtitles(FrameProcessor): - def __init__(self, language): - super().__init__() - self._language = language +class TranscriptHandler: + """Simple handler to demonstrate transcript processing. - # - # This doesn't do anything unless the receiver recognizes the message being - # sent. For example, in this case, we are sending a message to the transport - # so an application running at the other end of the transport could display - # subtitles. - # - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) + Maintains a list of conversation messages and logs them with timestamps. + """ - if isinstance(frame, TextFrame): - message = {"language": self._language, "text": frame.text} - await self.push_frame(DailyTransportMessageFrame(message)) + def __init__(self, in_language="English", out_language="Spanish"): + """Initialize the TranscriptHandler with an empty list of messages.""" + self.messages: List[TranscriptionMessage] = [] + self.in_language = in_language + self.out_language = out_language - await self.push_frame(frame) + 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 + """ + self.messages.extend(frame.messages) + + # Log the new messages + logger.info("New transcript messages:") + for msg in frame.messages: + timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" + message = { + "event": "translation", + "timestamp": msg.timestamp, + "role": msg.role, + "language": self.out_language if msg.role == "assistant" else self.in_language, + "text": msg.content, + } + logger.info(f"{timestamp}{msg.role}: {msg.content}") async def main(): + """Main function to set up and run the translation chatbot pipeline.""" async with aiohttp.ClientSession() as session: (room_url, token) = await configure(session) @@ -95,31 +137,60 @@ async def main(): "Translator", DailyParams( audio_out_enabled=True, - transcription_enabled=True, - transcription_settings=DailyTranscriptionSettings(extra={"interim_results": False}), + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, ), ) - tts = AzureTTSService( - api_key=os.getenv("AZURE_SPEECH_API_KEY"), - region=os.getenv("AZURE_SPEECH_REGION"), - voice="es-ES-AlvaroNeural", + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="34dbb662-8e98-413c-a1ef-1a3407675fe7", # Spanish Narrator Man + model="sonic-multilingual", ) + in_language = "English" + out_language = "Spanish" + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + context = OpenAILLMContext() + context_aggregator = llm.create_context_aggregator(context) - sa = SentenceAggregator() - tp = TranslationProcessor("Spanish") - lfra = LLMFullResponseAggregator() - ts = TranslationSubtitles("spanish") + tp = TranslationProcessor(in_language=in_language, out_language=out_language) - pipeline = Pipeline([transport.input(), sa, tp, llm, lfra, ts, tts, transport.output()]) + transcript = TranscriptProcessor() + transcript_handler = TranscriptHandler(in_language=in_language, out_language=out_language) + + # 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) + + pipeline = Pipeline( + [ + transport.input(), + stt, + transcript.user(), # User transcripts + tp, + llm, + tts, + transport.output(), + context_aggregator.assistant(), + transcript.assistant(), # Assistant transcripts + ] + ) task = PipelineTask(pipeline) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - await transport.capture_participant_transcription(participant["id"]) + logger.info("First participant joined") + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.queue_frame(EndFrame()) runner = PipelineRunner() diff --git a/examples/translation-chatbot/server.py b/examples/translation-chatbot/server.py index b19242ae2..1dd2ac966 100644 --- a/examples/translation-chatbot/server.py +++ b/examples/translation-chatbot/server.py @@ -121,7 +121,7 @@ if __name__ == "__main__": default_host = os.getenv("HOST", "0.0.0.0") default_port = int(os.getenv("FAST_API_PORT", "7860")) - parser = argparse.ArgumentParser(description="Daily Storyteller FastAPI server") + parser = argparse.ArgumentParser(description="Daily Translator FastAPI server") parser.add_argument("--host", type=str, default=default_host, help="Host address") parser.add_argument("--port", type=int, default=default_port, help="Port number") parser.add_argument("--reload", action="store_true", help="Reload code on change")