From 25f6ba76d68cc7a50961704a86931d24c13039db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 8 Jan 2026 21:24:21 -0800 Subject: [PATCH] add start timestamp to user and assistant turn messages --- .../foundational/28-user-assistant-turns.py | 8 +++---- .../aggregators/llm_response_universal.py | 24 ++++++++++++++++--- 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/examples/foundational/28-user-assistant-turns.py b/examples/foundational/28-user-assistant-turns.py index bb8f5df47..da405088f 100644 --- a/examples/foundational/28-user-assistant-turns.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index ca8870f60..c47579fba 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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."""