Tasks¶

conatus.tasks.decorator.task ¶

task() -> Callable[[Callable[Ps, R]], BaseTask[R, Ps]]

task(
    *,
    using: type[BaseTask[ParamType, ...]] = Task,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    agent_cls: (
        type[BaseAgent[ParamType, ...]] | None
    ) = None,
    config: TaskConfig | None = None
) -> Callable[[Callable[Ps, R]], BaseTask[R, Ps]]

task(fn: Callable[Ps, R]) -> BaseTask[R, Ps]

task(
    fn: Callable[Ps, R],
    *,
    using: type[BaseTask[ParamType, ...]] = Task,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    agent_cls: (
        type[BaseAgent[ParamType, ...]] | None
    ) = None,
    config: TaskConfig | None = None
) -> BaseTask[R, Ps]

task(
    fn: Callable[Ps, R] | None = None,
    *,
    using: type[BaseTask[ParamType, ...]] = Task,
    actions: (
        Sequence[RawAction] | ActionStarterPack | None
    ) = None,
    agent_cls: (
        type[BaseAgent[ParamType, ...]] | None
    ) = None,
    config: TaskConfig | None = None
) -> (
    BaseTask[R, Ps]
    | Callable[[Callable[Ps, R]], BaseTask[R, Ps]]
)

Create a Task from a function.

PARAMETER	DESCRIPTION
`fn`	The function to wrap in a task. TYPE: `Callable[Ps, R] \| None` DEFAULT: `None`
`actions`	The actions to add to the task. TYPE: `Sequence[RawAction] \| ActionStarterPack \| None` DEFAULT: `None`
`using`	If you want to mention a specific class of `BaseTask` to use, you can do so here. Doing this will be equivalent to calling this class's constructor with the same arguments. TYPE: `type[BaseTask[ParamType, ...]]` DEFAULT: `Task`
`agent_cls`	The class of the agent to use for the task. TYPE: `type[BaseAgent[ParamType, ...]] \| None` DEFAULT: `None`
`config`	The configuration for the task. TYPE: `TaskConfig \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`BaseTask[R, Ps] \| Callable[[Callable[Ps, R]], BaseTask[R, Ps]]`	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/decorator.py

def task(
    fn: Callable[Ps, R] | None = None,
    *,
    using: type[BaseTask[ParamType, ...]] = Task,
    actions: Sequence[RawAction] | ActionStarterPack | None = None,
    agent_cls: type[BaseAgent[ParamType, ...]] | None = None,
    config: TaskConfig | None = None,
) -> BaseTask[R, Ps] | Callable[[Callable[Ps, R]], BaseTask[R, Ps]]:
    """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.
        using: If you want to mention a specific class of [`BaseTask`
            ][conatus.tasks.base.BaseTask] to use, you can do so here. Doing
            this will be equivalent to calling this class's constructor with
            the same arguments.
        agent_cls: The class of the agent to use for the task.
        config: The 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[Ps, R]) -> BaseTask[R, Ps]:
        return using.from_function(
            fn,
            actions=actions,
            agent_cls=cast("type[BaseAgent[R, Ps]]", agent_cls),
            config=config,
        )

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

Default task¶

Task ¶

Task(func_or_description: Callable[Ps, R])

Task(
    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,
)

Task(
    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,
)

Task(
    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,
)

Task(
    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: BaseTask[R, Ps]

Default Task implementation.

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.

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.

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 .

task_runs `instance-attribute` ¶

task_runs: list[RunData] = []

The runs of the task.

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.

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.

call ¶

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

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

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

Source code in conatus/tasks/default.py

@override  # type: ignore[misc]
def __call__(  # pyright: ignore[reportInconsistentOverload]
    self,
    func: Callable[Ps2, R2] | None = None,
    *args: Ps.args,
    **kwargs: Ps.kwargs,
) -> Task[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):
            return cast(
                "Task[R2, Ps2]", super().__call__(func, *args, **kwargs)
            )
        args = (func, *args)  # pyright: ignore[reportUnreachable]
    return self.run(*args, **kwargs)

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,
    )

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)

Configuration¶

conatus.tasks.config.TaskConfig ¶

Bases: TypedDict

The configuration for a task, as passed by the user.

This is a dictionary that can be passed to the Task constructor:

from conatus import Task

task = Task(
    description="Actually, I don't want to do a task today.",
    name="no_task_for_you",
    config={
        "preferred_model": "openai:gpt-o1",
    }
)

Internally, this dictionary is consolidated with the default values, and the result is stored in the ConsolidatedTaskConfig class. If you want to write your own Task subclass, it is with that class that you will be dealing instead.

max_var_repr_len `instance-attribute` ¶

max_var_repr_len: NotRequired[int | None]

The maximum length of the string representation of a variable.

This is used to limit the length of the string representation of a variable in the agent's output. Think of this as a way to prevent your AI bill from exploding. This is particularly useful when you're dealing with variables with verbose string representations, like lists or dictionaries.

Possible values are:

None: No limit (dangerous!).
0: None, the variable will be represented as ....
> 0: Length of the string representation of the variable.

If not passed, defaults to ConsolidatedTaskConfig.max_var_repr_len , which is 300.

max_result_repr_len `instance-attribute` ¶

max_result_repr_len: NotRequired[int | None]

The maximum length of the string representation of a result.

It differs from TaskConfig.max_var_repr_len in a subtle but important way:

max_var_repr_len is all about ensuring that the quick repr of each variable in the state does not amount to a very lengthy text. In other words, it is designed for repetitive use in the conversation.

max_result_repr_len focuses on communicating the result of a tool call. In general, such tool responses will only focus on one or two results at a time, allowing it to be much longer than the quick repr of a variable.

preferred_provider `instance-attribute` ¶

preferred_provider: NotRequired[ProviderName | str]

The preferred AI provider to use, if any.

You do NOT have to use this attribute

Task and preloaded tasks never require this attribute, but figure it out on their own. This is purely meant for the user to specify which provider they want to use.

Possible values are:

"openai": OpenAI.

In general, BaseTask and BaseAgent implementations should strive to honor this attribute when deciding which provider to use, but it is not required.

If not passed, defaults to ConsolidatedTaskConfig.preferred_provider , which is "openai".

preferred_model `instance-attribute` ¶

preferred_model: NotRequired[BaseAIModel | ModelName | str]

The default model to use, if any.

You do NOT have to use this attribute

Task and preloaded tasks never require this attribute, but figure it out on their own.

This attribute is offered to the user so that they have the option to specify specifically which model they want to use. This is useful when, for instance, a new model is released and the user wants to use it.

In general, BaseTask and BaseAgent implementations should strive to honor this attribute when deciding which model to use, but it is not required.

The ideal format is a string that can be parsed by the ModelName type, which is something like "openai:gpt-4o".

The user can also pass an instance of a BaseAIModel if they want to instantiate a model with specific parameters.

If not passed, defaults to ConsolidatedTaskConfig.preferred_model , which is "openai:gpt-4o".

stream `instance-attribute` ¶

stream: NotRequired[bool]

Whether to stream the response or not.

Note that this should be honored by the AI model, but it is not guaranteed.

max_turns `instance-attribute` ¶

max_turns: NotRequired[int]

The maximum number of turns to run.

This is used to limit the number of turns that the agent can take. If the agent takes more turns than this, it will be stopped.

If not passed, defaults to ConsolidatedTaskConfig.max_turns , which is 10.

Type-checker friendly task¶

conatus.tasks.default.StrictlyTypedTask ¶

StrictlyTypedTask(
    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: StrictlyTypedBaseTask[R, Ps]

A strictly typed Task implementation.

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,
    )

Tasks¶

conatus.tasks.decorator.task ¶

Default task¶

Task ¶

Using generic parameters¶

definition instance-attribute ¶

config instance-attribute ¶

task_runs instance-attribute ¶

accepts_empty_actions_parameter class-attribute instance-attribute ¶

running class-attribute instance-attribute ¶

actions property writable ¶

inputs property writable ¶

inputs_form property writable ¶

outputs property writable ¶

outputs_form property writable ¶

starting_variables property ¶

description property ¶

__call__ ¶

run ¶

additional_actions ¶

prepare_state_and_runtime ¶

postprocess_run_data ¶

write_recipe ¶

from_function classmethod ¶

Configuration¶

conatus.tasks.config.TaskConfig ¶

max_var_repr_len instance-attribute ¶

max_result_repr_len instance-attribute ¶

preferred_provider instance-attribute ¶

preferred_model instance-attribute ¶

stream instance-attribute ¶

max_turns instance-attribute ¶

Type-checker friendly task¶

conatus.tasks.default.StrictlyTypedTask ¶

definition `instance-attribute` ¶

config `instance-attribute` ¶

task_runs `instance-attribute` ¶

accepts_empty_actions_parameter `class-attribute` `instance-attribute` ¶

running `class-attribute` `instance-attribute` ¶

actions `property` `writable` ¶

inputs `property` `writable` ¶

inputs_form `property` `writable` ¶

outputs `property` `writable` ¶

outputs_form `property` `writable` ¶

starting_variables `property` ¶

description `property` ¶

call ¶

from_function `classmethod` ¶

max_var_repr_len `instance-attribute` ¶

max_result_repr_len `instance-attribute` ¶

preferred_provider `instance-attribute` ¶

preferred_model `instance-attribute` ¶

stream `instance-attribute` ¶

max_turns `instance-attribute` ¶