diff --git a/examples/foundational/race_bot.py b/examples/foundational/race_bot.py new file mode 100644 index 000000000..af8e2834b --- /dev/null +++ b/examples/foundational/race_bot.py @@ -0,0 +1,87 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys +import time + +import aiohttp +from loguru import logger +from runner import configure + +from pipecat.frames.frames import ( + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, _) = await configure(session) + + transport = DailyTransport( + room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True) + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o") + + 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 and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + runner = PipelineRunner() + + task = PipelineTask( + Pipeline( + [ + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + ) + + # Register an event handler so we can play the audio when the + # participant joins. + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + frames = [ + UserStartedSpeakingFrame(), + TranscriptionFrame("Tell a joke about dogs.", "user_id", time.time()), + UserStoppedSpeakingFrame(), + ] + + await task.queue_frames(frames) + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 93829c8fc..cc3b27050 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -6,11 +6,10 @@ import asyncio import inspect + from enum import Enum from typing import Awaitable, Callable, Optional -from loguru import logger - from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( EndFrame, @@ -25,6 +24,8 @@ from pipecat.metrics.metrics import LLMTokenUsage, MetricsData from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics from pipecat.utils.utils import obj_count, obj_id +from loguru import logger + class FrameDirection(Enum): DOWNSTREAM = 1 @@ -219,16 +220,11 @@ class FrameProcessor: # async def _start_interruption(self): - try: - # Cancel the push frame task. This will stop pushing frames downstream. - await self.__cancel_push_task() + # Cancel the push frame task. This will stop pushing frames downstream. + await self.__cancel_push_task() - # Cancel the input task. This will stop processing queued frames. - await self.__cancel_input_task() - except Exception as e: - logger.exception(f"Uncaught exception in {self}: {e}") - await self.push_error(ErrorFrame(str(e))) - raise + # Cancel the input task. This will stop processing queued frames. + await self.__cancel_input_task() # Create a new input queue and task. self.__create_input_task() @@ -285,11 +281,7 @@ class FrameProcessor: self.__input_queue.task_done() except asyncio.CancelledError: - logger.trace(f"Cancelled input task in {self}") break - except Exception as e: - logger.exception(f"Uncaught exception in {self}: {e}") - await self.push_error(ErrorFrame(str(e))) def __create_push_task(self): self.__push_queue = asyncio.Queue() @@ -308,11 +300,7 @@ class FrameProcessor: running = not isinstance(frame, EndFrame) self.__push_queue.task_done() except asyncio.CancelledError: - logger.trace(f"Cancelled push task in {self}") break - except Exception as e: - logger.exception(f"Uncaught exception in {self}: {e}") - await self.push_error(ErrorFrame(str(e))) async def _call_event_handler(self, event_name: str, *args, **kwargs): try: