> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/labs-voice-agent/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/labs-voice-agent/_mcp/server.

# nemo_voice_agent.evaluation.bridge

Voice Agent Evaluation Bridge

Connects two voice agents via WebSocket and provides:

* Bidirectional audio routing
* Response latency measurement
* Dynamic system prompt updates via RTVI actions
* Conversation monitoring and metrics

## Module Contents

### Classes

| Name                                                                                           | Description                                                    |
| ---------------------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| [`EvaluationMetrics`](#nemo_voice_agent-evaluation-bridge-EvaluationMetrics)                   | Metrics collected during evaluation                            |
| [`ResponseLatency`](#nemo_voice_agent-evaluation-bridge-ResponseLatency)                       | Single response latency measurement                            |
| [`SegmentEntry`](#nemo_voice_agent-evaluation-bridge-SegmentEntry)                             | Entry for segLST format (segment list with timing)             |
| [`VoiceAgentEvaluationBridge`](#nemo_voice_agent-evaluation-bridge-VoiceAgentEvaluationBridge) | Evaluation bridge that connects two voice agents via WebSocket |

### Data

[`RTVI_BOT_SERVER_MESSAGE`](#nemo_voice_agent-evaluation-bridge-RTVI_BOT_SERVER_MESSAGE)

[`RTVI_BOT_STARTED_SPEAKING`](#nemo_voice_agent-evaluation-bridge-RTVI_BOT_STARTED_SPEAKING)

[`RTVI_BOT_STOPPED_SPEAKING`](#nemo_voice_agent-evaluation-bridge-RTVI_BOT_STOPPED_SPEAKING)

[`RTVI_BOT_TRANSCRIPTION`](#nemo_voice_agent-evaluation-bridge-RTVI_BOT_TRANSCRIPTION)

[`RTVI_BOT_TTS_TEXT`](#nemo_voice_agent-evaluation-bridge-RTVI_BOT_TTS_TEXT)

[`STOP_REASON_EXIT`](#nemo_voice_agent-evaluation-bridge-STOP_REASON_EXIT)

[`STOP_REASON_TIMEOUT`](#nemo_voice_agent-evaluation-bridge-STOP_REASON_TIMEOUT)

[`_RTVI_TYPES_ALREADY_TAGGED`](#nemo_voice_agent-evaluation-bridge-_RTVI_TYPES_ALREADY_TAGGED)

### API

```python
class nemo_voice_agent.evaluation.bridge.EvaluationMetrics(
    turns: list = list(),
    latencies: typing.List[nemo_voice_agent.evaluation.bridge.ResponseLatency] = list(),
    start_time: datetime.datetime = None,
    end_time: datetime.datetime = None,
    log_entries: typing.List[typing.Tuple[float, str]] = list(),
    user_last_audio_time: typing.Optional[float] = None,
    agent_last_audio_time: typing.Optional[float] = None,
    waiting_for_agent_response: bool = False,
    last_user_transcript: str = '',
    user_current_transcript: str = '',
    agent_current_transcript: str = '',
    thread_start_timestamp: typing.Optional[float] = None,
    segments: typing.List[nemo_voice_agent.evaluation.bridge.SegmentEntry] = list(),
    current_user_segment: typing.Optional[nemo_voice_agent.evaluation.bridge.SegmentEntry] = None,
    current_agent_segment: typing.Optional[nemo_voice_agent.evaluation.bridge.SegmentEntry] = None,
    agent_final_response: typing.List[str] = list(),
    agent_final_response_time: typing.List[float] = list()
)
```

Dataclass

Metrics collected during evaluation

**`agent_current_transcript`**

---

**`agent_final_response`**

---

**`agent_final_response_time`**

---

**`agent_last_audio_time`**

---

**`current_agent_segment`**

---

**`current_user_segment`**

---

**`end_time`**

---

**`last_user_transcript`**

---

**`latencies`**

---

**`log_entries`**

---

**`segments`**

---

**`start_time`**

---

**`thread_start_timestamp`**

---

**`turns`**

---

**`user_current_transcript`**

---

**`user_last_audio_time`**

---

**`waiting_for_agent_response`**

---

```python
nemo_voice_agent.evaluation.bridge.EvaluationMetrics.get_latency_stats()
```

Calculate latency statistics

```python
nemo_voice_agent.evaluation.bridge.EvaluationMetrics.reset()
```

Reset all metrics to prepare for a new scenario

```python
class nemo_voice_agent.evaluation.bridge.ResponseLatency(
    user_stop_time: float,
    agent_start_time: float,
    latency_ms: float,
    user_transcript: str = '',
    agent_transcript: str = ''
)
```

Dataclass

Single response latency measurement

**`agent_start_time`**

---

**`agent_transcript`**

---

**`latency_ms`**

---

**`user_stop_time`**

---

**`user_transcript`**

---

```python
class nemo_voice_agent.evaluation.bridge.SegmentEntry(
    start_time: float,
    end_time: float,
    speaker: str,
    transcript: str
)
```

Dataclass

Entry for segLST format (segment list with timing)

**`end_time`**

---

**`speaker`**

---

**`start_time`**

---

**`transcript`**

---

```python
class nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge(
    user_url: str,
    agent_url: str,
    output_dir: typing.Optional[str] = None,
    scenario_name: typing.Optional[str] = None,
    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 = 16000,
    audio_chunk_in_seconds: float = 0.016,
    use_burst_mode: bool = False,
    burst_size_range: typing.Tuple[int, int] = (3, 8),
    burst_delay_ms: int = 0,
    grace_period: float = 1.0,
    turn_start_offset_secs: float = -0.0,
    turn_end_offset_secs: float = -0.3,
    noise_config: typing.Optional[nemo_voice_agent.utils.audio.NoiseConfig] = None,
    log_level: str = 'DEBUG'
)
```

Evaluation bridge that connects two voice agents via WebSocket
and provides control through RTVI actions.

Key features:

* Routes audio bidirectionally between agents
* Monitors transcriptions and metrics
* Measures response latency by tracking audio frames
* Can send RTVI control messages to update prompts
* Works with distributed agents

**`agent_to_user_queue`**

---

**`exit_settle_delay`**

---

**`final_response_file`**

---

**`final_scenario_db_hash_file`**

---

**`metrics`**

---

**`scenario_summary`**

---

**`sent_to_agent_chunks`**

---

**`sent_to_user_chunks`**

---

**`serializer`**

---

**`shadow_state`**

---

**`shadow_tool_map`**

---

**`stop_event`**

---

**`sync_enabled`**

---

**`sync_lock`**

---

**`threads`**

---

**`token_usage`**

---

**`user_to_agent_queue`**

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._apply_initialization(
    scenario,
    user_shared_state_init: dict,
    agent_shared_state_init: dict
) -> None
```

async

Send `apply_initialization` to BOTH bots — always, even with no init actions.

Bot-side dispatch via the `apply_initialization` RTVI action.
The handler does three things per bot:

1. Merge `shared_state_init` JSON into the bot's
   `shared_state` (preserving runtime sentinels stashed by
   `update_system_prompt`).
2. If `db_path` is in the merged state, resolve it against
   `EVAL_DATA_ROOT` and replace with the loaded `db` dict.
3. Dispatch each per-side init function record (filtered by
   `side`) against the now-loaded `db`.

Always called once per bot, even when `scenario.initialization_actions`
is empty — steps 1 and 2 (state merge + DB load) must run for every
tau2 scenario regardless. Single-side domains (eva / airline / retail)
whose `setup_shared_state` populates `db_path` rely on this call
to trigger the DB load. The bot's handler is fast (\~tens of ms) when
actions list is empty.

**Failure handling:** any per-bot `success: false` response
(invalid JSON, missing DB file, missing init function in the bot's
registry, dispatcher exception) raises `RuntimeError` so the
calling `prepare_for_scenario` aborts before the conversation
starts. Partial seeding produces noisy / unscoreable runs.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._build_conversation_log()
```

Build conversation log entries from finalized segments with computed latencies.

Called after all segments are finalized so that latency calculation has access
to all user and agent segments. For each agent segment, latency is computed as:
agent.start\_time - previous\_user.end\_time
Positive = normal response delay, negative = agent interrupted/barged in early.

Applies turn\_start\_offset\_secs and turn\_end\_offset\_secs to match seglst timestamps.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._finalize_speaker_turn(
    speaker: str,
    timestamp: float
) -> typing.Optional[nemo_voice_agent.evaluation.bridge.SegmentEntry]
```

Finalize the current in-progress turn for the given speaker.

Sets end\_time, assigns transcript (or "\[INTERRUPTED]" if no TTS text was received),
appends the segment to self.metrics.segments, and clears accumulation state.

**Parameters:**

**`speaker`**

"user" or "agent"

---

**`timestamp`**

Absolute timestamp (asyncio loop time)

---

**Returns:** `Optional[SegmentEntry]`

The finalized SegmentEntry, or None if no segment was in progress.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._format_turn_log(
    role: str,
    text: str,
    start_time: float,
    end_time: float,
    latency_ms: float = None
) -> str
```

Format a turn entry for the conversation log.

**Parameters:**

**`role`**

"user" or "agent"

---

**`text`**

Transcript text

---

**`start_time`**

Turn start time (relative to scenario start)

---

**`end_time`**

Turn end time (relative to scenario start)

---

**`latency_ms`**

Optional response latency in milliseconds

---

**Returns:** `str`

Formatted log entry string

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._fresh_token_usage() -> dict
```

staticmethod

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._get_relative_time(
    timestamp: float
) -> float
```

Get time relative to scenario start (thread start time).

**Parameters:**

**`timestamp`**

Absolute timestamp (asyncio loop time)

---

**Returns:** `float`

Time in seconds relative to thread\_start\_timestamp, or 0 if not set

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._log_rtvi_event(
    side: str,
    message_type: str,
    data: dict
) -> None
```

Emit a uniform side-tagged DEBUG log line for an RTVI event.

Each monitor (`_monitor_user_message` / `_monitor_agent_message`) is
bound to one side by construction, so we know which side emitted the
event without time-correlation. This helper writes a one-line tag plus
expanded payload lines for the message types that carry analytically
useful structured data (TTFB values, token counts, transcription text,
RTVI action names).

Skipped types (`_RTVI_TYPES_ALREADY_TAGGED`): the monitors already
emit dedicated side-tagged lines like `[AGENT TTS] <text>` for these,
so adding `[AGENT EVENT] type=bot-tts-text` on top would just
duplicate. Their expanded payload (the TTS text) is already in the
existing line.

Skipping is by *event-tag emission only* — the underlying RTVI message
still flows through pipecat's `ProtobufFrameSerializer:deserialize`
DEBUG log, so nothing is lost from the bridge log.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._monitor_agent_message(
    frame
)
```

async

Monitor agent messages for timing and transcripts.

Turn lifecycle: BOT\_STARTED\_SPEAKING → BOT\_TTS\_TEXT (accumulate) → BOT\_STOPPED\_SPEAKING (finalize).
Latency is measured from user's last audio to agent's first audio frame.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._monitor_user_message(
    frame
)
```

async

Monitor user messages for timing and transcripts.

Turn lifecycle: BOT\_STARTED\_SPEAKING → BOT\_TTS\_TEXT (accumulate) → BOT\_STOPPED\_SPEAKING (finalize).

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._propagate_cross_side_sync(
    action: dict,
    source_side: str
) -> None
```

async

Replay one action onto the shadow DBs, run scenario.sync\_state, dispatch deltas.

Called from `_monitor_agent_message` / `_monitor_user_message`
when an `action-applied` RTVI server message arrives. `source_side`
is which side's monitor saw it.

No-op when `sync_enabled` is False (single-side scenarios).

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._receive_agent_to_queue(
    agent_ws,
    duration: float
)
```

async

Receive audio from agent WebSocket and put into queue for user thread.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._receive_to_queue(
    ws: websockets.WebSocketClientProtocol,
    duration: float,
    direction: str,
    queue: queue.Queue,
    monitor_func: typing.Callable
)
```

async

Receive audio from websocket and put into queue.

**Parameters:**

**`ws`**

Source websocket to receive from

---

**`duration`**

How long to run the receive loop in seconds

---

**`direction`**

For logging (e.g., "USER→AGENT", "AGENT→USER")

---

**`queue`**

Thread-safe queue to put audio chunks into

---

**`monitor_func`**

Async monitoring function for metrics (e.g., \_monitor\_user\_message)

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._receive_user_to_queue(
    user_ws,
    duration: float
)
```

async

Receive audio from user WebSocket and put into queue for agent thread.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._resample_audio(
    audio: numpy.ndarray,
    from_rate: int,
    to_rate: int
) -> numpy.ndarray
```

staticmethod

Resample audio array using soxr. Returns int16 array.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._retrieve_context_history(
    ws
) -> dict
```

async

Retrieve the context history from the WebSocket. First send a message to the ws to trigger the
`get_context_history` RTVI action, then wait for the response.
Args:
ws: WebSocket connection
Returns:
context\_history: context history as a dictionary with two keys: `context` and `logs`,
where `context` the LLM context history, and `logs` is the bot server logs.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._retrieve_scenario_summary(
    ws,
    include_db: bool = False
) -> dict
```

async

Retrieve the scenario summary from the bot via the
`get_scenario_summary` RTVI action. Mirrors `_retrieve_context_history`.

**Parameters:**

**`ws`**

WebSocket connection to the agent bot.

---

**`include_db`**

When `True`, ask the bot to inline the `db`
dict alongside the hash. Set by the runner via
`Scenario.db_state_assertions` (predicate evaluation needs
the actual DB, not just a hash). Off by default to preserve
the existing hash-out behavior for retail (whose 7 MB DB
would exceed pipecat's 1 MB WS frame cap).

---

**Returns:** `dict`

Dict with at minimum `{"actions": list, "db_hash": str|None}`.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_audio_log()
```

Save final sent audio chunks to disk as stereo WAV for debugging.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_bot_server_history(
    output_dir: typing.Union[str, pathlib.Path],
    context_history: dict,
    role: str = ''
)
```

Save the bot server context history to a JSON file under the output directory.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_conversation_log()
```

Build and write conversation log entries sorted by start time, with computed latencies.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_final_response()
```

Save the agent's final response to a JSON file under the output directory.

Two sources, in priority order:

1. **Pull** (`self.scenario_summary["actions"]`) — the bridge-pulled
   auto-aggregated action list. Used when the bot registered the
   `get_scenario_summary` action (post-commit-3 bots) and returned
   a non-empty actions list.
2. **Push** (`self.metrics.agent_final_response`) — `<final_response>`
   text messages captured during the conversation. Used by domains that
   still have an LLM-callable summary tool (restaurant / customer\_service
   / qa) or as a fallback when pull returned empty.

Output is always list-wrapped (`[{"actions": ...}]`) for shape compat
with the existing strict comparator and downstream consumers.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_scenario_db()
```

Save the post-run scenario DB hash(es) to `final_scenario_db_hash.txt`.

Sourced from the bridge-pulled `scenario_summary["db_hash"]` (and
`["user_db_hash"]` for telecom). Skipped if no pull happened (legacy
bots) or the hash is None. Used by the runner's DB-state matching
when `scenario.expected_scenario_db` is set — the runner compares
the expected-DB hash (computed in-process from its own gold replay)
against the bot's reported hash.

Hash-only design: the full DB stays on the bot server. See
`create_get_scenario_summary_action` for the rationale (WebSocket
frame size limit; tau2's DB is 7 MB while pipecat's default frame
cap is 1 MB).

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_seglst()
```

Save segLST transcript file with offset-adjusted timestamps.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._save_user_agent_history()
```

Save the user and agent context history to a JSON file under the output directory.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_agent_to_user(
    user_ws,
    audio_stream: nemo_voice_agent.utils.audio.AudioStream,
    duration: int
)
```

async

Get audio from queue, process through AudioStream, send to user WebSocket.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_apply_initialization(
    ws,
    side_label: str,
    domain: str,
    shared_state_init: dict,
    actions: typing.List[dict]
) -> None
```

async

Send one `apply_initialization` action and wait for the result.

`side_label` is the bot label (`"agent"` or `"user"`) used only
in log lines and error messages — the bot itself doesn't need to know
which side it is, since each action in the payload already carries
its own `side` field that the dispatcher routes by.

**Raises:**

* `RuntimeError`: if the bot returns `success: false` or the request
  times out. The caller (`_apply_initialization`) propagates
  this up so `prepare_for_scenario` aborts cleanly.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_apply_sync_delta(
    ws,
    side_label: str,
    domain: str,
    delta: dict
) -> None
```

async

Send a single `apply_sync_delta` RTVI action to a bot.

Mirrors `_send_apply_initialization`: builds an action
message with a unique id, awaits the response, logs success or
failure. Failures are warnings — sync drift is recoverable as
long as the next propagation cycle eventually catches up.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_audio_stream(
    audio_stream: nemo_voice_agent.utils.audio.AudioStream,
    dest_ws: websockets.WebSocketClientProtocol,
    direction: str,
    duration: int,
    source_queue: queue.Queue,
    sent_chunks_list: typing.List[bytes]
)
```

async

Send audio stream at fixed intervals from AudioStream with duration and grace period.

**Parameters:**

**`audio_stream`**

AudioStream containing buffered and resampled audio

---

**`dest_ws`**

Destination websocket to send to

---

**`direction`**

For logging (e.g., "USER→AGENT", "AGENT→USER")

---

**`duration`**

How long to run the send loop in seconds

---

**`source_queue`**

Queue to retrieve audio chunks from

---

**`sent_chunks_list`**

List to append sent chunks to for tracking

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_client_ready(
    ws
)
```

async

Send RTVI client-ready handshake and wait for bot-ready

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_reset_action(
    ws,
    agent_name: str
)
```

async

Send RTVI reset action to clear conversation history.

**Parameters:**

**`ws`**

WebSocket connection

---

**`agent_name`**

Name of agent (for logging)

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._send_user_to_agent(
    agent_ws,
    audio_stream: nemo_voice_agent.utils.audio.AudioStream,
    duration: int
)
```

async

Get audio from queue, process through AudioStream, send to agent WebSocket.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge._setup_cross_side_sync(
    scenario
) -> None
```

async

Prepare shadow DBs + tool map for cross-side state propagation.

Called from `prepare_for_scenario` once per scenario,
**after** `_apply_initialization` so the bot-side
live state is already at its post-init starting point. We
mirror that here by replaying the same init actions onto the
shadow DBs, then run a one-shot `sync_state` to propagate
any cross-side state that should be coherent at conversation
start (e.g. agent-side `set_data_usage(15.1)` flipping
user-side `surroundings.mobile_data_usage_exceeded` to True).

No-op when the scenario's `sync_state` is the inherited
default (single-side domains).

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.agent_websocket_thread(
    duration: int
)
```

Thread 2: Handle all agent WebSocket traffic (bidirectional).

This thread:

* Gets user audio from user\_to\_agent\_queue
* Sends user audio to agent WebSocket
* Receives audio from agent WebSocket
* Puts agent audio into agent\_to\_user\_queue for user thread

**Parameters:**

**`duration`**

How long to run (seconds)

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.connect(
    max_retries: int = 5,
    retry_delay: float = 1.0
)
```

async

Connect to both user and agent with retry logic

**Parameters:**

**`max_retries`**

Maximum number of connection attempts per endpoint

---

**`retry_delay`**

Initial delay between retries (doubles each retry)

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.disconnect(
    print_stats: bool = False
)
```

async

Disconnect from both user and agent.

**Parameters:**

**`print_stats`**

If True, print final latency statistics (default: True)
Set to False when disconnecting during scenario resets

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.get_metrics()
```

Get evaluation metrics

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.init_output_dir(
    output_dir: str,
    scenario_name: typing.Optional[str] = None,
    log_level: str = 'DEBUG'
)
```

Initialize the output directory and all derived log/audio file paths.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.prepare_for_scenario(
    scenario,
    output_dir: str,
    log_level: str = 'DEBUG'
)
```

async

Prepare the bridge for a scenario.

**Parameters:**

**`scenario`**

A `Scenario` instance. The bridge calls methods
on it directly (`get_user_prompt`, `get_agent_prompt`,
`setup_shared_state`, `initialization_actions`,
`sync_state`, etc.) — no intermediate dict
serialization step. Single source of truth.

---

**`output_dir`**

Per-scenario output directory.

---

**`log_level`**

Pipecat log level for the bot servers.

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.reset()
```

async

Reset metrics and both agents' conversation history

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.reset_agent()
```

async

Reset agent's conversation history.
Useful to clear context between evaluation scenarios.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.reset_user()
```

async

Reset user's conversation history.
Useful to clear context between evaluation scenarios.

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.run_scenario(
    duration: int = 300
)
```

async

Route audio between agents and monitor conversation.
Uses separate threads per WebSocket to eliminate asyncio contention.

**Parameters:**

**`duration`**

Duration of the evaluation in seconds

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.send_text_to_agent(
    text: str
)
```

async

Send a text message to the agent agent to trigger conversation.

**Parameters:**

**`text`**

Text to send to agent agent's LLM

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.send_text_to_user(
    text: str
)
```

async

Send a text message to the user agent to trigger conversation.

**Parameters:**

**`text`**

Text to send to user agent's LLM

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.set_noise_config(
    noise_config: typing.Optional[typing.Union[nemo_voice_agent.utils.audio.NoiseConfig, dict]] = None
)
```

Set the noise configuration

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.update_agent_prompt(
    prompt: str,
    tools: str,
    auto_reset: bool = False,
    add_suffix: bool = False,
    tool_domain: str = 'default'
)
```

async

Update agent's system prompt via RTVI action.

Scenario fixture data (`db_path`, custom keys from
`Scenario.setup_shared_state`) is NOT sent here — it flows via
the subsequent `apply_initialization` call instead.

**Parameters:**

**`prompt`**

New system prompt text

---

**`tools`**

New tools in json string format

---

**`auto_reset`**

If True, also sends reset action after updating prompt

---

**`add_suffix`**

If True, add previously configured system prompt suffix to the new prompt

---

**`tool_domain`**

Registry namespace the bot server should use to look up
tools by name (e.g., `"tau2_airline"`). Falls back to
`"default"` per-tool if the name isn't in the specified
domain (with a warning logged bot-side).

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.update_user_prompt(
    prompt: str,
    tools: str,
    auto_reset: bool = False,
    add_suffix: bool = False,
    tool_domain: str = 'default'
)
```

async

Update user's system prompt via RTVI action.

Scenario fixture data (`db_path`, custom keys from
`Scenario.setup_shared_state`) is NOT sent here — it flows via
the subsequent `apply_initialization` call instead.

**Parameters:**

**`prompt`**

New system prompt text

---

**`tools`**

New tools in json string format

---

**`auto_reset`**

If True, also sends reset action after updating prompt

---

**`add_suffix`**

If True, add previously configured system prompt suffix to the new prompt

---

**`tool_domain`**

Registry namespace the bot server should use to look up
tools by name. Stashed on bot-side `shared_state["__tool_domain__"]`
for write tools to read when emitting `action-applied` events.

---

```python
nemo_voice_agent.evaluation.bridge.VoiceAgentEvaluationBridge.user_websocket_thread(
    duration: int
)
```

Thread 1: Handle all user WebSocket traffic (bidirectional).

This thread:

* Receives audio from user WebSocket
* Puts user audio into user\_to\_agent\_queue for agent thread
* Gets agent audio from agent\_to\_user\_queue
* Sends agent audio to user WebSocket

**Parameters:**

**`duration`**

How long to run (seconds)

---

```python
nemo_voice_agent.evaluation.bridge.RTVI_BOT_SERVER_MESSAGE = RTVI.ServerMessage(data=(RTVI.TextMessageData(text=''))).type
```

```python
nemo_voice_agent.evaluation.bridge.RTVI_BOT_STARTED_SPEAKING = RTVI.BotStartedSpeakingMessage().type
```

```python
nemo_voice_agent.evaluation.bridge.RTVI_BOT_STOPPED_SPEAKING = RTVI.BotStoppedSpeakingMessage().type
```

```python
nemo_voice_agent.evaluation.bridge.RTVI_BOT_TRANSCRIPTION = RTVI.BotTranscriptionMessage(data=(RTVI.TextMessageData(text=''))).type
```

```python
nemo_voice_agent.evaluation.bridge.RTVI_BOT_TTS_TEXT = RTVI.BotTTSTextMessage(data=(RTVI.TextMessageData(text=''))).type
```

```python
nemo_voice_agent.evaluation.bridge.STOP_REASON_EXIT = '[EXIT]'
```

```python
nemo_voice_agent.evaluation.bridge.STOP_REASON_TIMEOUT = '[TIMEOUT]'
```

```python
nemo_voice_agent.evaluation.bridge._RTVI_TYPES_ALREADY_TAGGED = frozenset({RTVI_BOT_STARTED_SPEAKING, RTVI_BOT_TTS_TEXT, RTVI_BOT_STOPPED_SPEAKI...
```