State

RuntimeState

At its most basic level, this class essentially stores two things:

1. A dictionary of the currently available variables. Each variable is a RuntimeVariable object, which acts as a wrapper around the real thing.
2. Every step that has been executed. Each step is a RuntimeStateStep object, which contains a list of RuntimeInstruction objects. ¹ Note that instructions and steps are two different things, since we can execute multiple instructions in a single step.

Actions are not stored in the state

RuntimeState does not keep track of Actions. This has to do with avoiding circular imports.

Actions need to be able to access the RuntimeState if they want to modify it. This means that we can't import Action in this module.

Therefore, RuntimeState is a class that exists independently of the Runtime , which needs access to Actions to use them.

Handling starting and unnamed variables

RuntimeState also contains a significant amount of machinery to handle variables that are given by the user. The process_unstructured_inputs_or_outputs method is key to this.

The general idea is that the user is free to provide inputs with or without names. In the latter case, we generate names for the variables according to the type of the variable.

For instance, if the user provides two integers as an input, they will be named int_0 and int_1. We also keep track of these variable names, in case we need to create more variables without having explicit names attached to them.

The tracking of this scheme is done with the vars_fallback_counter attribute, which is a Counter that keeps track of the number of variables that have been given a fallback name.

conatus.runtime.state.RuntimeState dataclass

RuntimeState(
    starting_variables: (
        Collection[ParamType]
        | Mapping[str, ParamType]
        | OrderedDict[str, RuntimeVariable]
        | None
    ),
    *,
    config: ConsolidatedTaskConfig,
    expected_input_types: (
        OrderedDict[str, TypeOfType] | None
    ) = None
)

State of the the agent's Runtime.

PARAMETER	DESCRIPTION
starting_variables	The starting variables. TYPE: Collection[ParamType] | Mapping[str, ParamType] | OrderedDict[str, RuntimeVariable] | None
config	The configuration of the run. TYPE: ConsolidatedTaskConfig
expected_input_types	The expected input types, as a dictionary of variable names and their types. If provided, we will use this dictionary to compare the types of the starting variables with the expected input types. TYPE: OrderedDict[str, TypeOfType] | None DEFAULT: None

METHOD	DESCRIPTION
import_statements_as_code	Get the import statements as a Python call.
code	Print the steps as a Python call.
update_vars_fallback_counter	Update the vars_fallback_counter.
get_variable_type	Get a type string from a variable or a type.
process_unstructured_inputs_or_outputs	Process unstructured inputs or outputs to an OrderedDict.
get_first_step	Get the first step.
normalize_starting_variables	Initialize and normalize the starting variables.
repr_at_step	Get the repr (text, image) of a variable at a given step.
add_result	Add a result to the state.
add_json_instruction	Add a JSON instruction to the state.
add_python_instruction	Add a Python instruction to the state.
add_instruction	Add an instruction to the state.
dump_variables	Dump the variables.

ATTRIBUTE	DESCRIPTION
termination_sentinel	Sentinel to indicate that the program should terminate. TYPE: bool
last_step	Get the last step. TYPE: RuntimeStateStep
variables	A dictionary of the variables in the state. TYPE: OrderedDict[str, RuntimeVariable]
starting_variables	A dictionary of the starting variables in the state. TYPE: OrderedDict[str, RuntimeVariable]
vars_fallback_counter	Counter of the number of variables with fallback type names. TYPE: Counter[str]
step_count	The current step number. TYPE: int
steps	The steps taken by the Runtime. TYPE: OrderedDict[int, RuntimeStateStep]
pending_variable_updates	The variables that were modified by an [instruction TYPE: list[str]

Source code in conatus/runtime/state.py

def __init__(
    self,
    starting_variables: Collection[ParamType]
    | Mapping[str, ParamType]
    | OrderedDict[str, RuntimeVariable]
    | None,
    *,
    config: ConsolidatedTaskConfig,
    expected_input_types: OrderedDict[str, TypeOfType] | None = None,
) -> None:
    """Initialize the RuntimeState.

    Args:
        starting_variables: The starting variables.
        config: The configuration of the run.
        expected_input_types: The expected input types, as a dictionary of
            variable names and their types. If provided, we will use this
            dictionary to compare the types of the starting variables with
            the expected input types.
    """
    variables, vars_fallback_counter = self.normalize_starting_variables(
        starting_variables,
        max_var_repr_len=config.max_var_repr_len,
        expected_input_types=expected_input_types,
    )
    first_step = self.get_first_step(variables)
    self.variables = variables
    self.starting_variables = variables
    self.vars_fallback_counter = vars_fallback_counter
    self.step_count = 0
    self.steps = OrderedDict({0: first_step})
    self.pending_variable_updates = []

termination_sentinel class-attribute instance-attribute

termination_sentinel: bool = False

Sentinel to indicate that the program should terminate.

last_step property

last_step: RuntimeStateStep

Get the last step.

variables instance-attribute

variables: OrderedDict[str, RuntimeVariable] = variables

A dictionary of the variables in the state.

starting_variables instance-attribute

starting_variables: OrderedDict[str, RuntimeVariable] = (
    variables
)

A dictionary of the starting variables in the state.

This is a subset of the variables dictionary, and is here to indicate which variables were given to the Runtime before any instructions were executed.

vars_fallback_counter instance-attribute

vars_fallback_counter: Counter[str] = vars_fallback_counter

Counter of the number of variables with fallback type names.

If the name of a variable is not provided, we automatically generate a name of the format <type>_<int>, where <type> is the type of the variable and <int> is the number of variables of that type that have had that name format. If a variable of <type> is given a name, we do not increment the number.

We also try to handle variables of that format that were given by the user: if the user provides a variable of the form <type>_<int>, we update the counter so that <type> is at least <int> + 1. This means that we might start a run where the counter has a value for <type> that is higher than the number of variables of that type that have been seen so far.

Important note: the counter is 0-indexed. So if the counter for <type> is 0, it means that there is one variable <type>_0. If it is 1, there are two, and so on.

step_count instance-attribute

step_count: int = 0

The current step number.

Note that, in general, step 0 is reserved for setting up the initial state of the agent (e.g. importing the initial variables).

steps instance-attribute

steps: OrderedDict[int, RuntimeStateStep] = OrderedDict(
    {0: first_step}
)

The steps taken by the Runtime.

pending_variable_updates instance-attribute

pending_variable_updates: list[str] = []

The variables that were modified by an instruction .

This is a temporary buffer that is used to store the variables that were modified by an instruction . Runtime s should check this buffer after an instruction has been executed, and then clear it.

import_statements_as_code

import_statements_as_code() -> str

Get the import statements as a Python call.

RETURNS	DESCRIPTION
str	The import statements as a Python call. TYPE: str

Source code in conatus/runtime/state.py

def import_statements_as_code(self) -> str:
    """Get the import statements as a Python call.

    Returns:
        str: The import statements as a Python call.
    """
    all_imports: dict[str, set[str]] = {}
    for step in self.steps.values():
        for module, name in step.import_statements():
            if module not in all_imports:
                all_imports[module] = set()
            all_imports[module].add(name)
    return "\n".join(
        f"from {module} import {', '.join(names)}"
        for module, names in all_imports.items()
    )

code

code(*, include_failed: bool = False) -> str

Print the steps as a Python call.

PARAMETER	DESCRIPTION
include_failed	Whether to include the code of failed instructions. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str	The steps as a Python call. TYPE: str

Source code in conatus/runtime/state.py

def code(self, *, include_failed: bool = False) -> str:
    """Print the [`steps`][conatus.runtime.state.RuntimeState.steps] as a Python call.

    Args:
        include_failed: Whether to include the code of failed instructions.

    Returns:
        str: The steps as a Python call.
    """  # noqa: E501
    imports_code = self.import_statements_as_code()
    imports_code_str = imports_code + "\n" if imports_code else ""
    lines_of_code: list[str] = []
    for step in self.steps.values():
        code = step.code(include_failed=include_failed)
        if code is not None:
            lines_of_code.append(code)
    return imports_code_str + "\n".join(lines_of_code)

update_vars_fallback_counter staticmethod

update_vars_fallback_counter(
    variables: (
        OrderedDict[str, RuntimeVariable]
        | OrderedDict[str, TypeOfType]
        | OrderedDict[str, ParamType]
    ),
    vars_fallback_counter: Counter[str] | None = None,
) -> Counter[str]

Update the vars_fallback_counter.

We take either a predefined counter or an empty one, and update it with the variables that are given.

See the vars_fallback_counter attribute for more information.

PARAMETER	DESCRIPTION
variables	The variables. We don't really care about the types of the values in the dictionary, since we only care about the keys. TYPE: OrderedDict[str, RuntimeVariable] | OrderedDict[str, TypeOfType] | OrderedDict[str, ParamType]
vars_fallback_counter	The vars fallback counter. TYPE: Counter[str] | None DEFAULT: None

RETURNS	DESCRIPTION
Counter[str]	The updated counter.

Source code in conatus/runtime/state.py

@staticmethod
def update_vars_fallback_counter(
    variables: OrderedDict[str, RuntimeVariable]
    | OrderedDict[str, TypeOfType]
    | OrderedDict[str, ParamType],
    vars_fallback_counter: Counter[str] | None = None,
) -> Counter[str]:
    """Update the [`vars_fallback_counter`][conatus.runtime.state.RuntimeState.vars_fallback_counter].

    We take either a predefined counter or an empty one, and update it with
    the variables that are given.

    See the [`vars_fallback_counter`
    ][conatus.runtime.state.RuntimeState.vars_fallback_counter] attribute
    for more information.

    Args:
        variables: The variables. We don't really care about the types of
            the values in the dictionary, since we only care about the keys.
        vars_fallback_counter: The vars fallback counter.

    Returns:
        (Counter[str]): The updated counter.
    """  # noqa: E501
    if vars_fallback_counter is None:
        vars_fallback_counter = Counter()
    for v_name in variables:
        # NOTE: Variables of the form `<type>_<int>` could be created either
        # by the runtime automatically (if a variable is not given, say)
        # or manually. What we want is to ensure is that the automatically
        # created variables have a counter that is at least as high as the
        # highest value of the manually created variables of the same type.
        # In other words: if we have a `int_0` variable created by the
        # runtime, and then a `int_999` variable created manually, we want
        # to ensure that the counter for `int` is at least 999.
        if (
            len(sp := v_name.split("_")) > 1
            and (i := attempt_to_cast_to_int(sp[1])) is not False
            and i >= vars_fallback_counter[sp[0]]
        ):
            vars_fallback_counter[sp[0]] = i
    return vars_fallback_counter

get_variable_type staticmethod

get_variable_type(
    var: TypeOfType | ParamType,
    kind_of_in_outputs: Literal[
        "input_types", "output_types", "starting_variables"
    ],
) -> str

Get a type string from a variable or a type.

Examples

from conatus.runtime.state import RuntimeState

get_variable_type = RuntimeState.get_variable_type

class MyClass:
    pass

assert get_variable_type(int, "input_types") == "int"
assert get_variable_type(MyClass, "output_types") == "myclass"

# For starting variables, we care about the physical type of the
# variable, since it's not a type object.
assert get_variable_type(MyClass(), "starting_variables") == "myclass"

PARAMETER	DESCRIPTION
var	The input type or the actual argument. Whether it is a type or a variable depends on the value of kind_of_in_outputs . TYPE: TypeOfType | ParamType
kind_of_in_outputs	The kind of input/output. If it is "starting_variables", we will return the type of the variable as a string. Otherwise, we will return the name of the type as a string. TYPE: Literal['input_types', 'output_types', 'starting_variables']

RETURNS	DESCRIPTION
str	The processed type string.

Source code in conatus/runtime/state.py

@staticmethod
def get_variable_type(
    var: TypeOfType | ParamType,
    kind_of_in_outputs: Literal[
        "input_types", "output_types", "starting_variables"
    ],
) -> str:
    """Get a type string from a variable or a type.

    # Examples

    ```python
    from conatus.runtime.state import RuntimeState

    get_variable_type = RuntimeState.get_variable_type

    class MyClass:
        pass

    assert get_variable_type(int, "input_types") == "int"
    assert get_variable_type(MyClass, "output_types") == "myclass"

    # For starting variables, we care about the physical type of the
    # variable, since it's not a type object.
    assert get_variable_type(MyClass(), "starting_variables") == "myclass"
    ```

    Args:
        var: The input type or the actual argument. Whether it is a type
            or a variable depends on the value of [`kind_of_in_outputs`
            ][conatus.runtime.state.RuntimeState.get_variable_type(kind_of_in_outputs)].
        kind_of_in_outputs: The kind of input/output. If it is
            "starting_variables", we will return the type of the variable
            as a string. Otherwise, we will return the name of the type
            as a string.

    Returns:
        (str): The processed type string.
    """
    if kind_of_in_outputs == "starting_variables":
        type_str = type(cast("ParamType", var)).__name__.lower()  # type: ignore[redundant-cast]
    elif var is None:
        type_str = "none"
    else:
        type_str = (
            getattr(var, "__name__", "input").lower()
            if kind_of_in_outputs == "input_types"
            else getattr(var, "__name__", "output").lower()
        )
    return (
        RuntimeVariable._convert_type_str_to_potential_var_name(type_str)  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]
    )

process_unstructured_inputs_or_outputs staticmethod

process_unstructured_inputs_or_outputs(
    in_or_outputs: (
        list[TypeOfType]
        | dict[str, TypeOfType]
        | TypeOfType
        | None
    ),
    kind_of_in_outputs: Literal["input_types"],
    *,
    max_var_repr_len: int | None = None,
    expected_input_types: None = None
) -> tuple[
    OrderedDict[str, TypeOfType],
    Counter[str],
    ExpectedTaskInputType,
]

process_unstructured_inputs_or_outputs(
    in_or_outputs: (
        list[TypeOfType]
        | dict[str, TypeOfType]
        | TypeOfType
        | None
    ),
    kind_of_in_outputs: Literal["output_types"],
    *,
    max_var_repr_len: int | None = None,
    expected_input_types: None = None
) -> tuple[
    OrderedDict[str, TypeOfType],
    Counter[str],
    ExpectedTaskInputType,
]

process_unstructured_inputs_or_outputs(
    in_or_outputs: (
        list[ParamType] | dict[str, ParamType] | None
    ),
    kind_of_in_outputs: Literal["starting_variables"],
    *,
    max_var_repr_len: int | None = None,
    expected_input_types: (
        OrderedDict[str, TypeOfType] | None
    ) = None
) -> tuple[
    OrderedDict[str, RuntimeVariable],
    Counter[str],
    ExpectedTaskInputType,
]

process_unstructured_inputs_or_outputs(
    in_or_outputs: (
        list[TypeOfType]
        | dict[str, TypeOfType]
        | TypeOfType
        | list[ParamType]
        | dict[str, ParamType]
        | None
    ),
    kind_of_in_outputs: Literal[
        "input_types", "output_types", "starting_variables"
    ],
    *,
    max_var_repr_len: int | None = None,
    expected_input_types: (
        OrderedDict[str, TypeOfType] | None
    ) = None
) -> tuple[
    OrderedDict[str, RuntimeVariable]
    | OrderedDict[str, TypeOfType],
    Counter[str],
    ExpectedTaskInputType,
]

Process unstructured inputs or outputs to an OrderedDict.

This function exists to standardize the way we process unstructured inputs or outputs. Since we need the RuntimeState to start with an OrderedDict of types or variables, we need to have a function that can create it regardless of the way the user defines the inputs, the outputs, or the starting variables.

This function is meant to be common to all three kinds of inputs or outputs: input_types, output_types, and starting_variables:

• input_types: We pass the user input for the inputs fields, which is meant to be a singleton, list or dictionary of TypeOfTypes. We return an OrderedDict of TypeOfType s. That dict either keeps the keys of the input dictionary, or generates new keys based on the type of the variables.
• output_types: We pass the user input for the outputs fields, and perform the same operations as for input_types.
• starting_variables: We pass the user input for the starting_variables field, which can be a singleton, list or dictionary of ParamTypes. We return an OrderedDict of RuntimeVariable s. That dict either keeps the keys of the input dictionary, or generates new keys based on the type of the variables. Optionally, we can pass an expected_input_types dictionary to compare the types of the starting variables with the expected input types.

Examples

With a list of types

Note that a singleton type will be converted to a list.

from collections import Counter, OrderedDict
from conatus.runtime.state import RuntimeState

cleaned_types, counter, input_types = (
    RuntimeState.process_unstructured_inputs_or_outputs(
        in_or_outputs=[list[str], list[int]],
        kind_of_in_outputs="input_types",
    )
)
assert cleaned_types == OrderedDict(
    {"list_0": list[str], "list_1": list[int]}
)
assert counter == Counter({"list": 1})
assert input_types == "list"

With a dict of types

from collections import Counter, OrderedDict
from conatus.runtime.state import RuntimeState

cleaned_types, counter, input_types = (
    RuntimeState.process_unstructured_inputs_or_outputs(
        in_or_outputs={"x": list[str], "y": list[int]},
        kind_of_in_outputs="input_types",
    )
)
assert cleaned_types == OrderedDict(
    {"x": list[str], "y": list[int]}
)
assert counter == Counter()
assert input_types == "dict"

With a list of starting variables

Note that the process is similar to the previous example in the case of a dictionary of starting variables.

from collections import Counter, OrderedDict
from conatus.runtime.state import RuntimeState

cleaned_vars, counter, input_types = (
    RuntimeState.process_unstructured_inputs_or_outputs(
        in_or_outputs=[1, "hello", True],
        kind_of_in_outputs="starting_variables",
    )
)
assert list(cleaned_vars.keys()) == ["int_0", "str_0", "bool_0"]
assert cleaned_vars["int_0"].value == 1
assert cleaned_vars["str_0"].value == "hello"
assert cleaned_vars["bool_0"].value == True
assert counter == Counter({"int": 0, "str": 0, "bool": 0})
assert input_types == "list"

PARAMETER	DESCRIPTION
in_or_outputs	The input types, output types, or starting variables. If kind_of_in_outputs is "starting_variables", this should be a list or dictionary of ParamType . We don't accept a single ParamType because we can't tell if it's a list or a dictionary, and this could lead to confusion. If kind_of_in_outputs is "input_types" or "output_types", this should be a list or dictionary of TypeOfType , or a single `TypeOfType . TYPE: list[TypeOfType] | dict[str, TypeOfType] | TypeOfType | list[ParamType] | dict[str, ParamType] | None
kind_of_in_outputs	The kind of in_or_outputs. This should be one of "input_types", "output_types", or "starting_variables". TYPE: Literal['input_types', 'output_types', 'starting_variables']
expected_input_types	The expected input types, as an ordered dictionary of variable names and their types. If kind_of_in_outputs is "starting_variables", we will use this dictionary to compare the types of the starting variables with the expected input types. TYPE: OrderedDict[str, TypeOfType] | None DEFAULT: None
max_var_repr_len	The maximum length of the string representation of a variable. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
OrderedDict[str, RuntimeVariable] | OrderedDict[str, TypeOfType]	If the kind_of_in_outputs is "starting_variables", we return an OrderedDict of RuntimeVariable s. Otherwise, we return an OrderedDict of TypeOfType s.
Counter[str]	The counter of the number of variables of each type.
ExpectedTaskInputType	The type of the starting variables, as a string indicating whether they were a list, a dictionary, or None.

Source code in conatus/runtime/state.py

@staticmethod
def process_unstructured_inputs_or_outputs(
    in_or_outputs: list[TypeOfType]
    | dict[str, TypeOfType]
    | TypeOfType
    | list[ParamType]
    | dict[str, ParamType]
    | None,
    kind_of_in_outputs: Literal[
        "input_types", "output_types", "starting_variables"
    ],
    *,
    max_var_repr_len: int | None = None,
    expected_input_types: OrderedDict[str, TypeOfType] | None = None,
) -> tuple[
    OrderedDict[str, RuntimeVariable] | OrderedDict[str, TypeOfType],
    Counter[str],
    ExpectedTaskInputType,
]:
    """Process unstructured inputs or outputs to an [OrderedDict][collections.OrderedDict].

    This function exists to standardize the way we process unstructured
    inputs or outputs. Since we need the [`RuntimeState`
    ][conatus.runtime.state.RuntimeState] to start with an
    [`OrderedDict`][collections.OrderedDict] of types or variables, we
    need to have a function that can create it regardless of the way the
    user defines the inputs, the outputs, or the starting variables.

    This function is meant to be common to all three kinds of inputs or
    outputs: `input_types`, `output_types`, and `starting_variables`:

    - `input_types`: We pass the user input for the `inputs` fields,
        which is meant to be a singleton, list or dictionary of
        TypeOfTypes. We return an [`OrderedDict`
        ][collections.OrderedDict] of [`TypeOfType`
        ][conatus._types.TypeOfType]s. That dict either keeps the keys of
        the input dictionary, or generates new keys based on the type of
        the variables.
    - `output_types`: We pass the user input for the `outputs` fields,
        and perform the same operations as for `input_types`.
    - `starting_variables`: We pass the user input for the
        `starting_variables` field, which can be a singleton, list or
        dictionary of ParamTypes. We return an [`OrderedDict`
        ][collections.OrderedDict] of [`RuntimeVariable`
        ][conatus.runtime.state.RuntimeVariable]s. That dict either keeps
        the keys of the input dictionary, or generates new keys based on
        the type of the variables. Optionally, we can pass an
        `expected_input_types` dictionary to compare the types of the
        starting variables with the expected input types.

    # Examples

    ## With a list of types

    Note that a singleton type will be converted to a list.

    ```python
    from collections import Counter, OrderedDict
    from conatus.runtime.state import RuntimeState

    cleaned_types, counter, input_types = (
        RuntimeState.process_unstructured_inputs_or_outputs(
            in_or_outputs=[list[str], list[int]],
            kind_of_in_outputs="input_types",
        )
    )
    assert cleaned_types == OrderedDict(
        {"list_0": list[str], "list_1": list[int]}
    )
    assert counter == Counter({"list": 1})
    assert input_types == "list"
    ```

    ## With a dict of types

    ```python
    from collections import Counter, OrderedDict
    from conatus.runtime.state import RuntimeState

    cleaned_types, counter, input_types = (
        RuntimeState.process_unstructured_inputs_or_outputs(
            in_or_outputs={"x": list[str], "y": list[int]},
            kind_of_in_outputs="input_types",
        )
    )
    assert cleaned_types == OrderedDict(
        {"x": list[str], "y": list[int]}
    )
    assert counter == Counter()
    assert input_types == "dict"
    ```

    ## With a list of starting variables

    Note that the process is similar to the previous example in the case
    of a dictionary of starting variables.

    ```python
    from collections import Counter, OrderedDict
    from conatus.runtime.state import RuntimeState

    cleaned_vars, counter, input_types = (
        RuntimeState.process_unstructured_inputs_or_outputs(
            in_or_outputs=[1, "hello", True],
            kind_of_in_outputs="starting_variables",
        )
    )
    assert list(cleaned_vars.keys()) == ["int_0", "str_0", "bool_0"]
    assert cleaned_vars["int_0"].value == 1
    assert cleaned_vars["str_0"].value == "hello"
    assert cleaned_vars["bool_0"].value == True
    assert counter == Counter({"int": 0, "str": 0, "bool": 0})
    assert input_types == "list"
    ```

    Args:
        in_or_outputs: The input types, output types, or starting variables.
            If `kind_of_in_outputs` is "starting_variables", this should
            be a list or dictionary of [`ParamType`
            ][conatus._types.ParamType]. We don't accept a single
            [`ParamType`][conatus._types.ParamType] because we can't tell if
            it's a list or a dictionary, and this could lead to confusion.
            If `kind_of_in_outputs` is "input_types" or "output_types",
            this should be a list or dictionary of [`TypeOfType`
            ][conatus._types.TypeOfType], or a single [`TypeOfType
            ][conatus._types.TypeOfType].
        kind_of_in_outputs: The kind of `in_or_outputs`. This should be one
            of "input_types", "output_types", or "starting_variables".
        expected_input_types: The expected input types, as an ordered
            dictionary of variable names and their types. If
            kind_of_in_outputs is "starting_variables", we will use this
            dictionary to compare the types of the starting variables with
            the expected input types.
        max_var_repr_len: The maximum length of the string
            representation of a variable.

    Returns:
        (OrderedDict[str, RuntimeVariable] | OrderedDict[str, TypeOfType]):
            If the kind_of_in_outputs is "starting_variables", we return an
            [`OrderedDict`][collections.OrderedDict] of [`RuntimeVariable`
            ][conatus.runtime.state.RuntimeVariable]s. Otherwise, we return
            an [`OrderedDict`][collections.OrderedDict] of [`TypeOfType`
            ][conatus._types.TypeOfType]s.
        (Counter[str]): The counter of the number of variables of each type.
        (ExpectedTaskInputType): The type of the starting variables, as a
            string indicating whether they were a list, a dictionary, or
            None.
    """  # noqa: E501
    match in_or_outputs:
        case None:
            return RuntimeState._process_none()

        # Every variable is given a unique name based on its type and the
        # number of variables of that type that have been seen so far.
        case list():
            return RuntimeState._process_list(
                in_or_outputs=cast(
                    "list[TypeOfType | ParamType]", in_or_outputs
                ),
                kind_of_in_outputs=kind_of_in_outputs,
                max_var_repr_len=max_var_repr_len,
                expected_input_types=expected_input_types,
            )

        # If the starting variables are a dictionary, we use the keys as
        # variable names. We still use the counter to ensure homogeneity.
        case dict():  # pragma: no branch
            return RuntimeState._process_dict(
                inputs=cast(
                    "dict[str, TypeOfType | ParamType]", in_or_outputs
                ),
                kind_of_in_outputs=kind_of_in_outputs,
                max_var_repr_len=max_var_repr_len,
                expected_input_types=expected_input_types,
            )

        # If the starting variables are a single type or a single value,
        # we wrap it in a list and call the list case.
        case _:
            return RuntimeState._process_singleton(
                in_or_output=in_or_outputs,
                kind_of_in_outputs=kind_of_in_outputs,
                max_var_repr_len=max_var_repr_len,
                expected_input_types=expected_input_types,
            )

get_first_step staticmethod

get_first_step(
    variables: OrderedDict[str, RuntimeVariable],
) -> RuntimeStateStep

Get the first step.

PARAMETER	DESCRIPTION
variables	The variables. TYPE: OrderedDict[str, RuntimeVariable]

RETURNS	DESCRIPTION
RuntimeStateStep	The first step. TYPE: RuntimeStateStep

Source code in conatus/runtime/state.py

@staticmethod
def get_first_step(
    variables: OrderedDict[str, RuntimeVariable],
) -> RuntimeStateStep:
    """Get the first step.

    Args:
        variables: The variables.

    Returns:
        RuntimeStateStep: The first step.
    """
    instructions: list[
        RuntimeJSONInstruction | RuntimePythonInstruction
    ] = [
        RuntimeJSONInstruction(
            action_name="import_variable",
            arguments={"_": variable.name},
            returns=[(variable.name, variable.type_hint)],
            stdout_output=None,
            stderr_output=None,
            modified_variables=[variable.name],
            execution_successful=True,
        )
        for variable in variables.values()
    ]
    return RuntimeStateStep(instructions=instructions, step_number=0)

normalize_starting_variables staticmethod

normalize_starting_variables(
    starting_variables: (
        Collection[ParamType]
        | Mapping[str, ParamType]
        | OrderedDict[str, RuntimeVariable]
        | None
    ),
    *,
    max_var_repr_len: int | None = None,
    vars_fallback_counter: Counter[str] | None = None,
    expected_input_types: (
        OrderedDict[str, TypeOfType] | None
    ) = None
) -> tuple[OrderedDict[str, RuntimeVariable], Counter[str]]

Initialize and normalize the starting variables.

We normalize the starting variables to an OrderedDict of RuntimeVariable s and a Counter of the number of variables of each type.

If that's already the case, we just return the starting variables with the Counter.

PARAMETER	DESCRIPTION
starting_variables	The starting variables. TYPE: Collection[ParamType] | Mapping[str, ParamType] | OrderedDict[str, RuntimeVariable] | None
max_var_repr_len	The maximum length of the string representation of a variable. TYPE: int | None DEFAULT: None
vars_fallback_counter	The counter of the number of variables of each type. TYPE: Counter[str] | None DEFAULT: None
expected_input_types	The expected input types, as a dictionary of variable names and their types. If provided, we will use this dictionary to compare the types of the starting variables with the expected input types. TYPE: OrderedDict[str, TypeOfType] | None DEFAULT: None

RETURNS	DESCRIPTION
OrderedDict[str, RuntimeVariable]	The variables.
Counter[str]	The counter of the number of variables of each type.

Source code in conatus/runtime/state.py

@staticmethod
def normalize_starting_variables(
    starting_variables: Collection[ParamType]
    | Mapping[str, ParamType]
    | OrderedDict[str, RuntimeVariable]
    | None,
    *,
    max_var_repr_len: int | None = None,
    vars_fallback_counter: Counter[str] | None = None,
    expected_input_types: OrderedDict[str, TypeOfType] | None = None,
) -> tuple[OrderedDict[str, RuntimeVariable], Counter[str]]:
    """Initialize and normalize the starting variables.

    We normalize the starting variables to an [`OrderedDict`
    ][collections.OrderedDict] of [`RuntimeVariable`
    ][conatus.runtime.variable.RuntimeVariable]s and a
    [`Counter`][collections.Counter] of the number of variables of each
    type.

    If that's already the case, we just return the starting variables with
    the [`Counter`][collections.Counter].

    Args:
        starting_variables: The starting variables.
        max_var_repr_len: The maximum length of the string
            representation of a variable.
        vars_fallback_counter: The counter of the number of variables of
            each type.
        expected_input_types: The expected input types, as a dictionary of
            variable names and their types. If provided, we will use this
            dictionary to compare the types of the starting variables with
            the expected input types.

    Returns:
        (OrderedDict[str, RuntimeVariable]): The variables.
        (Counter[str]): The counter of the number of variables of each type.
    """
    # If we have a dictionary of RuntimeVariables, we just create the
    # counter and return the variables.
    if isinstance(starting_variables, OrderedDict) and all(
        isinstance(var, RuntimeVariable)
        for var in starting_variables.values()  # pyright: ignore[reportUnknownVariableType]
    ):
        starting_variables = cast(
            "OrderedDict[str, RuntimeVariable]", starting_variables
        )
        variables: OrderedDict[str, RuntimeVariable] = starting_variables
        vars_fallback_counter = RuntimeState.update_vars_fallback_counter(
            variables, vars_fallback_counter
        )
    # Otherwise, we need to process the starting variables.
    else:
        starting_variables = cast(
            "list[ParamType] | dict[str, ParamType]", starting_variables
        )
        variables, vars_fallback_counter, _ = (
            RuntimeState.process_unstructured_inputs_or_outputs(
                in_or_outputs=starting_variables,
                kind_of_in_outputs="starting_variables",
                max_var_repr_len=max_var_repr_len,
                expected_input_types=expected_input_types,
            )
        )
    return variables, vars_fallback_counter

repr_at_step

repr_at_step(
    var_name: str, step: int
) -> tuple[str, str | None]

Get the repr (text, image) of a variable at a given step.

PARAMETER	DESCRIPTION
var_name	The name of the variable. TYPE: str
step	The step number. TYPE: int

RETURNS	DESCRIPTION
str	The text repr of the value of the variable at the given step. TYPE: str
str | None	str | None: The image repr of the value of the variable at the given step, if applicable.

Source code in conatus/runtime/state.py

def repr_at_step(self, var_name: str, step: int) -> tuple[str, str | None]:
    """Get the repr (text, image) of a variable at a given step.

    Args:
        var_name: The name of the variable.
        step: The step number.

    Returns:
        str: The text repr of the value of the variable at the given step.
        str | None: The image repr of the value of the variable at the
            given step, if applicable.
    """
    return self.variables[var_name].repr_at_step(step)

add_result

add_result(
    result: ParamType,
    variable_name: str | None = None,
    *,
    imported: bool = False
) -> RuntimeVariable

Add a result to the state.

If you provide a variable_name argument:

• If that variable already exists, we will overwrite its value.
• If that variable does not exist, we will create a new variable with that name.

If you do not provide a variable_name argument, we will generate a name for the variable.

PARAMETER	DESCRIPTION
result	The result to add. TYPE: ParamType
variable_name	The name of the variable to store the result in. If not provided, we will generate a name. TYPE: str | None DEFAULT: None
imported	Whether the variable is imported. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
RuntimeVariable	The variable that was added. TYPE: RuntimeVariable

Source code in conatus/runtime/state.py

def add_result(
    self,
    result: ParamType,
    variable_name: str | None = None,
    *,
    imported: bool = False,
) -> RuntimeVariable:
    """Add a result to the state.

    If you provide a `variable_name` argument:

    * If that variable already exists, we will overwrite its value.
    * If that variable does not exist, we will create a new variable with
        that name.

    If you do not provide a `variable_name` argument, we will generate a
    name for the variable.

    Args:
        result: The result to add.
        variable_name: The name of the variable to store the result in. If
            not provided, we will generate a name.
        imported: Whether the variable is imported.

    Returns:
        RuntimeVariable: The variable that was added.
    """
    if variable_name is not None and variable_name in self.variables:
        _ = self.variables[variable_name].update(
            result, step=self.step_count, skip_if_equal=False
        )
        # For good measure, let's update the counter again.
        # That way, if a user provides a variable of the type "var_0",
        # we ensure that the counter is up-to-date.
        self.vars_fallback_counter = self.update_vars_fallback_counter(
            variables=self.variables,
            vars_fallback_counter=self.vars_fallback_counter,
        )
        return self.variables[variable_name]
    if variable_name is not None and not variable_name.isidentifier():
        msg = (
            f"Invalid variable name: '{variable_name}' \n"
            "We will create a new variable automatically."
        )
        logger.warning(msg)
        variable_name = None
    if variable_name is None:
        type_hint = type(result).__name__.lower()
        if type_hint not in self.vars_fallback_counter:
            self.vars_fallback_counter[type_hint] = 0
        else:
            self.vars_fallback_counter[type_hint] += 1
        variable_name = (
            f"{type_hint}_{self.vars_fallback_counter[type_hint]}"
        )
    self.variables[variable_name] = RuntimeVariable(
        name=variable_name,
        value=result,
        imported=imported,
        initial_step=self.step_count,
    )
    return self.variables[variable_name]

add_json_instruction

add_json_instruction(
    action_name: str,
    arguments: dict[str, JSONType | RuntimeVariable],
    returns: list[tuple[str, TypeOfType]],
    *,
    stdout_output: str | None = None,
    stderr_output: str | None = None,
    modified_variables: list[str] | None = None,
    execution_successful: bool = False
) -> None

Add a JSON instruction to the state.

This method does not modify the step number.

This method does not modify the step number. If you want to add an instruction to a new step, you need to explicitly increment the step number by incrementing step_count .

PARAMETER	DESCRIPTION
action_name	The name of the action to execute. TYPE: str
arguments	The arguments to pass to the action, as represented in the JSON passed by the AI model. If the AI model passes a variable, you should ensure that it is passed as a RuntimeVariable. TYPE: dict[str, JSONType | RuntimeVariable]
returns	The return variables of the action, as a list of tuples. The first element of the tuple is the name of the return variable, and the second element is the type hint of the return variable. Note that we're not passing the actual return values, but the names of the variables that will hold the return values. TYPE: list[tuple[str, TypeOfType]]
stdout_output	The output of the action to stdout, if any. TYPE: str | None DEFAULT: None
stderr_output	The output of the action to stderr, if any. TYPE: str | None DEFAULT: None
modified_variables	The variables that were modified by the action. TYPE: list[str] | None DEFAULT: None
execution_successful	Whether the action was executed successfully. TYPE: bool DEFAULT: False

Source code in conatus/runtime/state.py

def add_json_instruction(
    self,
    action_name: str,
    arguments: dict[str, JSONType | RuntimeVariable],
    returns: list[tuple[str, TypeOfType]],
    *,
    stdout_output: str | None = None,
    stderr_output: str | None = None,
    modified_variables: list[str] | None = None,
    execution_successful: bool = False,
) -> None:
    """Add a JSON instruction to the state.

    !!! warning "This method does not modify the step number."

        This method does not modify the step number. If you want to add an
        instruction to a new step, you need to explicitly increment the
        step number by incrementing [`step_count`
        ][conatus.runtime.state.RuntimeState.step_count].

    Args:
        action_name: The name of the action to execute.
        arguments: The arguments to pass to the action, as represented in
            the JSON passed by the AI model. If the AI model passes a
            variable, you should ensure that it is passed as a
            [`RuntimeVariable`][conatus.runtime.state.RuntimeVariable].
        returns: The return variables of the action, as a list of tuples.
            The first element of the tuple is the name of the return
            variable, and the second element is the type hint of the
            return variable. Note that we're not passing the actual
            return values, but the names of the variables that will hold
            the return values.
        stdout_output: The output of the action to stdout, if any.
        stderr_output: The output of the action to stderr, if any.
        modified_variables: The variables that were modified by the action.
        execution_successful: Whether the action was executed successfully.
    """
    self.add_instruction(
        RuntimeJSONInstruction(
            action_name=action_name,
            arguments=arguments,
            returns=returns,
            stdout_output=stdout_output,
            stderr_output=stderr_output,
            modified_variables=modified_variables or [],
            execution_successful=execution_successful,
        )
    )

add_python_instruction

add_python_instruction(
    snippet: str,
    *,
    stdout_output: str | None = None,
    stderr_output: str | None = None,
    modified_variables: list[str] | None = None,
    execution_successful: bool = False
) -> None

Add a Python instruction to the state.

This method does not modify the step number.

This method does not modify the step number. If you want to add an instruction to a new step, you need to explicitly increment the step number by incrementing step_count .

PARAMETER	DESCRIPTION
snippet	The snippet, as passed by the AI model. TYPE: str
stdout_output	The output of the action to stdout, if any. TYPE: str | None DEFAULT: None
stderr_output	The output of the action to stderr, if any. TYPE: str | None DEFAULT: None
modified_variables	The variables that were modified by the action. TYPE: list[str] | None DEFAULT: None
execution_successful	Whether the action was executed successfully. TYPE: bool DEFAULT: False

Source code in conatus/runtime/state.py

def add_python_instruction(
    self,
    snippet: str,
    *,
    stdout_output: str | None = None,
    stderr_output: str | None = None,
    modified_variables: list[str] | None = None,
    execution_successful: bool = False,
) -> None:
    """Add a Python instruction to the state.

    !!! warning "This method does not modify the step number."

        This method does not modify the step number. If you want to add an
        instruction to a new step, you need to explicitly increment the
        step number by incrementing [`step_count`
        ][conatus.runtime.state.RuntimeState.step_count].

    Args:
        snippet: The snippet, as passed by the AI model.
        stdout_output: The output of the action to stdout, if any.
        stderr_output: The output of the action to stderr, if any.
        modified_variables: The variables that were modified by the action.
        execution_successful: Whether the action was executed successfully.
    """
    self.add_instruction(
        RuntimePythonInstruction(
            snippet,
            stdout_output=stdout_output,
            stderr_output=stderr_output,
            modified_variables=modified_variables or [],
            execution_successful=execution_successful,
        )
    )

add_instruction

add_instruction(
    instruction: (
        RuntimeJSONInstruction | RuntimePythonInstruction
    ),
) -> None

Add an instruction to the state.

You can use this method if you have your instruction already in the correct format; otherwise, feel free to use add_json_instruction or add_python_instruction.

This method does not modify the step number.

This method does not modify the step number. If you want to add an instruction to a new step, you need to explicitly increment the step number by incrementing step_count .

PARAMETER	DESCRIPTION
instruction	The instruction to add. TYPE: RuntimeJSONInstruction | RuntimePythonInstruction

Source code in conatus/runtime/state.py

def add_instruction(
    self, instruction: RuntimeJSONInstruction | RuntimePythonInstruction
) -> None:
    """Add an instruction to the state.

    You can use this method if you have your instruction already in the
    correct format; otherwise, feel free to use [`add_json_instruction`
    ][conatus.runtime.state.RuntimeState.add_json_instruction] or
    [`add_python_instruction`][conatus.runtime.state.RuntimeState.add_python_instruction].

    !!! warning "This method does not modify the step number."

        This method does not modify the step number. If you want to add an
        instruction to a new step, you need to explicitly increment the
        step number by incrementing [`step_count`
        ][conatus.runtime.state.RuntimeState.step_count].

    Args:
        instruction: The instruction to add.
    """
    self.last_step.instructions.append(instruction)

dump_variables

dump_variables() -> dict[str, ParamType]

Dump the variables.

This is useful to define a locals dictionary for the exec function.

RETURNS	DESCRIPTION
dict[str, ParamType]	A dictionary of variable names and their values.

Source code in conatus/runtime/state.py

def dump_variables(self) -> dict[str, ParamType]:
    """Dump the variables.

    This is useful to define a `locals` dictionary for the
    [`exec`][exec] function.

    Returns:
        A dictionary of variable names and their values.
    """
    return {var.name: var.value for var in self.variables.values()}

RuntimeStateStep

conatus.runtime.state.RuntimeStateStep dataclass

RuntimeStateStep(
    instructions: list[
        RuntimeJSONInstruction | RuntimePythonInstruction
    ],
    step_number: int,
)

Step taken by the agent.

During the lifecycle of an Runtime, the agent can take multiple steps. We tend to think of each step as a single action taken by the agent, but this is not the case: a single step can contain multiple instructions. AI providers, in particular, offer parallel tool calling, which allows for multiple instructions to be executed in a single step.

instructions instance-attribute

instructions: list[
    RuntimeJSONInstruction | RuntimePythonInstruction
]

The instructions for the step.

step_number instance-attribute

step_number: int

The step number.

stdout property

stdout: str

Get the stdout of the instructions.

RETURNS	DESCRIPTION
str	The stdout of the instructions. TYPE: str

stderr property

stderr: str

Get the stderr of the instructions.

RETURNS	DESCRIPTION
str	The stderr of the instructions. TYPE: str

code

code(*, include_failed: bool = False) -> str | None

Get the code of the instructions.

PARAMETER	DESCRIPTION
include_failed	Whether to include the code of failed instructions. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
str | None	The code of the instructions. If there are no instructions, or if there are no instructions that have been executed, or if include_failed is False and there are failed instructions, return None.

Source code in conatus/runtime/state.py

def code(self, *, include_failed: bool = False) -> str | None:
    """Get the code of the instructions.

    Args:
        include_failed: Whether to include the code of failed instructions.

    Returns:
        The code of the instructions. If there are no instructions, or if
            there are no instructions that have been executed, or if
            `include_failed` is `False` and there are failed instructions,
            return `None`.
    """
    if self.step_number == 0:
        append = (
            "# Step 0\n"
            if len(self.instructions) > 0
            else "# Step 0 -- No variables imported\n"
        )
    else:
        append = f"# Step {self.step_number}\n"

    instructions_code_lines: list[str] = []
    for instruction in self.instructions:
        if instruction.execution_successful:
            instructions_code_lines.append(instruction.code())
        elif include_failed:
            instructions_code_lines.append("# Failed to execute:")
            instructions_code_lines.extend(
                f"# {line_of_code}"
                for line_of_code in instruction.code().split("\n")
            )

    if instructions_code_lines or self.step_number == 0:
        return append + "\n".join(instructions_code_lines)
    return None

import_statements

import_statements() -> list[tuple[str, str]]

Get the import statements for the instructions.

RETURNS	DESCRIPTION
list[tuple[str, str]]	list[tuple[str, str]]: The import statements for the instructions.

Source code in conatus/runtime/state.py

def import_statements(self) -> list[tuple[str, str]]:
    """Get the import statements for the instructions.

    Returns:
        list[tuple[str, str]]: The import statements for the instructions.
    """
    all_imports: list[tuple[str, str]] = []
    for instruction in self.instructions:
        all_imports.extend(instruction.import_statements())
    return all_imports

---

1. Technically, the type hint is a list of either RuntimeJSONInstruction or RuntimePythonInstruction , because some of the attributes shared by both of these subclasses (e.g. stdout_output) have default values and cannot be expressed in the RuntimeInstruction superclass. ↩