diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py deleted file mode 100644 index 2c9fe2917..000000000 --- a/src/pipecat/processors/transcript_processor.py +++ /dev/null @@ -1,370 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Transcript processing utilities for conversation recording and analysis. - -This module provides processors that convert speech and text frames into structured -transcript messages with timestamps, enabling conversation history tracking and analysis. -""" - -from typing import List, Optional - -from loguru import logger - -from pipecat.frames.frames import ( - BotStoppedSpeakingFrame, - CancelFrame, - EndFrame, - Frame, - InterruptionFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - ThoughtTranscriptionMessage, - TranscriptionFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, - TTSTextFrame, -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text -from pipecat.utils.time import time_now_iso8601 - - -class BaseTranscriptProcessor(FrameProcessor): - """Base class for processing conversation transcripts. - - Provides common functionality for handling transcript messages and updates. - """ - - def __init__(self, **kwargs): - """Initialize processor with empty message store. - - Args: - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - self._processed_messages: List[TranscriptionMessage] = [] - self._register_event_handler("on_transcript_update") - - async def _emit_update(self, messages: List[TranscriptionMessage]): - """Emit transcript updates for new messages. - - Args: - messages: New messages to emit in update. - """ - if messages: - self._processed_messages.extend(messages) - update_frame = TranscriptionUpdateFrame(messages=messages) - await self._call_event_handler("on_transcript_update", update_frame) - await self.push_frame(update_frame) - - -class UserTranscriptProcessor(BaseTranscriptProcessor): - """Processes user transcription frames into timestamped conversation messages.""" - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process TranscriptionFrames into user conversation messages. - - Args: - frame: Input frame to process. - direction: Frame processing direction. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, TranscriptionFrame): - message = TranscriptionMessage( - role="user", user_id=frame.user_id, content=frame.text, timestamp=frame.timestamp - ) - await self._emit_update([message]) - - await self.push_frame(frame, direction) - - -class AssistantTranscriptProcessor(BaseTranscriptProcessor): - """Processes assistant TTS text frames and LLM thought frames into timestamped messages. - - This processor aggregates both TTS text frames and LLM thought frames into - complete utterances and thoughts, emitting them as transcript messages. - - An assistant utterance is completed when: - - The bot stops speaking (BotStoppedSpeakingFrame) - - The bot is interrupted (InterruptionFrame) - - The pipeline ends (EndFrame, CancelFrame) - - A thought is completed when: - - The thought ends (LLMThoughtEndFrame) - - The bot is interrupted (InterruptionFrame) - - The pipeline ends (EndFrame, CancelFrame) - """ - - def __init__(self, *, process_thoughts: bool = False, **kwargs): - """Initialize processor with aggregation state. - - Args: - process_thoughts: Whether to process LLM thought frames. Defaults to False. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - - self._process_thoughts = process_thoughts - self._current_assistant_text_parts: List[TextPartForConcatenation] = [] - self._assistant_text_start_time: Optional[str] = None - - self._current_thought_parts: List[TextPartForConcatenation] = [] - self._thought_start_time: Optional[str] = None - self._thought_active = False - - async def _emit_aggregated_assistant_text(self): - """Aggregates and emits text fragments as a transcript message. - - This method aggregates text fragments that may arrive in multiple - TTSTextFrame instances and emits them as a single TranscriptionMessage. - """ - if self._current_assistant_text_parts and self._assistant_text_start_time: - content = concatenate_aggregated_text(self._current_assistant_text_parts) - if content: - logger.trace(f"Emitting aggregated assistant message: {content}") - message = TranscriptionMessage( - role="assistant", - content=content, - timestamp=self._assistant_text_start_time, - ) - await self._emit_update([message]) - else: - logger.trace("No content to emit after stripping whitespace") - - # Reset aggregation state - self._current_assistant_text_parts = [] - self._assistant_text_start_time = None - - async def _emit_aggregated_thought(self): - """Aggregates and emits thought text fragments as a thought transcript message. - - This method aggregates thought fragments that may arrive in multiple - LLMThoughtTextFrame instances and emits them as a single ThoughtTranscriptionMessage. - """ - if self._current_thought_parts and self._thought_start_time: - content = concatenate_aggregated_text(self._current_thought_parts) - if content: - logger.trace(f"Emitting aggregated thought message: {content}") - message = ThoughtTranscriptionMessage( - content=content, - timestamp=self._thought_start_time, - ) - await self._emit_update([message]) - else: - logger.trace("No thought content to emit after stripping whitespace") - - # Reset aggregation state - self._current_thought_parts = [] - self._thought_start_time = None - self._thought_active = False - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames into assistant conversation messages and thought messages. - - Handles different frame types: - - - TTSTextFrame: Aggregates text for current utterance - - LLMThoughtStartFrame: Begins aggregating a new thought - - LLMThoughtTextFrame: Aggregates text for current thought - - LLMThoughtEndFrame: Completes current thought - - BotStoppedSpeakingFrame: Completes current utterance - - InterruptionFrame: Completes current utterance and thought due to interruption - - EndFrame: Completes current utterance and thought at pipeline end - - CancelFrame: Completes current utterance and thought due to cancellation - - Args: - frame: Input frame to process. - direction: Frame processing direction. - """ - await super().process_frame(frame, direction) - - if isinstance(frame, (InterruptionFrame, CancelFrame)): - # Push frame first otherwise our emitted transcription update frame - # might get cleaned up. - await self.push_frame(frame, direction) - # Emit accumulated text and thought with interruptions - await self._emit_aggregated_assistant_text() - if self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - elif isinstance(frame, LLMThoughtStartFrame): - # Start a new thought - if self._process_thoughts: - self._thought_active = True - self._thought_start_time = time_now_iso8601() - self._current_thought_parts = [] - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, LLMThoughtTextFrame): - # Aggregate thought text if we have an active thought - if self._process_thoughts and self._thought_active: - self._current_thought_parts.append( - TextPartForConcatenation( - frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces - ) - ) - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, LLMThoughtEndFrame): - # Emit accumulated thought when thought ends - if self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, TTSTextFrame): - # Start timestamp on first text part - if not self._assistant_text_start_time: - self._assistant_text_start_time = time_now_iso8601() - - self._current_assistant_text_parts.append( - TextPartForConcatenation( - frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces - ) - ) - - # Push frame. - await self.push_frame(frame, direction) - elif isinstance(frame, (BotStoppedSpeakingFrame, EndFrame)): - # Emit accumulated text when bot finishes speaking or pipeline ends. - await self._emit_aggregated_assistant_text() - # Emit accumulated thought at pipeline end if still active - if isinstance(frame, EndFrame) and self._process_thoughts and self._thought_active: - await self._emit_aggregated_thought() - # Push frame. - await self.push_frame(frame, direction) - else: - await self.push_frame(frame, direction) - - -class TranscriptProcessor: - """Factory for creating and managing transcript processors. - - Provides unified access to user and assistant transcript processors - with shared event handling. The assistant processor handles both TTS text - and LLM thought frames. - - Example:: - - transcript = TranscriptProcessor() - - pipeline = Pipeline( - [ - transport.input(), - stt, - transcript.user(), # User transcripts - context_aggregator.user(), - llm, - tts, - transport.output(), - transcript.assistant(), # Assistant transcripts (including thoughts) - context_aggregator.assistant(), - ] - ) - - @transcript.event_handler("on_transcript_update") - async def handle_update(processor, frame): - print(f"New messages: {frame.messages}") - - .. deprecated:: 0.0.99 - `TranscriptProcessor` is deprecated and will be removed in a future version. - Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead. - """ - - def __init__(self, *, process_thoughts: bool = False): - """Initialize factory. - - Args: - process_thoughts: Whether the assistant processor should handle LLM thought - frames. Defaults to False. - """ - self._process_thoughts = process_thoughts - self._user_processor = None - self._assistant_processor = None - self._event_handlers = {} - - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "`TranscriptProcessor` is deprecated and will be removed in a future version. " - "Use `LLMUserAggregator`'s and `LLMAssistantAggregator`'s new events instead.", - DeprecationWarning, - ) - - def user(self, **kwargs) -> UserTranscriptProcessor: - """Get the user transcript processor. - - Args: - **kwargs: Arguments specific to UserTranscriptProcessor. - - Returns: - The user transcript processor instance. - """ - if self._user_processor is None: - self._user_processor = UserTranscriptProcessor(**kwargs) - # Apply any registered event handlers - for event_name, handler in self._event_handlers.items(): - - @self._user_processor.event_handler(event_name) - async def user_handler(processor, frame): - return await handler(processor, frame) - - return self._user_processor - - def assistant(self, **kwargs) -> AssistantTranscriptProcessor: - """Get the assistant transcript processor. - - Args: - **kwargs: Arguments specific to AssistantTranscriptProcessor. - - Returns: - The assistant transcript processor instance. - """ - if self._assistant_processor is None: - self._assistant_processor = AssistantTranscriptProcessor( - process_thoughts=self._process_thoughts, **kwargs - ) - # Apply any registered event handlers - for event_name, handler in self._event_handlers.items(): - - @self._assistant_processor.event_handler(event_name) - async def assistant_handler(processor, frame): - return await handler(processor, frame) - - return self._assistant_processor - - def event_handler(self, event_name: str): - """Register event handler for both processors. - - Args: - event_name: Name of event to handle. - - Returns: - Decorator function that registers handler with both processors. - """ - - def decorator(handler): - self._event_handlers[event_name] = handler - - # Apply handler to existing processors if they exist - if self._user_processor: - - @self._user_processor.event_handler(event_name) - async def user_handler(processor, frame): - return await handler(processor, frame) - - if self._assistant_processor: - - @self._assistant_processor.event_handler(event_name) - async def assistant_handler(processor, frame): - return await handler(processor, frame) - - return handler - - return decorator diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py deleted file mode 100644 index 1dfdd58a3..000000000 --- a/tests/test_transcript_processor.py +++ /dev/null @@ -1,798 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - - -import asyncio -import unittest -from datetime import datetime, timezone -from typing import List, Tuple, cast - -from pipecat.frames.frames import ( - AggregationType, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - CancelFrame, - InterruptionFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - ThoughtTranscriptionMessage, - TranscriptionFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, - TTSTextFrame, -) -from pipecat.processors.transcript_processor import ( - AssistantTranscriptProcessor, - UserTranscriptProcessor, -) -from pipecat.tests.utils import SleepFrame, run_test - - -class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase): - """Tests for UserTranscriptProcessor""" - - async def test_basic_transcription(self): - """Test basic transcription frame processing""" - # Create processor - processor = UserTranscriptProcessor() - - # Create test timestamp - timestamp = datetime.now(timezone.utc).isoformat() - - # Create frames to send - frames_to_send = [ - TranscriptionFrame(text="Hello, world!", user_id="test_user", timestamp=timestamp) - ] - - # Expected frames downstream - note the order: - # 1. TranscriptionUpdateFrame (processor emits the update first) - # 2. TranscriptionFrame (original frame is passed through) - expected_down_frames = [TranscriptionUpdateFrame, TranscriptionFrame] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify the content of the TranscriptionUpdateFrame - update_frame = cast( - TranscriptionUpdateFrame, received_frames[0] - ) # Note: now checking first frame - self.assertIsInstance(update_frame, TranscriptionUpdateFrame) - self.assertEqual(len(update_frame.messages), 1) - message = update_frame.messages[0] - self.assertEqual(message.role, "user") - self.assertEqual(message.content, "Hello, world!") - self.assertEqual(message.user_id, "test_user") - self.assertEqual(message.timestamp, timestamp) - - async def test_event_handler(self): - """Test that event handlers are called with transcript updates""" - # Create processor - processor = UserTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionMessage] = [] - - # Register event handler - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.extend(frame.messages) - - # Create test data - timestamp = datetime.now(timezone.utc).isoformat() - frames_to_send = [ - TranscriptionFrame(text="First message", user_id="test_user", timestamp=timestamp), - TranscriptionFrame(text="Second message", user_id="test_user", timestamp=timestamp), - ] - - expected_down_frames = [ - TranscriptionUpdateFrame, - TranscriptionFrame, # First message - TranscriptionUpdateFrame, - TranscriptionFrame, # Second message - ] - - # Run test - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify event handler received updates - self.assertEqual(len(received_updates), 2) - - # Check first message - self.assertEqual(received_updates[0].role, "user") - self.assertEqual(received_updates[0].content, "First message") - self.assertEqual(received_updates[0].timestamp, timestamp) - - # Check second message - self.assertEqual(received_updates[1].role, "user") - self.assertEqual(received_updates[1].content, "Second message") - self.assertEqual(received_updates[1].timestamp, timestamp) - - async def test_text_aggregation(self): - """Test that TTSTextFrames are properly aggregated into a single message""" - # Create processor - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create test frames simulating bot speaking multiple text chunks - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), # Wait for StartedSpeaking to process - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="How", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="are", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="you?", aggregated_by=AggregationType.WORD), - SleepFrame(), # Wait for text frames to queue - BotStoppedSpeakingFrame(), - ] - - # Expected order: - # 1. BotStartedSpeakingFrame (system frame, immediate) - # 2. All queued TTSTextFrames - # 3. BotStoppedSpeakingFrame (system frame, immediate) - # 4. TranscriptionUpdateFrame (after aggregation) - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - BotStoppedSpeakingFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify update was received - self.assertEqual(len(received_updates), 1) - - # Get the update frame - update_frame = received_updates[0] - - # Should have one aggregated message - self.assertEqual(len(update_frame.messages), 1) - - message = update_frame.messages[0] - self.assertEqual(message.role, "assistant") - self.assertEqual(message.content, "Hello world! How are you?") - - # Verify timestamp exists - self.assertIsNotNone(message.timestamp) - - # All frames should be passed through in order, with update at end - downstream_update = cast(TranscriptionUpdateFrame, received_frames[-2]) - self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?") - - async def test_empty_text_handling(self): - """Test that empty messages are not emitted""" - processor = AssistantTranscriptProcessor() - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="", aggregated_by=AggregationType.WORD), # Empty text - TTSTextFrame(text=" ", aggregated_by=AggregationType.WORD), # Just whitespace - TTSTextFrame(text="\n", aggregated_by=AggregationType.WORD), # Just newline - BotStoppedSpeakingFrame(), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - # From our earlier tests, we know BotStoppedSpeakingFrame comes before TTSTextFrames - expected_down_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, # empty - TTSTextFrame, # whitespace - TTSTextFrame, # newline - # No TranscriptionUpdateFrame since content is empty after stripping - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - self.assertEqual(len(received_updates), 0, "No updates should be emitted for empty content") - - async def test_interruption_handling(self): - """Test that messages are properly captured when bot is interrupted""" - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Simulate bot being interrupted mid-sentence - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD), - SleepFrame(), - InterruptionFrame(), # User interrupts here - SleepFrame(), - BotStartedSpeakingFrame(), - TTSTextFrame(text="New", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="response", aggregated_by=AggregationType.WORD), - SleepFrame(), - BotStoppedSpeakingFrame(), - ] - - # Actual order of frames: - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, # "Hello" - TTSTextFrame, # "world!" - InterruptionFrame, - TranscriptionUpdateFrame, # First message (emitted due to interruption) - BotStartedSpeakingFrame, - TTSTextFrame, # "New" - TTSTextFrame, # "response" - TranscriptionUpdateFrame, # Second message - BotStoppedSpeakingFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Should have received two updates - self.assertEqual(len(received_updates), 2) - - # First update should be interrupted message - first_message = received_updates[0].messages[0] - self.assertEqual(first_message.role, "assistant") - self.assertEqual(first_message.content, "Hello world!") - self.assertIsNotNone(first_message.timestamp) - - # Second update should be new response - second_message = received_updates[1].messages[0] - self.assertEqual(second_message.role, "assistant") - self.assertEqual(second_message.content, "New response") - self.assertIsNotNone(second_message.timestamp) - - # Verify timestamps are different - self.assertNotEqual(first_message.timestamp, second_message.timestamp) - - async def test_end_frame_handling(self): - """Test that final messages are captured when pipeline ends normally""" - processor = AssistantTranscriptProcessor() - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world", aggregated_by=AggregationType.WORD), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, # Final message emitted due to EndFrame - ] - - # Run test - EndFrame will be sent automatically - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertEqual(message.role, "assistant") - self.assertEqual(message.content, "Hello world") - - async def test_cancel_frame_handling(self): - """Test that messages are properly captured when pipeline is cancelled""" - processor = AssistantTranscriptProcessor() - - # Track updates with timestamps to verify order - received_updates: List[Tuple[str, float]] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - # Record message content and time received - received_updates.append((frame.messages[0].content, asyncio.get_event_loop().time())) - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="world", aggregated_by=AggregationType.WORD), - SleepFrame(), # Ensure messages are processed - CancelFrame(), - ] - - # We don't need to verify frame order, just that CancelFrame triggers message emission - expected_down_frames = [ - BotStartedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - CancelFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - send_end_frame=False, - ) - - # Verify that we received an update - self.assertEqual(len(received_updates), 1, "Should receive one update before cancellation") - content, _ = received_updates[0] - self.assertEqual(content, "Hello world") - - async def test_transcript_processor_factory(self): - """Test that factory properly manages processors and event handlers""" - from pipecat.processors.transcript_processor import TranscriptProcessor - - factory = TranscriptProcessor() - received_updates: List[TranscriptionMessage] = [] - - # Register handler with factory - @factory.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.extend(frame.messages) - - # Get processors and verify they're reused - user_proc1 = factory.user() - user_proc2 = factory.user() - self.assertIs(user_proc1, user_proc2, "User processor should be reused") - - asst_proc1 = factory.assistant() - asst_proc2 = factory.assistant() - self.assertIs(asst_proc1, asst_proc2, "Assistant processor should be reused") - - # Test user processor - timestamp = datetime.now(timezone.utc).isoformat() - frames_to_send = [ - TranscriptionFrame(text="User message", user_id="user1", timestamp=timestamp) - ] - - await run_test( - user_proc1, - frames_to_send=frames_to_send, - expected_down_frames=[TranscriptionUpdateFrame, TranscriptionFrame], - ) - - # Test assistant processor - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - TTSTextFrame(text="Assistant", aggregated_by=AggregationType.WORD), - TTSTextFrame(text="message", aggregated_by=AggregationType.WORD), - BotStoppedSpeakingFrame(), - ] - - # The actual order we see in the output: - await run_test( - asst_proc1, - frames_to_send=frames_to_send, - expected_down_frames=[ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - ], - ) - - # Verify both processors triggered the same handler - self.assertEqual(len(received_updates), 2) - self.assertEqual(received_updates[0].role, "user") - self.assertEqual(received_updates[0].content, "User message") - self.assertEqual(received_updates[1].role, "assistant") - self.assertEqual(received_updates[1].content, "Assistant message") - - async def test_text_fragments_with_spaces(self): - """Test aggregating text fragments with various spacing patterns""" - processor = AssistantTranscriptProcessor() - - # Track received updates - received_updates = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Test the specific pattern shared - def make_tts_text_frame(text: str) -> TTSTextFrame: - frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD) - frame.includes_inter_frame_spaces = True - return frame - - frames_to_send = [ - BotStartedSpeakingFrame(), - SleepFrame(), - make_tts_text_frame("Hello"), - make_tts_text_frame(" there"), - make_tts_text_frame("!"), - make_tts_text_frame(" How"), - make_tts_text_frame("'s"), - make_tts_text_frame(" it"), - make_tts_text_frame(" going"), - make_tts_text_frame("?"), - BotStoppedSpeakingFrame(), - ] - - expected_down_frames = [ - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TTSTextFrame, - TranscriptionUpdateFrame, - ] - - # Run test - received_frames, _ = await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify result - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertEqual(message.role, "assistant") - # Should be properly joined without extra spaces - self.assertEqual(message.content, "Hello there! How's it going?") - - -class TestThoughtTranscription(unittest.IsolatedAsyncioTestCase): - """Tests for thought transcription in AssistantTranscriptProcessor""" - - async def test_basic_thought_transcription(self): - """Test basic thought frame processing""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create frames for a simple thought - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Let me think about this..."), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify update was received - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Let me think about this...") - self.assertIsNotNone(message.timestamp) - - async def test_thought_aggregation(self): - """Test that thought text frames are properly aggregated""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Create frames simulating chunked thought text - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="The user "), - LLMThoughtTextFrame(text="is asking "), - LLMThoughtTextFrame(text="about electric "), - LLMThoughtTextFrame(text="cars."), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify aggregation - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "The user is asking about electric cars.") - - async def test_thought_with_interruption(self): - """Test that thoughts are properly captured when interrupted""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="I need to consider "), - LLMThoughtTextFrame(text="multiple factors"), - SleepFrame(), - InterruptionFrame(), # User interrupts - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - InterruptionFrame, - TranscriptionUpdateFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify thought was captured on interruption - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "I need to consider multiple factors") - - async def test_thought_with_cancel(self): - """Test that thoughts are properly captured when cancelled""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Starting analysis"), - SleepFrame(), - CancelFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - CancelFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - send_end_frame=False, - ) - - # Verify thought was captured on cancellation - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Starting analysis") - - async def test_thought_with_end_frame(self): - """Test that thoughts are captured when pipeline ends normally""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Final thought"), - # Pipeline ends here; run_test will automatically send EndFrame - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify thought was captured on EndFrame - self.assertEqual(len(received_updates), 1) - message = received_updates[0].messages[0] - self.assertIsInstance(message, ThoughtTranscriptionMessage) - self.assertEqual(message.content, "Final thought") - - async def test_multiple_thoughts(self): - """Test multiple separate thoughts in sequence""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - # First thought - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="First consideration"), - LLMThoughtEndFrame(), - # Second thought - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text="Second consideration"), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - LLMThoughtStartFrame, - LLMThoughtTextFrame, - TranscriptionUpdateFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify both thoughts were captured - self.assertEqual(len(received_updates), 2) - - first_message = received_updates[0].messages[0] - self.assertIsInstance(first_message, ThoughtTranscriptionMessage) - self.assertEqual(first_message.content, "First consideration") - - second_message = received_updates[1].messages[0] - self.assertIsInstance(second_message, ThoughtTranscriptionMessage) - self.assertEqual(second_message.content, "Second consideration") - - async def test_empty_thought_handling(self): - """Test that empty thoughts are not emitted""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - frames_to_send = [ - LLMThoughtStartFrame(), - LLMThoughtTextFrame(text=""), # Empty - LLMThoughtTextFrame(text=" "), # Just whitespace - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtStartFrame, - LLMThoughtTextFrame, - LLMThoughtTextFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify no updates emitted for empty content - self.assertEqual(len(received_updates), 0) - - async def test_thought_without_start_frame(self): - """Test that thought text without start frame is ignored""" - processor = AssistantTranscriptProcessor(process_thoughts=True) - - received_updates: List[TranscriptionUpdateFrame] = [] - - @processor.event_handler("on_transcript_update") - async def handle_update(proc, frame: TranscriptionUpdateFrame): - received_updates.append(frame) - - # Send thought text without start frame - frames_to_send = [ - LLMThoughtTextFrame(text="This should be ignored"), - LLMThoughtEndFrame(), - ] - - expected_down_frames = [ - LLMThoughtTextFrame, - LLMThoughtEndFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - # Verify no updates since thought wasn't properly started - self.assertEqual(len(received_updates), 0) - - -if __name__ == "__main__": - unittest.main()