conatus.tasks.base.BaseTask

BaseTask(func_or_description: Callable[Ps, R])

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

BaseTask(
    func_or_description: str,
    /,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    *,
    name: str,
    description: 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,
    agent_cls: type[BaseAgent[R, Ps]] | None = None,
    mock_responses: list[AIResponse] | None = None,
)

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

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

Bases: ABC, Generic[R, Ps]

Base class for Task implementations.

Not meant to be used directly

If you are an end-user, you should use the Task class instead. This class is meant for developers who want to create their own task classes.

The BaseTask 'contract'

If we use the design-by-contract approach, we can define the contract of the BaseTask class as follows:

Using generic parameters

If you want to benefit from the generic parameters, you just need to specific a specific type in the outputs parameter.

from conatus import Task

write_generics_poem = Task(
    description="Write me a poem about generics",
    name="write_generics_poem",
    outputs=str
)
# Type-checkers will infer 'result' to be of type 'str'
# result = write_generics_poem()

PARAMETER	DESCRIPTION
func_or_description	The function to use as the task, or the description of the task. 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 task. If you pass a string, you cannot pass a description argument. TYPE: str | Callable[Ps, R] | None DEFAULT: None
actions	The actions that are available to the task. TYPE: Sequence[RawAction] | ActionStarterPack | None DEFAULT: None
description	The description of the task. Note that if you a description in the func_or_description parameter, you cannot pass a description argument. We will throw an error if you do. TYPE: str | None DEFAULT: None
name	The name of the task. Even though it is optional in type, it is required in practice and we will throw an error if you don't provide it. TYPE: str | None DEFAULT: None
inputs	The inputs to the task. 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. TYPE: TypeCollection | TypeOfType | None DEFAULT: None
outputs	The expected output of the task. 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. TYPE: TypeCollection | TypeOfType | 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?) TYPE: list[ParamType] | dict[str, ParamType] | None DEFAULT: None
config	The task configuration, as a dictionary. Many other values can also be set in the keyword arguments, and will override the values set in the config dictionary. TYPE: TaskConfig | None DEFAULT: None
model_config	The model configuration, as a dictionary. This will override any model configuration set in the classes that will be used to generate the task. TYPE: ModelConfig | SimpleDict | None DEFAULT: None
agent_cls	The class of the agent to use for the task. TYPE: type[BaseAgent[R, Ps]] | None DEFAULT: None
mock_responses	The mock responses to use for the task. Useful for testing. TYPE: list[AIResponse] | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If the agent parameter is an instance of the agent class, since what should be passed is the class itself.
ValueError	If the description parameter is passed as a positional argument and as a keyword argument.
ValueError	If the name parameter is not passed in circumstances where it is required.

METHOD	DESCRIPTION
run	Run the task.
additional_actions	The additional actions to add to the task.
prepare_state_and_runtime	Prepare the state and runtime for the task.
postprocess_run_data	Postprocess the run data.
write_recipe	Write the recipe to the run writer.
__call__	Run the task.
from_function	Create a Task from a function.

ATTRIBUTE	DESCRIPTION
definition	The task definition. TYPE: TaskDefinition
accepts_empty_actions_parameter	Whether the task accepts an empty list of actions. TYPE: bool
running	Whether the task is currently running. TYPE: bool
config	The task configuration. TYPE: ConsolidatedTaskConfig
agent_cls	The class of the agent to use for the task. TYPE: type[BaseAgent[R, Ps]]
agent	The agent to use for the task. TYPE: BaseAgent[R, Ps]
task_runs	The runs of the task. TYPE: list[RunData]
actions	The actions to perform the task. TYPE: list[Action]
inputs	The inputs of the task. TYPE: OrderedDict[str, TypeOfType]
inputs_form	The form of the inputs of the task. TYPE: ExpectedInputForm | None
outputs	The outputs of the task. TYPE: OrderedDict[str, TypeOfType]
outputs_form	The form of the outputs of the task. TYPE: ExpectedOutputForm | None
starting_variables	The starting variables of the task. TYPE: dict[str, RuntimeVariable] | None
description	The description of the task. TYPE: str

Source code in conatus/tasks/base.py

def __init__(
    self,
    func_or_description: str | Callable[Ps, R] | None = None,
    /,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    *,
    description: str | None = None,
    name: str | None = None,
    inputs: TypeCollection | TypeOfType | None = None,
    outputs: TypeCollection | TypeOfType | type[R] | None = None,
    starting_variables: list[ParamType]
    | dict[str, ParamType]
    | None = None,
    config: TaskConfig | None = None,
    model_config: ModelConfig | SimpleDict | None = None,
    agent_cls: type[BaseAgent[R, Ps]] | None = None,
    mock_responses: list[AIResponse] | None = None,
) -> None:
    """Initialize the task.

    # Using generic parameters

    If you want to benefit from the generic parameters, you just need to
    specific a specific type in the `outputs` parameter.

    ```python
    from conatus import Task

    write_generics_poem = Task(
        description="Write me a poem about generics",
        name="write_generics_poem",
        outputs=str
    )
    # Type-checkers will infer 'result' to be of type 'str'
    # result = write_generics_poem()
    ```

    Args:
        func_or_description: The function to use as the task, or the
            description of the task. <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 task.
            <br/> If you pass a string, you cannot pass a `description`
            argument.
        actions: The actions that are available to the task.
        description: The description of the task. <br/> Note that if you
            a description in the `func_or_description` parameter, you
            cannot pass a `description` argument. We will throw an error
            if you do.
        name: The name of the task. Even though it is optional in type,
            it is required in practice and we will throw an error if you
            don't provide it.
        inputs: The inputs to the task. 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.
        outputs: The expected output of the task. 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.
        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?)
        config: The task configuration, as a dictionary. Many other
            values can also be set in the keyword arguments, and will
            override the values set in the `config` dictionary.
        model_config: The model configuration, as a dictionary. This will
            override any model configuration set in the classes that will
            be used to generate the task.
        agent_cls: The class of the agent to use for the task.
        mock_responses: The mock responses to use for the task. Useful for
            testing.

    Raises:
        TypeError: If the `agent` parameter is an instance of the agent
            class, since what should be passed is the class itself.
        ValueError: If the `description` parameter is passed as a
            positional argument and as a keyword argument.
        ValueError: If the `name` parameter is not passed in circumstances
            where it is required.
    """
    # 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, name, 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=agent_cls,  # type: ignore[arg-type] # pyright: ignore[reportArgumentType]
        )
        if any([inputs, outputs, starting_variables]):
            msg = (
                "You cannot provide any of the following arguments "
                "when using the `@Task` decorator: `inputs`, `outputs`, "
                "`starting_variables`."
            )
            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 task with description as a
    # keyword argument. We need both name and description here.
    else:
        self._two_step_initialization = False
        self._imported_via_decorator = False

    # Finish the initialization of the task for overload 1, 3, 4
    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, additional_actions=self.additional_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, outputs = parse_callable_for_task(
            cast("Callable[Ps, R]", func_or_description)
        )
        self.definition = init_definition_decorated(
            name=name,
            description=description,
            inputs=inputs,
            outputs=outputs,
            all_actions=all_actions,
            user_provided_actions_keys=user_provided_actions_keys,
        )

    # Overload 3 or 4
    else:
        if name is None or description is None:
            msg = (
                "You must pass a name and a description for the task."
                "By the way, you might find useful to use the `@Task` "
                "decorator."
            )
            raise ValueError(msg)
        self.definition = init_definition_not_decorated(
            name=name,
            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,
        )

    if isinstance(agent_cls, BaseAgent):
        msg = (
            "The 'agent_cls' parameter must be a class, not an instance. "
            "In other words, instead of doing "
            "`task(..., agent_cls=MyAgent())`,"
            " you should do `task(..., agent_cls=MyAgent)`."
        )
        raise TypeError(msg)

    self.agent_cls = agent_cls or self.agent_cls

    self.agent = self.agent_cls(  # type: ignore[abstract]
        task=self,
        mock_responses=mock_responses,
    )
    self.task_runs = []

definition instance-attribute

definition: TaskDefinition

The task definition.

It contains information about the task, such as the user prompt, the actions that can be used to perform the task, and the expected inputs and outputs of the task. It should be set by the constructor of the BaseTask subclass.

TaskDefinition can be subclassed to add additional information to the task; the downside, of course, is that lower classes such as BaseAgent will have to be updated to support the new information.

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.

running class-attribute instance-attribute

running: bool = False

Whether the task is currently running.

config instance-attribute

config: ConsolidatedTaskConfig = ConsolidatedTaskConfig(
    user_provided_config
)

The task configuration.

The user can provide a TaskConfig dictionary when instantiating a Task. The config attribute is a ConsolidatedTaskConfig instance, which contains the user-provided values, as well as the default values.

For more information, see ConsolidatedTaskConfig .

agent_cls instance-attribute

agent_cls: type[BaseAgent[R, Ps]] = agent_cls or agent_cls

The class of the agent to use for the task.

agent instance-attribute

agent: BaseAgent[R, Ps] = agent_cls(
    task=self, mock_responses=mock_responses
)

The agent to use for the task.

task_runs instance-attribute

task_runs: list[RunData] = []

The runs of the task.

actions property writable

actions: list[Action]

The actions to perform the task.

Note that you cannot modify this attribute while the task is running.

inputs property writable

inputs: OrderedDict[str, TypeOfType]

The inputs of the task.

Note that you cannot modify this attribute.

inputs_form property writable

inputs_form: ExpectedInputForm | None

The form of the inputs of the task.

Note that you cannot modify this attribute.

outputs property writable

outputs: OrderedDict[str, TypeOfType]

The outputs of the task.

Note that you cannot modify this attribute while the task is running.

outputs_form property writable

outputs_form: ExpectedOutputForm | None

The form of the outputs of the task.

Note that you cannot modify this attribute.

starting_variables property

starting_variables: dict[str, RuntimeVariable] | None

The starting variables of the task.

If the inputs are not predefined, this is set to None.

Note that you cannot modify this attribute.

description property

description: str

The description of the task.

run

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

Run the task.

Just treat this is a normal function call.

PARAMETER	DESCRIPTION
*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
R	The result of the task.

Source code in conatus/tasks/base.py

def run(self, *args: Ps.args, **kwargs: Ps.kwargs) -> R:
    """Run the task.

    Just treat this is a normal function call.

    Args:
        *args: The arguments to pass to the task.
        **kwargs: The keyword arguments to pass to the task.

    Returns:
        The result of the task.
    """
    run_writer = FileWriter()
    logger.debug("Running task %s", self.definition.user_prompt)
    state, runtime = generate_state_and_runtime(
        task=cast("BaseTask[ParamType, ...]", self),
        args=cast("list[ParamType]", args),  # pyright: ignore[reportInvalidCast]
        **kwargs,  # type: ignore[arg-type]
    )
    state, runtime = self.prepare_state_and_runtime(state, runtime)
    run_data = self.agent.run(runtime=runtime, run_writer=run_writer)
    run_data = self.postprocess_run_data(run_data)
    self.write_recipe(run_data, run_writer)
    logger.info("Cost of the task: %s", run_data.cost)
    self.task_runs.append(run_data)
    return cast(
        "R", convert_to_output_form(self.definition, run_data.result)
    )

additional_actions

additional_actions() -> list[RawAction] | ActionStarterPack

The additional actions to add to the task.

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

RETURNS	DESCRIPTION
list[RawAction] | ActionStarterPack	The additional actions to add to the task.

Source code in conatus/tasks/base.py

def additional_actions(self) -> list[RawAction] | ActionStarterPack:
    """The additional actions to add to the task.

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

    Returns:
        The additional actions to add to the task.
    """
    _ = self
    return []

prepare_state_and_runtime

prepare_state_and_runtime(
    state: RuntimeState, runtime: Runtime
) -> tuple[RuntimeState, Runtime]

Prepare the state and runtime for the task.

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

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

RETURNS	DESCRIPTION
tuple[RuntimeState, Runtime]	The state and runtime of the agent.

Source code in conatus/tasks/base.py

def prepare_state_and_runtime(
    self, state: RuntimeState, runtime: Runtime
) -> tuple[RuntimeState, Runtime]:
    """Prepare the state and runtime for the task.

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

    Args:
        state: The state of the agent.
        runtime: The runtime of the agent.

    Returns:
        The state and runtime of the agent.
    """
    _ = self
    return state, runtime

postprocess_run_data

postprocess_run_data(
    run_data: CompleteRunData,
) -> CompleteRunData

Postprocess the run data.

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

PARAMETER	DESCRIPTION
run_data	The run data to postprocess. TYPE: CompleteRunData

RETURNS	DESCRIPTION
CompleteRunData	The postprocessed run data.

Source code in conatus/tasks/base.py

def postprocess_run_data(
    self, run_data: CompleteRunData
) -> CompleteRunData:
    """Postprocess the run data.

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

    Args:
        run_data: The run data to postprocess.

    Returns:
        The postprocessed run data.
    """
    _ = self
    return run_data

write_recipe

write_recipe(
    run_data: CompleteRunData, run_writer: FileWriter
) -> None

Write the recipe to the run writer.

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

PARAMETER	DESCRIPTION
run_data	The run data to write the recipe to. TYPE: CompleteRunData
run_writer	The run writer to write the recipe to. TYPE: FileWriter

Source code in conatus/tasks/base.py

def write_recipe(
    self, run_data: CompleteRunData, run_writer: FileWriter
) -> None:
    """Write the recipe to the run writer.

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

    Args:
        run_data: The run data to write the recipe to.
        run_writer: The run writer to write the recipe to.
    """
    recipe_code = run_data.recipe.to_code(self.definition)
    if recipe_code is None:
        return
    run_writer.write(
        data=recipe_code,
        file_name="recipe.py",
        output_type=OutputType.OUT,
    )

__call__

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

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

__call__(
    func: Callable[Ps2, R2] | None = None,
    *args: args,
    **kwargs: kwargs
) -> BaseTask[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
BaseTask[R2, Ps2] | R	The result of the task.

Source code in conatus/tasks/base.py

def __call__(  # type: ignore[misc]  # pyright: ignore[reportInconsistentOverload]
    self,
    func: Callable[Ps2, R2] | None = None,
    *args: Ps.args,
    **kwargs: Ps.kwargs,
) -> BaseTask[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
            agent_cls = 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
                agent_cls = cast(
                    "type[BaseAgent[R2, Ps2]]",
                    self._uninitialized_args.agent_cls,
                )

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

from_function classmethod

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

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

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

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

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

Create a Task from a function.

PARAMETER	DESCRIPTION
fn	The function to wrap in a task. TYPE: Callable[Ps2, R2] | None DEFAULT: None
actions	The actions to add to the task. TYPE: Sequence[RawAction] | ActionStarterPack | None DEFAULT: None
agent_cls	The class of the agent to use for the task. TYPE: type[BaseAgent[R2, Ps2]] | None DEFAULT: None
config	The configuration for the task. TYPE: TaskConfig | None DEFAULT: None
model_config	The model configuration for the task. TYPE: ModelConfig | SimpleDict | None DEFAULT: None

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

Source code in conatus/tasks/base.py

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

    Args:
        fn: The function to wrap in a task.
        actions: The actions to add to the task.
        agent_cls: The class of the agent to use for the task.
        config: The configuration for the task.
        model_config: The model configuration for the task.

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

    def wrapper(fn: Callable[Ps2, R2]) -> BaseTask[R2, Ps2]:
        return cast(
            "BaseTask[R2, Ps2]",
            cls(
                cast("Callable[Ps, R]", fn),  # type: ignore[arg-type]
                actions=actions,
                agent_cls=agent_cls,  # pyright: ignore[reportArgumentType]
                config=config,
                model_config=model_config,
            ),
        )

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

Utilities

conatus.tasks.base.Ps module-attribute

Ps = ParamSpec('Ps')

Type variable for the task parameters (or inputs).

conatus.tasks.base.R module-attribute

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

Type variable for the task result.

conatus.tasks.base.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.

conatus.tasks.base.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.

conatus.tasks.base.RawAction module-attribute

RawAction: TypeAlias = (
    "Action | ActionBlueprint | ActionStarterPack | FunctionType"
)

A raw action that can be converted to an action.

Can be either:

• An instance of Action
• An instance of ActionBlueprint
• An instance of ActionStarterPack
• A function, as defined by the FunctionType type.

conatus.tasks.base.StrictlyTypedBaseTask

StrictlyTypedBaseTask(
    description: str,
    name: str = "my_task",
    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,
    agent_cls: type[BaseAgent[R, Ps]] | None = None,
    mock_responses: list[AIResponse] | None = None
)

Bases: BaseTask[R, Ps], ABC

A task that is strictly typed.

This is a convenience class to make type-checkers happy, but it is equivalent in functionality to the non-generic BaseTask .

The difference between this class and the BaseTask class is purely for type-checking purposes: it expects the actions to be instances or class instances of the Action class. It will not raise an error if that happens, though.

PARAMETER	DESCRIPTION
description	The description of the task. TYPE: str
name	The name of the task. TYPE: str DEFAULT: 'my_task'
actions	The actions to perform. TYPE: Sequence[RawAction] | ActionStarterPack | None DEFAULT: None
inputs	The inputs to the task. 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. TYPE: TypeCollection | TypeOfType | None DEFAULT: None
outputs	The expected output of the task. 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. 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?) TYPE: list[ParamType] | dict[str, ParamType] | None DEFAULT: None
config	The task configuration, as a dictionary. Many other values can also be set in the keyword arguments, and will override the values set in the config dictionary. TYPE: TaskConfig | None DEFAULT: None
model_config	The model configuration, as a dictionary. This will override any model configuration set in the classes that will be used to generate the task. TYPE: ModelConfig | SimpleDict | None DEFAULT: None
agent_cls	The class of the agent to use for the task. TYPE: type[BaseAgent[R, Ps]] | None DEFAULT: None
mock_responses	The mock responses to use for the task. Useful for testing. TYPE: list[AIResponse] | None DEFAULT: None

Source code in conatus/tasks/base.py

def __init__(
    self,
    description: str,
    name: str = "my_task",
    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,
    agent_cls: type[BaseAgent[R, Ps]] | None = None,
    mock_responses: list[AIResponse] | None = None,
) -> None:
    """Initialize the task.

    The difference between this class and the [`BaseTask`
    ][conatus.tasks.base.BaseTask] class is purely for type-checking
    purposes: it expects the actions to be instances or class instances
    of the [`Action`][conatus.actions.action.Action] class. It will
    not raise an error if that happens, though.

    Args:
        description: The description of the task.
        name: The name of the task.
        actions: The actions to perform.
        inputs: The inputs to the task. 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.
        outputs: The expected output of the task. 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.
        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?)
        config: The task configuration, as a dictionary. Many other
            values can also be set in the keyword arguments, and will
            override the values set in the `config` dictionary.
        model_config: The model configuration, as a dictionary. This will
            override any model configuration set in the classes that will
            be used to generate the task.
        agent_cls: The class of the agent to use for the task.
        mock_responses: The mock responses to use for the task. Useful for
            testing.
    """
    super().__init__(
        description=description,
        name=name,
        actions=actions,
        inputs=inputs,
        outputs=outputs,
        starting_variables=starting_variables,
        config=config,
        agent_cls=agent_cls,
        mock_responses=mock_responses,
        model_config=model_config,
    )