Measure per-turn latency in scripted evals and record in stereo

changelog/5714.added.md

@@ -0,0 +1,3 @@
+- Scripted eval turns now report when the bot's reply happened. Each `EvalScriptTurnResult` carries a `timing` (`EvalTurnTiming`), also written to `results.jsonl` under `turns[].timing`: milliseconds from the turn's input anchor to the bot's `llm_started`, its first LLM token, the end of the LLM response, its first function call, its own `bot_started_speaking` and `bot_stopped_speaking` reports, and the moment the harness's VAD heard its speech, plus the derived `voice_to_voice_ms` and `speech_padding_ms`. The anchor is the send for a text turn and the end of the utterance for a spoken one (`input_duration_ms` is its length); an expectation's `within_ms` keeps running from the send. The measures are taken by `EvalTimingObserver` (`pipecat.evals.timing`), an observer on the harness's own pipeline; the bot's RTVI `metrics` reports (TTFB, processing time, token usage) now deserialize into `MetricsFrame`s there and are kept per turn as `timing.bot_metrics`. `pipecat eval run -v` prints `ttfb`, and `v2v` for a spoken turn, under each turn.
+
+- Eval recordings (`--record-dir`) are now stereo: the user on the left channel and the bot on the right, each side laid out on its own aligned timeline, so overlaps, unprompted speech and onsets can be read off the file.

src/pipecat/cli/commands/eval.py

@@ -163,6 +163,10 @@ def _print_progress(session: EvalSession, p: EvalProgress) -> None:
     elif p.status == "turn":
         label = f'"{p.event_name}"' if p.event_name else "(observe)"
         print(f"      {_dim(f'turn {p.turn_index}')} → {label}")
+    elif p.status == "timing":
+        # The turn's latency, under its expectations: harness-measured time to
+        # the first LLM token and, for a spoken turn, voice-to-voice.
+        print(f"        {_dim(p.detail)}")
     else:
         badge = _green("✓") if p.status == "matched" else _red("✗")
         line = f"        {badge} {p.event_name}"

src/pipecat/evals/client.py

@@ -59,13 +59,15 @@
     LLMFullResponseStartFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
+    MetricsFrame,
     OutputTransportMessageUrgentFrame,
     TTSAudioRawFrame,
     TTSSpeakFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
     TTSTextFrame,
 )
+from pipecat.observers.base_observer import BaseObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.worker import PipelineParams, PipelineWorker
 from pipecat.processors.aggregators.llm_context import LLMContext
@@ -86,7 +88,9 @@
 # Frames the bot produced: the sink turns them into events and stops them here.
 # Everything downstream of the sink speaks for the user (a persona LLM, the user
 # TTS, the output) and must never see the bot's text or reports as its own
-# input. The aggregator's context frame and lifecycle frames pass.
+# input. The aggregator's context frame and lifecycle frames pass. The bot's
+# metrics make no event; the timing observer has read them by the time they
+# reach the sink.
 _BOT_FRAMES = (
     LLMFullResponseStartFrame,
     LLMTextFrame,
@@ -98,6 +102,7 @@
     FunctionCallResultFrame,
     FunctionCallCancelFrame,
     InputTransportMessageFrame,
+    MetricsFrame,
 )
 
 
@@ -303,6 +308,7 @@ def __init__(
         user_tts: CachingTTSService | None = None,
         bot_stt: STTService | None = None,
         persona: EvalPersona | None = None,
+        observers: list[BaseObserver] | None = None,
     ):
         """Initialize the client.
 
@@ -323,6 +329,9 @@ def __init__(
                 whose LLM rides in the pipeline and whose context the
                 aggregators keep up to date with both sides of the conversation;
                 ``None`` for a scripted scenario.
+            observers: Observers to attach to the eval pipeline's worker, such
+                as the :class:`~pipecat.evals.timing.EvalTimingObserver` that
+                times a scripted run's turns.
         """
         self._bot_url = bot_url
         self._stream = stream
@@ -332,6 +341,7 @@ def __init__(
         self._user_tts = user_tts
         self._bot_stt = bot_stt
         self._persona = persona
+        self._observers = list(observers or [])
 
         # The eval pipeline's worker (built by start()) and the runner task driving it.
         self._worker: PipelineWorker | None = None
@@ -340,8 +350,9 @@ def __init__(
         # Where the user's turns enter the pipeline (built with the processors).
         self._sink: _BotFrameSink | None = None
         self._run_task: asyncio.Task | None = None
-        # Records the conversation audio (bot + user) when record_path is set and
-        # the scenario is audio mode; fed raw audio by the transport, written on stop().
+        # Records the conversation audio (user left, bot right) when record_path is
+        # set and the scenario is audio mode; fed raw audio by the transport,
+        # written on stop().
         self._recorder: EvalClientRecorder | None = None
         # Set by the transport's on_bot_ready handler once the bot completes the
         # RTVI handshake; handshake() waits on it.
@@ -398,6 +409,7 @@ async def start(self) -> None:
             params=self._pipeline_params(),
             enable_rtvi=False,
             cancel_on_idle_timeout=False,
+            observers=self._observers,
         )
 
         @self._worker.event_handler("on_pipeline_error")
@@ -728,7 +740,7 @@ async def _send_cancel(self) -> None:
             pass
 
     async def _write_recording(self) -> None:
-        """Write the recorded conversation audio (bot + user) to ``record_path``."""
+        """Write the recorded conversation audio (user left, bot right) to ``record_path``."""
         if self._recorder is None or not self._record_path or not self._recorder.has_audio():
             return
         if await self._recorder.write(self._record_path):

src/pipecat/evals/client_transport.py

@@ -16,15 +16,16 @@
 
 The recording does not use the paced streams, whose jitter would make it
 stutter: :class:`EvalClientRecorder` is fed the raw audio on both edges and
-lays each side out on its own timeline.
+lays each side out on its own timeline, the user on the left channel and
+the bot on the right.
 """
 
 import asyncio
 import time
 import wave
 from pathlib import Path
 
-from pipecat.audio.utils import create_stream_resampler, mix_audio
+from pipecat.audio.utils import create_stream_resampler, interleave_stereo_audio
 from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
@@ -66,8 +67,10 @@ class EvalClientRecorder:
     Python cannot hold the 40 ms pacing tick precisely, and a recording of the
     paced streams stutters. So each side is recorded as it was produced or
     received, laid out on its own timeline with silence only where a real
-    pause was, and mixed to mono at :meth:`write`. Audio the bot sent past an
-    interruption is dropped, as a real client would drop it.
+    pause was, and written as stereo at :meth:`write`: the user on the left
+    channel, the bot on the right, so overlaps and onsets can be read off
+    the file. Audio the bot sent past an interruption is dropped, as a real
+    client would drop it.
     """
 
     # A chunk arriving later than its side's playout position by more than this
@@ -106,7 +109,10 @@ def has_audio(self) -> bool:
         return self._user.first is not None or self._bot.first is not None
 
     async def write(self, path: str) -> bool:
-        """Resample both sides to a common rate, align, mix to mono, and write a WAV.
+        """Resample both sides to a common rate, align, and write a stereo WAV.
+
+        The user is the left channel and the bot the right; the shorter side
+        is padded with silence to the longer.
 
         Returns:
             True if a file was written, False if nothing was recorded.
@@ -117,14 +123,15 @@ async def write(self, path: str) -> bool:
         start = min(firsts)
         user = await self._user.rendered(self._rate, start)
         bot = await self._bot.rendered(self._rate, start)
-        mixed = mix_audio(user, bot)
+        length = max(len(user), len(bot))
+        stereo = interleave_stereo_audio(user.ljust(length, b"\x00"), bot.ljust(length, b"\x00"))
         out = Path(path)
         out.parent.mkdir(parents=True, exist_ok=True)
         with wave.open(str(out), "wb") as wf:
-            wf.setnchannels(1)
+            wf.setnchannels(2)
             wf.setsampwidth(2)
             wf.setframerate(self._rate)
-            wf.writeframes(mixed)
+            wf.writeframes(stereo)
         return True
 
 
@@ -196,6 +203,13 @@ class EvalClientOutputTransport(WebsocketClientOutputTransport):
     A real-time task sends one 40 ms frame per tick: queued TTS audio when
     there is some, silence otherwise, so the bot's VAD and turn detection see
     the silence they need to end a turn. Runs only when audio output is on.
+
+    A write returns once the send task has sent its audio, the way a write to
+    a sound device returns once the device took it. The base transport's
+    ``BotStartedSpeakingFrame`` and ``BotStoppedSpeakingFrame`` (here: the
+    *user's* utterance going out) then bracket the audio as it was sent, and
+    the stop frame marks the end of the user's speech for whatever times the
+    turn.
     """
 
     def __init__(self, *args, recorder: "EvalClientRecorder | None" = None, **kwargs):
@@ -204,6 +218,11 @@ def __init__(self, *args, recorder: "EvalClientRecorder | None" = None, **kwargs
         self._pending = bytearray()
         self._send_task = None
         self._recorder = recorder
+        # Bytes queued for the send task and bytes it has consumed (sent, or
+        # dropped by an interruption); a write waits for its own to be consumed.
+        self._queued_bytes = 0
+        self._consumed_bytes = 0
+        self._consumed = asyncio.Event()
 
     async def start(self, frame: StartFrame):
         """Start the transport and, in audio mode, the real-time send stream."""
@@ -226,12 +245,14 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, InterruptionFrame) and self._pending:
             if self._recorder is not None:
                 self._recorder.drop_user_tail(len(self._pending))
-            self._pending.clear()
+            self._drop_pending()
         await super().process_frame(frame, direction)
 
     async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
-        """Queue the user audio for the send task.
+        """Queue the user audio for the send task, and wait until it has gone out.
 
+        The media sender writes one chunk at a time, so the wait is at most a
+        tick or two; an interruption or the transport stopping releases it.
         Returns False so the media sender does not push this un-paced frame
         downstream; the send task pushes the paced frames instead.
         """
@@ -242,6 +263,13 @@ async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         if self._recorder is not None:
             self._recorder.add_user(frame.audio, frame.sample_rate)
         self._pending.extend(frame.audio)
+        self._queued_bytes += len(frame.audio)
+        sent_by = self._queued_bytes
+        # With no send task (the transport stopping) nothing would send it, so
+        # nothing is waited for.
+        while self._send_task is not None and self._consumed_bytes < sent_by:
+            self._consumed.clear()
+            await self._consumed.wait()
         return False
 
     async def _send_task_handler(self):
@@ -265,6 +293,10 @@ async def _send_task_handler(self):
                 num_channels=self._params.audio_out_channels,
             )
             await self._send_frame(frame)
+            if pcm is not silence:
+                # The audio went out: the write that queued it may return.
+                self._consumed_bytes = self._queued_bytes - len(self._pending)
+                self._consumed.set()
             # Push every frame (audio and silence) downstream at this paced cadence:
             # the harness recorder aligns tracks by wall-clock, so a continuous
             # stream keeps the user turn at the right time (pushing only audio would
@@ -278,10 +310,18 @@ async def _send_frame(self, frame: OutputAudioRawFrame):
             return
         await self._write_frame(frame)
 
+    def _drop_pending(self):
+        """Forget the queued audio, releasing the write waiting on it."""
+        self._pending.clear()
+        self._consumed_bytes = self._queued_bytes
+        self._consumed.set()
+
     async def _cancel_send_task(self):
         if self._send_task is not None:
             await self.cancel_task(self._send_task)
             self._send_task = None
+        # Nothing will send what is queued now; a write waiting on it returns.
+        self._drop_pending()
 
 
 class EvalClientInputTransport(WebsocketClientInputTransport):

src/pipecat/evals/results.py

@@ -72,6 +72,77 @@ def __str__(self) -> str:
         )
 
 
+@dataclass
+class EvalTurnTiming:
+    """When the bot's reply to one turn happened, measured by the harness.
+
+    Every measure is milliseconds from the turn's *input anchor*: for a text
+    turn the moment the ``send-text`` message was sent; for a spoken turn
+    (synthesized or an ``audio:`` recording) the moment the last chunk of the
+    utterance went out to the bot, i.e. when the user stopped speaking. The
+    anchor is the end of speech when ``input_duration_ms`` is above zero and
+    the send otherwise. A turn that sends nothing is anchored where the
+    harness began observing it. A measure is ``None`` when its event did not
+    occur after the anchor.
+
+    This is not the clock an expectation's ``within_ms`` runs on: that budget
+    is anchored at the send for every turn, spoken or not (see
+    :mod:`pipecat.evals.script`).
+
+    Parameters:
+        input_duration_ms: Length of the user audio played for the turn; 0
+            for a text or DTMF turn.
+        llm_started_ms: The bot's LLM began a completion (``llm_started``).
+        first_token_ms: The first chunk of LLM text arrived.
+        llm_response_ms: The LLM response ended (``llm_response``).
+        function_call_ms: The turn's first function call (``function_call``).
+        bot_started_speaking_ms: The bot reported it started sending speech
+            (``bot_started_speaking``).
+        bot_speech_onset_ms: The harness's own VAD heard speech in the bot's
+            audio. Audio mode only; includes the VAD's start window.
+        bot_stopped_speaking_ms: The bot reported it stopped speaking
+            (``bot_stopped_speaking``); only counted after it started.
+        bot_metrics: The bot's own ``metrics`` reports that arrived during the
+            turn, in order, each a dict of ``processor``, ``ttfb_ms``,
+            ``processing_ms`` and ``tokens``, ``None`` for the parts a report
+            did not carry. The bot's token usage is reported without a
+            processor name.
+    """
+
+    input_duration_ms: int = 0
+    llm_started_ms: int | None = None
+    first_token_ms: int | None = None
+    llm_response_ms: int | None = None
+    function_call_ms: int | None = None
+    bot_started_speaking_ms: int | None = None
+    bot_speech_onset_ms: int | None = None
+    bot_stopped_speaking_ms: int | None = None
+    bot_metrics: list[dict] = field(default_factory=list)
+
+    @property
+    def voice_to_voice_ms(self) -> int | None:
+        """From the end of the user's speech to the bot's audible speech.
+
+        ``bot_speech_onset_ms`` when the turn was spoken (the anchor is then
+        the end of the utterance); ``None`` for a text turn, whose anchor is
+        the send, or when the bot's speech was not heard.
+        """
+        if not self.input_duration_ms:
+            return None
+        return self.bot_speech_onset_ms
+
+    @property
+    def speech_padding_ms(self) -> int | None:
+        """The silence the bot sent before audible speech.
+
+        The gap between the bot's own ``bot_started_speaking`` and the
+        harness hearing its speech; ``None`` unless both were seen.
+        """
+        if self.bot_speech_onset_ms is None or self.bot_started_speaking_ms is None:
+            return None
+        return self.bot_speech_onset_ms - self.bot_started_speaking_ms
+
+
 @dataclass
 class EvalScriptTurnResult:
     """Outcome of one turn within a scenario run.
@@ -86,14 +157,18 @@ class EvalScriptTurnResult:
             :attr:`~pipecat.evals.script.EvalScriptScenario.stop_on_failure`.
         failures: The turn's failed assertions, in order; empty unless ``status``
             is ``failed``.
-        duration_ms: Wall-clock time the turn took, in milliseconds; 0 when the
-            turn was not run.
+        duration_ms: Wall-clock time the turn took, in milliseconds, judge
+            latency included; 0 when the turn was not run.
+        timing: When the bot's reply happened, relative to the turn's input
+            (see :class:`EvalTurnTiming`); ``None`` when the turn was not
+            driven to a send.
     """
 
     turn_index: int
     status: str = "not_run"
     failures: list[EvalAssertionFailure] = field(default_factory=list)
     duration_ms: int = 0
+    timing: EvalTurnTiming | None = None
 
 
 @deprecated(
@@ -289,8 +364,11 @@ class EvalScriptTurnProgress:
         expectation_index: Index of the expectation, or -1 for turn-level records
             (the turn header, or a ``send_after`` that never fired).
         event_name: The expectation's event (or the user text for a turn header).
-        status: ``turn`` (header), ``matched``, ``failed``, or ``timeout``.
-        detail: Optional extra text (failure reason, user utterance, ...).
+        status: ``turn`` (header), ``matched``, ``failed``, ``timeout``, or
+            ``timing`` (the turn's latency summary, once its expectations
+            resolved; ``expectation_index`` is -1 and ``event_name`` empty).
+        detail: Optional extra text (failure reason, user utterance, the
+            timing summary, ...).
     """
 
     turn_index: int

src/pipecat/evals/script.py

@@ -50,8 +50,12 @@
     required — event type name
 
 ``within_ms: <int>``
-    latency budget from the most recent anchor (optional; defaults to 60s when
-    omitted)
+    latency budget from the turn's send (optional; defaults to 60s when
+    omitted). The anchor is the send for every turn, spoken or not: for a
+    spoken turn the budget runs while the utterance is still streaming to the
+    bot. A turn's measured latencies, the ``timing`` on its result
+    (:class:`~pipecat.evals.results.EvalTurnTiming`), use a different anchor:
+    the end of the user's speech for a spoken turn, the send for a text one.
 
 ``text_contains: <str>``
     substring check on the event's text content, ignoring whitespace differences
@@ -259,7 +263,8 @@ class EvalExpectation:
             all of a turn's expectations share that one anchor, so a stalled turn
             fails within a single budget rather than one per expectation. For audio
             turns the anchor is when the utterance was *sent*, not when it finishes
-            streaming to the bot. Defaults to 60s when omitted, so timing isn't
+            streaming to the bot (the turn's measured ``timing`` is anchored
+            there instead). Defaults to 60s when omitted, so timing isn't
             asserted unless set explicitly.
         text_contains: Optional substring check on the event's text content
             (``llm_response.text`` or ``user_transcription.transcript``).