nemo_voice_agent.evaluation.runner | NeMo Labs Voice Agent

Dynamic Voice Agent Evaluation Runner

Runs evaluation scenarios with dynamic system prompt updates. Accepts structured Scenario objects instead of raw dicts.

Module Contents

Classes

Name	Description
`ConversationEndPolicy`	Policies used to decide whether a scenario reached a valid end state.
`RunAggregator`	Run-level metric accumulator.

Functions

Name	Description
`_load_optional_trace_metrics`	-
`auto_retry_reason`	Decide whether a just-finished scenario attempt should be re-run.
`evaluate_conversation_end`	Return the policy verdict and the concrete evidence that produced it.
`run_dynamic_evaluation`	Run evaluation with dynamic scenario switching and latency measurement.

API

class nemo_voice_agent.evaluation.runner.ConversationEndPolicy

Bases: enum.Enum

Policies used to decide whether a scenario reached a valid end state.

TOOL_ONLY

= 'tool-only'

VALID_TERMINAL_STATE

= 'valid-terminal-state'

class nemo_voice_agent.evaluation.runner.RunAggregator(
    success_results: typing.List[bool] = list(),
    task_success_results: typing.List[bool] = list(),
    action_match_results: typing.List[bool] = list(),
    judge_score_results: typing.List[float] = list(),
    judge_pass_results: typing.List[bool] = list(),
    db_state_results: typing.List[bool] = list(),
    nl_assertion_results: typing.List[bool] = list(),
    db_state_assertion_results: typing.List[bool] = list(),
    clean_exit_results: typing.List[bool] = list(),
    insufficient_turns_skipped: typing.List[str] = list(),
    per_domain_success: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_task_success: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_action_match: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_judge_score: typing.Dict[str, typing.List[float]] = dict(),
    per_domain_judge_pass: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_db_state: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_nl_assertion: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_db_state_assertion: typing.Dict[str, typing.List[bool]] = dict(),
    per_domain_clean_exit: typing.Dict[str, typing.List[bool]] = dict(),
    run_token_usage: dict = (lambda: {'agent': {'n_call...
)

Dataclass

Run-level metric accumulator.

Holds per-signal pass-rate buckets + per-domain breakdowns + per-side token totals across all scenarios in a run. add_scenario(metrics, domain) folds a per-scenario metrics dict into all buckets uniformly — used by both the freshly-run path (after computing metrics) and the resume-skip path (after loading metrics.json from disk for an already-completed scenario). Centralizing this logic guarantees the final all_summary.txt aggregate is identical regardless of whether a scenario ran live or was loaded from a prior session.

Bucket lists are kept as flat List[bool|float] (not keyed by SuccessSignal) so the existing downstream consumers in run_dynamic_evaluation (summary writer, per-domain rollups) can reference them by name without changing 100+ call sites. Dict-key lookups on incoming metrics use SuccessSignal.* members directly (StrEnum makes them str-equal to their JSON-key values).

action_match_results

List[bool] = field(default_factory=list)

clean_exit_results

List[bool] = field(default_factory=list)

db_state_assertion_results

List[bool] = field(default_factory=list)

db_state_results

List[bool] = field(default_factory=list)

insufficient_turns_skipped

List[str] = field(default_factory=list)

judge_pass_results

List[bool] = field(default_factory=list)

judge_score_results

List[float] = field(default_factory=list)

nl_assertion_results

List[bool] = field(default_factory=list)

per_domain_action_match

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_clean_exit

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_db_state

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_db_state_assertion

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_judge_pass

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_judge_score

Dict[str, List[float]] = field(default_factory=dict)

per_domain_nl_assertion

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_success

Dict[str, List[bool]] = field(default_factory=dict)

per_domain_task_success

Dict[str, List[bool]] = field(default_factory=dict)

run_token_usage

dict

success_results

List[bool] = field(default_factory=list)

task_success_results

List[bool] = field(default_factory=list)

nemo_voice_agent.evaluation.runner.RunAggregator.add_scenario(
    metrics: dict,
    domain: str,
    min_agent_turns: int = 0
) -> None

Append one scenario’s metrics into all run-level + per-domain buckets.

Idempotent against incomplete metrics dicts — each signal is only appended when its corresponding key is present and well-typed in metrics. Signals not opted into by the scenario (no expected_scenario_db, no NL assertions, etc.) simply don’t contribute to their bucket.

SuccessSignal enum members are used as the dict keys when looking up signal values in metrics — StrEnum members compare equal to their string values, so this preserves byte-stability of the on-disk metrics.json format while pinning the lookup to a typo-resistant symbol.

When min_agent_turns > 0, scenarios with fewer agent LLM responses (stalled agent) are counted as failures in the composite success/task buckets, skipped in the per-signal buckets (those measurements are meaningless for a run that stalled), and recorded in insufficient_turns_skipped for the resume warning. Token usage is still rolled up so cost accounting stays accurate.

nemo_voice_agent.evaluation.runner._load_optional_trace_metrics(
    scenario_dir: str
) -> typing.Optional[dict]

nemo_voice_agent.evaluation.runner.auto_retry_reason(
    turn_count: typing.Optional[int],
    agent_llm_count: typing.Optional[int],
    min_agent_turns: int,
    on_stale: bool,
    on_insufficient_turns: bool
) -> typing.Optional[str]

Decide whether a just-finished scenario attempt should be re-run.

Returns a short reason string when the attempt should be retried, or None to accept it.

Two triggers, deliberately separated by how certain the failure is:

Hard failure (turn_count == 0): not a single conversation turn was exchanged, so the bots never talked and no measurement exists. Re-running cannot bias the score because there is no score to bias. This mirrors the "0 turns (bot crashed before audio)" case in classify_scenario_resume_state. Governed by on_stale (default True).
Soft heuristic (0 < agent_llm_count < min_agent_turns): the agent responded, but fewer times than the floor. This is only a proxy for a stalled backend — legitimately short scenarios (refusals, quick lookups) sit right on the floor. Because insufficient_turns forces is_successful=False, retrying converts a guaranteed failure into a fresh draw, which inflates scores whenever the heuristic fires on a scenario that would have passed. Governed by on_insufficient_turns, which defaults to False for that reason.

Turn count, not token_usage, gates the hard trigger on purpose. token_usage.agent.n_calls only increments when the agent emits an RTVI token-usage message, so a backend that does not report usage would read as zero on a perfectly healthy run — and with on_stale defaulting to True that would silently re-run every scenario. Turn count comes from the bridge’s own conversation record and does not depend on backend reporting.

A None count means the signal was unavailable; never retry on a guess.

nemo_voice_agent.evaluation.runner.evaluate_conversation_end(
    policy: nemo_voice_agent.evaluation.runner.ConversationEndPolicy | str,
    stop_reason: str,
    turns: list[dict],
    simulator_end_reported: bool = False
) -> tuple[bool, str]

Return the policy verdict and the concrete evidence that produced it.

nemo_voice_agent.evaluation.runner.run_dynamic_evaluation(
    user_url: str,
    agent_url: str,
    output_dir: str,
    scenarios: typing.List[nemo_voice_agent.evaluation.scenarios.classes.Scenario],
    audio_chunk_in_seconds: float = 0.016,
    duration_per_scenario: typing.Optional[int] = None,
    pause_between_scenarios: float = 0.5,
    user_output_sample_rate: int = 24000,
    agent_output_sample_rate: int = 24000,
    user_input_sample_rate: int = 16000,
    agent_input_sample_rate: int = 16000,
    output_sample_rate: int = 24000,
    global_timestamp: str = None,
    logger: nemo_voice_agent.utils.FileLogger = None,
    judge: typing.Optional[nemo_voice_agent.evaluation.utils.LLMJudge] = None,
    judge_threshold: typing.Optional[float] = None,
    judge_include_conversation: bool = False,
    strict_match: bool = False,
    min_agent_turns: int = 0,
    auto_resume_on_stale: bool = True,
    auto_resume_on_insufficient_turns: bool = False,
    max_auto_retries: int = 1,
    auto_retry_backoff_secs: float = 5.0,
    conversation_end_policy: nemo_voice_agent.evaluation.runner.ConversationEndPolicy | str = ConversationEndPolicy.TOOL_...,
    inactivity_timeout: typing.Optional[float] = 120.0
)

async

Run evaluation with dynamic scenario switching and latency measurement.

Parameters:

user_url

str

WebSocket URL of user (simulated user)

agent_url

str

WebSocket URL of agent being tested

output_dir

str

Output directory for results

scenarios

List[Scenario]

List of Scenario objects defining each evaluation scenario

audio_chunk_in_seconds

float" default="0.016

Audio chunk in seconds for the audio stream (default: 0.016)

duration_per_scenario

Optional[int]" default="None

Maximum duration per scenario in seconds, which overrides the scenario’s own max_duration if set.

pause_between_scenarios

float" default="0.5

Seconds to pause between scenarios

user_output_sample_rate

int" default="24000

User TTS output sample rate (default: 24000)

agent_output_sample_rate

int" default="24000

Agent TTS output sample rate (default: 24000)

user_input_sample_rate

int" default="16000

User STT input sample rate (default: 16000)

agent_input_sample_rate

int" default="16000

Agent STT input sample rate (default: 16000)

output_sample_rate

int" default="24000

Output sample rate for recorded audio (default: 24000)

global_timestamp

str" default="None

Timestamp string for output file naming

logger

FileLogger" default="None

FileLogger instance for logging

judge

Optional[LLMJudge]" default="None

LLMJudge instance for judging the scenario

judge_threshold

Optional[float]" default="None

Threshold for judging the scenario if binary result is desired, None for score based result

judge_include_conversation

bool" default="False

If True, include bridge transcript turns in the judge input. Disabled by default because interrupted/cross-talk segments can make these turns noisy.

strict_match

bool" default="False

If True, force disallow_extra_items=True on every scenario for this run, overriding each scenario’s own setting. Default False respects per-scenario flags.

min_agent_turns

int" default="0

scenarios with agent turns less than this number will be treated as incomplete

auto_resume_on_stale

bool" default="True

If True (default), immediately re-run a scenario that recorded zero conversation turns — an unambiguous infrastructure failure with no measurement to bias. Keyed on the bridge’s turn record rather than token_usage, which reads zero on a healthy run against a backend that does not report usage.

auto_resume_on_insufficient_turns

bool" default="False

If True, also re-run scenarios that merely fall below min_agent_turns. Default False: that floor is a heuristic, legitimately short scenarios sit on it, and retrying a forced failure biases the score upward.

max_auto_retries

int" default="1

Maximum automatic re-runs per scenario (default 1). Bounds the loop when the backend is persistently down.

auto_retry_backoff_secs

float" default="5.0

Seconds to wait before an automatic re-run (default 5.0). An immediate retry against a hung backend is the least likely moment to succeed.

conversation_end_policy

ConversationEndPolicy | str" default="ConversationEndPolicy.TOOL_ONLY

Policy for the CLEAN_EXIT gate. tool-only requires the agent’s EndConversationTool. valid-terminal-state also accepts a simulator-reported end or an inactivity timeout after the user’s final turn.

inactivity_timeout

Optional[float]" default="120.0

Seconds without conversational activity before the bridge stops the scenario.