diff --git a/examples/foundational/28a-transcription-update-openai.py b/examples/foundational/28a-transcription-update-openai.py index ec103ff82..390343465 100644 --- a/examples/foundational/28a-transcription-update-openai.py +++ b/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 - logger.info("Full transcript:") - for msg in self.messages: - logger.info(f"{msg.role}: {msg.content}") + # # Log the full transcript + # logger.info("Full transcript:") + # for msg in self.messages: + # timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" + # logger.info(f"{timestamp}{msg.role}: {msg.content}") async def main(): diff --git a/examples/foundational/28b-transcription-update-anthropic.py b/examples/foundational/28b-transcription-update-anthropic.py index 23ee93a21..1119efad2 100644 --- a/examples/foundational/28b-transcription-update-anthropic.py +++ b/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 - logger.info("Full transcript:") - for msg in self.messages: - logger.info(f"{msg.role}: {msg.content}") + # # Log the full transcript + # logger.info("Full transcript:") + # for msg in self.messages: + # timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" + # logger.info(f"{timestamp}{msg.role}: {msg.content}") async def main(): diff --git a/examples/foundational/28c-transcription-update-gemini.py b/examples/foundational/28c-transcription-update-gemini.py index 27291a7c9..bf9448199 100644 --- a/examples/foundational/28c-transcription-update-gemini.py +++ b/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 - logger.info("Full transcript:") - for msg in self.messages: - logger.info(f"{msg.role}: {msg.content}") + # # Log the full transcript + # logger.info("Full transcript:") + # for msg in self.messages: + # timestamp = f"[{msg.timestamp}] " if msg.timestamp else "" + # logger.info(f"{timestamp}{msg.role}: {msg.content}") async def main(): diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index f74d30371..ab8a6f6ad 100644 --- a/src/pipecat/frames/frames.py +++ b/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. diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 479746471..612375da2 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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() diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index 97173f967..fcd4bfe52 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/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 - content, filtering out system messages and function calls. When new messages - are detected, it emits a TranscriptionUpdateFrame containing only the new - messages. + This processor monitors OpenAILLMContextFrame frames and their corresponding + timestamp frames to build a chronological conversation transcript. Messages are + stored by role until their matching timestamp frame arrives, then emitted via + 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. - Args: TranscriptionUpdateFrame containing new messages. + on_transcript_update: Emitted when timestamped messages are available. + 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: - # Convert context messages to standard format - standard_messages = [] - for msg in frame.context.messages: - converted = frame.context.to_standard_messages(msg) - standard_messages.extend(converted) + # Extract and store messages by role + standard_messages = [] + for msg in frame.context.messages: + converted = frame.context.to_standard_messages(msg) + standard_messages.extend(converted) - # Extract and process messages - current_messages = self._extract_messages(standard_messages) - new_messages = self._find_new_messages(current_messages) + current_messages = self._extract_messages(standard_messages) + new_messages = self._find_new_messages(current_messages) - if new_messages: - # Update state and notify listeners - self._processed_messages.extend(new_messages) - update_frame = TranscriptionUpdateFrame(messages=new_messages) - await self._call_event_handler("on_transcript_update", update_frame) - await self.push_frame(update_frame) + # Store new messages by role + for msg in new_messages: + if msg.role == "user": + self._pending_user_messages.append(msg) + elif msg.role == "assistant": + self._pending_assistant_messages.append(msg) - except Exception as e: - logger.error(f"Error processing transcript in {self}: {e}") - await self.push_error(ErrorFrame(str(e))) + elif isinstance(frame, OpenAILLMContextUserTimestampFrame): + # Process pending user messages with timestamp + 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) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index d8f485296..93cff6f9e 100644 --- a/src/pipecat/services/anthropic.py +++ b/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}") diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 383dde624..c7d32eff3 100644 --- a/src/pipecat/services/google.py +++ b/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}") diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 85e1a95f0..159db8d2f 100644 --- a/src/pipecat/services/openai.py +++ b/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}")