wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add BotConnectedFrame and on_transport_timing_report event

Browse Source 
Add BotConnectedFrame (SystemFrame) pushed by SFU transports (Daily,
LiveKit, HeyGen, Tavus) when the bot joins the room. Replace the
on_transport_readiness_measured event with on_transport_timing_report
which includes both bot_connected_secs and client_connected_secs.
This commit is contained in:
Mark Backman
2026-03-02 10:41:05 -05:00
parent de87894778
commit 68e8732e72
12 changed files with 215 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/3881.added.3.md Normal file
View File

@@ -0,0 +1 @@
Added `BotConnectedFrame` for SFU transports and `on_transport_timing_report` event to `StartupTimingObserver` with bot and client connection timing.

8
examples/foundational/29-turn-tracking-observer.py
View File

@@ -111,9 +111,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
for timing in report.processor_timings:
logger.info(f" {timing.processor_name}: {timing.duration_secs:.3f}s")
@startup_observer.event_handler("on_transport_readiness_measured")
async def on_transport_readiness_measured(observer, report):
logger.info(f"Transport readiness: {report.readiness_secs:.3f}s")
@startup_observer.event_handler("on_transport_timing_report")
async def on_transport_timing_report(observer, report):
if report.bot_connected_secs is not None:
logger.info(f"Bot connected: {report.bot_connected_secs:.3f}s")
logger.info(f"Client connected: {report.client_connected_secs:.3f}s")
turn_observer = task.turn_tracking_observer
if turn_observer:

12
src/pipecat/frames/frames.py
View File

@@ -1910,6 +1910,18 @@ class StopFrame(ControlFrame, UninterruptibleFrame):
pass
@dataclass
class BotConnectedFrame(SystemFrame):
"""Frame indicating the bot has connected to the transport service.
Pushed downstream by SFU transports (Daily, LiveKit, HeyGen, Tavus)
when the bot successfully joins the room. Non-SFU transports do not
emit this frame.
"""
pass
@dataclass
class ClientConnectedFrame(SystemFrame):
"""Frame indicating that a client has connected to the transport.

93
src/pipecat/observers/startup_timing_observer.py
View File

@@ -12,9 +12,9 @@ when a ``StartFrame`` arrives at a processor (``on_process_frame``) versus
when it leaves (``on_push_frame``), giving the exact ``start()`` duration
for each processor in the pipeline.
It also measures transport readiness — the time from ``StartFrame`` to the
first ``ClientConnectedFrame`` — via a separate ``on_transport_readiness_measured``
event.
It also measures transport timing — the time from ``StartFrame`` to the
first ``BotConnectedFrame`` (SFU transports only) and ``ClientConnectedFrame``
— via a separate ``on_transport_timing_report`` event.
Example::
@@ -25,9 +25,11 @@ Example::
for t in report.processor_timings:
print(f"{t.processor_name}: {t.duration_secs:.3f}s")
@observer.event_handler("on_transport_readiness_measured")
async def on_readiness(observer, report):
print(f"Transport ready in {report.readiness_secs:.3f}s")
@observer.event_handler("on_transport_timing_report")
async def on_transport(observer, report):
if report.bot_connected_secs is not None:
print(f"Bot connected in {report.bot_connected_secs:.3f}s")
print(f"Client connected in {report.client_connected_secs:.3f}s")
task = PipelineTask(pipeline, observers=[observer])
"""
@@ -35,9 +37,7 @@ Example::
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple, Type
from loguru import logger
from pipecat.frames.frames import ClientConnectedFrame, StartFrame
from pipecat.frames.frames import BotConnectedFrame, ClientConnectedFrame, StartFrame
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.pipeline import PipelineSink, PipelineSource
@@ -74,14 +74,17 @@ class StartupTimingReport:
@dataclass
class TransportReadinessReport:
"""Time from pipeline start to first client connection.
class TransportTimingReport:
"""Time from pipeline start to transport connection milestones.
Parameters:
readiness_secs: Seconds from StartFrame to first ClientConnectedFrame.
bot_connected_secs: Seconds from StartFrame to first BotConnectedFrame
(only set for SFU transports).
client_connected_secs: Seconds from StartFrame to first ClientConnectedFrame.
"""
readiness_secs: float
bot_connected_secs: Optional[float] = None
client_connected_secs: Optional[float] = None
class StartupTimingObserver(BaseObserver):
@@ -92,9 +95,13 @@ class StartupTimingObserver(BaseObserver):
pushed downstream. This captures WebSocket connections, API authentication,
model loading, and other initialization work.
Also measures transport readiness — the time from ``StartFrame`` to the
first ``ClientConnectedFrame`` — indicating how long it takes for a client
to connect after the pipeline starts.
Also measures transport timing, the time from ``StartFrame`` to connection
milestones:
- ``bot_connected_secs``: When the bot joins the transport room
(SFU transports only, triggered by ``BotConnectedFrame``).
- ``client_connected_secs``: When a remote participant connects
(triggered by ``ClientConnectedFrame``).
By default, internal pipeline processors (``PipelineSource``, ``PipelineSink``,
``Pipeline``) are excluded from the report. Pass ``processor_types`` to
@@ -104,8 +111,9 @@ class StartupTimingObserver(BaseObserver):
- on_startup_timing_report: Called once after startup completes with the full
timing report.
- on_transport_readiness_measured: Called once when the first client connects with the
transport readiness timing.
- on_transport_timing_report: Called once when the first client connects with a
TransportTimingReport containing client_connected_secs and bot_connected_secs
(if available).
Example::
@@ -118,9 +126,11 @@ class StartupTimingObserver(BaseObserver):
for t in report.processor_timings:
logger.info(f"{t.processor_name}: {t.duration_secs:.3f}s")
@observer.event_handler("on_transport_readiness_measured")
async def on_readiness(observer, report):
logger.info(f"Transport ready in {report.readiness_secs:.3f}s")
@observer.event_handler("on_transport_timing_report")
async def on_transport(observer, report):
if report.bot_connected_secs is not None:
logger.info(f"Bot connected in {report.bot_connected_secs:.3f}s")
logger.info(f"Client connected in {report.client_connected_secs:.3f}s")
task = PipelineTask(pipeline, observers=[observer])
@@ -157,14 +167,17 @@ class StartupTimingObserver(BaseObserver):
# Whether we've already emitted the startup timing report.
self._startup_timing_reported = False
# Whether we've already measured transport readiness.
self._transport_readiness_measured = False
# Whether we've already measured transport timing.
self._transport_timing_reported = False
# Timestamp (ns) when we first see a StartFrame arrive at a processor.
self._start_frame_arrival_ns: Optional[int] = None
# Bot connected timing (stored for inclusion in the transport report).
self._bot_connected_secs: Optional[float] = None
self._register_event_handler("on_startup_timing_report")
self._register_event_handler("on_transport_readiness_measured")
self._register_event_handler("on_transport_timing_report")
def _should_track(self, processor: FrameProcessor) -> bool:
"""Check if a processor should be tracked for timing.
@@ -216,11 +229,16 @@ class StartupTimingObserver(BaseObserver):
async def on_push_frame(self, data: FramePushed):
"""Record when a StartFrame leaves a processor and compute the delta.
Also handles ``ClientConnectedFrame`` to measure transport readiness.
Also handles ``BotConnectedFrame`` and ``ClientConnectedFrame`` to
measure transport timing.
Args:
data: The frame push event data.
"""
if isinstance(data.frame, BotConnectedFrame):
self._handle_bot_connected(data)
return
if isinstance(data.frame, ClientConnectedFrame):
await self._handle_client_connected(data)
return
@@ -249,16 +267,27 @@ class StartupTimingObserver(BaseObserver):
)
)
async def _handle_client_connected(self, data: FramePushed):
"""Measure transport readiness on first client connection."""
if self._transport_readiness_measured or self._start_frame_arrival_ns is None:
def _handle_bot_connected(self, data: FramePushed):
"""Record bot connected timing on first BotConnectedFrame."""
if self._bot_connected_secs is not None or self._start_frame_arrival_ns is None:
return
self._transport_readiness_measured = True
delta_ns = data.timestamp - self._start_frame_arrival_ns
readiness_secs = delta_ns / 1e9
report = TransportReadinessReport(readiness_secs=readiness_secs)
await self._call_event_handler("on_transport_readiness_measured", report)
self._bot_connected_secs = delta_ns / 1e9
async def _handle_client_connected(self, data: FramePushed):
"""Emit transport timing report on first ClientConnectedFrame."""
if self._transport_timing_reported or self._start_frame_arrival_ns is None:
return
self._transport_timing_reported = True
delta_ns = data.timestamp - self._start_frame_arrival_ns
client_connected_secs = delta_ns / 1e9
report = TransportTimingReport(
bot_connected_secs=self._bot_connected_secs,
client_connected_secs=client_connected_secs,
)
await self._call_event_handler("on_transport_timing_report", report)
async def _emit_report(self):
"""Build and emit the startup timing report."""

7
src/pipecat/services/heygen/client.py
View File

@@ -62,10 +62,12 @@ class HeyGenCallbacks(BaseModel):
"""Callback handlers for HeyGen events.
Parameters:
on_participant_connected: Called when a participant connects
on_participant_disconnected: Called when a participant disconnects
on_connected: Called when the bot connects to the LiveKit room.
on_participant_connected: Called when a participant connects.
on_participant_disconnected: Called when a participant disconnects.
"""
on_connected: Callable[[], Awaitable[None]]
on_participant_connected: Callable[[str], Awaitable[None]]
on_participant_disconnected: Callable[[str], Awaitable[None]]
@@ -251,6 +253,7 @@ class HeyGenClient:
logger.debug(f"HeyGenClient send_interval: {self._send_interval}")
await self._ws_connect()
await self._livekit_connect()
self._call_event_callback(self._callbacks.on_connected)
async def stop(self) -> None:
"""Stop the client and terminate all connections.

5
src/pipecat/services/heygen/video.py
View File

@@ -128,6 +128,7 @@ class HeyGenVideoService(AIService):
session_request=self._session_request,
service_type=self._service_type,
callbacks=HeyGenCallbacks(
on_connected=self._on_connected,
on_participant_connected=self._on_participant_connected,
on_participant_disconnected=self._on_participant_disconnected,
),
@@ -144,6 +145,10 @@ class HeyGenVideoService(AIService):
await self._client.cleanup()
self._client = None
async def _on_connected(self):
"""Handle bot connected to LiveKit room."""
logger.info("HeyGen bot connected to LiveKit room")
async def _on_participant_connected(self, participant_id: str):
"""Handle participant connected events."""
logger.info(f"Participant connected {participant_id}")

5
src/pipecat/services/tavus/video.py
View File

@@ -94,6 +94,7 @@ class TavusVideoService(AIService):
"""
await super().setup(setup)
callbacks = TavusCallbacks(
on_joined=self._on_joined,
on_participant_joined=self._on_participant_joined,
on_participant_left=self._on_participant_left,
)
@@ -119,6 +120,10 @@ class TavusVideoService(AIService):
await self._client.cleanup()
self._client = None
async def _on_joined(self, data):
"""Handle bot joined the Daily room."""
logger.info("Tavus bot joined Daily room")
async def _on_participant_left(self, participant, reason):
"""Handle participant leaving the session."""
participant_id = participant["id"]

3
src/pipecat/transports/daily/transport.py
View File

@@ -24,6 +24,7 @@ from pydantic import BaseModel
from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams
from pipecat.frames.frames import (
BotConnectedFrame,
CancelFrame,
ClientConnectedFrame,
DataFrame,
@@ -2579,6 +2580,8 @@ class DailyTransport(BaseTransport):
if error:
await self._on_error(f"Unable to start transcription: {error}")
await self._call_event_handler("on_joined", data)
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_left(self):
"""Handle room left events."""

9
src/pipecat/transports/heygen/transport.py
View File

@@ -23,6 +23,7 @@ from loguru import logger
from pipecat.frames.frames import (
AudioRawFrame,
BotConnectedFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -340,6 +341,7 @@ class HeyGenTransport(BaseTransport):
session_request=session_request,
service_type=service_type,
callbacks=HeyGenCallbacks(
on_connected=self._on_connected,
on_participant_connected=self._on_participant_connected,
on_participant_disconnected=self._on_participant_disconnected,
),
@@ -350,9 +352,16 @@ class HeyGenTransport(BaseTransport):
# Register supported handlers. The user will only be able to register
# these handlers.
self._register_event_handler("on_connected")
self._register_event_handler("on_client_connected")
self._register_event_handler("on_client_disconnected")
async def _on_connected(self):
"""Handle bot connected to LiveKit room."""
await self._call_event_handler("on_connected")
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_participant_disconnected(self, participant_id: str):
logger.debug(f"HeyGen participant {participant_id} disconnected")
if participant_id != "heygen":

3
src/pipecat/transports/livekit/transport.py
View File

@@ -23,6 +23,7 @@ from pipecat.audio.utils import create_stream_resampler
from pipecat.audio.vad.vad_analyzer import VADAnalyzer
from pipecat.frames.frames import (
AudioRawFrame,
BotConnectedFrame,
CancelFrame,
ClientConnectedFrame,
EndFrame,
@@ -1132,6 +1133,8 @@ class LiveKitTransport(BaseTransport):
async def _on_connected(self):
"""Handle room connected events."""
await self._call_event_handler("on_connected")
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_disconnected(self):
"""Handle room disconnected events."""

12
src/pipecat/transports/tavus/transport.py
View File

@@ -21,6 +21,7 @@ from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
BotConnectedFrame,
CancelFrame,
ClientConnectedFrame,
EndFrame,
@@ -133,10 +134,12 @@ class TavusCallbacks(BaseModel):
"""Callback handlers for Tavus events.
Parameters:
on_joined: Called when the bot joins the Daily room.
on_participant_joined: Called when a participant joins the conversation.
on_participant_left: Called when a participant leaves the conversation.
"""
on_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]]
on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]]
@@ -271,6 +274,7 @@ class TavusTransportClient:
async def _on_joined(self, data):
"""Handle joined event."""
logger.debug("TavusTransportClient joined!")
await self._callbacks.on_joined(data)
async def _on_left(self):
"""Handle left event."""
@@ -703,6 +707,7 @@ class TavusTransport(BaseTransport):
self._params = params
callbacks = TavusCallbacks(
on_joined=self._on_joined,
on_participant_joined=self._on_participant_joined,
on_participant_left=self._on_participant_left,
)
@@ -721,9 +726,16 @@ class TavusTransport(BaseTransport):
# Register supported handlers. The user will only be able to register
# these handlers.
self._register_event_handler("on_joined")
self._register_event_handler("on_client_connected")
self._register_event_handler("on_client_disconnected")
async def _on_joined(self, data):
"""Handle bot joined room event."""
await self._call_event_handler("on_joined", data)
if self._input:
await self._input.push_frame(BotConnectedFrame())
async def _on_participant_left(self, participant, reason):
"""Handle participant left events."""
persona_name = await self._client.get_persona_name()

114
tests/test_startup_timing_observer.py
View File

@@ -1,11 +1,17 @@
import asyncio
import unittest
from pipecat.frames.frames import ClientConnectedFrame, Frame, StartFrame, TextFrame
from pipecat.frames.frames import (
BotConnectedFrame,
ClientConnectedFrame,
Frame,
StartFrame,
TextFrame,
)
from pipecat.observers.startup_timing_observer import (
StartupTimingObserver,
StartupTimingReport,
TransportReadinessReport,
TransportTimingReport,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.tests.utils import run_test
@@ -182,16 +188,16 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
f"Internal processor {t.processor_name} should be excluded by default",
)
async def test_transport_readiness_measured(self):
"""Test that ClientConnectedFrame after startup emits on_transport_readiness_measured."""
async def test_transport_timing_client_only(self):
"""Test that ClientConnectedFrame emits on_transport_timing_report."""
observer = StartupTimingObserver()
processor = FastProcessor()
readiness_reports = []
transport_reports = []
@observer.event_handler("on_transport_readiness_measured")
async def on_readiness(obs, report):
readiness_reports.append(report)
@observer.event_handler("on_transport_timing_report")
async def on_transport(obs, report):
transport_reports.append(report)
frames_to_send = [ClientConnectedFrame(), TextFrame(text="hello")]
@@ -202,21 +208,22 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
observers=[observer],
)
self.assertEqual(len(readiness_reports), 1)
report = readiness_reports[0]
self.assertIsInstance(report, TransportReadinessReport)
self.assertGreater(report.readiness_secs, 0)
self.assertEqual(len(transport_reports), 1)
report = transport_reports[0]
self.assertIsInstance(report, TransportTimingReport)
self.assertGreater(report.client_connected_secs, 0)
self.assertIsNone(report.bot_connected_secs)
async def test_transport_readiness_only_first(self):
async def test_transport_timing_only_first_client(self):
"""Test that only the first ClientConnectedFrame triggers the event."""
observer = StartupTimingObserver()
processor = FastProcessor()
readiness_reports = []
transport_reports = []
@observer.event_handler("on_transport_readiness_measured")
async def on_readiness(obs, report):
readiness_reports.append(report)
@observer.event_handler("on_transport_timing_report")
async def on_transport(obs, report):
transport_reports.append(report)
frames_to_send = [
ClientConnectedFrame(),
@@ -231,9 +238,9 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
observers=[observer],
)
self.assertEqual(len(readiness_reports), 1)
self.assertEqual(len(transport_reports), 1)
async def test_transport_readiness_without_start_frame(self):
async def test_transport_timing_without_start_frame(self):
"""Test that ClientConnectedFrame before StartFrame does not crash."""
observer = StartupTimingObserver()
@@ -253,7 +260,74 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
await observer.on_push_frame(data)
# No event should have been emitted.
self.assertFalse(observer._transport_readiness_measured)
self.assertFalse(observer._transport_timing_reported)
async def test_bot_and_client_connected(self):
"""Test that BotConnectedFrame timing is included in the transport report."""
observer = StartupTimingObserver()
processor = FastProcessor()
transport_reports = []
@observer.event_handler("on_transport_timing_report")
async def on_transport(obs, report):
transport_reports.append(report)
frames_to_send = [
BotConnectedFrame(),
ClientConnectedFrame(),
TextFrame(text="hello"),
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=[BotConnectedFrame, ClientConnectedFrame, TextFrame],
observers=[observer],
)
self.assertEqual(len(transport_reports), 1)
report = transport_reports[0]
self.assertGreater(report.client_connected_secs, 0)
self.assertIsNotNone(report.bot_connected_secs)
self.assertGreater(report.bot_connected_secs, 0)
# Client connected should be >= bot connected.
self.assertGreaterEqual(report.client_connected_secs, report.bot_connected_secs)
async def test_bot_connected_only_first(self):
"""Test that only the first BotConnectedFrame is recorded."""
observer = StartupTimingObserver()
processor = FastProcessor()
transport_reports = []
@observer.event_handler("on_transport_timing_report")
async def on_transport(obs, report):
transport_reports.append(report)
frames_to_send = [
BotConnectedFrame(),
BotConnectedFrame(),
ClientConnectedFrame(),
TextFrame(text="hello"),
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=[
BotConnectedFrame,
BotConnectedFrame,
ClientConnectedFrame,
TextFrame,
],
observers=[observer],
)
# Only one transport report, with bot timing from first frame.
self.assertEqual(len(transport_reports), 1)
self.assertIsNotNone(transport_reports[0].bot_connected_secs)
if __name__ == "__main__":