Add timestamp frames and include timestamps in the transcription event and frame

examples/foundational/28a-transcription-update-openai.py

@@ -32,7 +32,10 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class TranscriptHandler:
-    """Simple handler to demonstrate transcript processing."""
+    """Simple handler to demonstrate transcript processing.
+
+    Maintains a list of conversation messages and logs them with timestamps.
+    """
 
     def __init__(self):
         self.messages: List[TranscriptionMessage] = []
@@ -40,18 +43,25 @@ class TranscriptHandler:
     async def on_transcript_update(
         self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame
     ):
-        """Handle new transcript messages."""
+        """Handle new transcript messages.
+
+        Args:
+            processor: The TranscriptProcessor that emitted the update
+            frame: TranscriptionUpdateFrame containing new messages
+        """
         self.messages.extend(frame.messages)
 
         # Log the new messages
         logger.info("New transcript messages:")
         for msg in frame.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+            timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+            logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
-        # Log the full transcript
+        # # Log the full transcript
-        logger.info("Full transcript:")
+        # logger.info("Full transcript:")
-        for msg in self.messages:
+        # for msg in self.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+        #     timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+        #     logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
 
 async def main():

examples/foundational/28b-transcription-update-anthropic.py

@@ -32,7 +32,10 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class TranscriptHandler:
-    """Simple handler to demonstrate transcript processing."""
+    """Simple handler to demonstrate transcript processing.
+
+    Maintains a list of conversation messages and logs them with timestamps.
+    """
 
     def __init__(self):
         self.messages: List[TranscriptionMessage] = []
@@ -40,18 +43,25 @@ class TranscriptHandler:
     async def on_transcript_update(
         self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame
     ):
-        """Handle new transcript messages."""
+        """Handle new transcript messages.
+
+        Args:
+            processor: The TranscriptProcessor that emitted the update
+            frame: TranscriptionUpdateFrame containing new messages
+        """
         self.messages.extend(frame.messages)
 
         # Log the new messages
         logger.info("New transcript messages:")
         for msg in frame.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+            timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+            logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
-        # Log the full transcript
+        # # Log the full transcript
-        logger.info("Full transcript:")
+        # logger.info("Full transcript:")
-        for msg in self.messages:
+        # for msg in self.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+        #     timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+        #     logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
 
 async def main():

examples/foundational/28c-transcription-update-gemini.py

@@ -33,7 +33,10 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 class TranscriptHandler:
-    """Simple handler to demonstrate transcript processing."""
+    """Simple handler to demonstrate transcript processing.
+
+    Maintains a list of conversation messages and logs them with timestamps.
+    """
 
     def __init__(self):
         self.messages: List[TranscriptionMessage] = []
@@ -41,18 +44,25 @@ class TranscriptHandler:
     async def on_transcript_update(
         self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame
     ):
-        """Handle new transcript messages."""
+        """Handle new transcript messages.
+
+        Args:
+            processor: The TranscriptProcessor that emitted the update
+            frame: TranscriptionUpdateFrame containing new messages
+        """
         self.messages.extend(frame.messages)
 
         # Log the new messages
         logger.info("New transcript messages:")
         for msg in frame.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+            timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+            logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
-        # Log the full transcript
+        # # Log the full transcript
-        logger.info("Full transcript:")
+        # logger.info("Full transcript:")
-        for msg in self.messages:
+        # for msg in self.messages:
-            logger.info(f"{msg.role}: {msg.content}")
+        #     timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+        #     logger.info(f"{timestamp}{msg.role}: {msg.content}")
 
 
 async def main():

src/pipecat/frames/frames.py

@@ -207,6 +207,20 @@ class InterimTranscriptionFrame(TextFrame):
         return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
 
 
+@dataclass
+class OpenAILLMContextUserTimestampFrame(DataFrame):
+    """Timestamp information for user message in LLM context."""
+
+    timestamp: str
+
+
+@dataclass
+class OpenAILLMContextAssistantTimestampFrame(DataFrame):
+    """Timestamp information for assistant message in LLM context."""
+
+    timestamp: str
+
+
 @dataclass
 class TranscriptionMessage:
     """A message in a conversation transcript containing the role and content.

src/pipecat/processors/aggregators/llm_response.py

@@ -15,6 +15,7 @@ from pipecat.frames.frames import (
     LLMMessagesFrame,
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
+    OpenAILLMContextUserTimestampFrame,
     StartInterruptionFrame,
     TextFrame,
     TranscriptionFrame,
@@ -26,6 +27,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.time import time_now_iso8601
 
 
 class LLMResponseAggregator(FrameProcessor):
@@ -289,6 +291,10 @@ class LLMContextAggregator(LLMResponseAggregator):
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
+            # Push timestamp frame with current time
+            timestamp_frame = OpenAILLMContextUserTimestampFrame(timestamp=time_now_iso8601())
+            await self.push_frame(timestamp_frame)
+
             # Reset our accumulator state.
             self._reset()
 

src/pipecat/processors/transcript_processor.py

@@ -4,13 +4,15 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import List
+from typing import List, Optional
 
 from loguru import logger
 
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
+    OpenAILLMContextAssistantTimestampFrame,
+    OpenAILLMContextUserTimestampFrame,
     TranscriptionMessage,
     TranscriptionUpdateFrame,
 )
@@ -19,12 +21,12 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
 class TranscriptProcessor(FrameProcessor):
-    """Processes LLM context frames to generate conversation transcripts.
+    """Processes LLM context frames to generate timestamped conversation transcripts.
 
-    This processor monitors OpenAILLMContextFrame frames and extracts conversation
+    This processor monitors OpenAILLMContextFrame frames and their corresponding
-    content, filtering out system messages and function calls. When new messages
+    timestamp frames to build a chronological conversation transcript. Messages are
-    are detected, it emits a TranscriptionUpdateFrame containing only the new
+    stored by role until their matching timestamp frame arrives, then emitted via
-    messages.
+    TranscriptionUpdateFrame.
 
     Each LLM context (OpenAI, Anthropic, Google) provides conversion to the standard format:
     [
@@ -39,8 +41,8 @@ class TranscriptProcessor(FrameProcessor):
     ]
 
     Events:
-        on_transcript_update: Emitted when new transcript messages are available.
+        on_transcript_update: Emitted when timestamped messages are available.
-            Args: TranscriptionUpdateFrame containing new messages.
+            Args: TranscriptionUpdateFrame containing timestamped messages.
 
     Example:
         ```python
@@ -49,7 +51,7 @@ class TranscriptProcessor(FrameProcessor):
         @transcript_processor.event_handler("on_transcript_update")
         async def on_transcript_update(processor, frame):
             for msg in frame.messages:
-                print(f"{msg.role}: {msg.content}")
+                print(f"[{msg.timestamp}] {msg.role}: {msg.content}")
         ```
     """
 
@@ -62,6 +64,8 @@ class TranscriptProcessor(FrameProcessor):
         super().__init__(**kwargs)
         self._processed_messages: List[TranscriptionMessage] = []
         self._register_event_handler("on_transcript_update")
+        self._pending_user_messages: List[TranscriptionMessage] = []
+        self._pending_assistant_messages: List[TranscriptionMessage] = []
 
     def _extract_messages(self, messages: List[dict]) -> List[TranscriptionMessage]:
         """Extract conversation messages from standard format.
@@ -112,7 +116,16 @@ class TranscriptProcessor(FrameProcessor):
         return current[processed_len:]
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process incoming frames, watching for OpenAILLMContextFrame.
+        """Process frames to build a timestamped conversation transcript.
+
+        Handles three frame types in sequence:
+        1. OpenAILLMContextFrame: Contains new messages to be timestamped
+        2. OpenAILLMContextUserTimestampFrame: Timestamp for user messages
+        3. OpenAILLMContextAssistantTimestampFrame: Timestamp for assistant messages
+
+        Messages are stored by role until their corresponding timestamp frame arrives.
+        When a timestamp frame is received, the matching messages are timestamped and
+        emitted in chronological order via TranscriptionUpdateFrame.
 
         Args:
             frame: The frame to process
@@ -124,27 +137,42 @@ class TranscriptProcessor(FrameProcessor):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, OpenAILLMContextFrame):
-            try:
+            # Extract and store messages by role
-                # Convert context messages to standard format
+            standard_messages = []
-                standard_messages = []
+            for msg in frame.context.messages:
-                for msg in frame.context.messages:
+                converted = frame.context.to_standard_messages(msg)
-                    converted = frame.context.to_standard_messages(msg)
+                standard_messages.extend(converted)
-                    standard_messages.extend(converted)
 
-                # Extract and process messages
+            current_messages = self._extract_messages(standard_messages)
-                current_messages = self._extract_messages(standard_messages)
+            new_messages = self._find_new_messages(current_messages)
-                new_messages = self._find_new_messages(current_messages)
 
-                if new_messages:
+            # Store new messages by role
-                    # Update state and notify listeners
+            for msg in new_messages:
-                    self._processed_messages.extend(new_messages)
+                if msg.role == "user":
-                    update_frame = TranscriptionUpdateFrame(messages=new_messages)
+                    self._pending_user_messages.append(msg)
-                    await self._call_event_handler("on_transcript_update", update_frame)
+                elif msg.role == "assistant":
-                    await self.push_frame(update_frame)
+                    self._pending_assistant_messages.append(msg)
 
-            except Exception as e:
+        elif isinstance(frame, OpenAILLMContextUserTimestampFrame):
-                logger.error(f"Error processing transcript in {self}: {e}")
+            # Process pending user messages with timestamp
-                await self.push_error(ErrorFrame(str(e)))
+            if self._pending_user_messages:
+                for msg in self._pending_user_messages:
+                    msg.timestamp = frame.timestamp
+                self._processed_messages.extend(self._pending_user_messages)
+                update_frame = TranscriptionUpdateFrame(messages=self._pending_user_messages)
+                await self._call_event_handler("on_transcript_update", update_frame)
+                await self.push_frame(update_frame)
+                self._pending_user_messages = []
+
+        elif isinstance(frame, OpenAILLMContextAssistantTimestampFrame):
+            # Process pending assistant messages with timestamp
+            if self._pending_assistant_messages:
+                for msg in self._pending_assistant_messages:
+                    msg.timestamp = frame.timestamp
+                self._processed_messages.extend(self._pending_assistant_messages)
+                update_frame = TranscriptionUpdateFrame(messages=self._pending_assistant_messages)
+                await self._call_event_handler("on_transcript_update", update_frame)
+                await self.push_frame(update_frame)
+                self._pending_assistant_messages = []
 
-        # Always push the original frame downstream
         await self.push_frame(frame, direction)

src/pipecat/services/anthropic.py

@@ -26,6 +26,7 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMUpdateSettingsFrame,
+    OpenAILLMContextAssistantTimestampFrame,
     StartInterruptionFrame,
     TextFrame,
     UserImageRawFrame,
@@ -43,6 +44,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import LLMService
+from pipecat.utils.time import time_now_iso8601
 
 try:
     from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
@@ -791,8 +793,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
+            # Push timestamp frame with current time
+            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+            await self.push_frame(timestamp_frame)
+
         except Exception as e:
             logger.error(f"Error processing frame: {e}")

src/pipecat/services/google.py

@@ -23,6 +23,8 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMUpdateSettingsFrame,
+    OpenAILLMContextAssistantTimestampFrame,
+    OpenAILLMContextUserTimestampFrame,
     TextFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
@@ -41,6 +43,7 @@ from pipecat.services.openai import (
     OpenAIUserContextAggregator,
 )
 from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
 
 try:
     import google.ai.generativelanguage as glm
@@ -227,9 +230,14 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
             # if the tasks gets cancelled we won't be able to clear things up.
             self._aggregation = ""
 
+            # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
+            # Push timestamp frame with current time
+            timestamp_frame = OpenAILLMContextUserTimestampFrame(timestamp=time_now_iso8601())
+            await self.push_frame(timestamp_frame)
+
             # Reset our accumulator state.
             self._reset()
 
@@ -300,9 +308,14 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
+            # Push timestamp frame with current time
+            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+            await self.push_frame(timestamp_frame)
+
         except Exception as e:
             logger.exception(f"Error processing frame: {e}")
 

src/pipecat/services/openai.py

@@ -25,6 +25,7 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMUpdateSettingsFrame,
+    OpenAILLMContextAssistantTimestampFrame,
     StartInterruptionFrame,
     TextFrame,
     TTSAudioRawFrame,
@@ -46,6 +47,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import ImageGenService, LLMService, TTSService
+from pipecat.utils.time import time_now_iso8601
 
 try:
     from openai import (
@@ -597,8 +599,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             if run_llm:
                 await self._user_context_aggregator.push_context_frame()
 
+            # Push context frame
             frame = OpenAILLMContextFrame(self._context)
             await self.push_frame(frame)
 
+            # Push timestamp frame with current time
+            timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
+            await self.push_frame(timestamp_frame)
+
         except Exception as e:
             logger.error(f"Error processing frame: {e}")