ReferencePython SDK ReferenceNemoguardrailsColangV1 0Runtime

nemoguardrails.colang.v1_0.runtime.runtime

View as Markdown

Module Contents

Classes

NameDescription
RuntimeV1_0Runtime for executing the guardrails.

Data

log

API

class nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0()

Bases: Runtime

Runtime for executing the guardrails.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._compute_next_steps(
    events: typing.List[dict],
    processing_log: typing.List[dict]
) -> typing.List[dict]
async

Compute the next steps based on the current flow.

Parameters:

events
List[dict]

The list of events.

processing_log
List[dict]

The processing log so far. This will be mutated.

Returns: List[dict]

List[dict]: The list of computed next steps.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._get_action_resp(
    action_meta: typing.Dict[str, typing.Any],
    action_name: str,
    kwargs: typing.Dict[str, typing.Any]
) -> typing.Tuple[typing.Dict[str, typing.Any], str]
async

Interact with actions and get response from the action-server and system actions.

Parameters:

action_meta
Dict[str, Any]

Metadata for the action.

action_name
str

The name of the action.

kwargs
Dict[str, Any]

The action parameters.

Returns: Tuple[Dict[str, Any], str]

Tuple[Dict[str, Any], str]: The response and status.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._init_flow_configs()

Initialize the flow configurations.

Returns:

None

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._internal_error_action_result(
    message: str
)
staticmethod

Helper to construct an action result for an internal error.

Parameters:

message
str

The error message.

Returns:

The action result.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._load_flow_config(
    flow: dict
)

Load a flow configuration.

Parameters:

flow
dict

The flow data.

Returns:

None

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._process_start_action(
    events: typing.List[dict]
) -> typing.List[dict]
async

Start the specified action, wait for it to finish, and post back the result.

Parameters:

events
List[dict]

The list of events.

Returns: List[dict]

List[dict]: The list of next steps.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._process_start_flow(
    events: typing.List[dict],
    processing_log: typing.List[dict]
) -> typing.List[dict]
async

Start a flow.

Parameters:

events
List[dict]

The list of events.

processing_log
List[dict]

The processing log so far. This will be mutated.

Returns: List[dict]

List[dict]: The list of next steps.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._run_flows_in_parallel(
    flows: typing.List[str],
    events: typing.List[dict],
    pre_events: typing.Optional[typing.List[dict]] = None,
    post_events: typing.Optional[typing.List[dict]] = None
) -> nemoguardrails.actions.actions.ActionResult
async

Run flows in parallel.

Running flows in parallel is done by triggering a separate event loop with a start_flow event for each flow, in the context of the current event loop.

Parameters:

flows
List[str]

The list of flow names to run in parallel.

events
List[dict]

The current events.

pre_events
List[dict]Defaults to None

Events to be added before starting each flow.

post_events
List[dict]Defaults to None

Events to be added after finishing each flow.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._run_input_rails_in_parallel(
    flows: typing.List[str],
    events: typing.List[dict]
) -> nemoguardrails.actions.actions.ActionResult
async

Run the input rails in parallel.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._run_output_rails_in_parallel(
    flows: typing.List[str],
    events: typing.List[dict]
) -> nemoguardrails.actions.actions.ActionResult
async

Run the output rails in parallel.

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0._run_output_rails_in_parallel_streaming(
    flows_with_params: typing.Dict[str, dict],
    events: typing.List[dict]
) -> nemoguardrails.actions.actions.ActionResult
async

Run the output rails in parallel for streaming chunks.

This is a streamlined version that avoids the full flow state management which can cause issues with hide_prev_turn logic during streaming.

Parameters:

flows_with_params
Dict[str, dict]

Dictionary mapping flow_id to {“action_name”: str, “params”: dict}

events
List[dict]

The events list for context

nemoguardrails.colang.v1_0.runtime.runtime.RuntimeV1_0.generate_events(
    events: typing.List[dict],
    processing_log: typing.Optional[typing.List[dict]] = None
) -> typing.List[dict]
async

Generates the next events based on the provided history.

This is a wrapper around the process_events method, that will keep processing the events until the listen event is produced.

Parameters:

events
List[dict]

The list of events.

processing_log
Optional[List[dict]]Defaults to None

The processing log so far. This will be mutated.

Returns: List[dict]

List[dict]: The list of generated events.

nemoguardrails.colang.v1_0.runtime.runtime.log = logging.getLogger(__name__)