State
RuntimeState¶
At its most basic level, this class essentially stores two things:
- A dictionary of the currently available variables. Each variable is
a
RuntimeVariableobject, which acts as a wrapper around the real thing. - Every step that has been executed. Each step is a
RuntimeStateStepobject, which contains a list ofRuntimeInstructionobjects. 1 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_core.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 agent's Runtime.
| PARAMETER | DESCRIPTION |
|---|---|
|
The starting variables.
TYPE:
|
|
The configuration of the run.
TYPE:
|
|
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:
|
| METHOD | DESCRIPTION |
|---|---|
import_statements_as_code |
Get the import statements as a Python call. |
code |
Print the |
update_vars_fallback_counter |
Update the |
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:
|
last_step |
Get the last step.
TYPE:
|
variables |
A dictionary of the variables in the state.
TYPE:
|
starting_variables |
A dictionary of the starting variables in the state.
TYPE:
|
vars_fallback_counter |
Counter of the number of variables with fallback type names. |
step_count |
The current step number.
TYPE:
|
steps |
The steps taken by the
TYPE:
|
pending_variable_updates |
The variables that were modified by an [instruction |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
termination_sentinel
class-attribute
instance-attribute
¶
termination_sentinel: bool = False
Sentinel to indicate that the program should terminate.
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
¶
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
¶
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:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
code
¶
code(*, include_failed: bool = False) -> str
Print the steps as a Python call.
Instructions coming from passive actions (see Action.passive
) are omitted, unless one of their
return variables is referenced by another recorded instruction. See
RuntimeStateStep.code
for details.
| PARAMETER | DESCRIPTION |
|---|---|
|
Whether to include the code of failed instructions.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
str
|
The steps as a Python call.
TYPE:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 |
|---|---|
|
The variables. We don't really care about the types of the values in the dictionary, since we only care about the keys.
TYPE:
|
|
The vars fallback counter. |
| RETURNS | DESCRIPTION |
|---|---|
Counter[str]
|
The updated counter. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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_core.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 |
|---|---|
|
The input type or the actual argument. Whether it is a type
or a variable depends on the value of
TYPE:
|
|
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:
|
| RETURNS | DESCRIPTION |
|---|---|
str
|
The processed type string. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 theinputsfields, which is meant to be a singleton, list or dictionary of TypeOfTypes. We return anOrderedDictofTypeOfTypes. 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 theoutputsfields, and perform the same operations as forinput_types.starting_variables: We pass the user input for thestarting_variablesfield, which can be a singleton, list or dictionary of ParamTypes. We return anOrderedDictofRuntimeVariables. 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 anexpected_input_typesdictionary 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_core.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_core.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_core.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 |
|---|---|
|
The input types, output types, or starting variables.
If
TYPE:
|
|
The kind of
TYPE:
|
|
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:
|
|
The maximum length of the string representation of a variable.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
OrderedDict[str, RuntimeVariable] | OrderedDict[str, TypeOfType]
|
If the kind_of_in_outputs is "starting_variables", we return an
|
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 packages/conatus_core/src/conatus_core/runtime/state.py
720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 | |
get_first_step
staticmethod
¶
get_first_step(
variables: OrderedDict[str, RuntimeVariable],
) -> RuntimeStateStep
Get the first step.
| PARAMETER | DESCRIPTION |
|---|---|
|
The variables.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RuntimeStateStep
|
The first step.
TYPE:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 |
|---|---|
|
The starting variables.
TYPE:
|
|
The maximum length of the string representation of a variable.
TYPE:
|
|
The counter of the number of variables of each type. |
|
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:
|
| RETURNS | DESCRIPTION |
|---|---|
OrderedDict[str, RuntimeVariable]
|
The variables. |
Counter[str]
|
The counter of the number of variables of each type. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
repr_at_step
¶
Get the repr (text, image) of a variable at a given step.
| PARAMETER | DESCRIPTION |
|---|---|
|
The name of the variable.
TYPE:
|
|
The step number.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
str
|
The text repr of the value of the variable at the given step.
TYPE:
|
str | None
|
str | None: The image repr of the value of the variable at the given step, if applicable. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 |
|---|---|
|
The result to add.
TYPE:
|
|
The name of the variable to store the result in. If not provided, we will generate a name.
TYPE:
|
|
Whether the variable is imported.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RuntimeVariable
|
The variable that was added.
TYPE:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 | |
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,
passive: 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 |
|---|---|
|
The name of the action to execute.
TYPE:
|
|
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
TYPE:
|
|
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:
|
|
The output of the action to stdout, if any.
TYPE:
|
|
The output of the action to stderr, if any.
TYPE:
|
|
The variables that were modified by the action. |
|
Whether the action was executed successfully.
TYPE:
|
|
Whether the instruction comes from a passive action. See
TYPE:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 |
|---|---|
|
The snippet, as passed by the AI model.
TYPE:
|
|
The output of the action to stdout, if any.
TYPE:
|
|
The output of the action to stderr, if any.
TYPE:
|
|
The variables that were modified by the action. |
|
Whether the action was executed successfully.
TYPE:
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
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 |
|---|---|
|
The instruction to add. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
RuntimeStateStep¶
conatus_core.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.
stdout
property
¶
stdout: str
Get the stdout of the instructions.
| RETURNS | DESCRIPTION |
|---|---|
str
|
The stdout of the instructions.
TYPE:
|
stderr
property
¶
stderr: str
Get the stderr of the instructions.
| RETURNS | DESCRIPTION |
|---|---|
str
|
The stderr of the instructions.
TYPE:
|
referenced_variable_names
staticmethod
¶
referenced_variable_names(
steps: Iterable[RuntimeStateStep],
) -> frozenset[str]
Collect the names of every variable referenced by the steps.
A variable counts as referenced when a RuntimeJSONInstruction
receives it
as a RuntimeVariable
argument, or when it
appears as an identifier in the snippet of a
RuntimePythonInstruction
.
This is deliberately conservative -- an identifier inside a string literal of a snippet counts as a reference. The set is used to decide whether a passive instruction can safely be dropped from the generated code, and a false positive only means keeping an extra step.
| PARAMETER | DESCRIPTION |
|---|---|
steps
|
The steps to scan for variable references.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
frozenset[str]
|
The names of the variables referenced by the steps. |
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
code
¶
code(
*,
include_failed: bool = False,
include_passive: bool = False,
referenced_variables: Collection[str] = ()
) -> str | None
Get the code of the instructions.
Instructions coming from passive actions (see Action.passive
) are omitted: they exist to inform
the model mid-run and have no place in a deterministic replay script.
A passive instruction is still emitted if one of its return variables
appears in referenced_variables, since dropping it would break the
generated code.
| PARAMETER | DESCRIPTION |
|---|---|
include_failed
|
Whether to include the code of failed instructions.
TYPE:
|
include_passive
|
Whether to include the code of instructions coming from passive actions.
TYPE:
|
referenced_variables
|
The names of the variables referenced by the
recorded instructions, typically computed over all steps with
TYPE:
|
| 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
|
Source code in packages/conatus_core/src/conatus_core/runtime/state.py
import_statements
¶
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 packages/conatus_core/src/conatus_core/runtime/state.py
-
Technically, the type hint is a list of either
RuntimeJSONInstructionorRuntimePythonInstruction, because some of the attributes shared by both of these subclasses (e.g.stdout_output) have default values and cannot be expressed in theRuntimeInstructionsuperclass. ↩