diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 0e2c09ebc..6998fe200 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -4,131 +4,15 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Optional - -from pipecat.frames.frames import ( - Frame, - InterimTranscriptionFrame, - StartInterruptionFrame, - TextFrame, - TranscriptionFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.frames.frames import TextFrame +from pipecat.processors.aggregators.llm_response import LLMUserResponseAggregator -class ResponseAggregator(FrameProcessor): - """This frame processor aggregates frames between a start and an end frame - into complete text frame sentences. +class UserResponseAggregator(LLMUserResponseAggregator): + def __init__(self, **kwargs): + super().__init__(**kwargs) - For example, frame input/output: - UserStartedSpeakingFrame() -> None - TranscriptionFrame("Hello,") -> None - TranscriptionFrame(" world.") -> None - UserStoppedSpeakingFrame() -> TextFrame("Hello world.") - - Doctest: FIXME to work with asyncio - >>> async def print_frames(aggregator, frame): - ... async for frame in aggregator.process_frame(frame): - ... if isinstance(frame, TextFrame): - ... print(frame.text) - - >>> aggregator = ResponseAggregator(start_frame = UserStartedSpeakingFrame, - ... end_frame=UserStoppedSpeakingFrame, - ... accumulator_frame=TranscriptionFrame, - ... pass_through=False) - >>> asyncio.run(print_frames(aggregator, UserStartedSpeakingFrame())) - >>> asyncio.run(print_frames(aggregator, TranscriptionFrame("Hello,", 1, 1))) - >>> asyncio.run(print_frames(aggregator, TranscriptionFrame("world.", 1, 2))) - >>> asyncio.run(print_frames(aggregator, UserStoppedSpeakingFrame())) - Hello, world. - - """ - - def __init__( - self, - *, - start_frame, - end_frame, - accumulator_frame: TextFrame, - interim_accumulator_frame: Optional[TextFrame] = None, - ): - super().__init__() - - self._start_frame = start_frame - self._end_frame = end_frame - self._accumulator_frame = accumulator_frame - self._interim_accumulator_frame = interim_accumulator_frame - - # Reset our accumulator state. - self._reset() - - # - # Frame processor - # - - # Use cases implemented: - # - # S: Start, E: End, T: Transcription, I: Interim, X: Text - # - # S E -> None - # S T E -> X - # S I T E -> X - # S I E T -> X - # S I E I T -> X - # S E T -> X - # S E I T -> X - # - # The following case would not be supported: - # - # S I E T1 I T2 -> X - # - # and T2 would be dropped. - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - send_aggregation = False - - if isinstance(frame, self._start_frame): - self._aggregating = True - self._seen_start_frame = True - self._seen_end_frame = False - self._seen_interim_results = False - await self.push_frame(frame, direction) - elif isinstance(frame, self._end_frame): - self._seen_end_frame = True - self._seen_start_frame = False - - # We might have received the end frame but we might still be - # aggregating (i.e. we have seen interim results but not the final - # text). - self._aggregating = self._seen_interim_results or len(self._aggregation) == 0 - - # Send the aggregation if we are not aggregating anymore (i.e. no - # more interim results received). - send_aggregation = not self._aggregating - await self.push_frame(frame, direction) - elif isinstance(frame, self._accumulator_frame): - if self._aggregating: - self._aggregation += f" {frame.text}" - # We have recevied a complete sentence, so if we have seen the - # end frame and we were still aggregating, it means we should - # send the aggregation. - send_aggregation = self._seen_end_frame - - # We just got our final result, so let's reset interim results. - self._seen_interim_results = False - elif self._interim_accumulator_frame and isinstance(frame, self._interim_accumulator_frame): - self._seen_interim_results = True - else: - await self.push_frame(frame, direction) - - if send_aggregation: - await self._push_aggregation() - - async def _push_aggregation(self): + async def push_aggregation(self): if len(self._aggregation) > 0: frame = TextFrame(self._aggregation.strip()) @@ -139,21 +23,4 @@ class ResponseAggregator(FrameProcessor): await self.push_frame(frame) # Reset our accumulator state. - self._reset() - - def _reset(self): - self._aggregation = "" - self._aggregating = False - self._seen_start_frame = False - self._seen_end_frame = False - self._seen_interim_results = False - - -class UserResponseAggregator(ResponseAggregator): - def __init__(self): - super().__init__( - start_frame=UserStartedSpeakingFrame, - end_frame=UserStoppedSpeakingFrame, - accumulator_frame=TranscriptionFrame, - interim_accumulator_frame=InterimTranscriptionFrame, - ) + self.reset()