harness

Eval session: drives a bot over RTVI and asserts on the events it emits.

An EvalSession connects to a running bot’s eval transport (a SingleClientWebsocketServerTransport speaking RTVI via RTVIEvalSerializer), walks through a parsed EvalScenario, and verifies that the expected semantic events arrive in order, with the right payloads, within their latency budgets. It returns an EvalResult.

The session is a thin RTVI client. It builds outgoing messages with the RTVI models (pipecat.processors.frameworks.rtvi.models) and translates the RTVI server messages it receives back into a small set of friendly event names the scenario files assert on:

Matching semantics: expected events must appear in the specified order, but unmatched events may appear between them (so a scenario doesn’t have to enumerate every event the bot emits). The within_ms budget for each expectation is measured from the most recent send-text / raw-audio / dtmf send (default 60s when omitted).

An llm_response with a content check (text_contains / eval:) aggregates: the harness accumulates the text of successive response segments within the turn and re-checks on each one, so an interim filler (“Let me check on that.”) or the on-connect greeting is rolled past rather than mistaken for the turn’s answer. Responses that began before the turn’s input are skipped, so an interrupted prior turn doesn’t bleed in. The judge returns yes / no / continue; text_contains treats a missing substring as continue. The within_ms budget bounds the wait.

Example:

scenario = EvalScenario.load("scenarios/greeting.yaml")
result = await EvalSession.from_scenario(scenario, "ws://localhost:7860").run()
if result.passed:
    print("PASS")
else:
    for f in result.failures:
        print(f"  {f}")
class pipecat.evals.harness.EvalAssertionFailure(turn_index: int, expectation_index: int, event_name: str, reason: str)[source]

Bases: object

A single failed assertion within an eval.

Parameters:

  • turn_index – Index of the turn that failed.

  • expectation_index – Index of the expectation within the turn, or -1 for a turn-level failure (e.g. a send_after that never fired).

  • event_name – The expectation’s event name.

  • reason – Human-readable explanation of the failure.

turn_index: int
expectation_index: int
event_name: str
reason: str
class pipecat.evals.harness.EvalResult(scenario_name: str, passed: bool, failures: list[EvalAssertionFailure] = <factory>, duration_ms: int = 0, events_seen: list[dict] = <factory>, debug_log: list[str] = <factory>, skipped: str | None = None)[source]

Bases: object

Outcome of running a scenario in an EvalSession.

Parameters:

  • scenario_name – Name of the scenario that was run.

  • passed – Whether every assertion passed.

  • failures – The assertions that failed, in order.

  • duration_ms – Wall-clock time the run took, in milliseconds.

  • events_seen – Every friendly event observed, for diagnostics.

  • debug_log – Timestamped trace of the harness’s own decisions (events received, audio transcribed, matcher progress), for diagnosing flaky runs. Saved per-scenario by the orchestrator alongside the bot log.

  • skipped – When set, the scenario was not run (e.g. a tts_response assertion without audio mode); the string is the reason. Such a result is neither passed nor failed.

scenario_name: str
passed: bool
failures: list[EvalAssertionFailure]
duration_ms: int = 0
events_seen: list[dict]
debug_log: list[str]
skipped: str | None = None
class pipecat.evals.harness.EvalTurnProgress(turn_index: int, expectation_index: int, event_name: str, status: str, detail: str = '')[source]

Bases: object

A real-time progress record emitted while a turn runs (for verbose output).

Parameters:

  • turn_index – The turn being run.

  • 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, …).

turn_index: int
expectation_index: int
event_name: str
status: str
detail: str = ''
class pipecat.evals.harness.EvalSession(scenario: EvalScenario, bot_url: str, *, connect_timeout_s: float = 5.0, default_timeout_ms: int = 60000, on_progress: Callable[[EvalTurnProgress], None] | None = None, record_path: str | None = None, stop_bot: bool = False, trigger_disconnect: bool = False, judge: EvalJudge | None = None, speech: EvalSpeech | None = None, transcriber: EvalTranscriber | None = None)[source]

Bases: object

Runs one EvalScenario against a bot over a single WebSocket session.

Connects as an RTVI client, drives each turn (sending send-text, raw-audio, or dtmf), collects the RTVI events the bot emits, and asserts on them. Build one with from_scenario() (which constructs the judge, speech, and transcriber the scenario needs), then await run().

__init__(scenario: EvalScenario, bot_url: str, *, connect_timeout_s: float = 5.0, default_timeout_ms: int = 60000, on_progress: Callable[[EvalTurnProgress], None] | None = None, record_path: str | None = None, stop_bot: bool = False, trigger_disconnect: bool = False, judge: EvalJudge | None = None, speech: EvalSpeech | None = None, transcriber: EvalTranscriber | None = None)[source]

Initialize the eval session.

The judge, speech, and transcriber are injected pre-built: from_scenario() constructs the defaults from the scenario’s config (via the respective from_config) and passes them in. Construct and pass your own to override them (e.g. a custom judge LLM or TTS service).

Parameters:

  • scenario – The parsed scenario to run.

  • bot_url – WebSocket URL of the bot’s eval transport.

  • connect_timeout_s – How long to wait for the bot to accept the WS connection before giving up.

  • default_timeout_ms – Per-expectation latency budget for expectations without their own within_ms (the turn’s expectations share one deadline anchored at the send). Defaults to 60s.

  • on_progress – Optional callback invoked with a EvalTurnProgress as each turn and expectation resolves (used for verbose output).

  • record_path – When set (and the scenario is audio mode), asks the eval transport to record the conversation audio to this path (bot-side).

  • stop_bot – When True, ask the bot to cancel its pipeline (and exit) on teardown via eval-cancel. The suite enables it to clean up each spawned bot.

  • trigger_disconnect – When True (or when the scenario sets trigger_disconnect), ask the eval transport to fire the bot’s on_client_disconnected handler when this connection ends. Bots often cancel their pipeline there, so it is off by default to avoid that between scenarios.

  • judge – The EvalJudge for eval: assertions, or None if the scenario has none.

  • speech – The EvalSpeech for synthesizing user audio, or None for text-mode scenarios. Started and stopped by the session.

  • transcriber – The EvalTranscriber for the response event, or None when unused. Started and stopped by the session.

classmethod from_scenario(scenario: EvalScenario, bot_url: str, *, connect_timeout_s: float = 5.0, default_timeout_ms: int = 60000, on_progress: Callable[[EvalTurnProgress], None] | None = None, record_path: str | None = None, cache_dir: str | None = None, use_cache: bool = True, stop_bot: bool = False, trigger_disconnect: bool = False, judge: EvalJudge | None = None, speech: EvalSpeech | None = None, transcriber: EvalTranscriber | None = None) → EvalSession[source]

Build a ready-to-run session from a scenario, constructing what it needs.

Builds the judge, speech, and transcriber the scenario calls for — each via its from_config — and injects them into a new session. Pass judge / speech / transcriber to override any of them with your own pre-built instance. Then await run():

session = EvalSession.from_scenario(scenario, "ws://localhost:7860")
result = await session.run()
Parameters:

  • scenario – The parsed scenario to run.

  • bot_url – WebSocket URL of the bot’s eval transport.

  • connect_timeout_s – How long to wait for the bot to accept the WS connection before giving up.

  • default_timeout_ms – Per-expectation latency budget for expectations without their own within_ms. Defaults to 60s.

  • on_progress – Optional per-turn/expectation progress callback (verbose).

  • record_path – Optional path to record the conversation audio (audio mode).

  • cache_dir – Optional directory for cached synthesized user audio (default <user-cache-dir>/pipecat/tts).

  • use_cache – When False, ignore cached user audio and force fresh synthesis (no cache reads or writes). Defaults to True.

  • stop_bot – When True, ask the bot to cancel its pipeline (and exit) on teardown. Leave False to keep it running for more scenarios.

  • trigger_disconnect – When True, fire the bot’s on_client_disconnected handler when the connection ends (the scenario’s own trigger_disconnect field also opts in). Off by default.

  • judge – Override the judge (default: built from scenario.judge when the scenario has eval: assertions).

  • speech – Override the user-audio generator (default: built from scenario.user_audio in audio mode).

  • transcriber – Override the bot-audio transcriber (default: built from scenario.transcriber when the scenario asserts response).

Returns:

A configured session, ready for run().

async run() → EvalResult[source]

Connect, drive the scenario, and return the result.