Parsing callables

conatus.utils.callable_parsing.actions ¶

Utilities for parsing actions.

add_termination_action ¶

add_termination_action(
    existing_actions: list[Action],
    expected_outputs: OrderedDict[
        str, tuple[TypeOfType, str]
    ],
) -> list[Action]

Add the termination action to the list of actions.

PARAMETER	DESCRIPTION
`existing_actions`	The list of actions to which we will add the termination action. TYPE: `list[Action]`
`expected_outputs`	The expected outputs of the task. The keys are the names of the outputs, and the values are a tuple containing the type of the output and its description. TYPE: `OrderedDict[str, tuple[TypeOfType, str]]`

RETURNS	DESCRIPTION
`list[Action]`	The list of actions with the termination action added.

Source code in conatus/utils/callable_parsing/actions.py

def add_termination_action(
    existing_actions: list[Action],
    expected_outputs: OrderedDict[str, tuple[TypeOfType, str]],
) -> list[Action]:
    """Add the termination action to the list of actions.

    Args:
        existing_actions: The list of actions to which we will add the
            termination action.
        expected_outputs: The expected outputs of the task. The keys are
            the names of the outputs, and the values are a tuple
            containing the type of the output and its description.

    Returns:
        The list of actions with the termination action added.
    """
    termination_action = terminate(expected_outputs)
    return [*existing_actions, termination_action]

process_user_provided_actions ¶

process_user_provided_actions(
    user_provided_actions: (
        Collection[RawAction]
        | Sequence[RawAction]
        | ActionStarterPack
    ),
    *,
    accepts_empty_actions_parameter: bool
) -> list[Action]

Normalize the values of the 'actions' parameter to a list of actions.

Note that this function does not add the default actions to the list of actions.

PARAMETER	DESCRIPTION
`user_provided_actions`	The list of actions, or action-like objects that need to be cast to actions. TYPE: `Collection[RawAction] \| Sequence[RawAction] \| ActionStarterPack`
`accepts_empty_actions_parameter`	Whether the task accepts an empty list of actions. TYPE: `bool`

RETURNS	DESCRIPTION
`list[Action]`	The list of actions instances.

RAISES	DESCRIPTION
`EmptyActionListError`	If the list of actions is empty and the task does not accept empty lists of actions.

Source code in conatus/utils/callable_parsing/actions.py

def process_user_provided_actions(
    user_provided_actions: Collection[RawAction]
    | Sequence[RawAction]
    | ActionStarterPack,
    *,
    accepts_empty_actions_parameter: bool,
) -> list[Action]:
    """Normalize the values of the 'actions' parameter to a list of actions.

    Note that this function does **not** add the default actions to the
    list of actions.

    Args:
        user_provided_actions: The list of actions, or action-like objects
            that need to be cast to actions.
        accepts_empty_actions_parameter: Whether the task accepts an empty
            list of actions.

    Returns:
        The list of actions instances.

    Raises:
        EmptyActionListError: If the list of actions is empty and the task
            does not accept empty lists of actions.
    """
    new_actions_list: list[Action] = []

    # We prepare the groundwork for the warning message in case at least
    # one action is not retrievable.
    at_least_one_not_retrievable = False
    msg = ""

    maybe_actions = (
        user_provided_actions.actions
        if isinstance(user_provided_actions, ActionStarterPack)
        else user_provided_actions
    )

    if not accepts_empty_actions_parameter and len(maybe_actions) == 0:
        msg = (
            "You provided an empty list of actions. "
            "Please provide at least one action."
        )
        raise EmptyActionListError(msg)

    for i, maybe_action in enumerate(maybe_actions):
        # Adding starter packs to the list of actions
        if isinstance(maybe_action, ActionStarterPack):
            new_actions_list.extend(maybe_action.actions)
            continue

        # Convert the maybe_action to an action
        new_action = convert_to_action(maybe_action)
        new_actions_list.append(new_action)

        # We add the warnings to the message in case at least one action is
        # not retrievable.
        new_msg = new_action.get_maybe_recipe_retrievability_warnings(
            i=i, with_first_message=(i == 0)
        )
        logger.debug(new_msg)
        msg += new_msg if new_msg is not None else ""
        at_least_one_not_retrievable = (
            True if new_msg is not None else at_least_one_not_retrievable
        )

    if at_least_one_not_retrievable:
        logger.warning(msg + "\n")

    # Add default actions
    return new_actions_list

add_additional_and_default_actions ¶

add_additional_and_default_actions(
    user_provided_actions: list[Action],
    additional_actions: (
        list[RawAction] | ActionStarterPack | None
    ) = None,
    *extra_actions: RawAction
) -> list[Action]

Add the additional and default actions to the list of actions.

PARAMETER	DESCRIPTION
`user_provided_actions`	The list of actions provided by the user. TYPE: `list[Action]`
`additional_actions`	The additional actions to add to the list of actions. We expect this to be the product of the `additional_actions` method. TYPE: `list[RawAction] \| ActionStarterPack \| None` DEFAULT: `None`
`extra_actions`	The extra actions to add to the list of actions. This is really a last resort, and should be used sparingly. TYPE: `RawAction` DEFAULT: `()`

RETURNS	DESCRIPTION
`list[Action]`	The list of actions.

Source code in conatus/utils/callable_parsing/actions.py

def add_additional_and_default_actions(
    user_provided_actions: list[Action],
    additional_actions: list[RawAction] | ActionStarterPack | None = None,
    *extra_actions: RawAction,
) -> list[Action]:
    """Add the additional and default actions to the list of actions.

    Args:
        user_provided_actions: The list of actions provided by the user.
        additional_actions: The additional actions to add to the list of
            actions. We expect this to be the product of the
            [`additional_actions`][conatus.tasks.base.BaseTask.additional_actions]
            method.
        extra_actions: The extra actions to add to the list of actions. This
            is really a last resort, and should be used sparingly.

    Returns:
        The list of actions.
    """
    # Adding the actions defined in the task
    task_defined_actions: list[Action] = []
    if additional_actions is not None:
        processed_additional_actions = (
            additional_actions.actions
            if isinstance(additional_actions, ActionStarterPack)
            else additional_actions
        )
        task_defined_actions.extend(
            convert_to_actions(processed_additional_actions)
        )
    task_defined_actions.extend(convert_to_actions(extra_actions))

    # Return all of that
    return [*user_provided_actions, *task_defined_actions]

conatus.utils.callable_parsing.callable_parsing ¶

Utilities for parsing callables.

Ps `module-attribute` ¶

Ps = ParamSpec('Ps')

Type variable for the task parameters (or inputs).

R `module-attribute` ¶

R = TypeVar('R', bound=ParamType)

Type variable for the task result.

parse_callable_for_task ¶

parse_callable_for_task(
    fn: Callable[Ps, R],
) -> tuple[
    str,
    str,
    OrderedDict[
        str, tuple[TypeOfType, ParamKind, bool, str]
    ],
    OrderedDict[str, tuple[TypeOfType, str]],
]

Parse the callable for the task.

RETURNS	DESCRIPTION
`str`	The name of the task.
`str`	The description of the task.
`dict[str, tuple[TypeOfType, ParamKind, bool, str]]`	The inputs of the task with their type, kind (positional, keyword, or positional-only), whether they are required, and the description of the parameter.
`dict[str, TypeOfType]`	The outputs of the task.

Source code in conatus/utils/callable_parsing/callable_parsing.py

def parse_callable_for_task(
    fn: Callable[Ps, R],
) -> tuple[
    str,
    str,
    OrderedDict[str, tuple[TypeOfType, ParamKind, bool, str]],
    OrderedDict[str, tuple[TypeOfType, str]],
]:
    """Parse the callable for the task.

    Returns:
        (str): The name of the task.
        (str): The description of the task.
        (dict[str, tuple[TypeOfType, ParamKind, bool, str]]): The inputs of the
            task with their type, kind (positional, keyword, or
            positional-only), whether they are required, and the description
            of the parameter.
        (dict[str, TypeOfType]): The outputs of the task.
    """
    function_info = extract_function_info(fn)
    outputs: OrderedDict[str, tuple[TypeOfType, str]]
    if not isinstance(function_info.returns, list):
        if function_info.returns.type_hint in {None, NoneType}:
            outputs = OrderedDict[str, tuple[TypeOfType, str]]()
        elif function_info.returns.type_hint == Any:
            outputs = OrderedDict({"result": (Any, "The result of the task.")})
        else:
            outputs = OrderedDict(
                {
                    "result": (
                        function_info.returns.type_hint,
                        function_info.returns.description or "",
                    )
                }
            )
    else:
        raw_outputs = cast(
            "OrderedDict[str, TypeOfType]",
            RuntimeState._process_list(  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]
                in_or_outputs=[
                    return_.type_hint for return_ in function_info.returns
                ],
                kind_of_in_outputs="output_types",
                max_var_repr_len=None,
                expected_input_types=None,
            )[0],
        )
        outputs = OrderedDict(
            {
                name: (output_type, return_info.description or "")
                for (name, output_type), return_info in zip(
                    raw_outputs.items(), function_info.returns, strict=True
                )
            }
        )
    inputs = OrderedDict(
        {
            param_name: (
                param_info.type_hint,
                param_info.kind,
                param_info.is_required,
                param_info.description or "<no description>",
            )
            for param_name, param_info in function_info.parameters.items()
        }
    )
    description = function_info.description or ""
    return fn.__name__, description, inputs, outputs

conatus.utils.callable_parsing.definition ¶

Assemble the definition of a task.

TypeCollection `module-attribute` ¶

TypeCollection = dict[str, TypeOfType] | list[TypeOfType]

Type alias representing a dictionary or list of type hints.

init_definition_not_decorated ¶

init_definition_not_decorated(
    *,
    name: str,
    description: str,
    all_actions: list[Action],
    inputs: TypeCollection | TypeOfType | None,
    outputs: TypeCollection | TypeOfType | None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ),
    user_provided_actions_keys: list[str],
    max_var_repr_len: int | None,
    flatten_output_to_string: bool = False
) -> TaskDefinition

Initialize the task definition if not processed via a decorator.

PARAMETER	DESCRIPTION
`name`	The name of the task. TYPE: `str`
`description`	The description of the task. TYPE: `str`
`all_actions`	The list of actions to add to the task. TYPE: `list[Action]`
`inputs`	The inputs of the task. TYPE: `TypeCollection \| TypeOfType \| None`
`outputs`	The outputs of the task. TYPE: `TypeCollection \| TypeOfType \| None`
`starting_variables`	The starting variables of the task. TYPE: `list[ParamType] \| dict[str, ParamType] \| None`
`user_provided_actions_keys`	The keys of the user-provided actions. TYPE: `list[str]`
`max_var_repr_len`	The maximum length of the variable representation. TYPE: `int \| None`
`flatten_output_to_string`	Whether to flatten the output to a string. Will only be respected if `outputs` is `None`. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`TaskDefinition`	The task definition.

Source code in conatus/utils/callable_parsing/definition.py

def init_definition_not_decorated(
    *,
    name: str,
    description: str,
    all_actions: list[Action],
    inputs: TypeCollection | TypeOfType | None,
    outputs: TypeCollection | TypeOfType | None,
    starting_variables: list[ParamType] | dict[str, ParamType] | None,
    user_provided_actions_keys: list[str],
    max_var_repr_len: int | None,
    flatten_output_to_string: bool = False,
) -> TaskDefinition:
    """Initialize the task definition if not processed via a decorator.

    Args:
        name: The name of the task.
        description: The description of the task.
        all_actions: The list of actions to add to the task.
        inputs: The inputs of the task.
        outputs: The outputs of the task.
        starting_variables: The starting variables of the task.
        user_provided_actions_keys: The keys of the user-provided actions.
        max_var_repr_len: The maximum length of the variable representation.
        flatten_output_to_string: Whether to flatten the output to a string.
            Will only be respected if `outputs` is `None`.

    Returns:
        The task definition.
    """
    if inputs is None and flatten_output_to_string:
        outputs = Any
    max_var_repr_len = max_var_repr_len or DEFAULT_MAX_RESULT_REPR_LEN
    processed_inputs_outputs = process_inputs_outputs(
        inputs=inputs,
        outputs=outputs,
        starting_variables=starting_variables,
        max_var_repr_len=max_var_repr_len,
    )
    all_actions = add_termination_action(
        existing_actions=all_actions,
        expected_outputs=processed_inputs_outputs["outputs_with_positions"],
    )
    return TaskDefinition(
        name=name,
        user_prompt=description,
        actions=OrderedDict({action.name: action for action in all_actions}),
        user_provided_actions_keys=user_provided_actions_keys,
        flatten_output_to_string=flatten_output_to_string,
        **processed_inputs_outputs,
    )

init_definition_decorated ¶

init_definition_decorated(
    *,
    name: str,
    description: str,
    all_actions: list[Action],
    inputs: OrderedDict[
        str, tuple[TypeOfType, ParamKind, bool, str]
    ],
    outputs: OrderedDict[str, tuple[TypeOfType, str]],
    user_provided_actions_keys: list[str]
) -> TaskDefinition

Initialize the task definition if processed via a decorator.

PARAMETER	DESCRIPTION
`name`	The name of the task. TYPE: `str`
`description`	The description of the task. TYPE: `str`
`all_actions`	The list of actions to add to the task. TYPE: `list[Action]`
`inputs`	The inputs of the task. TYPE: `OrderedDict[str, tuple[TypeOfType, ParamKind, bool, str]]`
`outputs`	The outputs of the task. TYPE: `OrderedDict[str, tuple[TypeOfType, str]]`
`user_provided_actions_keys`	The keys of the user-provided actions. TYPE: `list[str]`

RETURNS	DESCRIPTION
`TaskDefinition`	The task definition.

Source code in conatus/utils/callable_parsing/definition.py

def init_definition_decorated(
    *,
    name: str,
    description: str,
    all_actions: list[Action],
    inputs: OrderedDict[str, tuple[TypeOfType, ParamKind, bool, str]],
    outputs: OrderedDict[str, tuple[TypeOfType, str]],
    user_provided_actions_keys: list[str],
) -> TaskDefinition:
    """Initialize the task definition if processed via a decorator.

    Args:
        name: The name of the task.
        description: The description of the task.
        all_actions: The list of actions to add to the task.
        inputs: The inputs of the task.
        outputs: The outputs of the task.
        user_provided_actions_keys: The keys of the user-provided actions.

    Returns:
        The task definition.
    """
    all_actions = add_termination_action(
        existing_actions=all_actions,
        expected_outputs=outputs,
    )
    return TaskDefinition(
        name=name,
        user_prompt=description,
        actions=OrderedDict({action.name: action for action in all_actions}),
        user_provided_actions_keys=user_provided_actions_keys,
        inputs_with_positions=inputs,
        inputs_form=None,
        inputs_are_predefined=False,
        predefined_inputs=None,
        outputs_with_positions=outputs,
        outputs_form=None,
    )

conatus.utils.callable_parsing.parse_inputs ¶

Utilities for parsing inputs.

ProcessedInputsOutputs ¶

Bases: TypedDict

The processed inputs and outputs of a task.

Convenience TypedDict to simplify the handling of inputs and outputs in the constructor of BaseTask .

inputs_with_positions `instance-attribute` ¶

inputs_with_positions: OrderedDict[
    str, tuple[TypeOfType, ParamKind, bool, str]
]

The normalized inputs of the task.

The structure of the dictionary is as follows:

{
    "input_name": (input_type, ParamKind, is_required, description),
}

Where input_type is the type of the input, ParamKind is the kind of the input (positional, keyword, or positional-only), is_required is a boolean indicating whether the input is required, and description is the description of the input.

inputs_are_predefined `instance-attribute` ¶

inputs_are_predefined: bool

Whether the inputs are predefined.

inputs_form `instance-attribute` ¶

inputs_form: ExpectedInputForm

The form of the inputs of the task (singleton, list, or dictionary).

outputs_with_positions `instance-attribute` ¶

outputs_with_positions: OrderedDict[
    str, tuple[TypeOfType, str]
]

The normalized outputs of the task.

The structure of the dictionary is as follows:

{
    "output_name": (output_type, description),
}

Where output_type is the type of the output, and description is the description of the output.

outputs_form `instance-attribute` ¶

outputs_form: ExpectedOutputForm

The form of the outputs of the task (singleton, list, or dictionary).

predefined_inputs `instance-attribute` ¶

predefined_inputs: OrderedDict[str, RuntimeVariable] | None

The predefined inputs of the task.

get_inputs_with_positions ¶

get_inputs_with_positions(
    inputs: OrderedDict[str, TypeOfType],
    inputs_form: ExpectedInputForm,
) -> OrderedDict[
    str, tuple[TypeOfType, ParamKind, bool, str]
]

Get the inputs with their positions.

PARAMETER	DESCRIPTION
`inputs`	The inputs to the task. TYPE: `OrderedDict[str, TypeOfType]`
`inputs_form`	The form of the inputs. TYPE: `ExpectedInputForm`

RETURNS	DESCRIPTION
`OrderedDict[str, tuple[TypeOfType, ParamKind, bool, str]]`	The inputs with their positions.

Source code in conatus/utils/callable_parsing/parse_inputs.py

def get_inputs_with_positions(
    inputs: OrderedDict[str, TypeOfType],
    inputs_form: ExpectedInputForm,
) -> OrderedDict[str, tuple[TypeOfType, ParamKind, bool, str]]:
    """Get the inputs with their positions.

    Args:
        inputs: The inputs to the task.
        inputs_form: The form of the inputs.

    Returns:
        The inputs with their positions.
    """
    kind_mapping = {
        "none": ParamKind.VAR_POSITIONAL,
        "singleton": ParamKind.POSITIONAL_OR_KEYWORD,
        "list": ParamKind.VAR_POSITIONAL,
        "dict": ParamKind.VAR_KEYWORD,
    }
    return OrderedDict(
        {
            input_name: (input_type, kind_mapping[inputs_form], True, "")
            for input_name, input_type in inputs.items()
        }
    )

process_inputs_outputs ¶

process_inputs_outputs(
    inputs: (
        dict[str, TypeOfType]
        | list[TypeOfType]
        | TypeOfType
        | None
    ),
    outputs: (
        dict[str, TypeOfType]
        | list[TypeOfType]
        | TypeOfType
        | None
    ),
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ),
    *,
    max_var_repr_len: int
) -> ProcessedInputsOutputs

Process the inputs and outputs of the task.

PARAMETER	DESCRIPTION
`inputs`	The inputs to the task. TYPE: `dict[str, TypeOfType] \| list[TypeOfType] \| TypeOfType \| None`
`outputs`	The outputs of the task. TYPE: `dict[str, TypeOfType] \| list[TypeOfType] \| TypeOfType \| None`
`starting_variables`	The starting variables of the task. TYPE: `list[ParamType] \| dict[str, ParamType] \| None`
`max_var_repr_len`	The maximum length of the variable representation. TYPE: `int`

RETURNS	DESCRIPTION
`ProcessedInputsOutputs`	The normalized inputs and outputs of the task.

RAISES	DESCRIPTION
`ValueError`	If both `inputs` and `starting_variables` are provided.

Source code in conatus/utils/callable_parsing/parse_inputs.py

def process_inputs_outputs(
    inputs: dict[str, TypeOfType] | list[TypeOfType] | TypeOfType | None,
    outputs: dict[str, TypeOfType] | list[TypeOfType] | TypeOfType | None,
    starting_variables: list[ParamType] | dict[str, ParamType] | None,
    *,
    max_var_repr_len: int,
) -> ProcessedInputsOutputs:
    """Process the inputs and outputs of the task.

    Args:
        inputs: The inputs to the task.
        outputs: The outputs of the task.
        starting_variables: The starting variables of the task.
        max_var_repr_len: The maximum length of the variable representation.

    Returns:
        The normalized inputs and outputs of the task.

    Raises:
        ValueError: If both `inputs` and `starting_variables` are
            provided.
    """
    if inputs is not None and starting_variables is not None:
        msg = (
            "You provided both 'inputs' and 'starting_variables'. "
            "Please provide only one of them."
        )
        raise ValueError(msg)
    if starting_variables is not None:
        inputs_are_predefined = True
        predefined_inputs, _, inputs_form = (
            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,
            )
        )
        normalized_inputs = OrderedDict(
            {
                input_.name: input_.type_hint
                for input_ in predefined_inputs.values()
            }
        )

    else:
        normalized_inputs, _, inputs_form = (
            RuntimeState.process_unstructured_inputs_or_outputs(
                in_or_outputs=inputs,
                kind_of_in_outputs="input_types",
                max_var_repr_len=max_var_repr_len,
            )
        )
        inputs_are_predefined = False
        predefined_inputs = None
    normalized_outputs, _, outputs_form = (
        RuntimeState.process_unstructured_inputs_or_outputs(
            in_or_outputs=outputs,
            kind_of_in_outputs="output_types",
            max_var_repr_len=max_var_repr_len,
        )
    )
    inputs_with_positions = get_inputs_with_positions(
        inputs=normalized_inputs,
        inputs_form=inputs_form,
    )
    if normalized_outputs == OrderedDict({"result": Any}):
        outputs_with_positions: OrderedDict[str, tuple[TypeOfType, str]] = (
            OrderedDict(
                {
                    "result": (Any, "Put here any result you want to return."),
                }
            )
        )
    else:
        outputs_with_positions = OrderedDict(
            {
                name: (
                    output_type,
                    "Result of the following type: "
                    f"{process_typehint(output_type, 'str')}",
                )
                for name, output_type in normalized_outputs.items()
            }
        )
    return ProcessedInputsOutputs(
        inputs_with_positions=inputs_with_positions,
        inputs_are_predefined=inputs_are_predefined,
        inputs_form=inputs_form,
        outputs_with_positions=outputs_with_positions,
        outputs_form=outputs_form,
        predefined_inputs=predefined_inputs,
    )

extract_starting_variables ¶

extract_starting_variables(
    definition: TaskDefinition,
    config: ConsolidatedTaskConfig,
    *args: ParamType,
    **kwargs: ParamType
) -> OrderedDict[str, RuntimeVariable]

Extract the starting variables from the arguments passed Agent.run.

We perform some checks to make sure that the arguments are correct:

No inputs: * If the agent expects no inputs, we check that the user did not provide any inputs.
Singleton inputs: * If the agent expects a single input, we check that the user provided only one input. * If the user provided a keyword argument, we check that the keyword argument is the same as the auto-generated name of the input.
List inputs: * If the agent expects a list of inputs, we check that the user provided the correct number of inputs. * If the user provided keyword arguments, we check that the keyword arguments are the same as the auto-generated names of the inputs.
Dictionary inputs: * If the agent expects a dictionary of inputs, we check that the user provided the correct number of inputs. * If the user provided keyword arguments, we check that the keyword arguments are the same as the auto-generated names of the inputs.

No type checking

At this time, we do not perform any type checking on the inputs.

PARAMETER	DESCRIPTION
`definition`	The definition of the task. TYPE: `TaskDefinition`
`config`	The configuration of the task. TYPE: `ConsolidatedTaskConfig`
`args`	The arguments. TYPE: `ParamType` DEFAULT: `()`
`kwargs`	The keyword arguments. TYPE: `ParamType` DEFAULT: `{}`

RETURNS	DESCRIPTION
`OrderedDict[str, RuntimeVariable]`	The starting variables.

RAISES	DESCRIPTION
`WrongArgsToAgentRunError`	If the arguments are not correct.

Source code in conatus/utils/callable_parsing/parse_inputs.py

def extract_starting_variables(
    definition: TaskDefinition,
    config: ConsolidatedTaskConfig,
    *args: ParamType,
    **kwargs: ParamType,
) -> OrderedDict[str, RuntimeVariable]:
    """Extract the starting variables from the arguments passed Agent.run.

    We perform some checks to make sure that the arguments are correct:

    * No inputs:
        * If the agent expects no inputs, we check that the user did not
        provide any inputs.
    * Singleton inputs:
        * If the agent expects a single input, we check that the user
        provided only one input.
        * If the user provided a keyword argument, we check that the
        keyword argument is the same as the auto-generated name of the
        input.
    * List inputs:
        * If the agent expects a list of inputs, we check that the user
        provided the correct number of inputs.
        * If the user provided keyword arguments, we check that the
        keyword arguments are the same as the auto-generated names of
        the inputs.
    * Dictionary inputs:
        * If the agent expects a dictionary of inputs, we check that the
        user provided the correct number of inputs.
        * If the user provided keyword arguments, we check that the
        keyword arguments are the same as the auto-generated names of
        the inputs.

    !!! warning "No type checking"

        At this time, we do not perform any type checking on the inputs.

    Args:
        definition: The definition of the task.
        config: The configuration of the task.
        args: The arguments.
        kwargs: The keyword arguments.

    Returns:
        The starting variables.

    Raises:
        WrongArgsToAgentRunError: If the arguments are not correct.
    """
    # We use this variable in order to duplicate our code too much
    dict_of_param_type: OrderedDict[str, ParamType]

    match definition.inputs_form:
        case "none":
            if len(args) > 0 or len(kwargs) > 0:
                msg = (
                    "You provided inputs, but the agent does not expect "
                    "any inputs. Please remove the inputs."
                )
                raise WrongArgsToAgentRunError(msg)
            return OrderedDict()
        case "singleton":
            return extract_singleton_inputs(
                definition,
                config,
                *args,
                **kwargs,
            )
        case "list":
            if len(kwargs) > 0:
                msg = (
                    "You provided keyword arguments, but the agent expects "
                    "a list of inputs. Please provide the inputs as a list."
                )
                raise WrongArgsToAgentRunError(msg)
            if len(args) != len(definition.inputs):
                msg = (
                    "The number of inputs you provided does not match the "
                    "expected number of inputs. Please provide the correct "
                    "number of inputs."
                )
                raise WrongArgsToAgentRunError(msg)
            dict_of_param_type = OrderedDict(
                zip(definition.inputs, args, strict=True)
            )
        case "dict":
            if set(definition.inputs.keys()) != set(kwargs.keys()):
                msg = (
                    "The inputs you provided do not match the expected "
                    "inputs. Please provide the inputs with the"
                    " correct names.\n"
                    f"Expected inputs: {list(definition.inputs.keys())}\n"
                    f"Provided inputs: {list(kwargs.keys())}"
                )
                raise WrongArgsToAgentRunError(msg)
            dict_of_param_type = OrderedDict(kwargs)
        case _:
            param_prepared_list = [
                (k, v[1], v[2])
                for k, v in definition.inputs_with_positions.items()
            ]
            dict_of_param_type = validate_params(
                param_prepared_list, args, kwargs
            )

    return OrderedDict(
        {
            k: RuntimeVariable(
                name=k,
                value=v,
                type_hint=definition.inputs[k],
                imported=True,
                max_var_repr_len=config.max_var_repr_len,
                initial_step=0,
            )
            for k, v in dict_of_param_type.items()
        }
    )

extract_singleton_inputs ¶

extract_singleton_inputs(
    definition: TaskDefinition,
    config: ConsolidatedTaskConfig,
    *args: ParamType,
    **kwargs: ParamType
) -> OrderedDict[str, RuntimeVariable]

Extract the singleton inputs from the arguments passed Agent.run.

See the documentation of extract_starting_variables for more information.

PARAMETER	DESCRIPTION
`definition`	The definition of the task. TYPE: `TaskDefinition`
`config`	The configuration of the task. TYPE: `ConsolidatedTaskConfig`
`args`	The arguments. TYPE: `ParamType` DEFAULT: `()`
`kwargs`	The keyword arguments. TYPE: `ParamType` DEFAULT: `{}`

RETURNS	DESCRIPTION
`OrderedDict[str, RuntimeVariable]`	The singleton inputs.

RAISES	DESCRIPTION
`WrongArgsToAgentRunError`	If the arguments are not correct.

Source code in conatus/utils/callable_parsing/parse_inputs.py

def extract_singleton_inputs(
    definition: TaskDefinition,
    config: ConsolidatedTaskConfig,
    *args: ParamType,
    **kwargs: ParamType,
) -> OrderedDict[str, RuntimeVariable]:
    """Extract the singleton inputs from the arguments passed Agent.run.

    See the documentation of [`extract_starting_variables`
    ][conatus.utils.callable_parsing.parse_inputs.extract_starting_variables]
    for more information.

    Args:
        definition: The definition of the task.
        config: The configuration of the task.
        args: The arguments.
        kwargs: The keyword arguments.

    Returns:
        The singleton inputs.

    Raises:
        WrongArgsToAgentRunError: If the arguments are not correct.
    """
    arg_key = next(iter(definition.inputs.keys()))
    dict_of_param_type: OrderedDict[str, ParamType]
    if (len(args) + len(kwargs)) > 1:
        msg = (
            "You provided more than one input, but the agent "
            "expects only one input. Please provide only one input."
        )
        raise WrongArgsToAgentRunError(msg)
    if len(args) > 0:
        dict_of_param_type = OrderedDict({arg_key: args[0]})
    elif len(kwargs) > 0:
        if kwargs.keys() != definition.inputs.keys():
            msg = (
                "Your agent expects one input, "
                f"whose auto-generated name is {arg_key}.\n"
                "You provided an input with name"
                f" {next(iter(kwargs.keys()))}.\n"
                "Please provide the input with the correct name,"
                " or just pass the input without the keyword."
            )
            raise WrongArgsToAgentRunError(msg)
        dict_of_param_type = OrderedDict(kwargs)
    else:
        msg = (
            "You provided no inputs, but the agent expects one input. "
            "Please provide the input."
        )
        raise WrongArgsToAgentRunError(msg)

    return OrderedDict(
        {
            k: RuntimeVariable(
                name=k,
                value=v,
                type_hint=definition.inputs[k],
                imported=True,
                max_var_repr_len=config.max_var_repr_len,
                initial_step=0,
            )
            for k, v in dict_of_param_type.items()
        }
    )

conatus.utils.callable_parsing.parse_outputs ¶

Module for parsing the outputs of a callable.

convert_to_output_form ¶

convert_to_output_form(
    definition: TaskDefinition,
    results: dict[str, RuntimeVariable],
) -> object

Convert the results to the output form.

This method can be overridden by the user in the task class.

PARAMETER	DESCRIPTION
`definition`	The definition of the task. TYPE: `TaskDefinition`
`results`	The results of the task. TYPE: `dict[str, RuntimeVariable]`

RETURNS	DESCRIPTION
`object`	The results of the task in the output form.

RAISES	DESCRIPTION
`ValueError`	If the output is not present in the results.

Source code in conatus/utils/callable_parsing/parse_outputs.py

def convert_to_output_form(  # noqa: PLR0911
    definition: TaskDefinition,
    results: dict[str, RuntimeVariable],
) -> object:
    """Convert the results to the output form.

    This method can be overridden by the user in the task class.

    Args:
        definition: The definition of the task.
        results: The results of the task.

    Returns:
        The results of the task in the output form.

    Raises:
        ValueError: If the output is not present in the results.
    """
    match definition.outputs_form:
        case "singleton":
            val = next(iter(results.values())).value
            return str(val) if definition.flatten_output_to_string else val
        case "list":
            return tuple(v.value for v in results.values())
        case "dict":
            return {k: v.value for k, v in results.items()}
        case "none":
            return None
        case None:
            # We need to make sure that the arguments match the output
            # format expected by the task.
            res: list[ParamType] = []
            for k in definition.outputs:
                if k not in results:
                    msg = (
                        f"The output '{k}' is expected by the task, "
                        "but it is not present in the results."
                    )
                    raise ValueError(msg)
                res.append(results[k].value)
            if len(res) == 0:
                return None
            if len(res) == 1:
                return res[0]
            return tuple(res)

conatus.utils.callable_parsing.runtime ¶

Assemble the runtime of a task.

generate_state_and_runtime ¶

generate_state_and_runtime(
    *,
    task: BaseTask[ParamType, ...],
    config: None = None,
    definition: None = None,
    args: list[ParamType] | None = None,
    **kwargs: ParamType
) -> tuple[RuntimeState, Runtime]

generate_state_and_runtime(
    *,
    config: ConsolidatedTaskConfig,
    definition: TaskDefinition,
    task: None = None,
    args: list[ParamType] | None = None,
    **kwargs: ParamType
) -> tuple[RuntimeState, Runtime]

generate_state_and_runtime(
    *,
    task: BaseTask[ParamType, ...] | None = None,
    config: ConsolidatedTaskConfig | None = None,
    definition: TaskDefinition | None = None,
    args: list[ParamType] | None = None,
    **kwargs: ParamType
) -> tuple[RuntimeState, Runtime]

Generate the state and runtime for the task.

PARAMETER	DESCRIPTION
`task`	The task to run. TYPE: `BaseTask[ParamType, ...] \| None` DEFAULT: `None`
`definition`	The definition of the task. TYPE: `TaskDefinition \| None` DEFAULT: `None`
`config`	The configuration for the task. TYPE: `ConsolidatedTaskConfig \| None` DEFAULT: `None`
`args`	The arguments to pass to the task. TYPE: `list[ParamType] \| None` DEFAULT: `None`
`**kwargs`	The keyword arguments to pass to the task. TYPE: `ParamType` DEFAULT: `{}`

RETURNS	DESCRIPTION
`tuple[RuntimeState, Runtime]`	The state and runtime for the task.

RAISES	DESCRIPTION
`ValueError`	If task is None and both definition and config are None.

Source code in conatus/utils/callable_parsing/runtime.py

def generate_state_and_runtime(
    *,
    task: BaseTask[ParamType, ...] | None = None,
    config: ConsolidatedTaskConfig | None = None,
    definition: TaskDefinition | None = None,
    args: list[ParamType] | None = None,
    **kwargs: ParamType,
) -> tuple[RuntimeState, Runtime]:
    """Generate the state and runtime for the task.

    Args:
        task: The task to run.
        definition: The definition of the task.
        config: The configuration for the task.
        args: The arguments to pass to the task.
        **kwargs: The keyword arguments to pass to the task.

    Returns:
        The state and runtime for the task.

    Raises:
        ValueError: If task is None and both definition and config are None.
    """
    args = args or []
    if task is not None:
        definition = task.definition
        config = task.config
    elif definition is None or config is None:
        msg = "Either task or both definition and config must be provided."
        raise ValueError(msg)
    extracted_starting_variables = (
        extract_starting_variables(
            definition,
            config,
            *args,
            **kwargs,
        )
        if not definition.inputs_are_predefined
        or not definition.predefined_inputs
        else definition.predefined_inputs
    )
    state = RuntimeState(
        starting_variables=extracted_starting_variables,
        config=config,
    )
    runtime = Runtime(
        state=state,
        actions=definition.actions_list,
        config=config,
    )
    return state, runtime

Parsing callables

conatus.utils.callable_parsing.actions ¶

add_termination_action ¶

process_user_provided_actions ¶

add_additional_and_default_actions ¶

conatus.utils.callable_parsing.callable_parsing ¶

Ps module-attribute ¶

R module-attribute ¶

parse_callable_for_task ¶

conatus.utils.callable_parsing.definition ¶

TypeCollection module-attribute ¶

init_definition_not_decorated ¶

init_definition_decorated ¶

conatus.utils.callable_parsing.parse_inputs ¶

ProcessedInputsOutputs ¶

inputs_with_positions instance-attribute ¶

inputs_are_predefined instance-attribute ¶

inputs_form instance-attribute ¶

outputs_with_positions instance-attribute ¶

outputs_form instance-attribute ¶

predefined_inputs instance-attribute ¶

get_inputs_with_positions ¶

process_inputs_outputs ¶

extract_starting_variables ¶

extract_singleton_inputs ¶

conatus.utils.callable_parsing.parse_outputs ¶

convert_to_output_form ¶

conatus.utils.callable_parsing.runtime ¶

generate_state_and_runtime ¶

Ps `module-attribute` ¶

R `module-attribute` ¶

TypeCollection `module-attribute` ¶

inputs_with_positions `instance-attribute` ¶

inputs_are_predefined `instance-attribute` ¶

inputs_form `instance-attribute` ¶

outputs_with_positions `instance-attribute` ¶

outputs_form `instance-attribute` ¶

predefined_inputs `instance-attribute` ¶