opto.trace.bundle

ALLOW_EXTERNAL_DEPENDENCIES module-attribute

ALLOW_EXTERNAL_DEPENDENCIES = None

x module-attribute

x = node('hello')

y module-attribute

y = test(x)

trace_nodes

trace_nodes()

This is a context manager for keeping track which nodes are read/used in an operator.

token instance-attribute

token = None

FunModule

FunModule(
    fun: Callable,
    description: str = None,
    traceable_code: bool = False,
    _process_inputs: bool = True,
    trainable=False,
    catch_execution_error=True,
    allow_external_dependencies=False,
    overwrite_python_recursion=False,
    projections=None,
    _ldict=None,
)

Bases: Module

This is a decorator to trace a function. The wrapped function returns a MessageNode.

Args: fun (callable): the operator to be traced. description (str): a description of the operator; see the MessageNode for syntax. _process_inputs (bool): if True, the input is extracted from the container of nodes; if False, the inputs are passed directly to the underlying function. trainable (bool): if True, the block of code is treated as a variable in the optimization traceable_code (bool): if True, the operator's code is traceable by Trace catch_execution_error (bool): if True, the operator catches the exception raised during the execution of the operator and return ExecutionError. allow_external_dependencies (bool): if True, the operator allows external dependencies to be used in the operator. Namely, not all nodes used to create the output are in the inputs. In this case, the extra dependencies are stored in the info dictionary with key 'extra_dependencies'. overwrite_python_recursion (bool): if True, the operator allows the python recursion behavior of calling the decorated function to be overwritten. When true, applying bundle on a recursive function, would be the same as calling the function directly. When False, the Python's oriignal recursion behavior of decorated functions is preserved. _ldict (dict): the local dictionary to execute the code block.

info instance-attribute

info = dict(
    fun=None,
    fun_name=__qualname__,
    _fun_name=__name__,
    doc=(
        cleandoc(docstring) if docstring is not None else ""
    ),
    signature=signature(fun),
    source=source,
    line_number=line_number,
    file=getfile(fun),
    error_comment=None,
    traceback=None,
    output=None,
    inputs={"args": [], "kwargs": {}},
    external_dependencies=None,
)

traceable_code instance-attribute

traceable_code = traceable_code

description instance-attribute

description = description

catch_execution_error instance-attribute

catch_execution_error = catch_execution_error

allow_external_dependencies instance-attribute

allow_external_dependencies = allow_external_dependencies

parameter instance-attribute

parameter = None

overwrite_python_recursion instance-attribute

overwrite_python_recursion = overwrite_python_recursion

trainable property

trainable

fun property

fun

Return a callable function. Return the decorated function if the parameter is None. Otherwise, return the function defined by the parameter. When exception happens during the defining the function with the parameter, raise a trace.ExecutionError.

name property

name

sync_call_fun

sync_call_fun(fun, *_args, **_kwargs)

Call the operator fun and return the output. Catch the exception if catch_execution_error is True.

async_call_fun async

async_call_fun(fun, *_args, **_kwargs)

preprocess_inputs

preprocess_inputs(args, kwargs, _args, _kwargs)

Preprocess the inputs for the operator fun.

Args: _args (list): the original positional arguments. This includes the default values. _kwargs (dict): the original keyword arguments. This includes the default values. args (list): the wrapped positional arguments. kwargs (dict): the wrapped keyword arguments.

postprocess_output

postprocess_output(
    output, fun, _args, _kwargs, used_nodes, inputs
)

Wrap the output as a MessageNode. Log the inputs and output of the function call.

Args: output (Any): the output of the operator fun. fun (callable): the operator fun. _args (list): the original positional arguments. This includes the default values. _kwargs (dict): the original keyword arguments. This includes the default values. used_nodes (List[Node]): the nodes used in the operator fun. inputs (Dict[str, Node]): the inputs of the operator fun.

forward

forward(*args, **kwargs)

sync_forward

sync_forward(fun, *args, **kwargs)

Call the operator fun and return a MessageNode. All nodes used in the operator fun are added to used_nodes during the execution. If the output is not a Node, we wrap it as a MessageNode, whose inputs are nodes in used_nodes. Sync version.

async_forward async

async_forward(fun, *args, **kwargs)

Call the operator fun and return a MessageNode. All nodes used in the operator fun are added to used_nodes during the execution. If the output is not a Node, we wrap it as a MessageNode, whose inputs are nodes in used_nodes. Async version.

wrap

wrap(
    output: Any,
    inputs: Union[List[Node], Dict[str, Node]],
    external_dependencies: List[Node],
)

Wrap the output as a MessageNode of inputs as the parents.

is_valid_output staticmethod

is_valid_output(output)

detach

detach()

generate_comment

generate_comment(
    code: str,
    comment: str,
    comment_line_number: int,
    base_line_number: int = 0,
)

get_source

get_source(obj: Any, bug_mode=False)

Get the source code of the function and its line number, excluding the @bundle decorator line. bug_mode=True means We are in the forward() function, but there is an error during execution. The error can be caused by a lambda function which does not have def in the source code. We turn off the error raising in the end of this function.

Allowable two types of usages:

Examples:

@blah ... @bundle # or @ ....bundle() def fun(...): # ... .... or inline usage bundle()(fun) # or ....bundle()(fun)

disable_external_dependencies_check

disable_external_dependencies_check(
    allow_external_dependencies: bool,
)

Set the global flag for allowing external dependencies.

bundle

bundle(
    description=None,
    traceable_code=False,
    _process_inputs=True,
    trainable=False,
    catch_execution_error=True,
    allow_external_dependencies=False,
    overwrite_python_recursion=False,
    projections=None,
    mlflow_kwargs=None,
)

Wrap a function as a FunModule which returns node objects.

The input signature to the wrapped function stays the same. bundle can be used with other decorators so long as they are not named 'bundle'.

Args: description (str, optional): Description of the operator. Defaults to None. traceable_code (bool, optional): Whether the operator's code is traceable by Trace. Defaults to False. _process_inputs (bool, optional): Whether to extract input from container of nodes. Defaults to True. trainable (bool, optional): Whether block of code is treated as variable in optimization. Defaults to False. catch_execution_error (bool, optional): Whether to catch exceptions during operator execution. Defaults to True. allow_external_dependencies (bool, optional): Whether to allow external dependencies. Defaults to False. overwrite_python_recursion (bool, optional): Whether to overwrite Python recursion behavior. Defaults to False. projections (List[Projection], optional): List of projections to be used in updating trainable parameter. Defaults to None. mlflow_kwargs (dict, optional): Additional keyword arguments to pass to mlflow.trace() when MLFlow autologging is enabled. Supports 'name', 'span_type', 'attributes', and 'output_reducer' parameters as defined in mlflow.trace(). Additionally supports: - 'silent': If True, disables tracing for this specific function - 'default_op': If True and disable_default_op_logging is enabled, silences tracing for this function Only used when mlflow_autologging is True and log_models config is enabled. Defaults to None.

Returns: FunModule: The wrapped function that returns node objects.

to_data

to_data(obj)

Extract the data from a node or a container of nodes.

wrap_node

wrap_node(obj)

Wrap a node on top of the original object

detach_inputs

detach_inputs(obj)

Detach a node or a container of nodes.

update_local

update_local(frame, name, value)

Update the value of a local variable in a frame.

test

test(x)