nemo_voice_agent.evaluation.initialization_functions

Initialization-function registry + dispatcher for Scenario.initialization_actions.

An initialization function is a state-mutating action replayed against the live bot-side DB before the scenario starts. They seed the environment to a known starting state so the agent and user simulator can have a meaningful conversation (e.g., upstream tau2-bench’s set_user_info, turn_roaming_off, enable_roaming).

Concretely:

Function signature: (db: dict, **arguments) -> None. Mutates the passed dict in place; return value ignored. Errors raise normally and are caught by the dispatcher, which returns {success: False, errors: [...]} so the bridge can abort the scenario cleanly. Functions are side-agnostic — they don’t know if they’re operating on the “agent DB” or “user DB”, and they don’t need to: the caller (bot handler) chose the right DB dict before invoking them.
Functions are registered per (domain, func_name) only — no side. Function names are unique within a domain by upstream construction (in tau2 telecom, the 16 user-side init names and 4 agent-side init names are disjoint), so adding side to the registry key would namespace against a non-collision.
Dispatch is bot-side (the opposite of predicates). The bot has the live shared_state["db"] (agent) or shared_state["user_db"] (user) and the toolkit instance methods that read/write them; init functions just mutate the dict directly. The bridge calls the apply_initialization RTVI action once per side (agent bot for side=="agent" actions, user bot for side=="user" actions); each bot’s handler selects its own DB based on the side it was told and passes that single dict to this dispatcher (apply_initialization_actions — the Python function below, which keeps its plural name for clarity that it dispatches a LIST).

Why bot-side dispatch (and not runner-side like db_state_assertions):

Init actions mutate state. The mutation has to land in the same dict instance the live LLM tools will see and modify during the conversation — that’s the bot’s shared_state, not a snapshot in the runner.
Mutations are imperative; pure-function dispatch on the runner side would require shipping the mutated DB back to the bot, doubling the transport cost and adding a serialization round trip.
Symmetric to upstream: tau2-bench’s Environment.run_env_function_call dispatches against the live self.tools / self.user_tools toolkit instances. Bot-side replay matches that semantics exactly.

Upstream tau2-bench’s initialization functions are methods on toolkit classes (TelecomUserTools.set_user_info, etc.), not module-level functions. The telecom port extracts them as module-level functions taking a plain dict — same approach as the predicate port — so the registry stays language-agnostic and testable.

Module Contents

Functions

Name	Description
`apply_initialization_actions`	Iterate `actions` and dispatch each against `db`.
`list_registered_initialization_functions`	Diagnostic helper. Returns `{domain: [func_names]}`.
`register_initialization_function`	Decorator: register an init function under `(domain, fn.__name__)`.

Data

ALL_INITIALIZATION_FUNCTIONS

InitializationFunction

API

nemo_voice_agent.evaluation.initialization_functions.apply_initialization_actions(
    domain: str,
    actions: typing.List[typing.Dict[str, typing.Any]],
    db: typing.Optional[typing.Dict[str, typing.Any]]
) -> typing.Dict[str, typing.Any]

Iterate actions and dispatch each against db.

Called bot-side by the apply_initialization RTVI handler. All actions in the list apply to the single db dict — the caller (bot handler) has already picked the correct side-specific dict from shared_state based on the side arg sent by the bridge. Each action’s per-record side field is therefore informational only and not consulted here.

Mutates db in place. Returns a result dict with the overall success status and per-action error log (if any) so the bridge can abort the scenario when seeding fails.

Parameters:

domain

str

Scenario domain (e.g., "tau2_telecom"). Determines which registry bucket to look up functions in. Sent by the bridge in the action payload.

actions

List[Dict[str, Any]]

List of {func_name, arguments, side?} records to replay. side is permitted (upstream JSON carries it for traceability) but ignored by this dispatcher.

Optional[Dict[str, Any]]

The DB dict to mutate. The bot handler picks this from shared_state["db"] (agent side) or ["user_db"] (user side) based on the side arg the bridge sent.

Returns: Dict[str, Any]

{"success": bool, "errors": [str, ...]}. success is True

nemo_voice_agent.evaluation.initialization_functions.list_registered_initialization_functions(
    domain: typing.Optional[str] = None
) -> typing.Dict[str, typing.Any]

Diagnostic helper. Returns {domain: [func_names]}.

With domain=None returns the whole registry; with a specific domain returns only that subtree.

nemo_voice_agent.evaluation.initialization_functions.register_initialization_function(
    domain: str
)

Decorator: register an init function under (domain, fn.__name__).

Usage::

@register_initialization_function(domain=“tau2_telecom”) def set_user_info(db: dict, name: str, phone_number: str) -> None: db[“surroundings”][“name”] = name db[“surroundings”][“phone_number”] = phone_number

The function name (fn.__name__) becomes the registry key — matching the upstream func_name field in task["initial_state"]["initialization_actions"]. Renames at the source require updating the upstream task JSON, so don’t.

No side parameter. Function names are unique within a domain; side is purely caller-side metadata (the bridge uses it to route each action to the right bot, the bot handler uses it to pick the right DB out of its shared_state). The function itself is side-agnostic.

Parameters:

domain

str

Registry namespace (matching Scenario.domain, e.g. "tau2_telecom").

Raises:

ValueError: if a function with the same name is already registered in this domain.

nemo_voice_agent.evaluation.initialization_functions.ALL_INITIALIZATION_FUNCTIONS: Dict[str, Dict[str, InitializationFunction]] = {}

nemo_voice_agent.evaluation.initialization_functions.InitializationFunction = Callable[..., None]