> ## Documentation Index
> Fetch the complete documentation index at: https://daily-docs-flows-declarative.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# Using the Library
> Run scripted scenarios and simulations from Python with the pipecat.evals API: sessions, results, custom judges and persona LLMs, and suites.
Everything the `pipecat eval` CLI does is available as a library under `pipecat.evals`. Use it to run evals from your own test runner (pytest, a CI script, a custom dashboard), to build scenarios in code instead of YAML, or to customize pieces like the judge LLM or the persona LLM.
A **session** is one conversation with a bot, driven to a result. `EvalSession.from_scenario()` builds the session for a loaded scenario of either kind: an `EvalScriptSession` for a [scripted scenario](/pipecat/evals/scripted-scenarios), which returns an `EvalScriptResult`, or an `EvalSimulationSession` for a [simulation](/pipecat/evals/simulated-scenarios), which returns an `EvalSimulationResult`. `EvalSession` is the base class of both, and `EvalSessionParams` is how a run behaves, whichever kind it is.
`pipecat.evals` itself exports nothing. Import each name from the submodule
that defines it, as the examples below do.
## Running a scripted scenario
`EvalScriptScenario.load()` parses a scenario file, and `EvalSession.from_scenario()` builds a ready-to-run session, constructing the judge, user speech, and transcriber the scenario calls for:
```python theme={null}
import asyncio
from pipecat.evals.script import EvalScriptScenario
from pipecat.evals.session import EvalSession
async def main():
scenario = EvalScriptScenario.load("scenarios/capital_question.yaml")
session = EvalSession.from_scenario(scenario, "ws://localhost:7860")
result = await session.run()
if result.passed:
print(f"PASS ({result.duration_ms}ms)")
else:
for failure in result.failures:
print(f" {failure}")
asyncio.run(main())
```
The agent must already be running with its eval transport (`python bot.py -t eval`), just as with `pipecat eval run`. `from_scenario()` is typed by the scenario it is given: a type checker sees a session built from an `EvalScriptScenario` as an `EvalScriptSession`, with no narrowing needed. `EvalScriptSession.from_scenario()` from `pipecat.evals.script_session` takes the same arguments.
### The result
`run()` returns an `EvalScriptResult`:
| Field | Description |
| --------------- | --------------------------------------------------------------------------------------------------- |
| `scenario_name` | Name of the scenario that ran. |
| `passed` | Whether every assertion passed. |
| `failures` | The failed assertions, each with the turn index, expectation index, event name, reason, and `kind`. |
| `turns` | One `EvalScriptTurnResult` per scenario turn, in order. See below. |
| `duration_ms` | Wall-clock time the run took. |
| `events_seen` | Every semantic event observed, for diagnostics. |
| `debug_log` | The harness's timestamped decision trace (what the CLI writes to `.eval.log`). |
| `skipped` | Set (with a reason) when the scenario was not run; such a result is neither pass nor fail. |
Each `EvalScriptTurnResult` carries its `turn_index`, a `status` of `passed`, `failed`, or `not_run`, the `failures` it produced, and its `duration_ms`. `EvalScriptResult.failures` is these turns' failures flattened, plus any that belong to no turn, such as a failed connect.
`not_run` is deliberately distinct from a pass, so a turn the scenario never reached doesn't inflate a rate:
```python theme={null}
result = await EvalSession.from_scenario(scenario, "ws://localhost:7860").run()
scored = [t for t in result.turns if t.status != "not_run"]
print(f"{sum(1 for t in scored if t.status == 'passed')}/{len(scored)} turns")
```
`failures` carry a `kind` alongside the reason (`timeout`, `judge_no`, `text_mismatch`, `missing_function_call`, and so on), which is the stable key to group by when scoring a sweep rather than reading one run.
This maps cleanly onto a pytest test:
```python theme={null}
import pytest
from pipecat.evals.script import EvalScriptScenario
from pipecat.evals.session import EvalSession
@pytest.mark.asyncio
async def test_capital_question():
scenario = EvalScriptScenario.load("scenarios/capital_question.yaml")
result = await EvalSession.from_scenario(scenario, "ws://localhost:7860").run()
assert result.passed, "\n".join(str(f) for f in result.failures)
```
## Running a simulation
A simulation loads and runs the same way. For a simulation, `EvalSession.from_scenario()` builds an `EvalSimulationSession`, constructing the persona LLM from the file's `simulator:` block and the judge from its `judge:` block, plus the user TTS and the transcriber in audio mode:
```python theme={null}
import asyncio
from pipecat.evals.session import EvalSession
from pipecat.evals.simulation import EvalSimulationScenario
async def main():
simulation = EvalSimulationScenario.load("scenarios/simulated/book_table.yaml")
session = EvalSession.from_scenario(simulation, "ws://localhost:7860")
result = await session.run()
if result.passed:
print(f"PASS: {result.reason}")
else:
print(f"FAIL: {result.failure}")
for metric in result.metrics:
print(f" {metric.name}: {metric.score} ({metric.reason})")
asyncio.run(main())
```
A session runs the simulation once. The `runs:` field is honored by the [suite](#orchestrating-suites), which spawns a fresh agent for each run. `EvalSimulationSession.from_scenario()` from `pipecat.evals.simulation_session` is the same call on the kind's own class.
### The result
`run()` returns an `EvalSimulationResult`:
| Field | Description |
| ----------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
| `simulation_name` | Name of the simulation that ran. |
| `passed` | Whether the run completed, the judge said `success:` was met, and no metric with a threshold fell short. |
| `succeeded` | The judge's verdict on `success:` alone. |
| `reason` | The judge's reason for its verdict, or the error. |
| `failure` | Why the run did not pass, or `None`: the error, then `goal not met: ...`, then the first failed metric. |
| `error` | Set when the run did not complete (a failed connect, a harness error). Such a run is neither a goal success nor a goal failure. |
| `metrics` | One `EvalSimulationMetricScore` per metric, in the order the file lists them. See below. |
| `messages` | The conversation, the persona's turns as `user` messages and the agent's as `assistant`. The `assistant` messages are the turns the metrics scored. |
| `turns` | The persona's turn count. |
| `ended_by` | How the run ended: `end_call`, `bot`, `max_turns`, `max_duration`, `silence`, or `error`. |
| `end_call` | The persona's own `end_call` claim (`success`, `reason`) when it made one. Advisory: the judge decides `succeeded`. |
| `duration_ms` | Wall-clock time the run took. |
| `events_seen` | Every semantic event observed, for diagnostics. |
| `debug_log` | The harness's timestamped decision trace. |
Each `EvalSimulationMetricScore` carries the metric's `name`, its `score` (the share of turns the judge said yes to for a judged metric, `1.0` or `0.0` for a measured one, `None` when there was nothing to judge or measure), whether it `passed`, its `reason`, its `min_score`, its measured `value` (the seconds, words, turns, or number of calls), and a `failure_kind` (`judge_no`, `judge_no_verdict`, `out_of_range`, or `function_calls`; `None` when it passed). A judged metric also carries one `EvalSimulationTurnVerdict` per agent turn with the 1-based `turn`, whether it `passed`, the judge's `verdict` (`yes`, `no`, or `none` when the judge gave none, which counts as a no), and its `reason`:
```python theme={null}
bot_turns = [m["content"] for m in result.messages if m["role"] == "assistant"]
for metric in result.metrics:
for verdict in metric.verdicts:
if not verdict.passed:
said = bot_turns[verdict.turn - 1]
print(f"{metric.name} turn {verdict.turn}: {said!r}: {verdict.reason}")
```
## Loading either kind
`load_scenario_file()` reads a file as whichever kind it is, and `EvalSession.from_scenario()` builds the matching session, so a list mixing both kinds runs through one call each:
```python theme={null}
from pipecat.evals.results import EvalScriptResult
from pipecat.evals.scenario import load_scenario_file
from pipecat.evals.session import EvalSession
async def run_any(path: str, bot_url: str) -> bool:
scenario = load_scenario_file(path)
result = await EvalSession.from_scenario(scenario, bot_url).run()
if isinstance(result, EvalScriptResult):
return result.passed and result.skipped is None
return result.passed
```
A file with both `turns:` and `persona:`, or neither, raises a `ValueError` naming the problem. `EvalKind`, from `pipecat.evals.scenario`, names the two kinds, `script` and `simulation`, as a suite run or a results record reports them.
## Run parameters
How a run behaves, whichever kind it is, is one `EvalSessionParams` from `pipecat.evals.session`, passed as `params=` to `from_scenario()` or to a session's constructor. It is plain configuration, so one instance serves many runs:
```python theme={null}
from pipecat.evals.session import EvalSession, EvalSessionParams
params = EvalSessionParams(
record_path="recordings/capital_question.wav",
stop_bot=True,
)
result = await EvalSession.from_scenario(scenario, "ws://localhost:7860", params=params).run()
```
| Field | Default | Description |
| -------------------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------ |
| `connect_timeout_s` | `5.0` | How long to wait for the bot to accept the WebSocket connection. |
| `default_timeout_ms` | `60000` | Scripted scenarios only: the latency budget for expectations without their own `within_ms`. |
| `record_path` | `None` | Where to save the conversation audio. Only an audio-mode run records. |
| `cache_dir` | `None` | Directory for cached synthesized user audio; `None` for the default under the user cache directory. |
| `use_cache` | `True` | When `False`, ignore cached user audio and synthesize every turn, with no cache reads or writes. |
| `stop_bot` | `False` | When `True`, ask the bot to cancel its pipeline, and exit, on teardown. |
| `trigger_disconnect` | `False` | When `True`, fire the bot's `on_client_disconnected` handler when the connection ends. A scenario's own `trigger_disconnect` field also opts in. |
Passing these as individual keyword arguments to `from_scenario()`
(`connect_timeout_s=`, `stop_bot=`, and so on) is deprecated since 1.9.0 and
will be removed in 2.0.0. They still work, override the `params` field of the
same name, and emit a `DeprecationWarning`.
## Building scenarios in code
Scenarios are plain dataclasses, so you can construct them programmatically, generating turns from a dataset, parameterizing a template, or skipping YAML entirely:
```python theme={null}
from pipecat.evals.script import EvalExpectation, EvalScriptScenario, EvalScriptTurn
scenario = EvalScriptScenario(
name="capital_question",
turns=[
EvalScriptTurn(
user="What is the capital of Germany?",
expect=[
EvalExpectation(
event="llm_response",
eval="the response says the capital of Germany is Berlin",
)
],
)
],
)
```
The modality-agnostic `response` event is resolved while parsing YAML. When
constructing scenarios in code, use `llm_response` for text mode directly (or
`response` only when you also configure audio judging).
A simulation is built the same way. Each metric is an `EvalSimulationMetric` with either a `criterion` or a `measure`, and `simulator=` takes a plain mapping with the same shape as the YAML block when the persona shouldn't run on the default local model:
```python theme={null}
from pipecat.evals.simulation import EvalSimulationMetric, EvalSimulationScenario
simulation = EvalSimulationScenario(
name="book_table",
persona="Jamie, booking dinner for two tonight at 6 PM. Gives a name and phone number when asked.",
goal="Book a table for two at 6 PM, then end the call.",
success="the bot confirmed a reservation for two at 6 PM",
metrics=[
EvalSimulationMetric(
name="politeness",
criterion="the reply is courteous, never curt or dismissive",
min_score=1.0,
),
EvalSimulationMetric(name="latency", measure="latency", max_value=5.0),
],
max_turns=8,
)
```
## Customizing the judge and the persona
`from_scenario()` builds the judge from the scenario's `judge:` block, but you can inject your own. `EvalJudge` works with any Pipecat LLM service backed by an OpenAI-compatible API, and `judge=` applies to either kind:
```python theme={null}
import os
from pipecat.evals.judge import EvalJudge
from pipecat.evals.session import EvalSession
from pipecat.services.openai.llm import OpenAILLMService
llm = OpenAILLMService(
api_key=os.environ["OPENAI_API_KEY"],
settings=OpenAILLMService.Settings(model="gpt-4o-mini"),
)
session = EvalSession.from_scenario(
scenario,
"ws://localhost:7860",
judge=EvalJudge(llm),
)
```
For a simulation, `persona_llm=` is the LLM that plays the caller. It runs inside the harness's own pipeline, so it can be any Pipecat `LLMService` that supports function calling, which the persona needs to hang up with its `end_call` tool. Passing it for a scripted scenario raises a `ValueError`:
```python theme={null}
session = EvalSession.from_scenario(
simulation,
"ws://localhost:7860",
persona_llm=OpenAILLMService(
api_key=os.environ["OPENAI_API_KEY"],
settings=OpenAILLMService.Settings(model="gpt-4o-mini"),
),
judge=EvalJudge(llm),
)
```
Passing `judge=None` explicitly to the `EvalSimulationSession` constructor runs the conversation without a judge: measured metrics are still computed, and the run reports no verdict on the goal.
### Custom audio services
`from_scenario()` also takes `user_tts=` for the user's synthesized voice and `bot_stt=` for transcribing the agent's spoken audio, for either kind. `bot_stt` is any Pipecat `STTService`. `user_tts` is a `CachingTTSService`, which wraps a `TTSService` and caches its audio on disk so repeated turns don't re-synthesize; give it a `cache_key` that identifies the voice configuration:
```python theme={null}
import os
from pipecat.evals.session import EvalSession
from pipecat.evals.tts import CachingTTSService
from pipecat.services.fal.stt import FalSTTService
from pipecat.services.rime.tts import RimeHttpTTSService
user_tts = CachingTTSService(
RimeHttpTTSService(
api_key=os.environ["RIME_API_KEY"],
settings=RimeHttpTTSService.Settings(voice="luna"),
sample_rate=16000,
),
cache_key="rime-luna-16000",
)
session = EvalSession.from_scenario(
scenario,
"ws://localhost:7860",
user_tts=user_tts,
bot_stt=FalSTTService(api_key=os.environ["FAL_KEY"]),
)
```
The wrapped services can be local models or HTTP-based; WebSocket-streaming services are rejected, since they need a running pipeline to manage their connection lifecycle. For the YAML-only route, the `factory:` escape hatch in the [Scenario Configuration](/pipecat/evals/configuration#custom-services-with-factory) page reaches the same services without code.
## Observing progress
Every session emits an `on_progress` event as the conversation advances. A scripted session reports an `EvalScriptTurnProgress` as each turn and expectation resolves; a simulation session reports an `EvalSimulationProgress` for each line as it is spoken, with a `status` of `bot` or `user`, the `text`, and the persona's `turn` count so far, then one with a `status` of `ended` whose `text` says how the conversation ended:
```python theme={null}
from pipecat.evals.results import EvalSimulationProgress
from pipecat.evals.session import EvalSession
session = EvalSession.from_scenario(simulation, url)
@session.event_handler("on_progress")
async def on_progress(session, progress: EvalSimulationProgress):
if progress.status == "ended":
print(f"ended by {progress.text} after {progress.turn} turn(s)")
else:
print(f"{progress.status}: {progress.text}")
```
For a scripted session the record has `turn_index`, `status`, `event_name`, and `detail`:
```python theme={null}
from pipecat.evals.session import EvalSession
session = EvalSession.from_scenario(scenario, url)
@session.event_handler("on_progress")
async def on_progress(session, progress):
print(f"turn {progress.turn_index} [{progress.status}] {progress.event_name} {progress.detail}")
```
The `on_progress` callback parameter is deprecated since 1.9.0 and will be
removed in 2.0.0. Use the `on_progress` event handler instead. It only ever
applied to scripted scenarios, so passing it for a simulation raises a
`ValueError`.
## Orchestrating suites
`EvalManifest` and `EvalSuite` are the library behind `pipecat eval suite`: the suite spawns each agent with its eval transport on its own port, runs its scenarios of either kind, each in its own process, and executes several runs concurrently:
```python theme={null}
import asyncio
from pathlib import Path
from pipecat.evals.scenario import EvalKind
from pipecat.evals.session import EvalSessionParams
from pipecat.evals.suite import EvalManifest, EvalSuite
async def main():
manifest = EvalManifest.load("manifest.yaml")
suite = EvalSuite(manifest)
# Optionally narrow the runs, like the CLI's -p / -s / -k flags.
suite.filter(pattern="support", kind=EvalKind.SIMULATION)
@suite.event_handler("on_update")
async def on_update(suite, run):
print(run.bot, run.scenario, run.kind, run.status)
await suite.run(Path("eval-runs/logs"), params=EvalSessionParams(use_cache=False))
for run in suite.runs:
verdict = run.error or ("passed" if run.result and run.result.passed else "failed")
print(f"{run.bot} / {run.scenario} #{run.attempt}: {verdict}")
asyncio.run(main())
```
Each run is mutated in place as it executes (`status`, `result`, `error`, `duration_ms`), so a live display can render directly from `suite.runs`. A run's `kind` is `script` or `simulation`, and its `result` is the matching result type. A simulation appears once per attempt: `attempts` is the manifest's `repeat`, or the simulation's own `runs`, and `sweep` says which. A sweep is a measurement, where a failure is data; a simulation's own runs are a requirement, where every attempt must pass.
The `on_update` callback parameter to `suite.run()` is deprecated since 1.9.0
and will be removed in 2.0.0. Use the `on_update` event handler instead. So
are its `use_cache` and `default_timeout_ms` keyword arguments: pass
`params=EvalSessionParams(...)`.
`EvalManifest.load()` accepts keyword overrides for every manifest value (`concurrency`, `base_port`, `spawn`, `scenarios_dir`, `repeat`, and so on), mirroring the CLI flags.
## Migrating from 1.8
Simulations arrived in 1.9.0 with a rename of the scripted API, so the two kinds sit side by side. `EvalSession` stays. It is now the base class of both kinds and the home of the `from_scenario()` that builds either, and it moved to `pipecat.evals.session`. Constructing `EvalSession(...)` directly is no longer supported; use the kind's own class. The old names keep working until 2.0.0 and emit a `DeprecationWarning`:
| Deprecated | Use instead |
| -------------------------------------------------- | ---------------------------------------------------------------------- |
| `pipecat.evals.harness` | `pipecat.evals.session` and `pipecat.evals.script_session` |
| `EvalScenario` | `EvalScriptScenario` (in `pipecat.evals.script`) |
| `EvalTurn` | `EvalScriptTurn` |
| `EvalResult` | `EvalScriptResult` (in `pipecat.evals.results`) |
| `EvalTurnResult` | `EvalScriptTurnResult` |
| `EvalTurnProgress` | `EvalScriptTurnProgress` |
| `RTVIEvalSerializer` | `EvalSerializer` |
| `on_progress=` / `on_update=` | The `on_progress` / `on_update` event handlers |
| `stop_bot=`, `use_cache=`, and the other run knobs | `params=EvalSessionParams(...)`, see [Run parameters](#run-parameters) |
`EvalSpeech` and `EvalTranscriber` are gone: pass `user_tts=` and `bot_stt=` as shown [above](#custom-audio-services). `SEND_CHUNK_MS` was removed with no replacement, since the harness now paces the user's audio through its own output transport.