Update the AssistantTranscriptProcessor to use TTSTextFrames in place of OpenAILLMContextFrames

src/pipecat/processors/transcript_processor.py

@@ -4,16 +4,21 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+from datetime import datetime, timezone
 from typing import List
 
+from loguru import logger
+
 from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
+    EndFrame,
     Frame,
-    OpenAILLMContextAssistantTimestampFrame,
+    StartInterruptionFrame,
     TranscriptionFrame,
     TranscriptionMessage,
     TranscriptionUpdateFrame,
+    TTSTextFrame,
 )
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
@@ -64,89 +69,81 @@ class UserTranscriptProcessor(BaseTranscriptProcessor):
 
 
 class AssistantTranscriptProcessor(BaseTranscriptProcessor):
-    """Processes assistant LLM context frames into timestamped conversation messages."""
+    """Processes assistant TTS text frames into timestamped conversation messages.
+
+    This processor aggregates TTS text frames into complete utterances and emits them as
+    transcript messages. Utterances are completed when:
+    - The bot stops speaking (BotStoppedSpeakingFrame)
+    - The bot is interrupted (StartInterruptionFrame)
+    - The pipeline ends (EndFrame)
+
+    Attributes:
+        _current_text_parts: List of text fragments being aggregated for current utterance
+        _aggregation_start_time: Timestamp when the current utterance began
+    """
 
     def __init__(self, **kwargs):
-        """Initialize processor with empty message stores."""
+        """Initialize processor with aggregation state."""
         super().__init__(**kwargs)
-        self._pending_assistant_messages: List[TranscriptionMessage] = []
+        self._current_text_parts: List[str] = []
+        self._aggregation_start_time: datetime | None = None
 
-    def _extract_messages(self, messages: List[dict]) -> List[TranscriptionMessage]:
+    async def _emit_aggregated_text(self):
-        """Extract assistant messages from the OpenAI standard message format.
+        """Emit aggregated text as a transcript message."""
+        if self._current_text_parts and self._aggregation_start_time:
+            content = " ".join(self._current_text_parts).strip()
+            if content:
+                # Format timestamp with 3 decimal places
+                formatted_timestamp = (
+                    self._aggregation_start_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "+00:00"
+                )
+                logger.debug(f"Emitting aggregated assistant message: {content}")
+                message = TranscriptionMessage(
+                    role="assistant",
+                    content=content,
+                    timestamp=formatted_timestamp,
+                )
+                await self._emit_update([message])
+            else:
+                logger.debug("No content to emit after stripping whitespace")
 
-        Args:
+            # Reset aggregation state
-            messages: List of messages in OpenAI format, which can be either:
+            self._current_text_parts = []
-                - Simple format: {"role": "user", "content": "Hello"}
+            self._aggregation_start_time = None
-                - Content list: {"role": "user", "content": [{"type": "text", "text": "Hello"}]}
-
-        Returns:
-            List[TranscriptionMessage]: Normalized conversation messages
-        """
-        result = []
-        for msg in messages:
-            if msg["role"] != "assistant":
-                continue
-
-            content = msg.get("content")
-            if isinstance(content, str):
-                if content:
-                    result.append(TranscriptionMessage(role="assistant", content=content))
-            elif isinstance(content, list):
-                text_parts = []
-                for part in content:
-                    if isinstance(part, dict) and part.get("type") == "text":
-                        text_parts.append(part["text"])
-
-                if text_parts:
-                    result.append(
-                        TranscriptionMessage(role="assistant", content=" ".join(text_parts))
-                    )
-
-        return result
-
-    def _find_new_messages(self, current: List[TranscriptionMessage]) -> List[TranscriptionMessage]:
-        """Find unprocessed messages from current list.
-
-        Args:
-            current: List of current messages
-
-        Returns:
-            List[TranscriptionMessage]: New messages not yet processed
-        """
-        if not self._processed_messages:
-            return current
-
-        processed_len = len(self._processed_messages)
-        if len(current) <= processed_len:
-            return []
-
-        return current[processed_len:]
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames into assistant conversation messages.
 
+        Handles different frame types:
+        - TTSTextFrame: Aggregates text for current utterance
+        - BotStoppedSpeakingFrame: Completes current utterance
+        - StartInterruptionFrame: Completes current utterance due to interruption
+        - EndFrame: Completes current utterance at pipeline end
+
         Args:
             frame: Input frame to process
             direction: Frame processing direction
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, OpenAILLMContextFrame):
+        if isinstance(frame, TTSTextFrame):
-            standard_messages = []
+            # Start timestamp on first text part
-            for msg in frame.context.messages:
+            if not self._aggregation_start_time:
-                converted = frame.context.to_standard_messages(msg)
+                self._aggregation_start_time = datetime.now(timezone.utc)
-                standard_messages.extend(converted)
 
-            current_messages = self._extract_messages(standard_messages)
+            self._current_text_parts.append(frame.text)
-            new_messages = self._find_new_messages(current_messages)
-            self._pending_assistant_messages.extend(new_messages)
 
-        elif isinstance(frame, OpenAILLMContextAssistantTimestampFrame):
+        elif isinstance(frame, BotStoppedSpeakingFrame):
-            if self._pending_assistant_messages:
+            # Emit accumulated text when bot finishes speaking
-                for msg in self._pending_assistant_messages:
+            await self._emit_aggregated_text()
-                    msg.timestamp = frame.timestamp
+
-                await self._emit_update(self._pending_assistant_messages)
+        elif isinstance(frame, StartInterruptionFrame):
-                self._pending_assistant_messages = []
+            # Emit any pending text when interrupted
+            await self._emit_aggregated_text()
+
+        elif isinstance(frame, EndFrame):
+            # Emit any remaining text when pipeline ends
+            await self._emit_aggregated_text()
 
         await self.push_frame(frame, direction)
 
@@ -170,8 +167,8 @@ class TranscriptProcessor:
                 llm,
                 tts,
                 transport.output(),
+                transcript.assistant_tts(),     # Assistant transcripts
                 context_aggregator.assistant(),
-                transcript.assistant(),         # Assistant transcripts
             ]
         )