BaseAgent

conatus.agents.base

Base abstract agent class.

Ps module-attribute

Ps = ParamSpec('Ps', default=...)

Type variable for the task parameters (or inputs).

R module-attribute

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

Type variable for the task result.

Ps2 module-attribute

Ps2 = ParamSpec('Ps2')

Type variable for the task parameters (or inputs).

This type variable is used when the user creates a task with from_function , while Ps is used when the user creates a task with the constructor.

R2 module-attribute

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

Type variable for the task result.

This type variable is used when the user creates a task with from_function , while R is used when the user creates a task with the constructor.

BaseAgent

BaseAgent(func_or_description: Callable[Ps, R])

BaseAgent(
    func_or_description: Callable[Ps, R] | None = None,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: None = None,
    description: None = None,
    inputs: None = None,
    outputs: None = None,
    starting_variables: None = None,
    task: None = None,
    previous_runs: None = None,
    cost: None = None,
)

BaseAgent(
    func_or_description: str,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    description: None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

BaseAgent(
    func_or_description: None = None,
    /,
    *,
    description: str,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

BaseAgent(
    func_or_description: None = None,
    /,
    *,
    task: BaseTask[R, Ps],
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
    actions: None = None,
    inputs: None = None,
    outputs: None = None,
    description: None = None,
    starting_variables: None = None,
    config: None = None,
    model_config: None = None,
)

BaseAgent(
    func_or_description: (
        str | Callable[Ps, R] | None
    ) = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
)

Bases: ABC, Generic[R, Ps]

Abstract agent class.

PARAMETER	DESCRIPTION
func_or_description	The function or description of the agent. Note that if you pass a function, every other argument will be ignored, since we will use the function signature to infer the inputs and outputs of the agent. If you pass a string, you cannot pass a description argument. TYPE: str | Callable[Ps, R] | None DEFAULT: None
task	The task to use for the agent. This argument is required if func_or_description is not provided. Note that if you pass a task, every other argument will be ignored, since we will use the task definition to infer the inputs and outputs of the agent. TYPE: BaseTask[R, Ps] | None DEFAULT: None
description	The description of the agent. To be used only if neither func_or_description nor task are provided. TYPE: str | None DEFAULT: None
actions	The actions to use for the agent. To be used only if func_or_description is provided. TYPE: Sequence[RawAction] | ActionStarterPack | None DEFAULT: None
inputs	The inputs to use for the agent. Can be a list of type hints, a dictionary of type hints, a single type hint, or None. If you don't provide any inputs, or pass None, you will not be able to pass any variables to the task, or the recipe that you will generate from it. If you want to specify specific variables instead of type hints, you can use the starting_variables parameter instead. To be used only if func_or_description is provided. TYPE: TypeCollection | TypeOfType | None DEFAULT: None
outputs	The outputs to use for the agent. It should either be a dictionary of format {output_name: output_type}, a list of output types, a single output type, or None. If None, it will be up to the LLM to determine the expected output. If you want the task to return None, you need to provide an empty list or dictionary. If you pass a singleton type, it will be considered as the expected output of the task. To be used only if func_or_description is provided. TYPE: TypeCollection | type[R] | None DEFAULT: None
starting_variables	The variables to start with. Can be a list of variables, or a dictionary of variables, or None. This parameter is mutually exclusive with the inputs parameter. We don't allow single values for the starting variables in order to reduce ambiguity (e.g. if you pass a list of two items, should we consider it as a single variable or as two variables?) To be used only if func_or_description is provided. TYPE: list[ParamType] | dict[str, ParamType] | None DEFAULT: None
config	The configuration to use for the agent. To be used only if func_or_description is provided. TYPE: TaskConfig | None DEFAULT: None
model_config	The model configuration to use for the agent. To be used only if func_or_description is provided. TYPE: ModelConfig | SimpleDict | None DEFAULT: None
mock_responses	The mock responses to use for the agent. TYPE: list[AIResponse] | None DEFAULT: None
previous_runs	The previous runs to use for the agent. To be used only if task is provided. TYPE: list[RunData] | None DEFAULT: None
cost	The cost of the agent. To be used only if task is provided. TYPE: float | None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the arguments are invalid.

Source code in conatus/agents/base.py

def __init__(  # noqa: PLR0913, C901, PLR0915
    self,
    func_or_description: str | Callable[Ps, R] | None = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: list[ParamType]
    | dict[str, ParamType]
    | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
) -> None:
    """Initialize the agent.

    Args:
        func_or_description: The function or description of the agent. <br/>
            Note that if you pass a function, every other argument will be
            ignored, since we will use the function signature to infer the
            inputs and outputs of the agent. <br/> If you pass a string,
            you cannot pass a `description` argument.
        task: The task to use for the agent. This argument is required if
            `func_or_description` is not provided. <br/> Note that if you
            pass a task, every other argument will be ignored, since we
            will use the task definition to infer the inputs and outputs of
            the agent.
        description: The description of the agent. To be used only if
            neither `func_or_description` nor `task` are provided.
        actions: The actions to use for the agent. To be used only if
            `func_or_description` is provided.
        inputs: The inputs to use for the agent. Can be a list of type
            hints, a dictionary of type hints, a single type hint, or None.
            If you don't provide any inputs, or pass None, you will not be
            able to pass any variables to the task, or the recipe that you
            will generate from it. If you want to specify specific variables
            instead of type hints, you can use the `starting_variables`
            parameter instead. <br/> To be used only if
            `func_or_description` is provided.
        outputs: The outputs to use for the agent. It should either be a
            dictionary of format {output_name: output_type}, a list of
            output types, a single output type, or None. If None, it will
            be up to the LLM to determine the expected output. If you want
            the task to return `None`, you need to provide an empty list or
            dictionary. If you pass a singleton type, it will be considered
            as the expected output of the task. <br/> To be used only if
            `func_or_description` is provided.
        starting_variables: The variables to start with. Can be a list of
            variables, or a dictionary of variables, or None. This
            parameter is mutually exclusive with the `inputs` parameter.
            We don't allow single values for the starting variables in order
            to reduce ambiguity (e.g. if you pass a list of two items,
            should we consider it as a single variable or as two variables?)
            <br/> To be used only if `func_or_description` is provided.
        config: The configuration to use for the agent. To be used only if
            `func_or_description` is provided.
        model_config: The model configuration to use for the agent. To be
            used only if `func_or_description` is provided.
        mock_responses: The mock responses to use for the agent.
        previous_runs: The previous runs to use for the agent. To be used
            only if `task` is provided.
        cost: The cost of the agent. To be used only if `task` is provided.

    Raises:
        ValueError: If the arguments are invalid.
    """
    # Overload 1 (or 2): A decorator without arguments
    # We parse the callable and we move on
    if callable(func_or_description):
        self._two_step_initialization = False
        self._imported_via_decorator = True

    # Overload 2: Decorator with arguments
    # We assume that the user will feed a function later on and parse it
    elif (func_or_description, task, description) == (None, None, None):
        self._two_step_initialization = True
        self._imported_via_decorator = True
        self._uninitialized_args = UninitializedTaskArguments(
            actions=actions,
            config=config,
            model_config=model_config,
            agent_cls=None,
        )
        if any([inputs, outputs, starting_variables, previous_runs, cost]):
            msg = (
                "You cannot provide any of the following arguments "
                "when using the `@Agent` decorator: `inputs`, `outputs`, "
                "`starting_variables`, `previous_runs`, `cost`."
            )
            raise ValueError(msg)
        return

    # Overload 3: Manual definition of the task with a positional
    # description; we move on...
    elif isinstance(func_or_description, str):
        self._two_step_initialization = False
        self._imported_via_decorator = False
        if description is not None:
            msg = (
                "You cannot pass a description as a positional argument "
                "and as a keyword argument at the same time."
            )
            raise ValueError(msg)
        description = func_or_description

    # Overload 4: Manual definition of the agent with a keyword argument
    # for description.
    elif description is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False

    # Overload 5: Passing a task as argument
    elif task is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False
        self.definition = task.definition
        self.config = task.config
        self.mock_responses = mock_responses
        self.previous_runs = previous_runs or []
        self.model_config = model_config
        self.cost = cost or 0.0
        return

    else:
        msg = "Either `task` or `func_or_description` must be provided"
        raise ValueError(msg)

    # Finish the initialization of the task for overload 1, 3, 4
    self.mock_responses = mock_responses
    self.previous_runs = previous_runs or []
    self.model_config = model_config
    self.cost = cost or 0.0

    user_provided_config = config or {}
    self.config = ConsolidatedTaskConfig(user_provided_config)
    user_actions = process_user_provided_actions(
        actions or [],
        accepts_empty_actions_parameter=self.accepts_empty_actions_parameter,
    )
    user_provided_actions_keys = [action.name for action in user_actions]
    all_actions = add_additional_and_default_actions(user_actions)

    # Overload 1
    if self._imported_via_decorator:
        if inputs is not None or outputs is not None:
            msg = (
                "You cannot provide either `inputs` "
                "or `outputs` to the `@Task` decorator"
            )
            raise ValueError(msg)
        name, description, inputs, new_outputs = parse_callable_for_task(
            cast("Callable[Ps, R]", func_or_description)
        )
        self.definition = init_definition_decorated(
            name=name,
            description=description,
            inputs=inputs,
            outputs=new_outputs,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
        )

    # Overload 3 or 4
    else:
        if description is None:
            msg = (
                "You must pass a description for the agent."
                "By the way, you might find useful to use the `@Agent` "
                "decorator."
            )
            raise ValueError(msg)
        self.definition = init_definition_not_decorated(
            name="my_agent",
            description=description,
            inputs=inputs,
            outputs=outputs,
            starting_variables=starting_variables,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
            max_var_repr_len=self.config.max_var_repr_len,
            flatten_output_to_string=(outputs is None),
        )

accepts_empty_actions_parameter class-attribute instance-attribute

accepts_empty_actions_parameter: bool = True

Whether the task accepts an empty list of actions.

Set this to True if you're already providing some pre-defined actions in the constructor.

__call__

__call__(func: Callable[Ps2, R2]) -> BaseAgent[R2, Ps2]

__call__(*args: args, **kwargs: kwargs) -> R

__call__(
    func: Callable[Ps2, R2] | None = None,
    *args: args,
    **kwargs: kwargs
) -> BaseAgent[R2, Ps2] | R

Run the task.

This is a convenience method to run the task.

PARAMETER	DESCRIPTION
func	The function to use as the task. TYPE: Callable[Ps2, R2] | None DEFAULT: None
*args	The arguments to pass to the task. TYPE: args DEFAULT: ()
**kwargs	The keyword arguments to pass to the task. TYPE: kwargs DEFAULT: {}

RETURNS	DESCRIPTION
BaseAgent[R2, Ps2] | R	The result of the task.

Source code in conatus/agents/base.py

def __call__(  # type: ignore[misc]  # pyright: ignore[reportInconsistentOverload]
    self,
    func: Callable[Ps2, R2] | None = None,
    *args: Ps.args,
    **kwargs: Ps.kwargs,
) -> BaseAgent[R2, Ps2] | R:
    """Run the task.

    This is a convenience method to run the task.

    Args:
        func: The function to use as the task.
        *args: The arguments to pass to the task.
        **kwargs: The keyword arguments to pass to the task.

    Returns:
        The result of the task.
    """
    if func is not None:
        if callable(func):
            actions = None
            config = None
            model_config = None
            if getattr(self, "_uninitialized_args", None) is not None:
                actions = self._uninitialized_args.actions
                config = self._uninitialized_args.config
                model_config = self._uninitialized_args.model_config

            return self.from_function(
                func,
                actions=actions,
                config=config,
                model_config=model_config,
            )
        args = (func, *args)  # pyright: ignore[reportUnreachable]
    return self._run_from_call(*args, **kwargs)

from_function classmethod

from_function() -> (
    Callable[[Callable[Ps2, R2]], BaseAgent[R2, Ps2]]
)

from_function(
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None
) -> Callable[[Callable[Ps2, R2]], BaseAgent[R2, Ps2]]

from_function(fn: Callable[Ps2, R2]) -> BaseAgent[R2, Ps2]

from_function(
    fn: Callable[Ps2, R2],
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
) -> BaseAgent[R2, Ps2]

from_function(
    fn: Callable[Ps2, R2] | None = None,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
) -> (
    BaseAgent[R2, Ps2]
    | Callable[[Callable[Ps2, R2]], BaseAgent[R2, Ps2]]
)

Create a Agent from a function.

PARAMETER	DESCRIPTION
fn	The function to wrap in an agent. TYPE: Callable[Ps2, R2] | None DEFAULT: None
actions	The actions to add to the agent. TYPE: Sequence[RawAction] | ActionStarterPack | None DEFAULT: None
config	The configuration for the agent. TYPE: TaskConfig | None DEFAULT: None
model_config	The model configuration for the agent. TYPE: ModelConfig | SimpleDict | None DEFAULT: None

RETURNS	DESCRIPTION
BaseAgent[R2, Ps2] | Callable[[Callable[Ps2, R2]], BaseAgent[R2, Ps2]]	The BaseAgent or a callable that returns a BaseAgent. Note that a BaseAgent will never actually be returned, since it's an abstract class; a subclass of BaseAgent will always be returned.

Source code in conatus/agents/base.py

@classmethod
def from_function(
    cls,
    fn: Callable[Ps2, R2] | None = None,
    /,
    *,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
) -> BaseAgent[R2, Ps2] | Callable[[Callable[Ps2, R2]], BaseAgent[R2, Ps2]]:
    """Create a [`Agent`][conatus.agents.base.Agent] from a function.

    Args:
        fn: The function to wrap in an agent.
        actions: The actions to add to the agent.
        config: The configuration for the agent.
        model_config: The model configuration for the agent.

    Returns:
        The [`BaseAgent`][conatus.agents.base.BaseAgent] or a callable that
            returns a [`BaseAgent`][conatus.agents.base.BaseAgent]. Note
            that a [`BaseAgent`][conatus.agents.base.BaseAgent] will never
            actually be returned, since it's an abstract class; a subclass
            of [`BaseAgent`][conatus.agents.base.BaseAgent] will always be
            returned.
    """

    def wrapper(fn: Callable[Ps2, R2]) -> BaseAgent[R2, Ps2]:
        return cast(
            "BaseAgent[R2, Ps2]",
            cls(  # type: ignore[abstract]
                cast("Callable[Ps, R]", fn),
                actions=actions,
                config=config,
                model_config=model_config,
            ),
        )

    if fn is None:
        return wrapper
    return wrapper(fn)

filter_output_variables

filter_output_variables(
    state: RuntimeState,
) -> dict[str, RuntimeVariable]

Filter the output variables.

At the end of a run, the state should contain the output variables. This method allows to return a dictionary with only the output variables.

PARAMETER	DESCRIPTION
state	The state of the agent. TYPE: RuntimeState

RETURNS	DESCRIPTION
dict[str, RuntimeVariable]	The filtered output variables.

RAISES	DESCRIPTION
ValueError	If the state does not contain one of the output variables.

Source code in conatus/agents/base.py

def filter_output_variables(
    self, state: RuntimeState
) -> dict[str, RuntimeVariable]:
    """Filter the output variables.

    At the end of a run, the state should contain the output variables.
    This method allows to return a dictionary with only the output
    variables.

    Args:
        state: The state of the agent.

    Returns:
        The filtered output variables.

    Raises:
        ValueError: If the state does not contain one of the output
            variables.
    """
    candidate_output_variables = {
        k: v
        for k, v in state.variables.items()
        if k in self.definition.outputs
    }
    missing_output_variables = set(self.definition.outputs) - set(
        candidate_output_variables
    )
    if missing_output_variables:
        msg = (
            "The state does not contain the following output variables: "
            f"{missing_output_variables}"
        )
        raise ValueError(msg)
    return candidate_output_variables

run abstractmethod

run(
    runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData

Run the agent.

PARAMETER	DESCRIPTION
runtime	The runtime of the agent. TYPE: Runtime
run_writer	The writer to use for the run. This writer will be used to write down logs related to the run. TYPE: FileWriter | None

RETURNS	DESCRIPTION
CompleteRunData	The data of the run.

Source code in conatus/agents/base.py

@abstractmethod
def run(
    self, runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData:
    """Run the agent.

    Args:
        runtime: The runtime of the agent.
        run_writer: The writer to use for the run. This writer will be used
            to write down logs related to the run.

    Returns:
        The data of the run.
    """

Agent

Agent(func_or_description: Callable[Ps, R])

Agent(
    func_or_description: Callable[Ps, R] | None = None,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: None = None,
    description: None = None,
    inputs: None = None,
    outputs: None = None,
    starting_variables: None = None,
    task: None = None,
    previous_runs: None = None,
    cost: None = None,
)

Agent(
    func_or_description: str,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    description: None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

Agent(
    func_or_description: None = None,
    /,
    *,
    description: str,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

Agent(
    func_or_description: None = None,
    /,
    *,
    task: BaseTask[R, Ps],
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
    actions: None = None,
    inputs: None = None,
    outputs: None = None,
    description: None = None,
    starting_variables: None = None,
    config: None = None,
    model_config: None = None,
)

Agent(
    func_or_description: (
        str | Callable[Ps, R] | None
    ) = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
)

Bases: BaseAgent[R, Ps]

Default agent.

Source code in conatus/agents/base.py

def __init__(  # noqa: PLR0913, C901, PLR0915
    self,
    func_or_description: str | Callable[Ps, R] | None = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: list[ParamType]
    | dict[str, ParamType]
    | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
) -> None:
    """Initialize the agent.

    Args:
        func_or_description: The function or description of the agent. <br/>
            Note that if you pass a function, every other argument will be
            ignored, since we will use the function signature to infer the
            inputs and outputs of the agent. <br/> If you pass a string,
            you cannot pass a `description` argument.
        task: The task to use for the agent. This argument is required if
            `func_or_description` is not provided. <br/> Note that if you
            pass a task, every other argument will be ignored, since we
            will use the task definition to infer the inputs and outputs of
            the agent.
        description: The description of the agent. To be used only if
            neither `func_or_description` nor `task` are provided.
        actions: The actions to use for the agent. To be used only if
            `func_or_description` is provided.
        inputs: The inputs to use for the agent. Can be a list of type
            hints, a dictionary of type hints, a single type hint, or None.
            If you don't provide any inputs, or pass None, you will not be
            able to pass any variables to the task, or the recipe that you
            will generate from it. If you want to specify specific variables
            instead of type hints, you can use the `starting_variables`
            parameter instead. <br/> To be used only if
            `func_or_description` is provided.
        outputs: The outputs to use for the agent. It should either be a
            dictionary of format {output_name: output_type}, a list of
            output types, a single output type, or None. If None, it will
            be up to the LLM to determine the expected output. If you want
            the task to return `None`, you need to provide an empty list or
            dictionary. If you pass a singleton type, it will be considered
            as the expected output of the task. <br/> To be used only if
            `func_or_description` is provided.
        starting_variables: The variables to start with. Can be a list of
            variables, or a dictionary of variables, or None. This
            parameter is mutually exclusive with the `inputs` parameter.
            We don't allow single values for the starting variables in order
            to reduce ambiguity (e.g. if you pass a list of two items,
            should we consider it as a single variable or as two variables?)
            <br/> To be used only if `func_or_description` is provided.
        config: The configuration to use for the agent. To be used only if
            `func_or_description` is provided.
        model_config: The model configuration to use for the agent. To be
            used only if `func_or_description` is provided.
        mock_responses: The mock responses to use for the agent.
        previous_runs: The previous runs to use for the agent. To be used
            only if `task` is provided.
        cost: The cost of the agent. To be used only if `task` is provided.

    Raises:
        ValueError: If the arguments are invalid.
    """
    # Overload 1 (or 2): A decorator without arguments
    # We parse the callable and we move on
    if callable(func_or_description):
        self._two_step_initialization = False
        self._imported_via_decorator = True

    # Overload 2: Decorator with arguments
    # We assume that the user will feed a function later on and parse it
    elif (func_or_description, task, description) == (None, None, None):
        self._two_step_initialization = True
        self._imported_via_decorator = True
        self._uninitialized_args = UninitializedTaskArguments(
            actions=actions,
            config=config,
            model_config=model_config,
            agent_cls=None,
        )
        if any([inputs, outputs, starting_variables, previous_runs, cost]):
            msg = (
                "You cannot provide any of the following arguments "
                "when using the `@Agent` decorator: `inputs`, `outputs`, "
                "`starting_variables`, `previous_runs`, `cost`."
            )
            raise ValueError(msg)
        return

    # Overload 3: Manual definition of the task with a positional
    # description; we move on...
    elif isinstance(func_or_description, str):
        self._two_step_initialization = False
        self._imported_via_decorator = False
        if description is not None:
            msg = (
                "You cannot pass a description as a positional argument "
                "and as a keyword argument at the same time."
            )
            raise ValueError(msg)
        description = func_or_description

    # Overload 4: Manual definition of the agent with a keyword argument
    # for description.
    elif description is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False

    # Overload 5: Passing a task as argument
    elif task is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False
        self.definition = task.definition
        self.config = task.config
        self.mock_responses = mock_responses
        self.previous_runs = previous_runs or []
        self.model_config = model_config
        self.cost = cost or 0.0
        return

    else:
        msg = "Either `task` or `func_or_description` must be provided"
        raise ValueError(msg)

    # Finish the initialization of the task for overload 1, 3, 4
    self.mock_responses = mock_responses
    self.previous_runs = previous_runs or []
    self.model_config = model_config
    self.cost = cost or 0.0

    user_provided_config = config or {}
    self.config = ConsolidatedTaskConfig(user_provided_config)
    user_actions = process_user_provided_actions(
        actions or [],
        accepts_empty_actions_parameter=self.accepts_empty_actions_parameter,
    )
    user_provided_actions_keys = [action.name for action in user_actions]
    all_actions = add_additional_and_default_actions(user_actions)

    # Overload 1
    if self._imported_via_decorator:
        if inputs is not None or outputs is not None:
            msg = (
                "You cannot provide either `inputs` "
                "or `outputs` to the `@Task` decorator"
            )
            raise ValueError(msg)
        name, description, inputs, new_outputs = parse_callable_for_task(
            cast("Callable[Ps, R]", func_or_description)
        )
        self.definition = init_definition_decorated(
            name=name,
            description=description,
            inputs=inputs,
            outputs=new_outputs,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
        )

    # Overload 3 or 4
    else:
        if description is None:
            msg = (
                "You must pass a description for the agent."
                "By the way, you might find useful to use the `@Agent` "
                "decorator."
            )
            raise ValueError(msg)
        self.definition = init_definition_not_decorated(
            name="my_agent",
            description=description,
            inputs=inputs,
            outputs=outputs,
            starting_variables=starting_variables,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
            max_var_repr_len=self.config.max_var_repr_len,
            flatten_output_to_string=(outputs is None),
        )

run

run(
    runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData

Run the agent.

PARAMETER	DESCRIPTION
runtime	The runtime of the agent. TYPE: Runtime
run_writer	The writer to use for the run. This writer will be used to write down logs related to the run. TYPE: FileWriter | None

RETURNS	DESCRIPTION
CompleteRunData	The data of the run.

Source code in conatus/agents/base.py

@override
def run(
    self, runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData:
    """Run the agent.

    Args:
        runtime: The runtime of the agent.
        run_writer: The writer to use for the run. This writer will be used
            to write down logs related to the run.

    Returns:
        The data of the run.
    """
    logger.info("Running DefaultAgent")
    logger.info("Starting with the planning phase")
    planning_ai_interface = PlanningAIInterface(
        runtime=runtime,
        task_definition=self.definition,
        task_config=self.config,
        run_writer=run_writer,
    )
    planning_payload = planning_ai_interface.run()
    if planning_payload.cost != -1:
        self.cost += planning_payload.cost
    execution_ai_interface = ExecutionAIInterface(
        runtime=runtime,
        task_definition=self.definition,
        task_config=self.config,
        instructions=planning_payload.result,
        computer_use_mode=False,
        run_writer=run_writer,
    )
    execution_payload = execution_ai_interface.run()
    if execution_payload.cost != -1:
        self.cost += execution_payload.cost

    final_state = execution_payload.state
    results = self.filter_output_variables(final_state)
    return CompleteRunData(
        result=OrderedDict(results),
        recipe=Recipe(
            steps=list[RuntimeStateStep](final_state.steps.values()),
            actions=runtime.actions,
        ),
        complete=True,
        cost=self.cost,
    )

CUAgent

CUAgent(func_or_description: Callable[Ps, R])

CUAgent(
    func_or_description: Callable[Ps, R] | None = None,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: None = None,
    description: None = None,
    inputs: None = None,
    outputs: None = None,
    starting_variables: None = None,
    task: None = None,
    previous_runs: None = None,
    cost: None = None,
)

CUAgent(
    func_or_description: str,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    description: None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

CUAgent(
    func_or_description: None = None,
    /,
    *,
    description: str,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

CUAgent(
    func_or_description: None = None,
    /,
    *,
    task: BaseTask[R, Ps],
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
    actions: None = None,
    inputs: None = None,
    outputs: None = None,
    description: None = None,
    starting_variables: None = None,
    config: None = None,
    model_config: None = None,
)

CUAgent(
    func_or_description: (
        str | Callable[Ps, R] | None
    ) = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
)

Bases: BaseAgent[R, Ps]

Computer use agent.

Source code in conatus/agents/base.py

def __init__(  # noqa: PLR0913, C901, PLR0915
    self,
    func_or_description: str | Callable[Ps, R] | None = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: list[ParamType]
    | dict[str, ParamType]
    | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
) -> None:
    """Initialize the agent.

    Args:
        func_or_description: The function or description of the agent. <br/>
            Note that if you pass a function, every other argument will be
            ignored, since we will use the function signature to infer the
            inputs and outputs of the agent. <br/> If you pass a string,
            you cannot pass a `description` argument.
        task: The task to use for the agent. This argument is required if
            `func_or_description` is not provided. <br/> Note that if you
            pass a task, every other argument will be ignored, since we
            will use the task definition to infer the inputs and outputs of
            the agent.
        description: The description of the agent. To be used only if
            neither `func_or_description` nor `task` are provided.
        actions: The actions to use for the agent. To be used only if
            `func_or_description` is provided.
        inputs: The inputs to use for the agent. Can be a list of type
            hints, a dictionary of type hints, a single type hint, or None.
            If you don't provide any inputs, or pass None, you will not be
            able to pass any variables to the task, or the recipe that you
            will generate from it. If you want to specify specific variables
            instead of type hints, you can use the `starting_variables`
            parameter instead. <br/> To be used only if
            `func_or_description` is provided.
        outputs: The outputs to use for the agent. It should either be a
            dictionary of format {output_name: output_type}, a list of
            output types, a single output type, or None. If None, it will
            be up to the LLM to determine the expected output. If you want
            the task to return `None`, you need to provide an empty list or
            dictionary. If you pass a singleton type, it will be considered
            as the expected output of the task. <br/> To be used only if
            `func_or_description` is provided.
        starting_variables: The variables to start with. Can be a list of
            variables, or a dictionary of variables, or None. This
            parameter is mutually exclusive with the `inputs` parameter.
            We don't allow single values for the starting variables in order
            to reduce ambiguity (e.g. if you pass a list of two items,
            should we consider it as a single variable or as two variables?)
            <br/> To be used only if `func_or_description` is provided.
        config: The configuration to use for the agent. To be used only if
            `func_or_description` is provided.
        model_config: The model configuration to use for the agent. To be
            used only if `func_or_description` is provided.
        mock_responses: The mock responses to use for the agent.
        previous_runs: The previous runs to use for the agent. To be used
            only if `task` is provided.
        cost: The cost of the agent. To be used only if `task` is provided.

    Raises:
        ValueError: If the arguments are invalid.
    """
    # Overload 1 (or 2): A decorator without arguments
    # We parse the callable and we move on
    if callable(func_or_description):
        self._two_step_initialization = False
        self._imported_via_decorator = True

    # Overload 2: Decorator with arguments
    # We assume that the user will feed a function later on and parse it
    elif (func_or_description, task, description) == (None, None, None):
        self._two_step_initialization = True
        self._imported_via_decorator = True
        self._uninitialized_args = UninitializedTaskArguments(
            actions=actions,
            config=config,
            model_config=model_config,
            agent_cls=None,
        )
        if any([inputs, outputs, starting_variables, previous_runs, cost]):
            msg = (
                "You cannot provide any of the following arguments "
                "when using the `@Agent` decorator: `inputs`, `outputs`, "
                "`starting_variables`, `previous_runs`, `cost`."
            )
            raise ValueError(msg)
        return

    # Overload 3: Manual definition of the task with a positional
    # description; we move on...
    elif isinstance(func_or_description, str):
        self._two_step_initialization = False
        self._imported_via_decorator = False
        if description is not None:
            msg = (
                "You cannot pass a description as a positional argument "
                "and as a keyword argument at the same time."
            )
            raise ValueError(msg)
        description = func_or_description

    # Overload 4: Manual definition of the agent with a keyword argument
    # for description.
    elif description is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False

    # Overload 5: Passing a task as argument
    elif task is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False
        self.definition = task.definition
        self.config = task.config
        self.mock_responses = mock_responses
        self.previous_runs = previous_runs or []
        self.model_config = model_config
        self.cost = cost or 0.0
        return

    else:
        msg = "Either `task` or `func_or_description` must be provided"
        raise ValueError(msg)

    # Finish the initialization of the task for overload 1, 3, 4
    self.mock_responses = mock_responses
    self.previous_runs = previous_runs or []
    self.model_config = model_config
    self.cost = cost or 0.0

    user_provided_config = config or {}
    self.config = ConsolidatedTaskConfig(user_provided_config)
    user_actions = process_user_provided_actions(
        actions or [],
        accepts_empty_actions_parameter=self.accepts_empty_actions_parameter,
    )
    user_provided_actions_keys = [action.name for action in user_actions]
    all_actions = add_additional_and_default_actions(user_actions)

    # Overload 1
    if self._imported_via_decorator:
        if inputs is not None or outputs is not None:
            msg = (
                "You cannot provide either `inputs` "
                "or `outputs` to the `@Task` decorator"
            )
            raise ValueError(msg)
        name, description, inputs, new_outputs = parse_callable_for_task(
            cast("Callable[Ps, R]", func_or_description)
        )
        self.definition = init_definition_decorated(
            name=name,
            description=description,
            inputs=inputs,
            outputs=new_outputs,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
        )

    # Overload 3 or 4
    else:
        if description is None:
            msg = (
                "You must pass a description for the agent."
                "By the way, you might find useful to use the `@Agent` "
                "decorator."
            )
            raise ValueError(msg)
        self.definition = init_definition_not_decorated(
            name="my_agent",
            description=description,
            inputs=inputs,
            outputs=outputs,
            starting_variables=starting_variables,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
            max_var_repr_len=self.config.max_var_repr_len,
            flatten_output_to_string=(outputs is None),
        )

run

run(
    runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData

Run the agent.

PARAMETER	DESCRIPTION
runtime	The runtime of the agent. TYPE: Runtime
run_writer	The writer to use for the run. This writer will be used to write down logs related to the run. TYPE: FileWriter | None

RETURNS	DESCRIPTION
CompleteRunData	The data of the run.

Source code in conatus/agents/base.py

@override
def run(
    self, runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData:
    """Run the agent.

    Args:
        runtime: The runtime of the agent.
        run_writer: The writer to use for the run. This writer will be used
            to write down logs related to the run.

    Returns:
        The data of the run.
    """
    cu_interface = ComputerUseAIInterface(
        runtime=runtime,
        task_definition=self.definition,
        task_config=self.config,
        model_config={"stdout_mode": "preview"},
        run_writer=run_writer,
    )
    cu_payload = cu_interface.run()
    if cu_payload.cost != -1:
        self.cost += cu_payload.cost
    results = self.filter_output_variables(runtime.state)
    return CompleteRunData(
        result=OrderedDict(results),
        recipe=Recipe(
            steps=list[RuntimeStateStep](runtime.state.steps.values()),
            actions=runtime.actions,
        ),
        complete=True,
        cost=self.cost,
    )

MockAgent

MockAgent(func_or_description: Callable[Ps, R])

MockAgent(
    func_or_description: Callable[Ps, R] | None = None,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: None = None,
    description: None = None,
    inputs: None = None,
    outputs: None = None,
    starting_variables: None = None,
    task: None = None,
    previous_runs: None = None,
    cost: None = None,
)

MockAgent(
    func_or_description: str,
    /,
    *,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    description: None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

MockAgent(
    func_or_description: None = None,
    /,
    *,
    description: str,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: None = None,
    task: None = None,
    cost: None = None,
)

MockAgent(
    func_or_description: None = None,
    /,
    *,
    task: BaseTask[R, Ps],
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
    actions: None = None,
    inputs: None = None,
    outputs: None = None,
    description: None = None,
    starting_variables: None = None,
    config: None = None,
    model_config: None = None,
)

MockAgent(
    func_or_description: (
        str | Callable[Ps, R] | None
    ) = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: (
        list[ParamType] | dict[str, ParamType] | None
    ) = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
)

Bases: Agent[R, Ps]

Mock agent.

Source code in conatus/agents/base.py

def __init__(  # noqa: PLR0913, C901, PLR0915
    self,
    func_or_description: str | Callable[Ps, R] | None = None,
    /,
    *,
    task: BaseTask[R, Ps] | None = None,
    description: str | None = None,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | type[R] | None = None,
    starting_variables: list[ParamType]
    | dict[str, ParamType]
    | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    mock_responses: list[AIResponse] | None = None,
    previous_runs: list[RunData] | None = None,
    cost: float | None = None,
) -> None:
    """Initialize the agent.

    Args:
        func_or_description: The function or description of the agent. <br/>
            Note that if you pass a function, every other argument will be
            ignored, since we will use the function signature to infer the
            inputs and outputs of the agent. <br/> If you pass a string,
            you cannot pass a `description` argument.
        task: The task to use for the agent. This argument is required if
            `func_or_description` is not provided. <br/> Note that if you
            pass a task, every other argument will be ignored, since we
            will use the task definition to infer the inputs and outputs of
            the agent.
        description: The description of the agent. To be used only if
            neither `func_or_description` nor `task` are provided.
        actions: The actions to use for the agent. To be used only if
            `func_or_description` is provided.
        inputs: The inputs to use for the agent. Can be a list of type
            hints, a dictionary of type hints, a single type hint, or None.
            If you don't provide any inputs, or pass None, you will not be
            able to pass any variables to the task, or the recipe that you
            will generate from it. If you want to specify specific variables
            instead of type hints, you can use the `starting_variables`
            parameter instead. <br/> To be used only if
            `func_or_description` is provided.
        outputs: The outputs to use for the agent. It should either be a
            dictionary of format {output_name: output_type}, a list of
            output types, a single output type, or None. If None, it will
            be up to the LLM to determine the expected output. If you want
            the task to return `None`, you need to provide an empty list or
            dictionary. If you pass a singleton type, it will be considered
            as the expected output of the task. <br/> To be used only if
            `func_or_description` is provided.
        starting_variables: The variables to start with. Can be a list of
            variables, or a dictionary of variables, or None. This
            parameter is mutually exclusive with the `inputs` parameter.
            We don't allow single values for the starting variables in order
            to reduce ambiguity (e.g. if you pass a list of two items,
            should we consider it as a single variable or as two variables?)
            <br/> To be used only if `func_or_description` is provided.
        config: The configuration to use for the agent. To be used only if
            `func_or_description` is provided.
        model_config: The model configuration to use for the agent. To be
            used only if `func_or_description` is provided.
        mock_responses: The mock responses to use for the agent.
        previous_runs: The previous runs to use for the agent. To be used
            only if `task` is provided.
        cost: The cost of the agent. To be used only if `task` is provided.

    Raises:
        ValueError: If the arguments are invalid.
    """
    # Overload 1 (or 2): A decorator without arguments
    # We parse the callable and we move on
    if callable(func_or_description):
        self._two_step_initialization = False
        self._imported_via_decorator = True

    # Overload 2: Decorator with arguments
    # We assume that the user will feed a function later on and parse it
    elif (func_or_description, task, description) == (None, None, None):
        self._two_step_initialization = True
        self._imported_via_decorator = True
        self._uninitialized_args = UninitializedTaskArguments(
            actions=actions,
            config=config,
            model_config=model_config,
            agent_cls=None,
        )
        if any([inputs, outputs, starting_variables, previous_runs, cost]):
            msg = (
                "You cannot provide any of the following arguments "
                "when using the `@Agent` decorator: `inputs`, `outputs`, "
                "`starting_variables`, `previous_runs`, `cost`."
            )
            raise ValueError(msg)
        return

    # Overload 3: Manual definition of the task with a positional
    # description; we move on...
    elif isinstance(func_or_description, str):
        self._two_step_initialization = False
        self._imported_via_decorator = False
        if description is not None:
            msg = (
                "You cannot pass a description as a positional argument "
                "and as a keyword argument at the same time."
            )
            raise ValueError(msg)
        description = func_or_description

    # Overload 4: Manual definition of the agent with a keyword argument
    # for description.
    elif description is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False

    # Overload 5: Passing a task as argument
    elif task is not None:
        self._two_step_initialization = False
        self._imported_via_decorator = False
        self.definition = task.definition
        self.config = task.config
        self.mock_responses = mock_responses
        self.previous_runs = previous_runs or []
        self.model_config = model_config
        self.cost = cost or 0.0
        return

    else:
        msg = "Either `task` or `func_or_description` must be provided"
        raise ValueError(msg)

    # Finish the initialization of the task for overload 1, 3, 4
    self.mock_responses = mock_responses
    self.previous_runs = previous_runs or []
    self.model_config = model_config
    self.cost = cost or 0.0

    user_provided_config = config or {}
    self.config = ConsolidatedTaskConfig(user_provided_config)
    user_actions = process_user_provided_actions(
        actions or [],
        accepts_empty_actions_parameter=self.accepts_empty_actions_parameter,
    )
    user_provided_actions_keys = [action.name for action in user_actions]
    all_actions = add_additional_and_default_actions(user_actions)

    # Overload 1
    if self._imported_via_decorator:
        if inputs is not None or outputs is not None:
            msg = (
                "You cannot provide either `inputs` "
                "or `outputs` to the `@Task` decorator"
            )
            raise ValueError(msg)
        name, description, inputs, new_outputs = parse_callable_for_task(
            cast("Callable[Ps, R]", func_or_description)
        )
        self.definition = init_definition_decorated(
            name=name,
            description=description,
            inputs=inputs,
            outputs=new_outputs,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
        )

    # Overload 3 or 4
    else:
        if description is None:
            msg = (
                "You must pass a description for the agent."
                "By the way, you might find useful to use the `@Agent` "
                "decorator."
            )
            raise ValueError(msg)
        self.definition = init_definition_not_decorated(
            name="my_agent",
            description=description,
            inputs=inputs,
            outputs=outputs,
            starting_variables=starting_variables,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
            max_var_repr_len=self.config.max_var_repr_len,
            flatten_output_to_string=(outputs is None),
        )

run

run(
    runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData

Run the agent.

PARAMETER	DESCRIPTION
runtime	The runtime of the agent. TYPE: Runtime
run_writer	The writer to use for the run. This writer will be used to write down logs related to the run. TYPE: FileWriter | None

RETURNS	DESCRIPTION
CompleteRunData	The data of the run.

RAISES	DESCRIPTION
ValueError	If the mock responses are not set.

Source code in conatus/agents/base.py

@override
def run(
    self, runtime: Runtime, run_writer: FileWriter | None
) -> CompleteRunData:
    """Run the agent.

    Args:
        runtime: The runtime of the agent.
        run_writer: The writer to use for the run. This writer will be used
            to write down logs related to the run.

    Returns:
        The data of the run.

    Raises:
        ValueError: If the mock responses are not set.
    """
    _ = run_writer
    if self.mock_responses is None:
        msg = "Mock responses are not set"
        raise ValueError(msg)
    for mock_response in self.mock_responses:
        _, _ = runtime.run(
            mock_response.code_snippets, mock_response.tool_calls
        )
    results = self.filter_output_variables(runtime.state)
    return CompleteRunData(
        result=OrderedDict(results),
        recipe=Recipe(
            steps=list[RuntimeStateStep](runtime.state.steps.values())
        ),
    )

initialize_ai_interfaces_dictionary

initialize_ai_interfaces_dictionary(
    default: dict[str, type[BaseAIInterface]] | None = None,
) -> dict[str, type[BaseAIInterface]]

Get the AI interfaces.

This is a helper function to make the ai_interfaces field of the BaseAgent class compatible with the default_factory argument of the field function.

Feel free to use this function in your own classes; that way, you ensure that the ai_interfaces attribute of your class is always a dictionary, and that the underlying dictionary is not shared across instances of the same class.

XXXX Example

PARAMETER	DESCRIPTION
default	The default AI interfaces. TYPE: dict[str, type[BaseAIInterface]] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, type[BaseAIInterface]]	The AI interfaces.

Source code in conatus/agents/base.py

def initialize_ai_interfaces_dictionary(
    default: dict[str, type[BaseAIInterface]] | None = None,
) -> dict[str, type[BaseAIInterface]]:
    """Get the AI interfaces.

    This is a helper function to make the `ai_interfaces` field of the
    [`BaseAgent`][conatus.agents.base.BaseAgent] class compatible with
    the `default_factory` argument of the `field` function.

    Feel free to use this function in your own classes; that way, you ensure
    that the `ai_interfaces` attribute of your class is always a dictionary,
    and that the underlying dictionary is not shared across instances of
    the same class.

    XXXX Example

    Args:
        default: The default AI interfaces.

    Returns:
        The AI interfaces.
    """
    return default or {}