motleycrew.tasks.simple

Functions

compose_simple_task_prompt_with_dependencies(...)

Compose a prompt for a simple task with upstream dependencies.

Classes

SimpleTask(crew, description[, name, agent, ...])

Simple task class.

SimpleTaskUnit(*[, status, output, ...])

Task unit for a simple task.
motleycrew.tasks.simple.compose_simple_task_prompt_with_dependencies(description: str, upstream_task_units: List[TaskUnit], prompt_template_with_upstreams: PromptTemplate, default_task_name: str = 'Unnamed task') → str

Compose a prompt for a simple task with upstream dependencies.

Parameters:

  • description – Description of the task, to be included in the prompt.

  • upstream_task_units – List of upstream task units whose results should be used.

  • prompt_template_with_upstreams – Prompt template to use for generating the prompt if the task has upstream dependencies. Otherwise, just the description is used. The template must have input variables ‘description’ and ‘upstream_results’.

  • default_task_name – Name to use for task units that don’t have a name attribute.

class motleycrew.tasks.simple.SimpleTaskUnit(*, status: str = 'pending', output: Any | None = None, name: str, prompt: str, additional_params: dict[str, Any] | None = None)

Bases: TaskUnit

Task unit for a simple task.

name

Name of the task unit.

Type:

str

prompt

Prompt for the task unit.

Type:

str

additional_params

Additional parameters for the task unit (can be used by the agent).

Type:

Optional[dict[str, Any]]

name: str
prompt: str
additional_params: dict[str, Any] | None
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'additional_params': FieldInfo(annotation=Union[dict[str, Any], NoneType], required=False, default=None), 'name': FieldInfo(annotation=str, required=True), 'output': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None), 'prompt': FieldInfo(annotation=str, required=True), 'status': FieldInfo(annotation=str, required=False, default='pending')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class motleycrew.tasks.simple.SimpleTask(crew: MotleyCrew, description: str, name: str | None = None, agent: MotleyAgentAbstractParent | None = None, tools: Sequence[MotleyTool] | None = None, additional_params: dict[str, Any] | None = None, prompt_template_with_upstreams: PromptTemplate = PromptTemplate(input_variables=['description', 'upstream_results'], input_types={}, partial_variables={}, template='{description}\n\nYou must use the results of these upstream tasks:\n\n{upstream_results}\n'))

Bases: Task

Simple task class.

A simple task consists of a description and an agent that can execute the task. It produces a single task unit with a prompt based on the description and the results of upstream tasks.

The task is considered done when the task unit is completed.

__init__(crew: MotleyCrew, description: str, name: str | None = None, agent: MotleyAgentAbstractParent | None = None, tools: Sequence[MotleyTool] | None = None, additional_params: dict[str, Any] | None = None, prompt_template_with_upstreams: PromptTemplate = PromptTemplate(input_variables=['description', 'upstream_results'], input_types={}, partial_variables={}, template='{description}\n\nYou must use the results of these upstream tasks:\n\n{upstream_results}\n'))

Initialize the simple task.

Parameters:

  • crew – Crew to which the task belongs.

  • description – Description of the task.

  • name – Name of the task (will be used as the name of the task unit).

  • agent – Agent to execute the task.

  • tools – Tools to use for the task.

  • additional_params – Additional parameters for the task.

  • prompt_template_with_upstreams – Prompt template to use for generating the prompt if the task has upstream dependencies. Otherwise, just the description is used. The template must have input variables ‘description’ and ‘upstream_results’.

on_unit_completion(unit: SimpleTaskUnit) → None

Handle completion of the task unit.

Sets the task as done and stores the output of the task unit.

Parameters:

unit – Task unit that has completed.

get_next_unit() → SimpleTaskUnit | None

Get the next task unit to run.

If all upstream tasks are done, returns a task unit with the prompt based on the description and the results of the upstream tasks. Otherwise, returns None (the task is not ready to run yet).

Returns:

Task unit to run if the task is ready, None otherwise.

get_worker(tools: List[MotleyTool] | None) → MotleyAgentAbstractParent

Get the worker for the task.

If the task is associated with a crew and an agent, returns the agent. Otherwise, raises an exception.

Parameters:

tools – Additional tools to add to the agent.

Returns:

Agent to run the task unit.