diff --git a/CHANGELOG.md b/CHANGELOG.md index 75dad0be3..98cab2a05 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `GatedOpenAILLMContextAggregator`. This aggregator keeps the last + received OpenAI LLM context frame and it doesn't let it through until the + notifier is notified. + +- Added `WakeNotifierFilter`. This processor expects a list of frame types and + will execute a given callback predicate when a frame of any of those type is + being processed. If the callback returns true the notifier will be notified. + +- Added `NullFilter`. A null filter doesn't push any frames upstream or + downstream. This is usually used to disable one of the pipelines in + `ParallelPipeline`. + +- Added `EventNotifier`. This can be used as a very simple synchronization + feature between processors. + - Added `TavusVideoService`. This is an integration for Tavus digital twins. (see https://www.tavus.io/) @@ -40,6 +55,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue with PlayHTTTSService, where the TTFB metrics were reporting very small time values. +### Other + +- Added a new foundational example 22-natural-conversation.py. This examples + shows how to achieve a more natural conversation detecting when the user ends + statement. + ## [0.0.47] - 2024-10-22 ### Added diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py new file mode 100644 index 000000000..73dfb003d --- /dev/null +++ b/examples/foundational/22-natural-conversation.py @@ -0,0 +1,168 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMMessagesFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.gated_openai_llm_context import GatedOpenAILLMContextAggregator +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.filters.null_filter import NullFilter +from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter +from pipecat.processors.user_idle_processor import UserIdleProcessor +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.openai import OpenAILLMService +from pipecat.sync.event_notifier import EventNotifier +from pipecat.transports.services.daily import DailyParams, DailyTransport + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv + +load_dotenv(override=True) + +logger.remove(0) +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, + "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 + ) + + # This is the LLM that will be used to detect if the user has finished a + # statement. This doesn't really need to be an LLM, we could use NLP + # libraries for that, but it was easier as an example because we + # leverage the context aggregators. + statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + statement_messages = [ + { + "role": "system", + "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.", + }, + ] + + statement_context = OpenAILLMContext(statement_messages) + statement_context_aggregator = statement_llm.create_context_aggregator(statement_context) + + # This is the regular LLM. + llm = OpenAILLMService(api_key=os.getenv("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) + + # We have instructed the LLM to return 'YES' if it thinks the user + # completed a sentence. So, if it's 'YES' we will return true in this + # predicate which will wake up the notifier. + async def wake_check_filter(frame): + return frame.text == "YES" + + # This is a notifier that we use to synchronize the two LLMs. + notifier = EventNotifier() + + # This a filter that will wake up the notifier if the given predicate + # (wake_check_filter) returns true. + completness_check = WakeNotifierFilter( + notifier, types=(TextFrame,), filter=wake_check_filter + ) + + # This processor keeps the last context and will let it through once the + # notifier is woken up. + gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier) + + # Notify if the user hasn't said anything. + async def user_idle_notifier(frame): + await notifier.notify() + + # Sometimes the LLM will fail detecting if a user has completed a + # sentence, this will wake up the notifier if that happens. + user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0) + + # The ParallePipeline input are the user transcripts. We have two + # contexts. The first one will be used to determine if the user finished + # a statement and if so the notifier will be woken up. The second + # context is simply the regular context but it's gated waiting for the + # notifier to be woken up. + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + ParallelPipeline( + [ + statement_context_aggregator.user(), + statement_llm, + completness_check, + NullFilter(), + ], + [context_aggregator.user(), gated_context_aggregator, llm], + ), + user_idle, + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=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. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/aggregators/gated_openai_llm_context.py b/src/pipecat/processors/aggregators/gated_openai_llm_context.py new file mode 100644 index 000000000..71a540dd4 --- /dev/null +++ b/src/pipecat/processors/aggregators/gated_openai_llm_context.py @@ -0,0 +1,55 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio + +from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.sync.base_notifier import BaseNotifier + + +class GatedOpenAILLMContextAggregator(FrameProcessor): + """This aggregator keeps the last received OpenAI LLM context frame and it + doesn't let it through until the notifier is notified. + + """ + + def __init__(self, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._notifier = notifier + self._last_context_frame = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartFrame): + await self.push_frame(frame) + await self._start() + if isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + await self.push_frame(frame) + elif isinstance(frame, OpenAILLMContextFrame): + self._last_context_frame = frame + else: + await self.push_frame(frame, direction) + + async def _start(self): + self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) + + async def _stop(self): + self._gate_task.cancel() + await self._gate_task + + async def _gate_task_handler(self): + while True: + try: + await self._notifier.wait() + if self._last_context_frame: + await self.push_frame(self._last_context_frame) + self._last_context_frame = None + except asyncio.CancelledError: + break diff --git a/src/pipecat/processors/filters/frame_filter.py b/src/pipecat/processors/filters/frame_filter.py index 45927a604..f4c4b0f61 100644 --- a/src/pipecat/processors/filters/frame_filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -4,14 +4,14 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import List +from typing import Tuple, Type from pipecat.frames.frames import AppFrame, ControlFrame, Frame, SystemFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class FrameFilter(FrameProcessor): - def __init__(self, types: List[type]): + def __init__(self, types: Tuple[Type[Frame]]): super().__init__() self._types = types @@ -20,9 +20,8 @@ class FrameFilter(FrameProcessor): # def _should_passthrough_frame(self, frame): - for t in self._types: - if isinstance(frame, t): - return True + if isinstance(frame, self._types): + return True return ( isinstance(frame, AppFrame) diff --git a/src/pipecat/processors/filters/null_filter.py b/src/pipecat/processors/filters/null_filter.py new file mode 100644 index 000000000..7e9ca6725 --- /dev/null +++ b/src/pipecat/processors/filters/null_filter.py @@ -0,0 +1,14 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from pipecat.processors.frame_processor import FrameProcessor + + +class NullFilter(FrameProcessor): + """This filter doesn't allow passing any frames up or downstream.""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) diff --git a/src/pipecat/processors/filters/wake_notifier_filter.py b/src/pipecat/processors/filters/wake_notifier_filter.py new file mode 100644 index 000000000..a7f074ccb --- /dev/null +++ b/src/pipecat/processors/filters/wake_notifier_filter.py @@ -0,0 +1,40 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Awaitable, Callable, Tuple, Type + +from pipecat.frames.frames import Frame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.sync.base_notifier import BaseNotifier + + +class WakeNotifierFilter(FrameProcessor): + """This processor expects a list of frame types and will execute a given + callback predicate when a frame of any of those type is being processed. If + the callback returns true the notifier will be notified. + + """ + + def __init__( + self, + notifier: BaseNotifier, + *, + types: Tuple[Type[Frame]], + filter: Callable[[Frame], Awaitable[bool]], + **kwargs, + ): + super().__init__(**kwargs) + self._notifier = notifier + self._types = types + self._filter = filter + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, self._types) and await self._filter(frame): + await self._notifier.notify() + + await self.push_frame(frame, direction) diff --git a/src/pipecat/sync/__init__.py b/src/pipecat/sync/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/sync/base_notifier.py b/src/pipecat/sync/base_notifier.py new file mode 100644 index 000000000..c7770ab26 --- /dev/null +++ b/src/pipecat/sync/base_notifier.py @@ -0,0 +1,17 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod + + +class BaseNotifier(ABC): + @abstractmethod + async def notify(self): + pass + + @abstractmethod + async def wait(self): + pass diff --git a/src/pipecat/sync/event_notifier.py b/src/pipecat/sync/event_notifier.py new file mode 100644 index 000000000..f02dcbdae --- /dev/null +++ b/src/pipecat/sync/event_notifier.py @@ -0,0 +1,21 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio + +from pipecat.sync.base_notifier import BaseNotifier + + +class EventNotifier(BaseNotifier): + def __init__(self): + self._event = asyncio.Event() + + async def notify(self): + self._event.set() + + async def wait(self): + await self._event.wait() + self._event.clear()