add start timestamp to user and assistant turn messages

examples/foundational/28-user-assistant-turns.py

@@ -34,7 +34,6 @@ from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
 from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
-from pipecat.utils.time import time_now_iso8601
 
 load_dotenv(override=True)
 
@@ -61,7 +60,7 @@ class TranscriptHandler:
             f"TranscriptHandler initialized {'with output_file=' + output_file if output_file else 'with log output only'}"
         )
 
-    async def save_message(self, role: str, content: str):
+    async def save_message(self, role: str, content: str, timestamp: str):
         """Save a single transcript message.
 
         Outputs the message to the log and optionally to a file.
@@ -70,7 +69,6 @@ class TranscriptHandler:
             role: Who generated this transcript
             content: The transcript to save
         """
-        timestamp = time_now_iso8601()
         line = f"[{timestamp}] {role}: {content}"
 
         # Always log the message
@@ -91,7 +89,7 @@ class TranscriptHandler:
             message: The new user message
         """
         logger.debug(f"Received user transcript update")
-        await self.save_message("user", message.content)
+        await self.save_message("user", message.content, message.timestamp)
 
     async def on_assistant_transcript(self, message: AssistantTurnStoppedMessage):
         """Handle new assistant transcript message.
@@ -100,7 +98,7 @@ class TranscriptHandler:
             message: The new assistant message
         """
         logger.debug(f"Received assistant transcript update")
-        await self.save_message("assistant", message.content)
+        await self.save_message("assistant", message.content, message.timestamp)
 
 
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -110,11 +110,13 @@ class UserTurnStoppedMessage:
 
     Parameters:
         content: The message content/text.
+        timestamp: When the user turn started.
         user_id: Optional identifier for the user.
 
     """
 
     content: str
+    timestamp: str
     user_id: Optional[str] = None
 
 
@@ -127,10 +129,12 @@ class AssistantTurnStoppedMessage:
 
     Parameters:
         content: The message content/text.
+        timestamp: When the assistant turn started.
 
     """
 
     content: str
+    timestamp: str
 
 
 @dataclass
@@ -142,7 +146,7 @@ class AssistantThoughtMessage:
 
     Parameters:
         content: The message content/text.
-        timestamp: When the thought was started.
+        timestamp: When the thought started.
 
     """
 
@@ -328,6 +332,7 @@ class LLMUserAggregator(LLMContextAggregator):
         user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
 
         self._user_is_muted = False
+        self._user_turn_start_timestamp = ""
 
         self._user_turn_controller = UserTurnController(
             user_turn_strategies=user_turn_strategies,
@@ -527,6 +532,8 @@ class LLMUserAggregator(LLMContextAggregator):
     ):
         logger.debug(f"{self}: User started speaking (user turn start strategy: {strategy})")
 
+        self._user_turn_start_timestamp = time_now_iso8601()
+
         if params.enable_user_speaking_frames:
             await self.broadcast_frame(UserStartedSpeakingFrame)
 
@@ -549,8 +556,11 @@ class LLMUserAggregator(LLMContextAggregator):
         # Always push context frame.
         aggregation = await self.push_aggregation()
 
-        message = UserTurnStoppedMessage(content=aggregation)
+        message = UserTurnStoppedMessage(
+            content=aggregation, timestamp=self._user_turn_start_timestamp
+        )
         await self._call_event_handler("on_user_turn_stopped", strategy, message)
+        self._user_turn_start_timestamp = ""
 
     async def _on_user_turn_stop_timeout(self, controller):
         await self._call_event_handler("on_user_turn_stop_timeout")
@@ -633,6 +643,8 @@ class LLMAssistantAggregator(LLMContextAggregator):
         self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
         self._context_updated_tasks: Set[asyncio.Task] = set()
 
+        self._assistant_turn_start_timestamp = ""
+
         self._thought_append_to_context = False
         self._thought_llm: str = ""
         self._thought_aggregation: List[TextPartForConcatenation] = []
@@ -960,14 +972,20 @@ class LLMAssistantAggregator(LLMContextAggregator):
         self._context_updated_tasks.discard(task)
 
     async def _trigger_assistant_turn_started(self):
+        self._assistant_turn_start_timestamp = time_now_iso8601()
+
         await self._call_event_handler("on_assistant_turn_started")
 
     async def _trigger_assistant_turn_stopped(self):
         aggregation = await self.push_aggregation()
         if aggregation:
-            message = AssistantTurnStoppedMessage(content=aggregation)
+            message = AssistantTurnStoppedMessage(
+                content=aggregation, timestamp=self._assistant_turn_start_timestamp
+            )
             await self._call_event_handler("on_assistant_turn_stopped", message)
 
+            self._assistant_turn_start_timestamp = ""
+
 
 class LLMContextAggregatorPair:
     """Pair of LLM context aggregators for updating context with user and assistant messages."""