processors(llm_response): unify new use cases into base class

CHANGELOG.md

@@ -75,6 +75,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed LLM response aggregators to support more uses cases such as delayed
+  transcriptions.
+
 - Fixed an issue that could cause the bot to stop talking if there was a user
   interruption before getting any audio from the TTS service.
 

src/pipecat/processors/aggregators/llm_response.py

@@ -7,6 +7,7 @@
 from typing import List, Type
 
 from pipecat.frames.frames import (
+    BotInterruptionFrame,
     Frame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
@@ -40,6 +41,7 @@ class LLMResponseAggregator(FrameProcessor):
         interim_accumulator_frame: Type[TextFrame] | None = None,
         handle_interruptions: bool = False,
         expect_stripped_words: bool = True,  # if True, need to add spaces between words
+        interrupt_double_accumulator: bool = True,  # if True, interrupt if two or more accumulators are received
     ):
         super().__init__()
 
@@ -51,8 +53,8 @@ class LLMResponseAggregator(FrameProcessor):
         self._interim_accumulator_frame = interim_accumulator_frame
         self._handle_interruptions = handle_interruptions
         self._expect_stripped_words = expect_stripped_words
+        self._interrupt_double_accumulator = interrupt_double_accumulator
 
-        # Reset our accumulator state.
         self._reset()
 
     @property
@@ -69,21 +71,20 @@ class LLMResponseAggregator(FrameProcessor):
 
     # Use cases implemented:
     #
-    # S: Start, E: End, T: Transcription, I: Interim, X: Text
+    # S: Start, E: End, T: Transcription, I: Interim
     #
-    #        S E -> None
+    #    S E             -> None                 -> User started speaking but no transcription.
-    #      S T E -> X
+    #    S T E           -> T                    -> Transcription between user started and stopped speaking.
-    #    S I T E -> X
+    #    S E T           -> T                    -> Transcription after user stopped speaking.
-    #    S I E T -> X
+    #    S I T E         -> T                    -> Transcription between user started and stopped speaking (with interims).
-    #  S I E I T -> X
+    #    S I E T         -> T                    -> Transcription after user stopped speaking (with interims).
-    #      S E T -> X
+    #    S I E I T       -> T                    -> Transcription after user stopped speaking (with interims).
-    #    S E I T -> X
+    #    S E I T         -> T                    -> Transcription after user stopped speaking (with interims).
-    #
+    #    S T1 I E S T2 E -> "T1 T2"              -> Merge two transcriptions if we got a first interim.
-    # The following case would not be supported:
+    #    S I E T1 I T2   -> T1 [Interruption] T2 -> Single user started/stopped, double transcription.
-    #
+    #    S T1 E T2       -> T1 [Interruption] T2 -> Single user started/stopped, double transcription.
-    #    S I E T1 I T2 -> X
+    #    S E T1 B T2     -> T1 [Interruption] T2 -> Single user started/stopped, double transcription.
-    #
+    #    S E T1 T2       -> T1 [Interruption] T2 -> Single user started/stopped, double transcription.
-    # and T2 would be dropped.
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
@@ -91,11 +92,9 @@ class LLMResponseAggregator(FrameProcessor):
         send_aggregation = False
 
         if isinstance(frame, self._start_frame):
-            self._aggregation = ""
             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
@@ -109,23 +108,36 @@ class LLMResponseAggregator(FrameProcessor):
             # 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:
+            if (
-                if self._expect_stripped_words:
+                self._interrupt_double_accumulator
-                    self._aggregation += f" {frame.text}" if self._aggregation else frame.text
+                and self._sent_aggregation_after_last_interruption
-                else:
+            ):
-                    self._aggregation += frame.text
+                await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
-                # We have recevied a complete sentence, so if we have seen the
+                self._sent_aggregation_after_last_interruption = False
-                # end frame and we were still aggregating, it means we should
+
-                # send the aggregation.
+            if self._expect_stripped_words:
-                send_aggregation = self._seen_end_frame
+                self._aggregation += f" {frame.text}" if self._aggregation else frame.text
+            else:
+                self._aggregation += frame.text
+
+            # If we haven't seen the start frame but we got an accumulator frame
+            # it means two things: it was develiver before the end frame or it
+            # was delivered late. In both cases so we want to send the
+            # aggregation.
+            send_aggregation = not self._seen_start_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):
+            if (
+                self._interrupt_double_accumulator
+                and self._sent_aggregation_after_last_interruption
+            ):
+                await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+                self._sent_aggregation_after_last_interruption = False
             self._seen_interim_results = True
-        elif self._handle_interruptions and isinstance(frame, StartInterruptionFrame):
+        elif isinstance(frame, StartInterruptionFrame) and self._handle_interruptions:
             await self._push_aggregation()
             # Reset anyways
             self._reset()
@@ -142,6 +154,9 @@ class LLMResponseAggregator(FrameProcessor):
         if send_aggregation:
             await self._push_aggregation()
 
+        if isinstance(frame, self._end_frame):
+            await self.push_frame(frame, direction)
+
     async def _push_aggregation(self):
         if len(self._aggregation) > 0:
             self._messages.append({"role": self._role, "content": self._aggregation})
@@ -150,6 +165,8 @@ class LLMResponseAggregator(FrameProcessor):
             # if the tasks gets cancelled we won't be able to clear things up.
             self._aggregation = ""
 
+            self._sent_aggregation_after_last_interruption = True
+
             frame = LLMMessagesFrame(self._messages)
             await self.push_frame(frame)
 
@@ -172,22 +189,11 @@ class LLMResponseAggregator(FrameProcessor):
         self._seen_start_frame = False
         self._seen_end_frame = False
         self._seen_interim_results = False
-
+        self._sent_aggregation_after_last_interruption = False
-
-class LLMAssistantResponseAggregator(LLMResponseAggregator):
-    def __init__(self, messages: List[dict] = []):
-        super().__init__(
-            messages=messages,
-            role="assistant",
-            start_frame=LLMFullResponseStartFrame,
-            end_frame=LLMFullResponseEndFrame,
-            accumulator_frame=TextFrame,
-            handle_interruptions=True,
-        )
 
 
 class LLMUserResponseAggregator(LLMResponseAggregator):
-    def __init__(self, messages: List[dict] = []):
+    def __init__(self, messages: List[dict] = [], **kwargs):
         super().__init__(
             messages=messages,
             role="user",
@@ -195,61 +201,21 @@ class LLMUserResponseAggregator(LLMResponseAggregator):
             end_frame=UserStoppedSpeakingFrame,
             accumulator_frame=TranscriptionFrame,
             interim_accumulator_frame=InterimTranscriptionFrame,
+            **kwargs,
         )
 
 
-class LLMFullResponseAggregator(FrameProcessor):
+class LLMAssistantResponseAggregator(LLMResponseAggregator):
-    """This class aggregates Text frames until it receives a
+    def __init__(self, messages: List[dict] = [], **kwargs):
-    LLMFullResponseEndFrame, then emits the concatenated text as
+        super().__init__(
-    a single text frame.
+            messages=messages,
-
+            role="assistant",
-    given the following frames:
+            start_frame=LLMFullResponseStartFrame,
-
+            end_frame=LLMFullResponseEndFrame,
-        TextFrame("Hello,")
+            accumulator_frame=TextFrame,
-        TextFrame(" world.")
+            handle_interruptions=True,
-        TextFrame(" I am")
+            **kwargs,
-        TextFrame(" an LLM.")
+        )
-        LLMFullResponseEndFrame()]
-
-    this processor will yield nothing for the first 4 frames, then
-
-        TextFrame("Hello, world. I am an LLM.")
-        LLMFullResponseEndFrame()
-
-    when passed the last frame.
-
-    >>> async def print_frames(aggregator, frame):
-    ...     async for frame in aggregator.process_frame(frame):
-    ...         if isinstance(frame, TextFrame):
-    ...             print(frame.text)
-    ...         else:
-    ...             print(frame.__class__.__name__)
-
-    >>> aggregator = LLMFullResponseAggregator()
-    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello,")))
-    >>> asyncio.run(print_frames(aggregator, TextFrame(" world.")))
-    >>> asyncio.run(print_frames(aggregator, TextFrame(" I am")))
-    >>> asyncio.run(print_frames(aggregator, TextFrame(" an LLM.")))
-    >>> asyncio.run(print_frames(aggregator, LLMFullResponseEndFrame()))
-    Hello, world. I am an LLM.
-    LLMFullResponseEndFrame
-    """
-
-    def __init__(self):
-        super().__init__()
-        self._aggregation = ""
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, TextFrame):
-            self._aggregation += frame.text
-        elif isinstance(frame, LLMFullResponseEndFrame):
-            await self.push_frame(TextFrame(self._aggregation))
-            await self.push_frame(frame)
-            self._aggregation = ""
-        else:
-            await self.push_frame(frame, direction)
 
 
 class LLMContextAggregator(LLMResponseAggregator):
@@ -286,15 +252,14 @@ class LLMContextAggregator(LLMResponseAggregator):
             # if the tasks gets cancelled we won't be able to clear things up.
             self._aggregation = ""
 
+            self._sent_aggregation_after_last_interruption = True
+
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
-            # Reset our accumulator state.
-            self._reset()
-
 
 class LLMAssistantContextAggregator(LLMContextAggregator):
-    def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True):
+    def __init__(self, context: OpenAILLMContext, **kwargs):
         super().__init__(
             messages=[],
             context=context,
@@ -303,12 +268,12 @@ class LLMAssistantContextAggregator(LLMContextAggregator):
             end_frame=LLMFullResponseEndFrame,
             accumulator_frame=TextFrame,
             handle_interruptions=True,
-            expect_stripped_words=expect_stripped_words,
+            **kwargs,
         )
 
 
 class LLMUserContextAggregator(LLMContextAggregator):
-    def __init__(self, context: OpenAILLMContext):
+    def __init__(self, context: OpenAILLMContext, **kwargs):
         super().__init__(
             messages=[],
             context=context,
@@ -317,131 +282,69 @@ class LLMUserContextAggregator(LLMContextAggregator):
             end_frame=UserStoppedSpeakingFrame,
             accumulator_frame=TranscriptionFrame,
             interim_accumulator_frame=InterimTranscriptionFrame,
+            **kwargs,
         )
-        # CUSTOM CODE: this variable remembers if we prompted the LLM
-        self.sent_aggregation_after_last_interruption = False
 
-    # Relevant functions:
-    # LLMContextAggregator.async def _push_aggregation(self)
-    # and
-    # def LLMResponseAggregator._reset(self):
 
-    # The original pipecat implementation is in:
+class LLMFullResponseAggregator(FrameProcessor):
-    # LLMResponseAggregator.process_frame
+    """This class aggregates Text frames between LLMFullResponseStartFrame and
+    LLMFullResponseEndFrame, then emits the concatenated text as a single text
+    frame.
 
-    # Use cases implemented:
+    given the following frames:
-    #
-    # S: Start, E: End, T: Transcription, I: Interim, X: Text
-    #
-    #        S E -> None
-    #      S T E -> T
-    #    S I T E -> T
-    #    S I E T -> T
-    #  S I E I T -> T
-    #      S E T -> T
-    #    S E I T -> T
-    #
-    #    S I E T1 I T2 -> T1
-    #
-    # and T2 would be dropped.
 
-    # We have:
+        LLMFullResponseStartFrame()
-    # S = UserStartedSpeakingFrame,
+        TextFrame("Hello,")
-    # E = UserStoppedSpeakingFrame,
+        TextFrame(" world.")
-    # T = TranscriptionFrame,
+        TextFrame(" I am")
-    # I = InterimTranscriptionFrame
+        TextFrame(" an LLM.")
+        LLMFullResponseEndFrame()
 
-    # Cases we want to handle:
+    this processor will push,
-    #  - Make sure we never delete some aggregation as it is something said by the user
+
-    #   - Solves case: S T1 I E S T2 E where we lose T1
+        LLMFullResponseStartFrame()
-    #  - Solve case: S T E Bot T (without E S) as the VAD is not activated (yeah case)
+        TextFrame("Hello, world. I am an LLM.")
-    #  - Solve case: S E T1 T2 where T2 is lost. (variation from above)
+        LLMFullResponseEndFrame()
-    #  For the last case we also send StartInterruptionFrame for making sure that the reprompt of the LLM does not make weird repeating messages.
+
+    when passed the last frame.
+
+    >>> async def print_frames(aggregator, frame):
+    ...     async for frame in aggregator.process_frame(frame):
+    ...         if isinstance(frame, TextFrame):
+    ...             print(frame.text)
+    ...         else:
+    ...             print(frame.__class__.__name__)
+
+    >>> aggregator = LLMFullResponseAggregator()
+    >>> asyncio.run(print_frames(aggregator, LLMFullResponseStartFrame()))
+    >>> asyncio.run(print_frames(aggregator, TextFrame("Hello,")))
+    >>> asyncio.run(print_frames(aggregator, TextFrame(" world.")))
+    >>> asyncio.run(print_frames(aggregator, TextFrame(" I am")))
+    >>> asyncio.run(print_frames(aggregator, TextFrame(" an LLM.")))
+    >>> asyncio.run(print_frames(aggregator, LLMFullResponseEndFrame()))
+    LLMFullResponseStartFrame
+    Hello, world. I am an LLM.
+    LLMFullResponseEndFrame
+
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._aggregation = ""
+        self._seen_start_frame = False
 
-    # So the cases would be:
-    #    S E             -> None
-    #    S T E           -> T
-    #    S I T E         -> T
-    #    S I E T         -> T
-    #    S I E I T       -> T
-    #    S E T           -> T
-    #    S E I T         -> T
-    #    S T1 I E S T2 E -> (T1 T2)
-    #    S I E T1 I T2   -> T1 Interruption T2
-    #    S T1 E T2       -> T1 Interruption T2
-    #    S E T1 B T2     -> T1 Bot Interruption T2
-    #    S E T1 T2       -> T1 Interruption T2
-    # see the tests at test_LLM_user_context_aggregator
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        await FrameProcessor.process_frame(self, frame, direction)
+        await super().process_frame(frame, direction)
 
-        send_aggregation = False
+        if isinstance(frame, LLMFullResponseStartFrame):
-
-        if isinstance(frame, self._start_frame):
-            # CUSTOM CODE: dont _aggregation = ""
-            # self._aggregation = ""
-            self._aggregating = True
             self._seen_start_frame = True
-            self._seen_end_frame = False
-            # CUSTOM CODE: _seen_interim_results should be updated by interimframe and accumulator frame only
-            # self._seen_interim_results = False
             await self.push_frame(frame, direction)
-        elif isinstance(frame, self._end_frame):
+        elif isinstance(frame, LLMFullResponseEndFrame):
-            self._seen_end_frame = True
             self._seen_start_frame = False
-
+            await self.push_frame(TextFrame(self._aggregation))
-            # 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):
-            # CUSTOM CODE: send interruption without VAD
-            if self.sent_aggregation_after_last_interruption:
-                await self.push_frame(StartInterruptionFrame())
-                self.sent_aggregation_after_last_interruption = False
-
-            # CUSTOM CODE: do not require _aggregating so we do not lose frames
-            self._aggregation += f" {frame.text}" if self._aggregation else 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.
-            # CUSTOM CODE: important thing is not see start frame and not end frame (so user is still speaking)
-            send_aggregation = not self._seen_start_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):
-            # CUSTOM CODE: send interruption without VAD
-            if self.sent_aggregation_after_last_interruption:
-                await self.push_frame(StartInterruptionFrame())
-                self.sent_aggregation_after_last_interruption = False
-            self._seen_interim_results = True
-        elif self._handle_interruptions and isinstance(frame, StartInterruptionFrame):
-            # CUSTOM CODE: manage new interruptions
-            self.sent_aggregation_after_last_interruption = False
-            await self._push_aggregation()
-            # Reset anyways
-            self._reset()
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, LLMMessagesAppendFrame):
-            self._messages.extend(frame.messages)
-            messages_frame = LLMMessagesFrame(self._messages)
-            await self.push_frame(messages_frame)
-        elif isinstance(frame, LLMMessagesUpdateFrame):
-            # We push the frame downstream so the assistant aggregator gets
-            # updated as well.
             await self.push_frame(frame)
-            # We can now reset this one.
+            self._aggregation = ""
-            self._reset()
+        elif isinstance(frame, TextFrame) and self._seen_start_frame:
-            self._messages = frame.messages
+            self._aggregation += frame.text
-            messages_frame = LLMMessagesFrame(self._messages)
-            await self.push_frame(messages_frame)
         else:
             await self.push_frame(frame, direction)
-
-        if send_aggregation:
-            await self._push_aggregation()

src/pipecat/processors/aggregators/test_LLM_user_context_aggregator.py

@@ -1,16 +1,20 @@
 # tests/test_custom_user_context.py
 
-"""Tests for CustomLLMUserContextAggregator""" 
+"""Tests for CustomLLMUserContextAggregator"""
 
+import asyncio
 import unittest
 
+from dataclasses import dataclass
+from typing import List
 
+from pipecat.clocks.system_clock import SystemClock
 from pipecat.frames.frames import (
+    ControlFrame,
     Frame,
+    StartFrame,
     TranscriptionFrame,
     InterimTranscriptionFrame,
-    StartInterruptionFrame,
-    StopInterruptionFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -35,121 +39,238 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 #    S E T1 T2       -> T1 Interruption T2
 
 
-class StoreFrameProcessor(FrameProcessor):
+@dataclass
-    def __init__(self, storage: list[Frame]) -> None:
+class EndTestFrame(ControlFrame):
+    pass
+
+
+class QueuedFrameProcessor(FrameProcessor):
+    def __init__(self, queue: asyncio.Queue, ignore_start: bool = True):
         super().__init__()
-        self.storage = storage
+        self._queue = queue
+        self._ignore_start = ignore_start
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        self.storage.append(frame)
+        await super().process_frame(frame, direction)
- 
+        if self._ignore_start and isinstance(frame, StartFrame):
-async def make_test(frames_to_send, expected_returned_frames):
+            return
-    context_aggregator = LLMUserContextAggregator(OpenAILLMContext(
+        await self._queue.put(frame)
-            messages=[{"role": "", "content": ""}]
+
-        ))
+
-    storage = []
+async def make_test(
-    storage_processor = StoreFrameProcessor(storage)
+    frames_to_send: List[Frame], expected_returned_frames: List[type]
-    context_aggregator.link(storage_processor)
+) -> List[Frame]:
+    context_aggregator = LLMUserContextAggregator(
+        OpenAILLMContext(messages=[{"role": "", "content": ""}])
+    )
+
+    received = asyncio.Queue()
+    test_processor = QueuedFrameProcessor(received)
+    context_aggregator.link(test_processor)
+
+    await context_aggregator.queue_frame(StartFrame(clock=SystemClock()))
     for frame in frames_to_send:
         await context_aggregator.process_frame(frame, direction=FrameDirection.DOWNSTREAM)
-    print("storage")
+    await context_aggregator.queue_frame(EndTestFrame())
-    for x in storage:
+
-        print(x)
+    received_frames: List[Frame] = []
-    print("expected_returned_frames")
+    running = True
-    for x in expected_returned_frames:
+    while running:
-        print(x)
+        frame = await received.get()
-    assert len(storage) == len(expected_returned_frames)
+        running = not isinstance(frame, EndTestFrame)
-    for expected, real in zip(expected_returned_frames, storage):
+        if running:
+            received_frames.append(frame)
+
+    assert len(received_frames) == len(expected_returned_frames)
+    for real, expected in zip(received_frames, expected_returned_frames):
         assert isinstance(real, expected)
-    return storage
+    return received_frames
+
 
 class TestFrameProcessing(unittest.IsolatedAsyncioTestCase):
-
+    # S E ->
-    # S E -> 
     async def test_s_e(self):
         """S E case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), StopInterruptionFrame(), UserStoppedSpeakingFrame()]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame]
+            UserStartedSpeakingFrame(),
+            UserStoppedSpeakingFrame(),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
     #    S T E           -> T
     async def test_s_t_e(self):
         """S T E case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), TranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame()]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            TranscriptionFrame("Hello", "", ""),
+            UserStoppedSpeakingFrame(),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
     #    S I T E         -> T
     async def test_s_i_t_e(self):
         """S I T E case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), InterimTranscriptionFrame("", "", ""), TranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame()]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            InterimTranscriptionFrame("This", "", ""),
+            TranscriptionFrame("This is a test", "", ""),
+            UserStoppedSpeakingFrame(),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
     #    S I E T         -> T
     async def test_s_i_e_t(self):
         """S I E T case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), InterimTranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame(),  TranscriptionFrame("", "", "")]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            InterimTranscriptionFrame("This", "", ""),
+            UserStoppedSpeakingFrame(),
+            TranscriptionFrame("This is a test", "", ""),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
-
     #    S I E I T       -> T
     async def test_s_i_e_i_t(self):
         """S I E I T case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), InterimTranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame(),  InterimTranscriptionFrame("", "", ""), TranscriptionFrame("", "", "")]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            InterimTranscriptionFrame("This", "", ""),
+            UserStoppedSpeakingFrame(),
+            InterimTranscriptionFrame("This is", "", ""),
+            TranscriptionFrame("This is a test", "", ""),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
     #    S E T           -> T
     async def test_s_e_t(self):
         """S E case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), StopInterruptionFrame(), UserStoppedSpeakingFrame(), TranscriptionFrame("", "", "")]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            UserStoppedSpeakingFrame(),
+            TranscriptionFrame("This is a test", "", ""),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
     #    S E I T         -> T
     async def test_s_e_i_t(self):
         """S E I T case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), StopInterruptionFrame(), UserStoppedSpeakingFrame(), InterimTranscriptionFrame("", "", ""), TranscriptionFrame("", "", "")]
+        frames_to_send = [
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            UserStartedSpeakingFrame(),
+            UserStoppedSpeakingFrame(),
+            InterimTranscriptionFrame("This", "", ""),
+            TranscriptionFrame("This is a test", "", ""),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         await make_test(frames_to_send, expected_returned_frames)
 
-    #    S T1 I E S T2 E -> (T1 T2)
+    #    S T1 I E S T2 E -> "T1 T2"
     async def test_s_t1_i_e_s_t2_e(self):
         """S T1 I E S T2 E case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), TranscriptionFrame("T1", "", ""), InterimTranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame(),
+        frames_to_send = [
-                        StartInterruptionFrame(), UserStartedSpeakingFrame(), TranscriptionFrame("T2", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame()]
+            UserStartedSpeakingFrame(),
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame,
+            TranscriptionFrame("T1", "", ""),
-                                    StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame, OpenAILLMContextFrame]
+            InterimTranscriptionFrame("", "", ""),
+            UserStoppedSpeakingFrame(),
+            UserStartedSpeakingFrame(),
+            TranscriptionFrame("T2", "", ""),
+            UserStoppedSpeakingFrame(),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+        ]
         result = await make_test(frames_to_send, expected_returned_frames)
-        assert result[-1].context.messages[-1]["content"] == " T1 T2"
+        assert result[-1].context.messages[-1]["content"] == "T1 T2"
 
     #    S I E T1 I T2   -> T1 Interruption T2
     async def test_s_i_e_t1_i_t2(self):
         """S I E T1 I T2 case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), InterimTranscriptionFrame("", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame(),
+        frames_to_send = [
-                        TranscriptionFrame("T1", "", ""), InterimTranscriptionFrame("", "", ""), TranscriptionFrame("T2", "", ""),]
+            UserStartedSpeakingFrame(),
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame,
+            InterimTranscriptionFrame("", "", ""),
-                                    OpenAILLMContextFrame, StartInterruptionFrame, OpenAILLMContextFrame]
+            UserStoppedSpeakingFrame(),
+            TranscriptionFrame("T1", "", ""),
+            InterimTranscriptionFrame("", "", ""),
+            TranscriptionFrame("T2", "", ""),
+        ]
+        expected_returned_frames = [
+            UserStartedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            OpenAILLMContextFrame,
+            OpenAILLMContextFrame,
+        ]
         result = await make_test(frames_to_send, expected_returned_frames)
-        assert result[-1].context.messages[-1]["content"] == " T1 T2"
+        assert result[-2].context.messages[-2]["content"] == "T1"
+        assert result[-1].context.messages[-1]["content"] == "T2"
 
-    #    S T1 E T2       -> T1 Interruption T2
+    # #    S T1 E T2       -> T1 Interruption T2
-    async def test_s_t1_e_t2(self):
+    # async def test_s_t1_e_t2(self):
-        """S T1 E T2 case"""
+    #     """S T1 E T2 case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), TranscriptionFrame("T1", "", ""), StopInterruptionFrame(), UserStoppedSpeakingFrame(),
+    #     frames_to_send = [
-                        TranscriptionFrame("T2", "", ""),]
+    #         UserStartedSpeakingFrame(),
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame,
+    #         TranscriptionFrame("T1", "", ""),
-                                    OpenAILLMContextFrame, StartInterruptionFrame, OpenAILLMContextFrame]
+    #         UserStoppedSpeakingFrame(),
-        result = await make_test(frames_to_send, expected_returned_frames)
+    #         TranscriptionFrame("T2", "", ""),
-        assert result[-1].context.messages[-1]["content"] == " T1 T2"
+    #     ]
+    #     expected_returned_frames = [
+    #         UserStartedSpeakingFrame,
+    #         UserStoppedSpeakingFrame,
+    #         OpenAILLMContextFrame,
+    #         OpenAILLMContextFrame,
+    #     ]
+    #     result = await make_test(frames_to_send, expected_returned_frames)
+    #     assert result[-1].context.messages[-1]["content"] == " T1 T2"
 
-    #    S E T1 T2       -> T1 Interruption T2
+    # #    S E T1 T2       -> T1 Interruption T2
-    async def test_s_e_t1_t2(self):
+    # async def test_s_e_t1_t2(self):
-        """S E T1 T2 case"""
+    #     """S E T1 T2 case"""
-        frames_to_send = [StartInterruptionFrame(), UserStartedSpeakingFrame(), StopInterruptionFrame(), UserStoppedSpeakingFrame(),
+    #     frames_to_send = [
-                        TranscriptionFrame("T1", "", ""), TranscriptionFrame("T2", "", ""),]
+    #         UserStartedSpeakingFrame(),
-        expected_returned_frames = [StartInterruptionFrame, UserStartedSpeakingFrame, StopInterruptionFrame, UserStoppedSpeakingFrame,
+    #         UserStoppedSpeakingFrame(),
-                                    OpenAILLMContextFrame, StartInterruptionFrame, OpenAILLMContextFrame]
+    #         TranscriptionFrame("T1", "", ""),
-        result = await make_test(frames_to_send, expected_returned_frames)
+    #         TranscriptionFrame("T2", "", ""),
-        assert result[-1].context.messages[-1]["content"] == " T1 T2"
+    #     ]
+    #     expected_returned_frames = [
+    #         UserStartedSpeakingFrame,
+    #         UserStoppedSpeakingFrame,
+    #         OpenAILLMContextFrame,
+    #         OpenAILLMContextFrame,
+    #     ]
+    #     result = await make_test(frames_to_send, expected_returned_frames)
+    #     assert result[-1].context.messages[-1]["content"] == " T1 T2"