wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes for proposed judge pipeline

Browse Source
This commit is contained in:
Kwindla Hultman Kramer
2024-11-04 13:28:35 -08:00
parent bd435d9e62
commit b56c789ae4
2 changed files with 51 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
Frame, Frame,
StartFrame, StartFrame,
StartInterruptionFrame, StartInterruptionFrame,
StopInterruptionFrame,
SystemFrame, SystemFrame,
TranscriptionFrame, TranscriptionFrame,
UserStartedSpeakingFrame, UserStartedSpeakingFrame,
@@ -39,6 +40,7 @@ from pipecat.frames.frames import (
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.base_notifier import BaseNotifier
from pipecat.processors.filters.function_filter import FunctionFilter from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.processors.user_idle_processor import UserIdleProcessor
from runner import configure from runner import configure
@@ -53,10 +55,14 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG") logger.add(sys.stderr, level="DEBUG")
classifier_statement = "Determine if the user's statement ends with a complete sentence or question. The user text is transcribed speech. It may contain multiple fragments concatentated together. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought." classifier_statement = "Determine if the user's statement ends with a complete thought and you should respond. The user text is transcribed speech. It may contain multiple fragments concatentated together. You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought."
class StatementJudgeContextFilter(FrameProcessor): class StatementJudgeContextFilter(FrameProcessor):
def __init__(self, notifier: BaseNotifier, **kwargs):
super().__init__(**kwargs)
self._notifier = notifier
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
# We must not block system frames. # We must not block system frames.
@@ -64,7 +70,12 @@ class StatementJudgeContextFilter(FrameProcessor):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
return return
# We only want to handle OpenAILLMContextFrames, and only want to push a simple # Just treat an LLMMessagesFrame as complete, no matter what.
if isinstance(frame, LLMMessagesFrame):
await self._notifier.notify()
return
# Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple
# messages frame that contains a system prompt and the most recent user messages, # messages frame that contains a system prompt and the most recent user messages,
# concatenated. # concatenated.
if isinstance(frame, OpenAILLMContextFrame): if isinstance(frame, OpenAILLMContextFrame):
@@ -102,31 +113,26 @@ class StatementJudgeContextFilter(FrameProcessor):
class CompletenessCheck(FrameProcessor): class CompletenessCheck(FrameProcessor):
def __init__(self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier): def __init__(self, notifier: BaseNotifier):
super().__init__() super().__init__()
self._complete_notifier = complete_notifier self._notifier = notifier
self._incomplete_notifier = incomplete_notifier
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if isinstance(frame, TextFrame) and frame.text == "YES": if isinstance(frame, TextFrame) and frame.text == "YES":
logger.debug("Completeness check YES") logger.debug("Completeness check YES")
await self.push_frame(UserStoppedSpeakingFrame()) await self.push_frame(UserStoppedSpeakingFrame())
await self._complete_notifier.notify() await self._notifier.notify()
elif isinstance(frame, TextFrame) and frame.text == "NO": elif isinstance(frame, TextFrame) and frame.text == "NO":
logger.debug("Completeness check NO") logger.debug("Completeness check NO")
await self._incomplete_notifier.notify()
class OutputGate(FrameProcessor): class OutputGate(FrameProcessor):
def __init__( def __init__(self, notifier: BaseNotifier, **kwargs):
self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier, **kwargs
):
super().__init__(**kwargs) super().__init__(**kwargs)
self._gate_open = False self._gate_open = False
self._frames_buffer = [] self._frames_buffer = []
self._complete_notifier = complete_notifier self._notifier = notifier
self._incomplete_notifier = incomplete_notifier
def close_gate(self): def close_gate(self):
self._gate_open = False self._gate_open = False
@@ -163,7 +169,6 @@ class OutputGate(FrameProcessor):
async def _start(self): async def _start(self):
self._frames_buffer = [] self._frames_buffer = []
self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) self._gate_task = self.get_event_loop().create_task(self._gate_task_handler())
self._interrupt_task = self.get_event_loop().create_task(self._interrupt_task_handler())
async def _stop(self): async def _stop(self):
self._gate_task.cancel() self._gate_task.cancel()
@@ -172,7 +177,7 @@ class OutputGate(FrameProcessor):
async def _gate_task_handler(self): async def _gate_task_handler(self):
while True: while True:
try: try:
await self._complete_notifier.wait() await self._notifier.wait()
self.open_gate() self.open_gate()
for frame, direction in self._frames_buffer: for frame, direction in self._frames_buffer:
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
@@ -180,16 +185,6 @@ class OutputGate(FrameProcessor):
except asyncio.CancelledError: except asyncio.CancelledError:
break break
async def _interrupt_task_handler(self):
while True:
try:
await self._incomplete_notifier.wait()
await self.push_frame(StartInterruptionFrame(), FrameDirection.UPSTREAM)
self._frames_buffer = []
except asyncio.CancelledError:
break
async def main(): async def main():
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
@@ -241,37 +236,39 @@ async def main():
# This is a notifier that we use to synchronize the two LLMs. # This is a notifier that we use to synchronize the two LLMs.
notifier = EventNotifier() notifier = EventNotifier()
# rename/comment?
interrupt_notifier = EventNotifier() # This turns the LLM context into an inference request to classify the user's speech
# as complete or incomplete.
statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier)
# This sends a UserStoppedSpeakingFrame and triggers the notifier event # This sends a UserStoppedSpeakingFrame and triggers the notifier event
completeness_check = CompletenessCheck( completeness_check = CompletenessCheck(notifier=notifier)
complete_notifier=notifier, incomplete_notifier=interrupt_notifier
)
# # Notify if the user hasn't said anything. # # Notify if the user hasn't said anything.
# async def user_idle_notifier(frame): async def user_idle_notifier(frame):
# await notifier.notify() await notifier.notify()
# # Sometimes the LLM will fail detecting if a user has completed a # Sometimes the LLM will fail detecting if a user has completed a
# # sentence, this will wake up the notifier if that happens. # sentence, this will wake up the notifier if that happens.
# user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=10.0) user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
bot_output_gate = OutputGate( bot_output_gate = OutputGate(notifier=notifier)
complete_notifier=notifier, incomplete_notifier=interrupt_notifier
)
async def block_user_stopped_speaking(frame): async def block_user_stopped_speaking(frame):
return not isinstance(frame, UserStoppedSpeakingFrame) return not isinstance(frame, UserStoppedSpeakingFrame)
async def pass_only_llm_trigger_frames(frame): async def pass_only_llm_trigger_frames(frame):
return isinstance(frame, OpenAILLMContextFrame) or isinstance(frame, LLMMessagesFrame) return (
isinstance(frame, OpenAILLMContextFrame)
or isinstance(frame, LLMMessagesFrame)
or isinstance(frame, StartInterruptionFrame)
or isinstance(frame, StopInterruptionFrame)
)
pipeline = Pipeline( pipeline = Pipeline(
[ [
transport.input(), transport.input(),
stt, stt,
# user_idle,
context_aggregator.user(), context_aggregator.user(),
ParallelPipeline( ParallelPipeline(
[ [
@@ -283,7 +280,7 @@ async def main():
# Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
# LLMMessagesFrame to the statement classifier LLM. The only frame this # LLMMessagesFrame to the statement classifier LLM. The only frame this
# sub-pipeline will output is a UserStoppedSpeakingFrame. # sub-pipeline will output is a UserStoppedSpeakingFrame.
StatementJudgeContextFilter(), statement_judge_context_filter,
statement_llm, statement_llm,
completeness_check, completeness_check,
], ],
@@ -291,10 +288,11 @@ async def main():
# Block everything except OpenAILLMContextFrame and LLMMessagesFrame # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
FunctionFilter(filter=pass_only_llm_trigger_frames), FunctionFilter(filter=pass_only_llm_trigger_frames),
llm, llm,
tts,
bot_output_gate, # Buffer all llm/tts output until notified. bot_output_gate, # Buffer all llm/tts output until notified.
], ],
), ),
tts,
user_idle,
transport.output(), transport.output(),
context_aggregator.assistant(), context_aggregator.assistant(),
] ]

16
src/pipecat/processors/filters/function_filter.py
View File

@@ -11,19 +11,27 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class FunctionFilter(FrameProcessor): class FunctionFilter(FrameProcessor):
def __init__(self, filter: Callable[[Frame], Awaitable[bool]]): def __init__(
self,
filter: Callable[[Frame], Awaitable[bool]],
direction: FrameDirection = FrameDirection.DOWNSTREAM,
):
super().__init__() super().__init__()
self._filter = filter self._filter = filter
self._direction = direction
# #
# Frame processor # Frame processor
# #
def _should_passthrough_frame(self, frame): # Ignore system frames and frames that are not following the direction of this gate
return isinstance(frame, SystemFrame) def _should_passthrough_frame(self, frame, direction):
return isinstance(frame, SystemFrame) or direction != self._direction
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
passthrough = self._should_passthrough_frame(frame) await super().process_frame(frame, direction)
passthrough = self._should_passthrough_frame(frame, direction)
allowed = await self._filter(frame) allowed = await self._filter(frame)
if passthrough or allowed: if passthrough or allowed:
await self.push_frame(frame, direction) await self.push_frame(frame, direction)