revive.computation package

Submodules

revive.computation.dists module

revive.computation.dists.all_equal(iterable)

Define a function to check if all elements in an iterable are equal

revive.computation.dists.exportable_broadcast(tensor1: Tensor, tensor2: Tensor) → Tuple[Tensor, Tensor]

Broadcast tensors to the same shape using onnx exportable operators

class revive.computation.dists.ReviveDistributionMixin

Bases: object

Define revive distribution API

property mode

return the most likely sample of the distributions

property std

return the standard deviation of the distributions

sample_with_logprob(sample_shape=())

class revive.computation.dists.ReviveDistribution(batch_shape: Size = (), event_shape: Size = (), validate_args: bool | None = None)

Bases: TorchDistribution, ReviveDistributionMixin

class revive.computation.dists.ExportableNormal(loc, scale, validate_args)

Bases: Normal

class revive.computation.dists.ExportableCategorical(probs=None, logits=None, validate_args=None)

Bases: Categorical

log_prob(value)

Use exportable_broadcast() aligns the shape of the tensor

class revive.computation.dists.DiagnalNormal(loc, scale, validate_args=False)

Bases: ReviveDistribution

sample(sample_shape=())

Generates a sample_shape shaped sample or sample_shape shaped batch of samples if the distribution parameters are batched.

rsample(sample_shape=())

Generates a sample_shape shaped reparameterized sample or sample_shape shaped batch of reparameterized samples if the distribution parameters are batched.

log_prob(sample)

Returns the log of the probability density/mass function evaluated at value.

Args:

value (Tensor):

entropy()

Returns entropy of distribution, batched over batch_shape.

Returns:

Tensor of shape batch_shape.

shift(mu_shift)

shift the distribution, useful in local mode transition

property mode

Returns the mode of the distribution.

property std

return the standard deviation of the distributions

class revive.computation.dists.TransformedDistribution(base_distribution, transforms, validate_args=None)

Bases: TransformedDistribution

property mode

Returns the mode of the distribution.

property std

entropy(num=(100,))

Returns entropy of distribution, batched over batch_shape.

Returns:

Tensor of shape batch_shape.

class revive.computation.dists.DiscreteLogistic(loc, scale, num, *, validate_args=False)

Bases: ReviveDistribution

Model discrete variable with Logistic distribution, inspired from: https://github.com/openai/vdvae/blob/main/vae_helpers.py

As far as I know, the trick was proposed in: Salimans, Tim, Andrej Karpathy, Xi Chen, and Diederik P. Kingma “Pixelcnn++: Improving the pixelcnn with discretized logistic mixture likelihood and other modifications.” arXiv preprint arXiv:1701.05517 (2017).

Parameters:

• loc – Location parameter, assert it have been normalized to [-1, 1]

• scale – Scale parameter.

• num – Number of possible value for each dimension.

arg_constraints = {'loc': Real(), 'scale': GreaterThan(lower_bound=0.0)}

support = Real()

has_rsample = True

log_prob(value)

Returns the log of the probability density/mass function evaluated at value.

Args:

value (Tensor):

sample(sample_shape=())

Generates a sample_shape shaped sample or sample_shape shaped batch of samples if the distribution parameters are batched.

rsample(sample_shape=())

Generates a sample_shape shaped reparameterized sample or sample_shape shaped batch of reparameterized samples if the distribution parameters are batched.

cdf(value)

Returns the cumulative density/mass function evaluated at value.

Args:

value (Tensor):

icdf(value)

Returns the inverse cumulative density/mass function evaluated at value.

Args:

value (Tensor):

round(value)

property mode

Returns the mode of the distribution.

property std

return the standard deviation of the distributions

entropy()

Returns entropy of distribution, batched over batch_shape.

Returns:

Tensor of shape batch_shape.

class revive.computation.dists.Onehot(logits=None, validate_args=False)

Bases: OneHotCategorical, TorchDistributionMixin, ReviveDistributionMixin

Differentiable Onehot Distribution

has_rsample = True

rsample(sample_shape=())

Generates a sample_shape shaped reparameterized sample or sample_shape shaped batch of reparameterized samples if the distribution parameters are batched.

property mode

Returns the mode of the distribution.

property std

return the standard deviation of the distributions

class revive.computation.dists.GaussianMixture(locs, coord_scale, component_logits)

Bases: MixtureOfDiagNormals, ReviveDistributionMixin

property mode

Returns the mode of the distribution.

property std

return the standard deviation of the distributions

shift(mu_shift)

shift the distribution, useful in local mode transition

entropy()

Returns entropy of distribution, batched over batch_shape.

Returns:

Tensor of shape batch_shape.

class revive.computation.dists.MixDistribution(dists)

Bases: ReviveDistribution

Collection of multiple distributions

arg_constraints = {}

sample(num=())

Generates a sample_shape shaped sample or sample_shape shaped batch of samples if the distribution parameters are batched.

rsample(num=())

Generates a sample_shape shaped reparameterized sample or sample_shape shaped batch of reparameterized samples if the distribution parameters are batched.

entropy()

Returns entropy of distribution, batched over batch_shape.

Returns:

Tensor of shape batch_shape.

log_prob(x)

Returns the log of the probability density/mass function evaluated at value.

Args:

value (Tensor):

property mode

Returns the mode of the distribution.

property std

return the standard deviation of the distributions

property dists_type

shift(mu_shift)

shift the distribution, useful in local mode transition

revive.computation.funs_parser module

revive.computation.funs_parser.get_nodes(yaml_file_path)

revive.computation.funs_parser.matching_bracket(string, idx, brackets=[])

revive.computation.funs_parser.strip(s)

revive.computation.funs_parser.find_all(s, sub_s)

revive.computation.funs_parser.find_node(line)

revive.computation.funs_parser.find_column(line)

revive.computation.funs_parser.find_operator(line)

revive.computation.funs_parser.convert_operator(oral_operator_name)

revive.computation.funs_parser.convert_column(oral_column, flag)

revive.computation.funs_parser.convert_node(bracket_start_index)

revive.computation.funs_parser.convert_operators(line)

Convert operator

Example: add -> opt.add

revive.computation.funs_parser.convert_columns(line)

Convert column name to column index

Example: “obs_1” -> 1

revive.computation.funs_parser.convert_nodes(line)

Convert node to tensor

Example: obs[1,2] -> obs[…,[1,2]]

revive.computation.funs_parser.convert_line(line)

revive.computation.funs_parser.checkt_convert(line)

revive.computation.funs_parser.convert_fn_def(origin_code: list)

find the intent of original code

revive.computation.funs_parser.get_fn_list(origin_code_list)

revive.computation.funs_parser.parser(input_file: str, output_file: str, yaml_file: str)

revive.computation.graph module

class revive.computation.graph.DesicionNode(name: str, input_names: List[str], input_descriptions: List[Dict[str, Dict[str, Any]]])

Bases: object

An abstract node for making decisions

node_type: str = None

__call__(data: Dict[str, Tensor], *args, **kwargs) → Tensor | ReviveDistribution

Run a forward computation of this node. NOTE: The input data was transferred by self.processor. You can use self.processor.deprocess_torch(data) to get the original data.

register_processor(processor: DataProcessor)

register the global data processor to this node

remove_processor()

remove the registered data processor

get_inputs(data: Dict[str, Tensor]) → OrderedDict

get only input variables

before_compute_node()

after_compute_node()

to(device: str) → DesicionNode

change the device of this node

requires_grad_(mode: bool = False) → DesicionNode

change the requirement of gradient for this node

train() → DesicionNode

set the state of this node to training

eval() → DesicionNode

set the state of this node to evaluation

reset() → None

reset the state of this node, useful when node is an RNN

export2onnx(onnx_file: str, verbose: bool = True)

export the node to onnx file, with input from original data space

class revive.computation.graph.NetworkDecisionNode(name: str, input_names: List[str], input_descriptions: List[Dict[str, Dict[str, Any]]])

Bases: DesicionNode

node_type: str = 'network'

set_network(network: Module)

set the network from a different source

get_network() → Module

return all the network in this node

initialize_network(input_dim: int | dict, output_dim: int, hidden_features: int, hidden_layers: int, backbone_type: str, dist_config: list, is_transition: bool = False, hidden_activation: str = 'leakyrelu', norm: str | None = None, transition_mode: str | None = None, obs_dim: int | None = None, input_dim_dict: dict = {}, *args, **kwargs)

initialize the network of this node

call(data: Dict[str, Tensor], *args, **kwargs) → ReviveDistribution

Run a forward computation of this node. NOTE: The input data was transferred by self.processor. You can use self.processor.deprocess_torch(data) to get the original data.

to(device: str) → DesicionNode

change the device of this node

requires_grad_(mode: bool = False) → DesicionNode

change the requirement of gradient for this node

train() → DesicionNode

set the state of this node to training

eval() → DesicionNode

set the state of this node to evaluation

reset() → None

reset the state of this node, useful when node is an RNN

class revive.computation.graph.FunctionDecisionNode(name: str, input_names: List[str], input_descriptions: List[Dict[str, Dict[str, Any]]])

Bases: DesicionNode

node_type: str = 'function'

register_node_function(node_function: Callable[[Dict[str, ndarray]], ndarray] | Callable[[Dict[str, Tensor]], Tensor], node_function_type: str)

remove_node_function()

delta_node_function(data)

call(data: Dict[str, Tensor], *args, **kwargs) → Tensor | ReviveDistribution

NOTE: if there is any provided function defined in numpy, this process cannot maintain gradients

export2onnx(onnx_file: str, verbose: bool = True)

export the node to onnx file, with input from original data space

reset() → None

reset the state of this node, useful when node is an RNN

class revive.computation.graph.DesicionGraph(graph_dict: Dict[str, List[str]], descriptions: Dict[str, List[Dict[str, Dict[str, Any]]]], fit, metric_nodes)

Bases: object

A collection of DecisionNodes

register_node(node_name: str, node_class)

Register a node with given node class

property learnable_node_names: List[str]

A list of names for learnable nodes the graph

register_target_nodes()

del_target_nodes()

use_target_network()

not_use_target_network()

update_target_network(polyak=0.99)

mark_tunable(node_name: str) → None

Mark a leaf variable as tunable

register_processor(processor: DataProcessor)

Register data processor to the graph and nodes

get_node(node_name: str, use_target: bool = False) → DesicionNode

get the node by name

compute_node(node_name: str, inputs: Dict[str, Tensor], use_target: bool = False, *args, **kwargs)

compute the node by name

get_relation_node_names() → List[str]

get all the nodes that related to the learning (network) nodes. NOTE: this is the default list if you have matcher and value functions.

get_node_value_net_node_names() → List[str]

get all the nodes that related to the learning (network) nodes. NOTE: this is the default list if you have matcher and value functions.

summary_nodes() → Dict[str, int]

collect_models() → List[Module]

return all the network that registered in this graph

is_equal_venv(source_graph: DesicionGraph, policy_node) → bool

check if new graph shares the same virtual environments

is_equal_structure(source_graph: DesicionGraph) → bool

check if new graph shares the same structure

copy_graph_node(source_graph: DesicionGraph) → bool

try copy all the node from source graph

get_leaf(graph: Dict[str, List[str]] | None = None) → List[str]

return the leaf of the graph in alphabet order

sort_graph(graph_dict: dict) → OrderedDict

Sort arbitrary computation graph to the topological order

to(device: str) → DesicionGraph

requires_grad_(mode: bool = False) → DesicionGraph

eval() → DesicionGraph

reset() → DesicionGraph

reset graph, useful for stateful graph

__getitem__(name: str) → List[str]

keys() → Iterable[str]

values() → Iterable[List[str]]

items() → Iterable[Dict[str, List[str]]]

__len__() → int

__call__(state: Dict[str, Tensor]) → Dict[str, Tensor]

compute the whole graph, from leaf node to all output nodes

state_transition(state: Dict[str, Tensor], copy: bool = False) → Dict[str, Tensor]

export2onnx(onnx_file: str, verbose: bool = True)

export the graph to onnx file, with input from original data space

revive.computation.inference module

revive.computation.inference.find_key_precedings(graph: DesicionGraph, node_name: str, preceding_type: str = 'network') → List

revive.computation.inference.get_matcher_input(batch_data: Batch, target_node_name: str, candidate_num: int, node_names: List[str], nodes_fit_dict: dict | None = None) → Tensor

return all data of node_names from batch_data Args:

batch_data (Batch): Batch of data node_names (List): list of node names to get data nodes_fit_index (Dict): dict of fixed index for nodel_names

Return:

data to get

class revive.computation.inference.VirtualEnvDev(graph: DesicionGraph, train_algo: str | None = None, info: dict = {}, **kwargs)

Bases: Module

to(device)

Moves and/or casts the parameters and buffers.

This can be called as

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

Its signature is similar to torch.Tensor.to(), but only accepts floating point or complex dtypes. In addition, this method will only cast the floating point or complex parameters and buffers to dtype (if given). The integral parameters and buffers will be moved device, if that is given, but with dtypes unchanged. When non_blocking is set, it tries to convert/move asynchronously with respect to the host if possible, e.g., moving CPU Tensors with pinned memory to CUDA devices.

See below for examples.

Note

This method modifies the module in-place.

Args:

device (torch.device): the desired device of the parameters

and buffers in this module

dtype (torch.dtype): the desired floating point or complex dtype of

the parameters and buffers in this module

tensor (torch.Tensor): Tensor whose dtype and device are the desired

dtype and device for all parameters and buffers in this module

memory_format (torch.memory_format): the desired memory

format for 4D parameters and buffers in this module (keyword only argument)

Returns:

Module: self

Examples:

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

check_version()

reset() → None

set_target_policy_name(target_policy_name: list) → None

load_filter_components(**kwargs)

init_ensemble_matcher(ensemble_size: int = 1, ensemble_choosing_interval: int = 1)

reset_ensemble_matcher(ensemble_size: int, ensemble_choosing_interval: int)

infer_k_steps(states: List[Dict[str, ndarray]], deterministic: bool = True, clip: bool = True) → List[Dict[str, ndarray]]

infer_one_step(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True, **kwargs) → Dict[str, ndarray]

node_infer(node_name: str, state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

node_dist(node_name: str, state: Dict[str, ndarray]) → Dict[str, ndarray]

node_pre_computation(node_name: str, node_data: ndarray)

node_post_computation(node_name: str, node_data: ndarray)

forward(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False) → Dict[str, Tensor]

run the target node

pre_computation(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False, policy_index: int = 0, disturb_node_list: Dict = {}, **kwargs) → Dict[str, Tensor]

run all the node before target node. skip if the node value is already available.

filter_penalty(penalty_type: str, data: Dict[str, Tensor], sample_num: int, clip: bool = False, policy_index: int = 0, **kwargs) → Dict[str, Tensor]

run all the node after target node

post_computation(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False, policy_index: int = 0, disturb_node_list: Dict = {}, **kwargs) → Dict[str, Tensor]

run all the node after target node

export2onnx(onnx_file: str, verbose: bool = True)

class revive.computation.inference.VirtualEnv(env_list: List[VirtualEnvDev])

Bases: object

to(device)

Move model to the device specified by the parameter.

Examples:

>>> venv_model.to("cpu")
>>> venv_model.to("cuda")
>>> venv_model.to("cuda:1")

check_version()

Check if the revive version of the saved model and the current revive version match.

reset() → None

When using RNN for model training, this method needs to be called before model reuse to reset the hidden layer information.

set_env(env_id) → None

property target_policy_name: str

Get the target policy name.

set_target_policy_name(target_policy_name) → None

Set the target policy name.

replace_policy(policy: PolicyModel) → None

Replace the target policy with the given policy.

infer_k_steps(states: Dict[str, ndarray] | List[Dict[str, ndarray]], k: int | None = None, deterministic: bool = True, clip: bool = True) → List[Dict[str, ndarray]]

Generate k steps interactive data.

Args:

states:

a dict of initial input nodes

k:

how many steps to generate

deterministic:

if True, the most likely actions are generated; if False, actions are generated by sample. Default: True

Return:

k steps interactive data dict

Examples:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> ten_step_output = venv_model.infer_k_steps(state, k=10)

infer_one_step(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True, **kwargs) → Dict[str, ndarray]

Generate one step interactive data given action.

Args:

state:

a dict of input nodes

deterministic:

if True, the most likely actions are generated; if False, actions are generated by sample. Default: True

Return:

one step outputs

Examples:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> one_step_output = venv_model.infer_one_step(state)

reset_ensemble_matcher(ensemble_size: int = 10, ensemble_choosing_interval: int = 10)

node_pre_computation(node_name: str, node_data: ndarray)

node_post_computation(node_name: str, node_data: ndarray)

node_infer(node_name: str, state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

Generate one step interactive data given node_name.

Args:

state:

a dict of input nodes

deterministic:

if True, the most likely actions are generated; if False, actions are generated by sample. Default: True

Return:

one step node output

Examples:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> action_output = venv_model.node_infer("action", state)

node_dist(node_name: str, state: Dict[str, ndarray]) → Dict[str, ndarray]

Generate one step interactive dist given node_name.

Args:

state:

a dict of input nodes

deterministic:

if True, the most likely actions are generated; if False, actions are generated by sample. Default: True

Return:

one step node output

Examples:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> action_output = venv_model.node_infer("action", state)

export2onnx(onnx_file: str, verbose: bool = True)

Exporting the model to onnx mode.

Reference: https://pytorch.org/docs/stable/onnx.html

Args:

onnx_file:

the onnx model file save path.

verbose:

if True, prints a description of the model being exported to stdout. In addition, the final ONNX graph will include the field doc_string` from the exported model which mentions the source code locations for model.

class revive.computation.inference.PolicyModelDev(nodes: List[DesicionNode])

Bases: Module

to(device)

Moves and/or casts the parameters and buffers.

This can be called as

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

Its signature is similar to torch.Tensor.to(), but only accepts floating point or complex dtypes. In addition, this method will only cast the floating point or complex parameters and buffers to dtype (if given). The integral parameters and buffers will be moved device, if that is given, but with dtypes unchanged. When non_blocking is set, it tries to convert/move asynchronously with respect to the host if possible, e.g., moving CPU Tensors with pinned memory to CUDA devices.

See below for examples.

Note

This method modifies the module in-place.

Args:

device (torch.device): the desired device of the parameters

and buffers in this module

dtype (torch.dtype): the desired floating point or complex dtype of

the parameters and buffers in this module

tensor (torch.Tensor): Tensor whose dtype and device are the desired

dtype and device for all parameters and buffers in this module

memory_format (torch.memory_format): the desired memory

format for 4D parameters and buffers in this module (keyword only argument)

Returns:

Module: self

Examples:

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

check_version()

reset()

infer(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = False) → ndarray

export2onnx(onnx_file: str, verbose: bool = True)

class revive.computation.inference.PolicyModel(policy_model_dev: PolicyModelDev, post_process: Callable[[Dict[str, ndarray], ndarray], ndarray] | None = None)

Bases: object

to(device: str)

Move model to the device specified by the parameter.

Examples:

>>> policy_model.to("cpu")
>>> policy_model.to("cuda")
>>> policy_model.to("cuda:1")

check_version()

Check if the revive version of the saved model and the current revive version match.

reset()

When using RNN for model training, this method needs to be called before model reuse to reset the hidden layer information.

property target_policy_name: None

Get the target policy name.

infer(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True, additional_info: Dict[str, ndarray] | None = None) → ndarray

Generate action according policy.

Args:

state:

a dict contain ALL the input nodes of the policy node

deterministic:

if True, the most likely actions are generated; if False, actions are generated by sample. Default: True

clip:

if True, The output will be cropped to the range set in the yaml file; if False, actions are generated by sample. Default: True

additional_info:

a dict of additional info for post process

Return:

action

Examples:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> action = policy_model.infer(state)

export2onnx(onnx_file: str, verbose: bool = True)

Exporting the model to onnx mode.

Reference: https://pytorch.org/docs/stable/onnx.html

Args:

onnx_file:

the onnx model file save path.

verbose:

if True, prints a description of the model being exported to stdout. In addition, the final ONNX graph will include the field doc_string` from the exported model which mentions the source code locations for model.

revive.computation.inference_cn module

class revive.computation.inference_cn.VirtualEnvDev(graph: DesicionGraph)

Bases: Module

to(device)

Moves and/or casts the parameters and buffers.

This can be called as

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

Its signature is similar to torch.Tensor.to(), but only accepts floating point or complex dtypes. In addition, this method will only cast the floating point or complex parameters and buffers to dtype (if given). The integral parameters and buffers will be moved device, if that is given, but with dtypes unchanged. When non_blocking is set, it tries to convert/move asynchronously with respect to the host if possible, e.g., moving CPU Tensors with pinned memory to CUDA devices.

See below for examples.

Note

This method modifies the module in-place.

Args:

device (torch.device): the desired device of the parameters

and buffers in this module

dtype (torch.dtype): the desired floating point or complex dtype of

the parameters and buffers in this module

tensor (torch.Tensor): Tensor whose dtype and device are the desired

dtype and device for all parameters and buffers in this module

memory_format (torch.memory_format): the desired memory

format for 4D parameters and buffers in this module (keyword only argument)

Returns:

Module: self

Examples:

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

check_version()

reset() → None

set_target_policy_name(target_policy_name: list) → None

infer_k_steps(states: List[Dict[str, ndarray]], deterministic: bool = True, clip: bool = True) → List[Dict[str, ndarray]]

infer_one_step(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

node_infer(node_name: str, state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

forward(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False) → Dict[str, Tensor]

run the target node

pre_computation(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False, policy_index: int = 0) → Dict[str, Tensor]

run all the node before target node. skip if the node value is already available.

post_computation(data: Dict[str, Tensor], deterministic: bool = True, clip: bool = False, policy_index: int = 0) → Dict[str, Tensor]

run all the node after target node

export2onnx(onnx_file: str, verbose: bool = True)

class revive.computation.inference_cn.VirtualEnv(env_list: List[VirtualEnvDev])

Bases: object

to(device)

切换模型所在设备，可以指定cpu或cuda。

示例:

>>> venv_model.to("cpu")
>>> venv_model.to("cuda")
>>> venv_model.to("cuda:1")

check_version()

检查训练模型使用的REVIVE SDK版本与当前安装的REVIVE SDK版本是否一致。

reset() → None

重置模型隐藏层信息，使用RNN训练的模型需要在每次开始使用时调用该方法。

property target_policy_name: str

获得策略节点的名称。

set_target_policy_name(target_policy_name) → None

设置策略节点的名称。

replace_policy(policy: PolicyModel) → None

使用给定的策略节点模型代替当前的策略节点模型。

infer_one_step(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

生成1步交互数据，1步表示决策流图完整的运行一遍。

参数:

states:

包含初始输入节点数据的字典，初始的节点数据应包括决策流图的所有叶子节点。

deterministic:

如果参数是True, 进行确定性的数据生成; 如果参数是False, 从分布中进行采样生成数据。 默认值: True

返回值:

字典，含有1步交互数据，key是节点名，value节点数据数组。

示例:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> one_step_output = venv_model.infer_one_step(state)

infer_k_steps(states: Dict[str, ndarray] | List[Dict[str, ndarray]], k: int | None = None, deterministic: bool = True, clip: bool = True) → List[Dict[str, ndarray]]

生成k步交互数据，每一步表示决策流图完整的运行一遍。

参数:

states:

包含初始输入节点数据的字典，初始的节点数据应包括决策流图的所有叶子节点。

k:

正整数，如果是1，则返回一步的交互数据；如果是10，则决策流图迭代的运行10次，返回10次的数据。

deterministic:

如果参数是True, 进行确定性的数据生成; 如果参数是False, 从分布中进行采样生成数据。 默认值: True

返回值:

字典，含有k步交互数据，key是节点名，value是含有k步该节点数据的数组。

示例:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> ten_step_output = venv_model.infer_k_steps(state, k=10)

node_infer(node_name: str, state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True) → Dict[str, ndarray]

使用指定节点模型进行推理.

参数:

state:

包含节点所有输入数据的字典。

deterministic:

如果参数是True, 进行确定性的数据生成; 如果参数是False, 从分布中进行采样生成数据。 默认值: True。

返回值:

节点输出。

示例:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> action_output = venv_model.node_infer("action", state)

export2onnx(onnx_file: str, verbose: bool = True)

导出环境模型为onnx格式。

参考: https://pytorch.org/docs/stable/onnx.html

参数:

onnx_file:

存储onnx模型的文件地址。

verbose:

默认为True。 如果为True，打印导出到的模型的描述，最终的ONNX图将包括导出模型中的字段doc_string，其中提到model的源代码位置。.

class revive.computation.inference_cn.PolicyModelDev(nodes: List[DesicionNode])

Bases: Module

to(device)

Moves and/or casts the parameters and buffers.

This can be called as

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

Its signature is similar to torch.Tensor.to(), but only accepts floating point or complex dtypes. In addition, this method will only cast the floating point or complex parameters and buffers to dtype (if given). The integral parameters and buffers will be moved device, if that is given, but with dtypes unchanged. When non_blocking is set, it tries to convert/move asynchronously with respect to the host if possible, e.g., moving CPU Tensors with pinned memory to CUDA devices.

See below for examples.

Note

This method modifies the module in-place.

Args:

device (torch.device): the desired device of the parameters

and buffers in this module

dtype (torch.dtype): the desired floating point or complex dtype of

the parameters and buffers in this module

tensor (torch.Tensor): Tensor whose dtype and device are the desired

dtype and device for all parameters and buffers in this module

memory_format (torch.memory_format): the desired memory

format for 4D parameters and buffers in this module (keyword only argument)

Returns:

Module: self

Examples:

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

check_version()

reset()

infer(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = False) → ndarray

export2onnx(onnx_file: str, verbose: bool = True)

class revive.computation.inference_cn.PolicyModel(policy_model_dev: PolicyModelDev, post_process: Callable[[Dict[str, ndarray], ndarray], ndarray] | None = None)

Bases: object

to(device: str)

切换模型所在设备，可以指定cpu或cuda。

示例:

>>> policy_model.to("cpu")
>>> policy_model.to("cuda")
>>> policy_model.to("cuda:1")

check_version()

检查训练模型使用的REVIVE SDK版本与当前安装的REVIVE SDK版本是否一致。

reset()

重置模型隐藏层信息，使用RNN训练的模型需要在每次开始使用时调用该方法。

property target_policy_name: None

获得策略节点的名称。

infer(state: Dict[str, ndarray], deterministic: bool = True, clip: bool = True, additional_info: Dict[str, ndarray] | None = None) → ndarray

使用策略模型进行推理，输出动作.

参数:

state:

包含策略节点所有输入数据的字典

deterministic:

如果参数是True, 进行确定性的数据生成; 如果参数是False, 从分布中进行采样生成数据。 默认值: True。

clip:

如果为True，输出的动作数值将会被裁剪到YAML文件中配置的范围； 如果为False，不对输出的动作数值进行裁剪，输出值可能存在越界的情况。 默认值：True

additional_info:

默认为None即可。

返回值:

动作。

示例:

>>> state = {"obs": obs_array, "static_obs": static_obs_array}
>>> action = policy_model.infer(state)

export2onnx(onnx_file: str, verbose: bool = True)

导出策略模型为onnx格式。

参考: https://pytorch.org/docs/stable/onnx.html

参数:

onnx_file:

存储onnx模型的文件地址。

verbose:

默认为True。 如果为True，打印导出到的模型的描述，最终的ONNX图将包括导出模型中的字段doc_string，其中提到model的源代码位置。.

revive.computation.modules module

revive.computation.modules.reglu(x)

revive.computation.modules.geglu(x)

class revive.computation.modules.Swish(*args, **kwargs)

Bases: Module

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.Value_Net(input_dim_dict, value_hidden_features, value_hidden_layers, value_normalization, value_activation, *args, **kwargs)

Bases: Module

Initializes a vectorized linear layer instance. Args:

in_features (int): The number of input features. out_features (int): The number of output features. ensemble_size (int): The number of ensembles to use.

forward(state: Tensor | Dict[str, Tensor]) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.ConvBlock(in_features: int, out_features: int, conv_time_step: int, output_activation: str = 'identity')

Bases: Module

Initializes a vectorized linear layer instance. Args:

in_features (int): The number of input features. out_features (int): The number of output features. ensemble_size (int): The number of ensembles to use.

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.VectorizedLinear(in_features: int, out_features: int, ensemble_size: int)

Bases: Module

Initializes a vectorized linear layer instance. Args:

in_features (int): The number of input features. out_features (int): The number of output features. ensemble_size (int): The number of ensembles to use.

reset_parameters()

forward(x: Tensor) → Tensor

input: [ensemble_size, batch_size, input_size] weight: [ensemble_size, input_size, out_size] out: [ensemble_size, batch_size, out_size]

class revive.computation.modules.EnsembleLinear(input_dim: int, output_dim: int, num_ensemble: int, weight_decay: float = 0.0)

Bases: Module

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

load_save() → None

update_save(indexes: List[int]) → None

get_decay_loss() → Tensor

to(device)

Moves and/or casts the parameters and buffers.

This can be called as

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

Its signature is similar to torch.Tensor.to(), but only accepts floating point or complex dtypes. In addition, this method will only cast the floating point or complex parameters and buffers to dtype (if given). The integral parameters and buffers will be moved device, if that is given, but with dtypes unchanged. When non_blocking is set, it tries to convert/move asynchronously with respect to the host if possible, e.g., moving CPU Tensors with pinned memory to CUDA devices.

See below for examples.

Note

This method modifies the module in-place.

Args:

device (torch.device): the desired device of the parameters

and buffers in this module

dtype (torch.dtype): the desired floating point or complex dtype of

the parameters and buffers in this module

tensor (torch.Tensor): Tensor whose dtype and device are the desired

dtype and device for all parameters and buffers in this module

memory_format (torch.memory_format): the desired memory

format for 4D parameters and buffers in this module (keyword only argument)

Returns:

Module: self

Examples:

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

class revive.computation.modules.MLP(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str | None = None, hidden_activation: str = 'leakyrelu', output_activation: str = 'identity')

Bases: Module

Multi-layer Perceptron

Args:

in_features : int, features numbers of the input

out_features : int, features numbers of the output

hidden_features : int, features numbers of the hidden layers

hidden_layers : int, numbers of the hidden layers

norm : str, normalization method between hidden layers, default : None

hidden_activation : str, activation function used in hidden layers, default : ‘leakyrelu’

output_activation : str, activation function used in output layer, default : ‘identity’

forward(x: Tensor) → Tensor

forward method of MLP only assume the last dim of x matches in_features

class revive.computation.modules.VectorizedMLP(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, norm: str | None = None, hidden_activation: str = 'leakyrelu', output_activation: str = 'identity')

Bases: Module

Vectorized MLP

Args:

in_features : int, features numbers of the input

out_features : int, features numbers of the output

hidden_features : int, features numbers of the hidden layers

hidden_layers : int, numbers of the hidden layers

norm : str, normalization method between hidden layers, default : None

hidden_activation : str, activation function used in hidden layers, default : ‘leakyrelu’

output_activation : str, activation function used in output layer, default : ‘identity’

forward(x: Tensor) → Tensor

forward method of MLP only assume the last dim of x matches in_features

class revive.computation.modules.ResBlock(input_feature: int, output_feature: int, norm: str = 'ln')

Bases: Module

Initializes a residual block instance. Args:

input_feature (int): The number of input features to the block. output_feature (int): The number of output features from the block. norm (str, optional): The type of normalization to apply to the block. Default is ‘ln’ for layer normalization.

forward(x: Tensor) → Tensor

x should be a 2D Tensor due to batchnorm

class revive.computation.modules.VectorizedResBlock(in_features: int, out_features: int, ensemble_size: int, norm: str = 'ln')

Bases: Module

Initializes a vectorized linear layer instance. Args:

in_features (int): The number of input features. out_features (int): The number of output features. ensemble_size (int): The number of ensembles to use.

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.ResNet(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str = 'bn', output_activation: str = 'identity')

Bases: Module

Initializes a residual neural network (ResNet) instance.

Args:

in_features (int): The number of input features to the ResNet. out_features (int): The number of output features from the ResNet. hidden_features (int): The number of hidden features in each residual block. hidden_layers (int): The number of residual blocks in the ResNet. norm (str, optional): The type of normalization to apply to the ResNet. Default is ‘bn’ for batch normalization. output_activation (str, optional): The type of activation function to apply to the output of the ResNet. Default is ‘identity’.

forward(x: Tensor) → Tensor

NOTE: reshape is needed since resblock only support 2D Tensor

class revive.computation.modules.VectorizedResNet(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, norm: str = 'bn', output_activation: str = 'identity')

Bases: Module

Initializes a residual neural network (ResNet) instance.

Args:

in_features (int): The number of input features to the ResNet. out_features (int): The number of output features from the ResNet. hidden_features (int): The number of hidden features in each residual block. hidden_layers (int): The number of residual blocks in the ResNet. norm (str, optional): The type of normalization to apply to the ResNet. Default is ‘bn’ for batch normalization. output_activation (str, optional): The type of activation function to apply to the output of the ResNet. Default is ‘identity’.

forward(x: Tensor) → Tensor

NOTE: reshape is needed since resblock only support 2D Tensor

class revive.computation.modules.TAPE(d_model, max_len=200, scale_factor=1.0)

Bases: Module

forward(x)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TAttention(d_model, nhead, dropout)

Bases: Module

forward(x)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TsTransformer(in_features, out_features, hidden_layers, hidden_features, dropout=0.0)

Bases: Module

forward(x)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.Transformer1D(in_features: int, out_features: int, transformer_features: int = 16, transformer_heads: int = 8, transformer_layers: int = 4)

Bases: Module

This is an experimental backbone.

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.Tokenizer(d_numerical: int, categories: List[int] | None, d_token: int, bias: bool)

Bases: Module

category_offsets: Tensor | None

property n_tokens: int

forward(x_num: Tensor, x_cat: Tensor | None)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.MultiheadAttention(d: int, n_heads: int, dropout: float, initialization: str)

Bases: Module

forward(x_q, x_kv, key_compression: Linear | None, value_compression: Linear | None)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.FT_Transformer(in_features: int, out_features: int, hidden_features: int = 256, hidden_layers: int = 5, categories: List[int] | None = None, token_bias: bool = True, n_heads: int = 8, d_ffn_factor: float = 1.33, attention_dropout: float = 0.0, ffn_dropout: float = 0.0, residual_dropout: float = 0.0, activation: str = 'reglu', prenormalization: bool = True, initialization: str = 'kaiming', kv_compression: float | None = None, kv_compression_sharing: str | None = None)

Bases: Module

References: - https://github.com/Yura52/tabular-dl-revisiting-models/blob/main/bin/ft_transformer.py

forward(x_num, x_cat=None)

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.DistributionWrapper(distribution_type: str = 'normal', **params)

Bases: Module

wrap output of Module to distribution

BASE_TYPES = ['normal', 'gmm', 'onehot', 'discrete_logistic']

SUPPORTED_TYPES = ['normal', 'gmm', 'onehot', 'discrete_logistic', 'mix']

forward(x: Tensor, adapt_std: Tensor | None = None, payload: Tensor | None = None, **kwargs) → ReviveDistribution

Warp the given tensor to distribution

:param adapt_std : it will overwrite the std part of the distribution (optional) :param payload : payload will be applied to the output distribution after built (optional)

extra_repr() → str

Set the extra representation of the module

To print customized extra information, you should re-implement this method in your own modules. Both single-line and multi-line strings are acceptable.

class revive.computation.modules.FeedForwardPolicy(in_features: int | dict, out_features: int, hidden_features: int, hidden_layers: int, dist_config: list, norm: str | None = None, hidden_activation: str = 'leakyrelu', backbone_type: str | str_ = 'mlp', use_multihead: bool = False, **kwargs)

Bases: Module

Policy for using mlp, resnet and transformer backbone.

Args:

in_features : The number of input features, or a dictionary describing the input distribution out_features : The number of output features hidden_features : The number of hidden features in each layer hidden_layers : The number of hidden layers dist_config : A list of configurations for the distributions to use in the model norm : The type of normalization to apply to the input features hidden_activation : The activation function to use in the hidden layers backbone_type : The type of backbone to use in the model use_multihead : Whether to use a multihead model use_feature_embed : Whether to use feature embedding

forward(state: Tensor | Dict[str, Tensor], adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

reset()

get_action(state: Tensor | Dict[str, Tensor], deterministic: bool = True)

class revive.computation.modules.RecurrentPolicy(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, backbone_type: str | str_ = 'gru', **kwargs)

Bases: Module

Initializes a recurrent policy network instance.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features in the RNN. hidden_layers (int): The number of layers in the RNN. dist_config (list): The configuration for the distributions used in the model. backbone_type (Union[str, np.str_], optional): The type of RNN to use (‘gru’ or ‘lstm’). Default is ‘gru’.

reset()

forward(x: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TsRecurrentPolicy(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, backbone_type: str | str_ = 'gru', ts_input_names: dict = {}, bidirectional: bool = True, dropout: float = 0, **kwargs)

Bases: Module

input_is_dict = True

Initializes a recurrent policy network instance for ts_node.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features in the RNN. hidden_layers (int): The number of layers in the RNN. dist_config (list): The configuration for the distributions used in the model. backbone_type (Union[str, np.str_], optional): The type of RNN to use (‘gru’ or ‘lstm’). Default is ‘gru’.

reset()

forward(x: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TsTransformerPolicy(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, ts_input_names: dict = {}, dropout: float = 0, **kwargs)

Bases: Module

input_is_dict = True

Initializes a recurrent policy network instance for ts_node.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features in the RNN. hidden_layers (int): The number of layers in the RNN. dist_config (list): The configuration for the distributions used in the model. backbone_type (Union[str, np.str_], optional): The type of RNN to use (‘gru’ or ‘lstm’). Default is ‘gru’.

reset()

forward(x: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.RecurrentRESPolicy(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, dist_config: list, backbone_type: str | str_ = 'gru', rnn_hidden_features: int = 64, window_size: int = 0, **kwargs)

Bases: Module

reset()

preprocess_window(a_embed)

postprocess_window(rnn_output, hidden_cat, cell_cat)

rnn_forward(a_embed, joint_train: bool)

forward(x: Tensor, adapt_std: Tensor | None = None, field: str = 'mail', **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.ContextualPolicy(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, dist_config: list, backbone_type: str | str_ = 'contextual_gru', **kwargs)

Bases: Module

reset()

forward(x: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.EnsembleFeedForwardPolicy(in_features: int | dict, out_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, dist_config: list, norm: str | None = None, hidden_activation: str = 'leakyrelu', backbone_type: str | str_ = 'EnsembleRES', use_multihead: bool = False, **kwargs)

Bases: Module

Policy for using mlp, resnet and transformer backbone.

Args:

in_features : The number of input features, or a dictionary describing the input distribution out_features : The number of output features hidden_features : The number of hidden features in each layer hidden_layers : The number of hidden layers dist_config : A list of configurations for the distributions to use in the model norm : The type of normalization to apply to the input features hidden_activation : The activation function to use in the hidden layers backbone_type : The type of backbone to use in the model use_multihead : Whether to use a multihead model use_feature_embed : Whether to use feature embedding

forward(state: Tensor | Dict[str, Tensor], adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

reset()

get_action(state: Tensor | Dict[str, Tensor], deterministic: bool = True)

class revive.computation.modules.FeedForwardTransition(in_features: int | dict, out_features: int, hidden_features: int, hidden_layers: int, dist_config: list, norm: str | None = None, hidden_activation: str = 'leakyrelu', backbone_type: str | str_ = 'mlp', mode: str = 'global', obs_dim: int | None = None, **kwargs)

Bases: FeedForwardPolicy

Initializes a feedforward transition instance.

Args:

in_features (Union[int, dict]): The number of input features or a dictionary of input feature sizes. out_features (int): The number of output features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. dist_config (list): The configuration of the distribution to use. norm (Optional[str]): The normalization method to use. None if no normalization is used. hidden_activation (str): The activation function to use for the hidden layers. backbone_type (Union[str, np.str_]): The type of backbone to use. mode (str): The mode to use for the transition. Either ‘global’ or ‘local’. obs_dim (Optional[int]): The dimension of the observation for ‘local’ mode. use_feature_embed (bool): Whether to use feature embedding.

reset()

forward(state: Tensor | Dict[str, Tensor], adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.RecurrentTransition(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, backbone_type: str | str_ = 'gru', mode: str = 'global', obs_dim: int | None = None, **kwargs)

Bases: RecurrentPolicy

Initializes a recurrent transition instance.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. dist_config (list): The distribution configuration. backbone_type (Union[str, np.str_]): The type of backbone to use. mode (str): The mode of the transition. Either ‘global’ or ‘local’. obs_dim (Optional[int]): The dimension of the observation.

forward(state: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TsRecurrentTransition(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, backbone_type: str | str_ = 'gru', mode: str = 'global', obs_dim: int | None = None, **kwargs)

Bases: TsRecurrentPolicy

input_is_dict = True

Initializes a recurrent transition instance.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. dist_config (list): The distribution configuration. backbone_type (Union[str, np.str_]): The type of backbone to use. mode (str): The mode of the transition. Either ‘global’ or ‘local’. obs_dim (Optional[int]): The dimension of the observation.

forward(state: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.TsTransformerTransition(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, norm: str, dist_config: list, mode: str = 'global', obs_dim: int | None = None, ts_input_names: dict = {}, dropout: float = 0, **kwargs)

Bases: TsTransformerPolicy

input_is_dict = True

Initializes a recurrent transition instance.

Args:

in_features (int): The number of input features. out_features (int): The number of output features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. dist_config (list): The distribution configuration. backbone_type (Union[str, np.str_]): The type of backbone to use. mode (str): The mode of the transition. Either ‘global’ or ‘local’. obs_dim (Optional[int]): The dimension of the observation.

forward(state: Tensor, adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.RecurrentRESTransition(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, dist_config: list, backbone_type: str | str_ = 'mlp', mode: str = 'global', obs_dim: int | None = None, rnn_hidden_features: int = 64, window_size: int = 0, **kwargs)

Bases: RecurrentRESPolicy

forward(state: Tensor, adapt_std: Tensor | None = None, field: str = 'mail', **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.EnsembleFeedForwardTransition(in_features: int | dict, out_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, dist_config: list, norm: str | None = None, hidden_activation: str = 'leakyrelu', backbone_type: str | str_ = 'EnsembleRES', mode: str = 'global', obs_dim: int | None = None, **kwargs)

Bases: EnsembleFeedForwardPolicy

Initializes a feedforward transition instance.

Args:

in_features (Union[int, dict]): The number of input features or a dictionary of input feature sizes. out_features (int): The number of output features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. dist_config (list): The configuration of the distribution to use. norm (Optional[str]): The normalization method to use. None if no normalization is used. hidden_activation (str): The activation function to use for the hidden layers. backbone_type (Union[str, np.str_]): The type of backbone to use. mode (str): The mode to use for the transition. Either ‘global’ or ‘local’. obs_dim (Optional[int]): The dimension of the observation for ‘local’ mode. use_feature_embed (bool): Whether to use feature embedding.

reset()

forward(state: Tensor | Dict[str, Tensor], adapt_std: Tensor | None = None, **kwargs) → ReviveDistribution

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.FeedForwardMatcher(in_features: int, hidden_features: int, hidden_layers: int, hidden_activation: str = 'leakyrelu', norm: str | None = None, backbone_type: str | str_ = 'mlp', **kwargs)

Bases: Module

Initializes a feedforward matcher instance.

Args:

in_features (int): The number of input features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. hidden_activation (str): The activation function to use for the hidden layers. norm (str): The normalization method to use. None if no normalization is used. backbone_type (Union[str, np.str_]): The type of backbone to use.

forward(x: Tensor) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.RecurrentMatcher(in_features: int, hidden_features: int, hidden_layers: int, backbone_type: str | str_ = 'gru', bidirect: bool = False)

Bases: Module

forward(*inputs: List[Tensor]) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.HierarchicalMatcher(in_features: list, hidden_features: int, hidden_layers: int, hidden_activation: int, norm: str | None = None)

Bases: Module

forward(*inputs: List[Tensor]) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.VectorizedMatcher(in_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, hidden_activation: str = 'leakyrelu', norm: str | None = None, backbone_type: str | str_ = 'mlp', **kwargs)

Bases: Module

Initializes a feedforward matcher instance.

Args:

in_features (int): The number of input features. hidden_features (int): The number of hidden features. hidden_layers (int): The number of hidden layers. hidden_activation (str): The activation function to use for the hidden layers. norm (str): The normalization method to use. None if no normalization is used. backbone_type (Union[str, np.str_]): The type of backbone to use.

forward(*inputs: List[Tensor]) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class revive.computation.modules.EnsembleMatcher(matcher_record: Dict[str, deque], ensemble_size: int = 50, ensemble_choosing_interval: int = 1, config: dict | None = None)

Bases: object

load_min_max(expert_scores: deque, generate_scores: deque)

init_ensemble_network()

run_raw_scores(inputs: Tensor, aggregation=None) → Tensor

run_scores(inputs: Tensor, aggregation: str | None = None, clip: bool = True) → Tensor

to(device)

class revive.computation.modules.VectorizedCritic(in_features: int, out_features: int, hidden_features: int, hidden_layers: int, ensemble_size: int, norm: str | None = None, hidden_activation: str = 'leakyrelu', output_activation: str = 'identity')

Bases: VectorizedMLP

forward(x: Tensor) → Tensor

forward method of MLP only assume the last dim of x matches in_features

class revive.computation.modules.Value_Net_VectorizedCritic(input_dim_dict, q_hidden_features, q_hidden_layers, num_q_net, *args, **kwargs)

Bases: Module

Initializes a vectorized linear layer instance. Args:

in_features (int): The number of input features. out_features (int): The number of output features. ensemble_size (int): The number of ensembles to use.

forward(state: Tensor | Dict[str, Tensor]) → Tensor

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

revive.computation.operators module

revive.computation.operators.sum(*args)

revive.computation.operators.cat(*args)

revive.computation.operators.sub(arg1, arg2)

revive.computation.operators.mul(*args)

revive.computation.operators.div(arg1, arg2)

revive.computation.operators.min(*args)

revive.computation.operators.mean(*args)

revive.computation.operators.max(*args)

revive.computation.operators.abs(arg)

revive.computation.operators.clip(arg, min_v=None, max_v=None)

revive.computation.operators.exp(arg)

revive.computation.operators.log(arg)

revive.computation.utils module

revive.computation.utils.soft_clamp(x: Tensor, _min=None, _max=None) → Tensor

revive.computation.utils.maintain_gradient_hard_clamp(x: Tensor, _min=None, _max=None) → Tensor

revive.computation.utils.safe_atanh(x)

revive.computation.utils.get_input_from_names(batch: Batch, names: list)

revive.computation.utils.get_input_from_graph(graph, output_name: str, batch_data: Batch)

revive.computation.utils.get_input_dict_from_graph(graph, output_name: str, batch_data: Batch)

revive.computation.utils.get_sample_function(deterministic: bool) → Callable[[ReviveDistribution], Tensor]

revive.computation.utils.to_numpy(x: Tuple[ndarray, Tensor]) → ndarray

Return an object without torch.Tensor.

revive.computation.utils.to_torch(x: Tuple[ndarray, Tensor], dtype=torch.float32, device: str = 'cpu') → Tensor

Return an object without torch.Tensor.

Module contents