From 41ffe91e207d8f1281710e980a53a4e9446eb649 Mon Sep 17 00:00:00 2001 From: Sabiha Khan Date: Tue, 7 Jul 2026 13:23:19 +0530 Subject: [PATCH] fix: address enhanced transcript review issues --- api/services/pipecat/in_memory_buffers.py | 36 +++++++++---- .../pipecat/realtime_feedback_observer.py | 19 +++++++ api/tests/test_realtime_feedback_observer.py | 53 +++++++++++++++++++ ui/src/types/workflow-configurations.ts | 1 + 4 files changed, 98 insertions(+), 11 deletions(-) diff --git a/api/services/pipecat/in_memory_buffers.py b/api/services/pipecat/in_memory_buffers.py index 1179e4a8..9b10ebde 100644 --- a/api/services/pipecat/in_memory_buffers.py +++ b/api/services/pipecat/in_memory_buffers.py @@ -109,6 +109,8 @@ class InMemoryLogsBuffer: self._current_node_name: Optional[str] = None self._user_speech_start_timestamp: Optional[str] = None self._user_speech_end_timestamp: Optional[str] = None + self._user_speech_start_from_vad = False + self._user_speech_end_from_vad = False self._bot_speech_start_timestamp: Optional[str] = None self._bot_speech_end_timestamp: Optional[str] = None @@ -131,37 +133,50 @@ class InMemoryLogsBuffer: def _now_iso() -> str: return datetime.now(UTC).isoformat(timespec="milliseconds") - def mark_user_started_speaking(self): + def mark_user_started_speaking( + self, timestamp: Optional[str] = None, *, from_vad: bool = False + ): """Record when the user started speaking for the current turn.""" - self._user_speech_start_timestamp = self._now_iso() + if self._user_speech_start_from_vad and not from_vad: + return + + self._user_speech_start_timestamp = timestamp or self._now_iso() self._user_speech_end_timestamp = None + self._user_speech_start_from_vad = from_vad + self._user_speech_end_from_vad = False self._update_latest_payload_start_timestamp( RealtimeFeedbackType.USER_TRANSCRIPTION.value, self._user_speech_start_timestamp, require_final=True, ) - def mark_user_stopped_speaking(self): + def mark_user_stopped_speaking( + self, timestamp: Optional[str] = None, *, from_vad: bool = False + ): """Record when the user stopped speaking and update the latest user event.""" - self._user_speech_end_timestamp = self._now_iso() + if self._user_speech_end_from_vad and not from_vad: + return + + self._user_speech_end_timestamp = timestamp or self._now_iso() + self._user_speech_end_from_vad = from_vad self._update_latest_payload_end_timestamp( RealtimeFeedbackType.USER_TRANSCRIPTION.value, self._user_speech_end_timestamp, require_final=True, ) - def mark_bot_started_speaking(self): + def mark_bot_started_speaking(self, timestamp: Optional[str] = None): """Record when the bot started speaking for the current assistant turn.""" - self._bot_speech_start_timestamp = self._now_iso() + self._bot_speech_start_timestamp = timestamp or self._now_iso() self._bot_speech_end_timestamp = None self._update_latest_payload_start_timestamp( RealtimeFeedbackType.BOT_TEXT.value, self._bot_speech_start_timestamp, ) - def mark_bot_stopped_speaking(self): + def mark_bot_stopped_speaking(self, timestamp: Optional[str] = None): """Record when the bot stopped speaking and update the latest bot event.""" - self._bot_speech_end_timestamp = self._now_iso() + self._bot_speech_end_timestamp = timestamp or self._now_iso() self._update_latest_payload_end_timestamp( RealtimeFeedbackType.BOT_TEXT.value, self._bot_speech_end_timestamp, @@ -217,10 +232,9 @@ class InMemoryLogsBuffer: if self._user_speech_end_timestamp: payload_with_timestamps["end_timestamp"] = self._user_speech_end_timestamp elif event_type == RealtimeFeedbackType.BOT_TEXT.value: - if self._bot_speech_start_timestamp: + bot_interval_is_active = self._bot_speech_end_timestamp is None + if bot_interval_is_active and self._bot_speech_start_timestamp: payload_with_timestamps["timestamp"] = self._bot_speech_start_timestamp - if self._bot_speech_end_timestamp: - payload_with_timestamps["end_timestamp"] = self._bot_speech_end_timestamp if payload_with_timestamps == payload: return event diff --git a/api/services/pipecat/realtime_feedback_observer.py b/api/services/pipecat/realtime_feedback_observer.py index 4579711d..0644edbe 100644 --- a/api/services/pipecat/realtime_feedback_observer.py +++ b/api/services/pipecat/realtime_feedback_observer.py @@ -21,6 +21,7 @@ node changes. """ import json +from datetime import UTC, datetime from typing import TYPE_CHECKING, Awaitable, Callable, Optional, Set from loguru import logger @@ -56,6 +57,8 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, UserMuteStartedFrame, UserMuteStoppedFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import TTFBMetricsData from pipecat.observers.base_observer import BaseObserver, FramePushed @@ -64,6 +67,10 @@ from pipecat.transports.base_output import BaseOutputTransport from pipecat.utils.enums import RealtimeFeedbackType +def _epoch_seconds_to_utc_iso(timestamp: float) -> str: + return datetime.fromtimestamp(timestamp, UTC).isoformat(timespec="milliseconds") + + class RealtimeFeedbackObserver(BaseObserver): """Observer that sends real-time events via WebSocket and persists final transcripts. @@ -157,6 +164,18 @@ class RealtimeFeedbackObserver(BaseObserver): elif isinstance(frame, UserStoppedSpeakingFrame): if self._logs_buffer: self._logs_buffer.mark_user_stopped_speaking() + elif isinstance(frame, VADUserStartedSpeakingFrame): + if self._logs_buffer: + self._logs_buffer.mark_user_started_speaking( + _epoch_seconds_to_utc_iso(frame.timestamp - frame.start_secs), + from_vad=True, + ) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + if self._logs_buffer: + self._logs_buffer.mark_user_stopped_speaking( + _epoch_seconds_to_utc_iso(frame.timestamp - frame.stop_secs), + from_vad=True, + ) # User mute state - WS only (ephemeral state signals, not persisted) elif isinstance(frame, UserMuteStartedFrame): await self._send_ws( diff --git a/api/tests/test_realtime_feedback_observer.py b/api/tests/test_realtime_feedback_observer.py index 909b8f92..e70c81b8 100644 --- a/api/tests/test_realtime_feedback_observer.py +++ b/api/tests/test_realtime_feedback_observer.py @@ -9,6 +9,8 @@ from pipecat.frames.frames import ( TTSTextFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) from pipecat.observers.base_observer import FramePushed from pipecat.processors.frame_processor import FrameDirection @@ -212,3 +214,54 @@ async def test_observer_attaches_backend_speaking_intervals_to_logged_transcript bot_event["payload"]["timestamp"], ) assert bot_event["payload"]["end_timestamp"] + + +@pytest.mark.asyncio +async def test_observer_uses_vad_user_speech_times_over_turn_closure_times(): + async def ws_sender(_message): + pass + + logs_buffer = InMemoryLogsBuffer(workflow_run_id=123) + observer = RealtimeFeedbackObserver(ws_sender=ws_sender, logs_buffer=logs_buffer) + user_aggregator = _FakeAggregator() + assistant_aggregator = _FakeAggregator() + register_turn_log_handlers(logs_buffer, user_aggregator, assistant_aggregator) + + await observer.on_push_frame( + _frame_pushed( + VADUserStartedSpeakingFrame(start_secs=0.2, timestamp=1000.2), + FrameDirection.DOWNSTREAM, + ) + ) + await observer.on_push_frame( + _frame_pushed( + VADUserStoppedSpeakingFrame(stop_secs=0.5, timestamp=1002.5), + FrameDirection.DOWNSTREAM, + ) + ) + await observer.on_push_frame( + _frame_pushed(UserStoppedSpeakingFrame(), FrameDirection.DOWNSTREAM) + ) + + await user_aggregator.handlers["on_user_turn_message_added"]( + user_aggregator, + SimpleNamespace(content="Hello", timestamp="aggregator-user-start"), + ) + + user_event = logs_buffer.get_events()[0] + assert user_event["payload"]["timestamp"] == "1970-01-01T00:16:40.000+00:00" + assert user_event["payload"]["end_timestamp"] == "1970-01-01T00:16:42.000+00:00" + + +@pytest.mark.asyncio +async def test_completed_bot_speech_interval_is_not_reused_for_next_pre_playback_text(): + logs_buffer = InMemoryLogsBuffer(workflow_run_id=123) + + logs_buffer.mark_bot_started_speaking("2026-01-01T00:00:01.000+00:00") + await logs_buffer.append({"type": "rtf-bot-text", "payload": {"text": "First"}}) + logs_buffer.mark_bot_stopped_speaking("2026-01-01T00:00:02.000+00:00") + + await logs_buffer.append({"type": "rtf-bot-text", "payload": {"text": "Second"}}) + second_event = logs_buffer.get_events()[-1] + + assert second_event["payload"] == {"text": "Second"} diff --git a/ui/src/types/workflow-configurations.ts b/ui/src/types/workflow-configurations.ts index db6692f1..dffc985b 100644 --- a/ui/src/types/workflow-configurations.ts +++ b/ui/src/types/workflow-configurations.ts @@ -198,6 +198,7 @@ export function resolveWorkflowConfigurations( ?? FALLBACK_WORKFLOW_CONFIGURATIONS.context_compaction_enabled, transcript_configuration: { ...DEFAULT_TRANSCRIPT_CONFIGURATION, + ...(defaults?.transcript_configuration as Partial | undefined), ...(configurations?.transcript_configuration as Partial | undefined), }, };