apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-07-25 12:01:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat: add additional timestamps in call transcript optionally

Browse source
This commit is contained in:
Sabiha Khan 2026-07-06 09:12:50 +05:30
parent b3e09be9b0
commit 44053d0a6d
12 changed files with 317 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

5
api/services/pipecat/event_handlers.py
View file

@ -70,6 +70,7 @@ def register_event_handlers(
pre_call_fetch_task: asyncio.Task | None = None,
user_provider_id: str | None = None,
integration_runtime_sessions: list[IntegrationRuntimeSession] | None = None,
include_transcript_end_timestamps: bool = False,
):
"""Register all event handlers for transport and task events.
@ -386,7 +387,9 @@ def register_event_handlers(
else:
logger.debug("Bot audio buffer is empty, skipping upload")
transcript_text = in_memory_logs_buffer.generate_transcript_text()
transcript_text = in_memory_logs_buffer.generate_transcript_text(
include_end_timestamps=include_transcript_end_timestamps
)
if not transcript_text:
logger.debug("No transcript events in logs buffer, skipping upload")

112
api/services/pipecat/in_memory_buffers.py
View file

@ -107,6 +107,10 @@ class InMemoryLogsBuffer:
self._turn_counter = 0
self._current_node_id: Optional[str] = None
self._current_node_name: Optional[str] = None
self._user_speech_start_timestamp: Optional[str] = None
self._user_speech_end_timestamp: Optional[str] = None
self._bot_speech_start_timestamp: Optional[str] = None
self._bot_speech_end_timestamp: Optional[str] = None
def set_current_node(self, node_id: str, node_name: str):
"""Set the current node ID and name to be injected into subsequent events."""
@ -123,11 +127,111 @@ class InMemoryLogsBuffer:
"""Get the current node name."""
return self._current_node_name
@staticmethod
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
def mark_user_started_speaking(self):
"""Record when the user started speaking for the current turn."""
self._user_speech_start_timestamp = self._now_iso()
self._user_speech_end_timestamp = None
self._update_latest_payload_start_timestamp(
RealtimeFeedbackType.USER_TRANSCRIPTION.value,
self._user_speech_start_timestamp,
require_final=True,
)
def mark_user_stopped_speaking(self):
"""Record when the user stopped speaking and update the latest user event."""
self._user_speech_end_timestamp = self._now_iso()
self._update_latest_payload_end_timestamp(
RealtimeFeedbackType.USER_TRANSCRIPTION.value,
self._user_speech_end_timestamp,
require_final=True,
)
def mark_bot_started_speaking(self):
"""Record when the bot started speaking for the current assistant turn."""
self._bot_speech_start_timestamp = 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):
"""Record when the bot stopped speaking and update the latest bot event."""
self._bot_speech_end_timestamp = self._now_iso()
self._update_latest_payload_end_timestamp(
RealtimeFeedbackType.BOT_TEXT.value,
self._bot_speech_end_timestamp,
)
def _find_latest_open_payload(
self, event_type: str, *, require_final: bool = False
) -> dict | None:
for event in reversed(self._events):
if event.get("type") != event_type:
continue
payload = event.get("payload")
if not isinstance(payload, dict):
continue
if require_final and payload.get("final") is not True:
continue
if payload.get("end_timestamp"):
continue
return payload
return None
def _update_latest_payload_start_timestamp(
self, event_type: str, start_timestamp: str, *, require_final: bool = False
):
payload = self._find_latest_open_payload(
event_type, require_final=require_final
)
if payload is not None:
payload["timestamp"] = start_timestamp
def _update_latest_payload_end_timestamp(
self, event_type: str, end_timestamp: str, *, require_final: bool = False
):
payload = self._find_latest_open_payload(
event_type, require_final=require_final
)
if payload is not None:
payload["end_timestamp"] = end_timestamp
def _event_with_speech_timestamps(self, event: dict) -> dict:
event_type = event.get("type")
payload = event.get("payload")
if not isinstance(payload, dict):
return event
payload_with_timestamps = dict(payload)
if (
event_type == RealtimeFeedbackType.USER_TRANSCRIPTION.value
and payload.get("final") is True
):
if self._user_speech_start_timestamp:
payload_with_timestamps["timestamp"] = self._user_speech_start_timestamp
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:
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
return {**event, "payload": payload_with_timestamps}
async def append(self, event: dict):
"""Append a feedback event to the buffer with timestamp and current node."""
event = self._event_with_speech_timestamps(event)
timestamped_event = stamp_realtime_feedback_event(
event,
timestamp=datetime.now(UTC).isoformat(),
timestamp=self._now_iso(),
turn=self._turn_counter,
node_id=self._current_node_id,
node_name=self._current_node_name,
@ -166,13 +270,15 @@ class InMemoryLogsBuffer:
return True
return False
def generate_transcript_text(self) -> str:
def generate_transcript_text(self, *, include_end_timestamps: bool = False) -> str:
"""Generate transcript text from logged events.
Filters for rtf-user-transcription (final) and rtf-bot-text events,
formats them as '[timestamp] user/assistant: text\\n'.
"""
return _generate_transcript_text(self._sorted_events())
return _generate_transcript_text(
self._sorted_events(), include_end_timestamps=include_end_timestamps
)
@property
def is_empty(self) -> bool:

6
api/services/pipecat/realtime_feedback_events.py
View file

@ -30,6 +30,7 @@ def build_user_transcription_event(
text: str,
final: bool,
timestamp: str | None = None,
end_timestamp: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
@ -38,6 +39,8 @@ def build_user_transcription_event(
}
if timestamp is not None:
payload["timestamp"] = timestamp
if end_timestamp is not None:
payload["end_timestamp"] = end_timestamp
if user_id is not None:
payload["user_id"] = user_id
return {
@ -50,10 +53,13 @@ def build_bot_text_event(
*,
text: str,
timestamp: str | None = None,
end_timestamp: str | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {"text": text}
if timestamp is not None:
payload["timestamp"] = timestamp
if end_timestamp is not None:
payload["end_timestamp"] = end_timestamp
return {
"type": RealtimeFeedbackType.BOT_TEXT.value,
"payload": payload,

14
api/services/pipecat/realtime_feedback_observer.py
View file

@ -52,6 +52,8 @@ from pipecat.frames.frames import (
TranscriptionFrame,
TTSSpeakFrame,
TTSTextFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
UserMuteStartedFrame,
UserMuteStoppedFrame,
)
@ -138,13 +140,23 @@ class RealtimeFeedbackObserver(BaseObserver):
return
# Bot speaking state - WS only (ephemeral state signals, not persisted)
elif isinstance(frame, BotStartedSpeakingFrame):
if self._logs_buffer:
self._logs_buffer.mark_bot_started_speaking()
await self._send_ws(
{"type": RealtimeFeedbackType.BOT_STARTED_SPEAKING.value, "payload": {}}
)
elif isinstance(frame, BotStoppedSpeakingFrame):
if self._logs_buffer:
self._logs_buffer.mark_bot_stopped_speaking()
await self._send_ws(
{"type": RealtimeFeedbackType.BOT_STOPPED_SPEAKING.value, "payload": {}}
)
elif isinstance(frame, UserStartedSpeakingFrame):
if self._logs_buffer:
self._logs_buffer.mark_user_started_speaking()
elif isinstance(frame, UserStoppedSpeakingFrame):
if self._logs_buffer:
self._logs_buffer.mark_user_stopped_speaking()
# User mute state - WS only (ephemeral state signals, not persisted)
elif isinstance(frame, UserMuteStartedFrame):
await self._send_ws(
@ -314,6 +326,7 @@ def register_turn_log_handlers(
text=message.content,
final=True,
timestamp=message.timestamp,
end_timestamp=getattr(message, "end_timestamp", None),
)
)
except Exception as e:
@ -327,6 +340,7 @@ def register_turn_log_handlers(
build_bot_text_event(
text=message.content,
timestamp=message.timestamp,
end_timestamp=getattr(message, "end_timestamp", None),
)
)
except Exception as e:

5
api/services/pipecat/run_pipeline.py
View file

@ -545,6 +545,10 @@ async def _run_pipeline_impl(
max_call_duration_seconds = DEFAULT_MAX_CALL_DURATION_SECONDS
max_user_idle_timeout = DEFAULT_MAX_USER_IDLE_TIMEOUT_SECONDS
keyterms = None # Dictionary words for STT boosting
transcript_config = run_configs.get("transcript_configuration") or {}
include_transcript_end_timestamps = bool(
transcript_config.get("include_end_timestamps", False)
)
if run_configs:
if "max_call_duration" in run_configs:
@ -1045,6 +1049,7 @@ async def _run_pipeline_impl(
pre_call_fetch_task=pre_call_fetch_task,
user_provider_id=user_provider_id,
integration_runtime_sessions=integration_runtime_sessions,
include_transcript_end_timestamps=include_transcript_end_timestamps,
)
register_audio_data_handler(audio_buffer, workflow_run_id, in_memory_audio_buffer)

37
api/tests/test_realtime_feedback_events.py
View file

@ -1,10 +1,12 @@
from api.services.pipecat.realtime_feedback_events import (
build_bot_text_event,
build_function_call_end_event,
build_user_transcription_event,
build_node_transition_event,
realtime_feedback_event_sort_key,
stamp_realtime_feedback_event,
)
from api.utils.transcript import generate_transcript_text
def test_build_function_call_end_event_serializes_results():
@ -51,3 +53,38 @@ def test_stamp_and_sort_realtime_feedback_events():
assert events == [bot_text, node_transition]
assert node_transition["node_id"] == "node-1"
assert node_transition["node_name"] == "Greeting"
def test_transcript_can_include_end_timestamps_without_changing_default_format():
events = [
stamp_realtime_feedback_event(
build_bot_text_event(
text="Can you confirm your date of birth?",
timestamp="2026-01-01T00:00:01+00:00",
end_timestamp="2026-01-01T00:00:04+00:00",
),
timestamp="2026-01-01T00:00:05+00:00",
turn=0,
),
stamp_realtime_feedback_event(
build_user_transcription_event(
text="January fifth",
final=True,
timestamp="2026-01-01T00:00:06+00:00",
end_timestamp="2026-01-01T00:00:08+00:00",
),
timestamp="2026-01-01T00:00:09+00:00",
turn=1,
),
]
assert generate_transcript_text(events) == (
"[2026-01-01T00:00:01+00:00] assistant: Can you confirm your date of birth?\n"
"[2026-01-01T00:00:06+00:00] user: January fifth\n"
)
assert generate_transcript_text(events, include_end_timestamps=True) == (
"[2026-01-01T00:00:01+00:00 -> 2026-01-01T00:00:04+00:00] "
"assistant: Can you confirm your date of birth?\n"
"[2026-01-01T00:00:06+00:00 -> 2026-01-01T00:00:08+00:00] "
"user: January fifth\n"
)

61
api/tests/test_realtime_feedback_observer.py
View file

@ -1,7 +1,14 @@
from types import SimpleNamespace
import pytest
from pipecat.frames.frames import TranscriptionFrame, TTSTextFrame
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
TranscriptionFrame,
TTSTextFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.observers.base_observer import FramePushed
from pipecat.processors.frame_processor import FrameDirection
from pipecat.transports.base_output import BaseOutputTransport
@ -144,3 +151,55 @@ async def test_turn_log_handlers_persist_user_message_added_events():
"timestamp": "2026-01-01T00:00:00+00:00",
}
assert events[0]["turn"] == 1
@pytest.mark.asyncio
async def test_observer_attaches_backend_speaking_intervals_to_logged_transcript_events():
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(UserStartedSpeakingFrame(), 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="January fifth",
timestamp="aggregator-user-start",
),
)
await observer.on_push_frame(
_frame_pushed(BotStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
)
await assistant_aggregator.handlers["on_assistant_turn_stopped"](
assistant_aggregator,
SimpleNamespace(
content="Thank you",
timestamp="aggregator-bot-start",
),
)
await observer.on_push_frame(
_frame_pushed(BotStoppedSpeakingFrame(), FrameDirection.DOWNSTREAM)
)
user_event, bot_event = [
event
for event in logs_buffer.get_events()
if event["type"] in {"rtf-user-transcription", "rtf-bot-text"}
]
assert user_event["payload"]["timestamp"] != "aggregator-user-start"
assert user_event["payload"]["end_timestamp"]
assert bot_event["payload"]["timestamp"] != "aggregator-bot-start"
assert bot_event["payload"]["end_timestamp"]

23
api/utils/transcript.py
View file

@ -3,11 +3,26 @@ from typing import List
from pipecat.utils.enums import RealtimeFeedbackType
def generate_transcript_text(events: List[dict]) -> str:
def _format_timestamp_range(payload: dict, event: dict, include_end_timestamps: bool) -> str:
start_timestamp = payload.get("timestamp") or event.get("timestamp", "")
if not include_end_timestamps:
return start_timestamp
end_timestamp = payload.get("end_timestamp")
if end_timestamp:
return f"{start_timestamp} -> {end_timestamp}" if start_timestamp else end_timestamp
return start_timestamp
def generate_transcript_text(
events: List[dict], *, include_end_timestamps: bool = False
) -> str:
"""Generate transcript text from realtime feedback events.
Filters for rtf-user-transcription (final) and rtf-bot-text events,
formats them as '[timestamp] user/assistant: text\\n'.
formats them as '[timestamp] user/assistant: text\\n'. When
include_end_timestamps is True, formats as
'[start_timestamp -> end_timestamp] user/assistant: text\\n'.
"""
lines: List[str] = []
for event in events:
@ -18,11 +33,11 @@ def generate_transcript_text(events: List[dict]) -> str:
event_type == RealtimeFeedbackType.USER_TRANSCRIPTION.value
and payload.get("final") is True
):
timestamp = payload.get("timestamp") or event.get("timestamp", "")
timestamp = _format_timestamp_range(payload, event, include_end_timestamps)
prefix = f"[{timestamp}] " if timestamp else ""
lines.append(f"{prefix}user: {payload.get('text', '')}\n")
elif event_type == RealtimeFeedbackType.BOT_TEXT.value:
timestamp = payload.get("timestamp") or event.get("timestamp", "")
timestamp = _format_timestamp_range(payload, event, include_end_timestamps)
prefix = f"[{timestamp}] " if timestamp else ""
lines.append(f"{prefix}assistant: {payload.get('text', '')}\n")