nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions

Mutable handle to the per-scenario shared_state dict.

The same dict that’s passed to tool constructors is also published here, so other RTVI action handlers (specifically get_scenario_summary) can read shared_state["actions"] and shared_state["db"] without needing tool references. state is reset (re-pointed at a new dict) every time update_system_prompt runs.

Mutable handle to a PipelineWorker and its running flag.

Construct early, hand to RTVI action factories, then populate once the task exists. running is flipped by the bot runner during shutdown so handlers can avoid queueing frames onto a dead task.

True if key names an audio/image/file container (e.g. input_audio, image_url, file, assistant audio).

Pick a human-readable placeholder tag for a stripped media/binary blob.

Keyed by the nearest media-container field name so audio/image/file payloads read naturally in the saved context; anything else → generic <binary>.

Build the context.apply_initialization action.

Bot-side scenario-state initializer. The bridge calls this once per scenario, immediately after update_system_prompt, to populate the empty shared_state dict with everything tools need at call time. Does three things in order:

1. Parse shared_state_init JSON into shared_state_ref.state (merge, not overwrite — preserves bot-side sentinels stashed by update_system_prompt like __rtvi__ and __tool_domain__). Custom keys from Scenario.setup_shared_state flow through here.
2. Load DB: if the merged state has a db_path key, resolve it against EVAL_DATA_ROOT and replace with the loaded db dict. Idempotent — skipped when db is already present (e.g., when the bridge sent an inline DB instead of a path).
3. Apply init functions: dispatch each {func_name, arguments} record against the now-loaded db. Side-agnostic — the bridge pre-filters by side before sending.

Defensive create-if-missing: if shared_state_ref.state is None (e.g., in tests that skip update_system_prompt), the handler creates an empty dict and proceeds. In production the bridge always calls update_system_prompt first, which creates + clears the dict and stashes runtime sentinels.

Payload shape:

.. code-block:: json

{ “domain”: “tau2_telecom”, “shared_state_init”: ”{“db_path”: “tau2_telecom/db.json”}”, “actions”: [ {“side”: “user”, “func_name”: “set_user_info”, “arguments”: {“name”: “John Smith”, “phone_number”: “555-…”}}, {“side”: “agent”, “func_name”: “enable_roaming”, “arguments”: {“customer_id”: “C1001”, “line_id”: “L1002”}} ] }

The bridge sends this action per-bot. shared_state_init carries per-side scenario fixture data (path or inline DB, optional initial actions list, etc.); actions carries the per-side filtered subset of upstream init mutations. Both are independently optional — a scenario with no init functions and an inline DB still benefits from a single apply_initialization call that does only the DB-load step.

Returns {"success": bool, "errors": list[str]}. success is True only when DB load and every action dispatch completed cleanly. The bridge treats any success=False as a framework-class failure and aborts the scenario without scoring it — partial seeding produces noise, not signal.

Build the context.apply_sync_delta action.

Bot-side endpoint of the cross-side state-propagation pipeline. The bridge calls this after an action-applied event from the other bot — it runs scenario.sync_state(agent_db, user_db) on its in-process shadow DBs, then pushes any non-empty per-side delta to the corresponding bot via this action.

Payload shape:

.. code-block:: json

{ “domain”: “tau2_telecom”, “delta”: { “surroundings.payment_request”: {“bill_id”: “B1002”, …}, “surroundings.line_active”: true } }

The bot’s handler dispatches via nemo_voice_agent.evaluation.sync_appliers.apply_sync_delta — looks up the domain-specific applier (falls back to the generic dotted-path setter) and mutates shared_state["db"] in place.

For single-side domains (eva / airline / retail) this action would never be called by the bridge — only scenarios overriding Scenario.sync_state trigger the pipeline. But the action is registered domain-agnostically so the same bot binary can run any scenario.

Returns {"success": bool, "errors": list[str]}. success is True when the delta applied cleanly. Errors are informational — the bridge logs them and continues (a malformed delta should not crash the bot or stall the conversation).

Build the context.get_context_history action.

Returns the assistant aggregator’s full message list, stringified to match the shape evaluation clients expect.

Race-safety against in-flight function calls. When the bridge fetches the context immediately after receiving the <exit> server message (the SendExitMessageTool / EndConversationTool flow), the agent’s final tool_call may still be in-flight in the aggregator’s pipeline — the exit message arrives via the bot_server_message channel while the tool_call frame is still being committed via the FunctionCallInProgressFrame → FunctionCallResultFrame cycle. If we read the context before those frames commit, the captured message list is missing the final assistant turn. The judge then sees a stale context that doesn’t contain the EndConversationTool tool_call and (incorrectly) deducts points for “didn’t call EndConversationTool”.

Fix: poll the aggregator until its in-progress map drains, with a hard deadline so a stuck tool can never deadlock scenario cleanup. The common case (no pending calls) returns immediately — zero added latency.

Build the context.get_scenario_summary action.

Returns {"actions": [...], "db_hash": "<sha>"} from the per-scenario shared state by default. With include_db=true in the request payload, also returns the inline db dict alongside the hash — used by the runner’s db_state_assertions aggregation (predicates need the actual DB values, not just the hash). Auto-aggregating tools (e.g. WriteScenarioTool subclasses) populate shared_state["actions"] on each successful mutation; the inbound fixture-loading flow populates shared_state["db"]. The bridge calls this action after <exit> (or scenario timeout) to retrieve the final artifacts without depending on any LLM-callable summary tool.

One DB per bot. Each bot’s shared_state["db"] IS this bot’s DB — the agent bot’s db is the agent-facing DB; the user bot’s db is the user-facing DB. The naming distinction (db vs user_db) lives at the bridge/runner boundary: the bridge calls get_scenario_summary once per bot and labels the responses by which WS it pulled from (the user-side pull is added when the first telecom scenario is ported). The bot itself doesn’t know its own side and doesn’t need to.

Hash-only outbound by default (not inline DB). The DB itself stays on the bot server; only the SHA-256 of the canonicalized DB travels through the WebSocket. This keeps the response payload under a few KB regardless of DB size (tau2’s airline DB is 7 MB inline; serialized via the previous inline-DB scheme it exceeded pipecat’s 1 MB WebSocket frame limit and closed the connection with code 1009). Both the bot and the runner import the same get_dict_hash from nemo_voice_agent.evaluation.db_hash so the canonical hashing rule (float normalization, order-independent list fields, excluded keys) is identical on both sides.

Inline DB opt-in (include_db=true). Telecom’s db_state_assertions surface needs the runner to invoke predicate functions on the actual DB state, not just compare hashes. The bridge sets include_db=True when scenario.db_state_assertions is truthy. Telecom’s per-bot DBs are small (~5 KB MockPhone state on the user side, modest customer/line data on the agent side) so the WS frame limit is not a concern; for retail (7 MB DB) the bridge leaves the flag at the default false and the existing hash-out behavior is preserved.

Trade-off (when include_db=false): the runner can no longer compute a per-field compute_db_diff on mismatch since it never sees the actual DB. For debugging hash mismatches in non-telecom domains, set include_db=true temporarily on the bridge call.

Mirrors how get_context_history is consumed by the bridge.

Build the context.reset action.

original_messages is captured by reference so the action always resets to whatever update_system_prompt last wrote.

Build the context.update_system_prompt action.

Tool registration is optional. When enable_tool_calling is True and a tools JSON string is supplied by the caller, tool_factory is invoked per tool to produce schema tools, then register_schema_tools swaps them onto llm / context. This keeps the factory decoupled from evaluation-specific tool registries.

Scenario-start lifecycle gate. This handler is the bot-side signal that a new scenario is starting. In addition to swapping the prompt + tool surface, it RESETS shared_state_ref.state so any prior scenario’s data (db, actions log, etc.) doesn’t bleed into the new one. The reset is done via dict.clear() rather than reassignment so the dict identity is preserved — any tool that already holds a reference to shared_state continues to see the same object after the clear, and subsequent mutations by apply_initialization propagate correctly.

Scenario fixture data (db, db_path, actions list, custom keys from Scenario.setup_shared_state) is NOT loaded here — that moves to create_apply_initialization_action which the bridge calls immediately after this one. This handler only stashes bot-side runtime sentinels (__rtvi__, __tool_domain__) into the freshly-cleared dict so write tools have them available at call time.

Install one on_client_message dispatcher for handlers.

This is the pipecat 1.x replacement for calling rtvi.register_action(...) once per action. Handler return values are sent back as the d payload of a server-response keyed to the request’s message id; an unknown type or a raised exception produces an error-response instead, so callers fail loudly rather than blocking until their read timeout expires.

Return a deep copy of an LLM-context structure with raw media blobs dropped.

Omni models keep audio (and, in future, image/file) payloads inline in their message list. Serializing those verbatim produces a multi-MB payload that overflows pipecat’s WebSocket frame cap, so get_context_history silently fails and the saved agent context ends up empty.

Only the raw encoded data is dropped (replaced with an <audio> / <image> / <file> / <binary> tag) — small metadata is kept so the snapshot stays useful for debugging:

• dropped: bytes / bytearray values; data: base64 URI strings (e.g. image_url.url); file_data / b64_json leaves; and the data leaf when nested inside a media container (input_audio.data, assistant audio.data).
• kept: text / refusal strings, format, filename, file_id, detail, audio transcript, id, and plain url / file-path strings (only base64 data: URIs are stripped, not real URLs/paths).

_media_key carries the nearest enclosing media-container key name down the recursion so a stripped leaf gets the right tag. Pure and non-mutating — the caller’s live context is never altered; a fresh structure is returned.