From 2c32cc2f27b7cc0b9921def4d2492cc167afb020 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 20 May 2024 22:24:24 -0700 Subject: [PATCH 1/6] improved wake word filter --- examples/foundational/14-wake-phrase.py | 172 ++++++++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 examples/foundational/14-wake-phrase.py diff --git a/examples/foundational/14-wake-phrase.py b/examples/foundational/14-wake-phrase.py new file mode 100644 index 000000000..b6d19116c --- /dev/null +++ b/examples/foundational/14-wake-phrase.py @@ -0,0 +1,172 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys +import re +import time +from enum import Enum + +from pipecat.frames.frames import ( + Frame, SystemFrame, TranscriptionFrame, TextFrame +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +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 ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +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") + + +class WakeCheckFilter(FrameProcessor): + """ + This filter looks for wake phrases in the transcription frames and only passes through frames + after a wake phrase has been detected. It also has a keepalive timeout to allow for a brief + period of continued conversation after a wake phrase has been detected. + """ + class WakeState(Enum): + IDLE = 1 + AWAKE = 2 + + class ParticipantState: + def __init__(self, participant_id: str): + self.participant_id = participant_id + self.state = WakeCheckFilter.WakeState.IDLE + self.wake_timer = 0 + self.accumulator = "" + + def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 2): + super().__init__() + self._participant_states = {} + self._keepalive_timeout = keepalive_timeout + self._wake_patterns = [] + for name in wake_phrases: + pattern = re.compile(r'\b' + r'\s*'.join(re.escape(word) + for word in name.split()) + r'\b', re.IGNORECASE) + self._wake_patterns.append(pattern) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + try: + if isinstance(frame, TranscriptionFrame): + p = self._participant_states.get(frame.user_id) + if p is None: + p = WakeCheckFilter.ParticipantState(frame.user_id) + self._participant_states[frame.user_id] = p + + # If we have been AWAKE within the last keepalive_timeout seconds, pass + # the frame through + if p.state == WakeCheckFilter.WakeState.AWAKE: + if time.time() - p.wake_timer < self._keepalive_timeout: + logger.debug( + f"Wake phrase keepalive timeout has not expired. Passing frame through.") + p.wake_timer = time.time() + await self.push_frame(frame) + return + else: + p.state = WakeCheckFilter.WakeState.IDLE + + p.accumulator += frame.text + for pattern in self._wake_patterns: + match = pattern.search(p.accumulator) + if match: + logger.debug(f"Wake phrase triggered: {match.group()}") + # Found the wake word. Discard from the accumulator up to the start of the match + # and modify the frame in place. + p.state = WakeCheckFilter.WakeState.AWAKE + p.wake_timer = time.time() + frame.text = p.accumulator[match.start():] + p.accumulator = "" + await self.push_frame(frame) + else: + pass + else: + await self.push_frame(frame, direction) + except Exception as e: + logger.error(f"Error in wake word filter: {e}") + + +async def main(room_url: str, token): + + async with aiohttp.ClientSession() as session: + transport = DailyTransport( + room_url, + token, + "Robot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = ElevenLabsTTSService( + aiohttp_session=session, + api_key=os.getenv("ELEVENLABS_API_KEY"), + voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful assistant. Respond to what the user said in a creative and helpful way. Keep your responses brief.", + }, + ] + + hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"]) + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + hey_robot_filter, # Filter out speech not directed at the robot + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, allow_interruptions=True) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.") + + # 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__": + (url, token) = configure() + asyncio.run(main(url, token)) From 661aa79b7cce0bfccf12f60733c25537a84b5243 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 20 May 2024 22:45:05 -0700 Subject: [PATCH 2/6] fix user_id str field name in TranscriptionFrame --- src/pipecat/frames/frames.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8eb32664c..61f4d3fe2 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -132,7 +132,7 @@ class TranscriptionFrame(TextFrame): timestamp: str def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})" + return f"{self.name}(user_id: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})" @dataclass From 0ddfa3de5be7fe3b66efadc14f00d4eb854c2c62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 15:36:18 -0700 Subject: [PATCH 3/6] move WakeCheckFilter to processors/filters --- examples/foundational/14-wake-phrase.py | 77 +---------------- .../{filter.py => filters/frame_filter.py} | 2 +- .../processors/filters/wake_check_filter.py | 82 +++++++++++++++++++ src/pipecat/processors/frame_processor.py | 2 +- 4 files changed, 86 insertions(+), 77 deletions(-) rename src/pipecat/processors/{filter.py => filters/frame_filter.py} (96%) create mode 100644 src/pipecat/processors/filters/wake_check_filter.py diff --git a/examples/foundational/14-wake-phrase.py b/examples/foundational/14-wake-phrase.py index b6d19116c..283768460 100644 --- a/examples/foundational/14-wake-phrase.py +++ b/examples/foundational/14-wake-phrase.py @@ -8,14 +8,8 @@ import asyncio import aiohttp import os import sys -import re -import time -from enum import Enum -from pipecat.frames.frames import ( - Frame, SystemFrame, TranscriptionFrame, TextFrame -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.filters.wake_check_filter import WakeCheckFilter from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -37,73 +31,6 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -class WakeCheckFilter(FrameProcessor): - """ - This filter looks for wake phrases in the transcription frames and only passes through frames - after a wake phrase has been detected. It also has a keepalive timeout to allow for a brief - period of continued conversation after a wake phrase has been detected. - """ - class WakeState(Enum): - IDLE = 1 - AWAKE = 2 - - class ParticipantState: - def __init__(self, participant_id: str): - self.participant_id = participant_id - self.state = WakeCheckFilter.WakeState.IDLE - self.wake_timer = 0 - self.accumulator = "" - - def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 2): - super().__init__() - self._participant_states = {} - self._keepalive_timeout = keepalive_timeout - self._wake_patterns = [] - for name in wake_phrases: - pattern = re.compile(r'\b' + r'\s*'.join(re.escape(word) - for word in name.split()) + r'\b', re.IGNORECASE) - self._wake_patterns.append(pattern) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - try: - if isinstance(frame, TranscriptionFrame): - p = self._participant_states.get(frame.user_id) - if p is None: - p = WakeCheckFilter.ParticipantState(frame.user_id) - self._participant_states[frame.user_id] = p - - # If we have been AWAKE within the last keepalive_timeout seconds, pass - # the frame through - if p.state == WakeCheckFilter.WakeState.AWAKE: - if time.time() - p.wake_timer < self._keepalive_timeout: - logger.debug( - f"Wake phrase keepalive timeout has not expired. Passing frame through.") - p.wake_timer = time.time() - await self.push_frame(frame) - return - else: - p.state = WakeCheckFilter.WakeState.IDLE - - p.accumulator += frame.text - for pattern in self._wake_patterns: - match = pattern.search(p.accumulator) - if match: - logger.debug(f"Wake phrase triggered: {match.group()}") - # Found the wake word. Discard from the accumulator up to the start of the match - # and modify the frame in place. - p.state = WakeCheckFilter.WakeState.AWAKE - p.wake_timer = time.time() - frame.text = p.accumulator[match.start():] - p.accumulator = "" - await self.push_frame(frame) - else: - pass - else: - await self.push_frame(frame, direction) - except Exception as e: - logger.error(f"Error in wake word filter: {e}") - - async def main(room_url: str, token): async with aiohttp.ClientSession() as session: @@ -152,7 +79,7 @@ async def main(room_url: str, token): task = PipelineTask(pipeline, allow_interruptions=True) - @ transport.event_handler("on_first_participant_joined") + @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): transport.capture_participant_transcription(participant["id"]) await tts.say("Hi! If you want to talk to me, just say 'Hey Robot'.") diff --git a/src/pipecat/processors/filter.py b/src/pipecat/processors/filters/frame_filter.py similarity index 96% rename from src/pipecat/processors/filter.py rename to src/pipecat/processors/filters/frame_filter.py index 0f7026b4a..42f0e9dd8 100644 --- a/src/pipecat/processors/filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -10,7 +10,7 @@ from pipecat.frames.frames import AppFrame, ControlFrame, Frame, SystemFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -class Filter(FrameProcessor): +class FrameFilter(FrameProcessor): def __init__(self, types: List[type]): super().__init__() diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py new file mode 100644 index 000000000..459ee5f75 --- /dev/null +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -0,0 +1,82 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import re +import time + +from enum import Enum + +from pipecat.frames.frames import Frame, TranscriptionFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +from loguru import logger + + +class WakeCheckFilter(FrameProcessor): + """ + This filter looks for wake phrases in the transcription frames and only passes through frames + after a wake phrase has been detected. It also has a keepalive timeout to allow for a brief + period of continued conversation after a wake phrase has been detected. + """ + class WakeState(Enum): + IDLE = 1 + AWAKE = 2 + + class ParticipantState: + def __init__(self, participant_id: str): + self.participant_id = participant_id + self.state = WakeCheckFilter.WakeState.IDLE + self.wake_timer = 0.0 + self.accumulator = "" + + def __init__(self, wake_phrases: list[str], keepalive_timeout: float = 2): + super().__init__() + self._participant_states = {} + self._keepalive_timeout = keepalive_timeout + self._wake_patterns = [] + for name in wake_phrases: + pattern = re.compile(r'\b' + r'\s*'.join(re.escape(word) + for word in name.split()) + r'\b', re.IGNORECASE) + self._wake_patterns.append(pattern) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + try: + if isinstance(frame, TranscriptionFrame): + p = self._participant_states.get(frame.user_id) + if p is None: + p = WakeCheckFilter.ParticipantState(frame.user_id) + self._participant_states[frame.user_id] = p + + # If we have been AWAKE within the last keepalive_timeout seconds, pass + # the frame through + if p.state == WakeCheckFilter.WakeState.AWAKE: + if time.time() - p.wake_timer < self._keepalive_timeout: + logger.debug( + "Wake phrase keepalive timeout has not expired. Passing frame through.") + p.wake_timer = time.time() + await self.push_frame(frame) + return + else: + p.state = WakeCheckFilter.WakeState.IDLE + + p.accumulator += frame.text + for pattern in self._wake_patterns: + match = pattern.search(p.accumulator) + if match: + logger.debug(f"Wake phrase triggered: {match.group()}") + # Found the wake word. Discard from the accumulator up to the start of the match + # and modify the frame in place. + p.state = WakeCheckFilter.WakeState.AWAKE + p.wake_timer = time.time() + frame.text = p.accumulator[match.start():] + p.accumulator = "" + await self.push_frame(frame) + else: + pass + else: + await self.push_frame(frame, direction) + except Exception as e: + logger.error(f"Error in wake word filter: {e}") diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index a79352f7a..3bb750218 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,7 +8,7 @@ import asyncio from asyncio import AbstractEventLoop from enum import Enum -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame +from pipecat.frames.frames import ErrorFrame, Frame from pipecat.utils.utils import obj_count, obj_id from loguru import logger From 0e5711e62dd04c32de5592db0ed473f70d002c04 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 15:41:33 -0700 Subject: [PATCH 4/6] examples: update 10-wake-work.py to use WakeCheckFilter --- examples/foundational/10-wake-word.py | 55 ++++++--------------------- 1 file changed, 11 insertions(+), 44 deletions(-) diff --git a/examples/foundational/10-wake-word.py b/examples/foundational/10-wake-word.py index cc1829046..4d0c0a16d 100644 --- a/examples/foundational/10-wake-word.py +++ b/examples/foundational/10-wake-word.py @@ -12,14 +12,7 @@ import sys from PIL import Image -from pipecat.frames.frames import ( - Frame, - SystemFrame, - TextFrame, - ImageRawFrame, - SpriteFrame, - TranscriptionFrame, -) +from pipecat.frames.frames import Frame, ImageRawFrame, SpriteFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -27,6 +20,7 @@ from pipecat.processors.aggregators.llm_context import ( LLMUserContextAggregator, LLMAssistantContextAggregator, ) +from pipecat.processors.filters.wake_check_filter import WakeCheckFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.openai import OpenAILLMService from pipecat.services.elevenlabs import ElevenLabsTTSService @@ -84,33 +78,6 @@ thinking_list = [ thinking_frame = SpriteFrame(thinking_list) -class NameCheckFilter(FrameProcessor): - def __init__(self, names: list[str]): - super().__init__() - self._names = names - self._sentence = "" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - if isinstance(frame, SystemFrame): - await self.push_frame(frame, direction) - return - - content: str = "" - - # TODO: split up transcription by participant - if isinstance(frame, TranscriptionFrame): - content = frame.text - self._sentence += content - if self._sentence.endswith((".", "?", "!")): - if any(name in self._sentence for name in self._names): - await self.push_frame(TextFrame(self._sentence)) - self._sentence = "" - else: - self._sentence = "" - else: - await self.push_frame(frame, direction) - - class ImageSyncAggregator(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -155,17 +122,17 @@ async def main(room_url: str, token): tma_in = LLMUserContextAggregator(messages) tma_out = LLMAssistantContextAggregator(messages) - ncf = NameCheckFilter(["Santa Cat", "Santa"]) + wcf = WakeCheckFilter(["Santa Cat", "Santa"]) pipeline = Pipeline([ - transport.input(), - isa, - ncf, - tma_in, - llm, - tts, - transport.output(), - tma_out + transport.input(), # Transport user input + isa, # Cat talking/quiet images + wcf, # Filter out speech not directed at Santa Cat + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Santa Cat spoken responses ]) @transport.event_handler("on_first_participant_joined") From e1169a4e82d18fb7f369943d6e2ff35dc49ee05d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 15:43:52 -0700 Subject: [PATCH 5/6] processors(WakeCheckFilter): push error --- src/pipecat/processors/filters/wake_check_filter.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py index 459ee5f75..4f976fc72 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -9,7 +9,7 @@ import time from enum import Enum -from pipecat.frames.frames import Frame, TranscriptionFrame +from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from loguru import logger @@ -79,4 +79,6 @@ class WakeCheckFilter(FrameProcessor): else: await self.push_frame(frame, direction) except Exception as e: - logger.error(f"Error in wake word filter: {e}") + error_msg = f"Error in wake word filter: {e}" + logger.error(error_msg) + await self.push_error(ErrorFrame(error_msg)) From 32d2f0db6623efc4e682eef9d5ca467b0c060f7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 22 May 2024 15:46:13 -0700 Subject: [PATCH 6/6] update CHANGELOG.ms with filters updates --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index cccd516bb..daf55d3ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `WakeCheckFilter` which allows you to pass information downstream only + if you say a certain phrase/word. + +### Changed + +- `Filter` has been renamed to `FrameFilter` and it's now under + `processors/filters`. + ### Fixed - Re-add exponential smoothing after volume calculation. This makes sure the