camel.tasks package

Contents

camel.tasks package#

Submodules#

camel.tasks.task module#

class camel.tasks.task.Task(*, content: str, id: str = '', state: TaskState = TaskState.OPEN, type: str | None = None, parent: Task | None = None, subtasks: List[Task] = [], result: str | None = '', failure_count: int = 0, additional_info: str | None = None)[source]#

Bases: BaseModel

Task is specific assignment that can be passed to a agent.

content#

string content for task.

Type:

str

id#

An unique string identifier for the task. This should

Type:

str

ideally be provided by the provider/model which created the task.
state#

The state which should be OPEN, RUNNING, DONE or DELETED.

Type:

camel.tasks.task.TaskState

type#

task type

Type:

str | None

parent#

The parent task, None for root task.

Type:

camel.tasks.task.Task | None

subtasks#

The childrent sub-tasks for the task.

Type:

List[camel.tasks.task.Task]

result#

The answer for the task.

Type:

str | None

add_subtask(task: Task)[source]#
additional_info: str | None#
compose(agent: ChatAgent, template: TextPrompt = 'As a Task composer with the role of {role_name}, your objective is to gather result from all sub tasks to get the final answer.\nThe root task is:\n\n{content}\n\nThe additional information of the task is:\n\n{additional_info}\n\nThe related tasks result and status:\n\n{other_results}\n\nso, the final answer of the root task is: \n', result_parser: Callable[[str], str] | None = None)[source]#

compose task result by the sub-tasks.

Parameters:
  • agent (ChatAgent) – An agent that used to compose the task result.

  • template (TextPrompt, optional) – The prompt template to compose task. If not provided, the default template will be used.

  • result_parser (Callable[[str, str], List[Task]], optional) – A function to extract Task from response.

content: str#
decompose(agent: ~camel.agents.chat_agent.ChatAgent, prompt: str | None = None, task_parser: ~typing.Callable[[str, str], ~typing.List[~camel.tasks.task.Task]] = <function parse_response>) List[Task][source]#

Decompose a task to a list of sub-tasks. It can be used for data generation and planner of agent.

Parameters:
  • agent (ChatAgent) – An agent that used to decompose the task.

  • prompt (str, optional) – A prompt to decompose the task. If not provided, the default prompt will be used.

  • task_parser (Callable[[str, str], List[Task]], optional) – A function to extract Task from response. If not provided, the default parse_response will be used.

Returns:

A list of tasks which are Task instances.

Return type:

List[Task]

failure_count: int#
classmethod from_message(message: BaseMessage) Task[source]#

Create a task from a message.

Parameters:

message (BaseMessage) – The message to the task.

Returns:

Task

get_depth() int[source]#

Get current task depth.

get_result(indent: str = '') str[source]#

Get task result to a sting.

Parameters:

indent (str) – The ident for hierarchical tasks.

Returns:

The printable task string.

Return type:

str

get_running_task() Task | None[source]#

Get RUNNING task.

id: str#
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_info': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'content': FieldInfo(annotation=str, required=True), 'failure_count': FieldInfo(annotation=int, required=False, default=0), 'id': FieldInfo(annotation=str, required=False, default=''), 'parent': FieldInfo(annotation=Union[Task, NoneType], required=False, default=None), 'result': FieldInfo(annotation=Union[str, NoneType], required=False, default=''), 'state': FieldInfo(annotation=TaskState, required=False, default=<TaskState.OPEN: 'OPEN'>), 'subtasks': FieldInfo(annotation=List[Task], required=False, default=[]), 'type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}#

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.

parent: Task | None#
remove_subtask(id: str)[source]#
reset()[source]#

Reset Task to initial state.

result: str | None#
set_id(id: str)[source]#
set_state(state: TaskState)[source]#

Recursively set the state of the task and its subtasks.

Parameters:

state (TaskState) – The giving state.

state: TaskState#
subtasks: List[Task]#
static to_message()[source]#

Convert a Task to a Message.

to_string(indent: str = '', state: bool = False) str[source]#

Convert task to a sting.

Parameters:
  • indent (str) – The ident for hierarchical tasks.

  • state (bool) – Include or not task state.

Returns:

The printable task string.

Return type:

str

type: str | None#
update_result(result: str)[source]#

Set task result and mark the task as DONE.

Parameters:

result (str) – The task result.

class camel.tasks.task.TaskManager(task: Task)[source]#

Bases: object

TaskManager is used to manage tasks.

root_task#

The root task.

tasks#

The ordered tasks.

task_map#

A map for task.id to Task.

current_task_id#

The current “RUNNING” task.id.

Parameters:

task (Task) – The root Task.

add_tasks(tasks: Task | List[Task]) None[source]#

self.tasks and self.task_map will be updated by the input tasks.

property current_task: Task | None#

Get the current task.

evolve(task: Task, agent: ChatAgent, template: TextPrompt | None = None, task_parser: Callable[[str, str], List[Task]] | None = None) Task | None[source]#
Evolve a task to a new task.

Evolve is only used for data generation.

Parameters:
  • task (Task) – A given task.

  • agent (ChatAgent) – An agent that used to evolve the task.

  • template (TextPrompt, optional) – A prompt template to evolve task. If not provided, the default template will be used.

  • task_parser (Callable, optional) – A function to extract Task from response. If not provided, the default parser will be used.

Returns:

The created Task instance or None.

Return type:

Task

exist(task_id: str) bool[source]#

Check if a task with the given id exists.

gen_task_id() str[source]#

Generate a new task id.

static set_tasks_dependence(root: Task, others: List[Task], type: Literal['serial', 'parallel'] = 'parallel')[source]#

Set relationship between root task and other tasks. Two relationships are currently supported: serial and parallel. serial : root -> other1 -> other2 parallel: root -> other1

-> other2

Parameters:
  • root (Task) – A root task.

  • others (List[Task]) – A list of tasks.

static topological_sort(tasks: List[Task]) List[Task][source]#

Sort a list of tasks by topological way.

Parameters:

tasks (List[Task]) – The giving list of tasks.

Returns:

The sorted list of tasks.

class camel.tasks.task.TaskState(value)[source]#

Bases: str, Enum

An enumeration.

DELETED = 'DELETED'#
DONE = 'DONE'#
FAILED = 'FAILED'#
OPEN = 'OPEN'#
RUNNING = 'RUNNING'#
classmethod states()[source]#
camel.tasks.task.parse_response(response: str, task_id: str | None = None) List[Task][source]#

Parse Tasks from a response.

Parameters:
  • response (str) – The model response.

  • task_id (str, optional) – a parent task id, the default value is “0”

Returns:

A list of tasks which is Task instance.

Return type:

List[Task]

camel.tasks.task_prompt module#

Module contents#

class camel.tasks.Task(*, content: str, id: str = '', state: TaskState = TaskState.OPEN, type: str | None = None, parent: Task | None = None, subtasks: List[Task] = [], result: str | None = '', failure_count: int = 0, additional_info: str | None = None)[source]#

Bases: BaseModel

Task is specific assignment that can be passed to a agent.

content#

string content for task.

Type:

str

id#

An unique string identifier for the task. This should

Type:

str

ideally be provided by the provider/model which created the task.
state#

The state which should be OPEN, RUNNING, DONE or DELETED.

Type:

camel.tasks.task.TaskState

type#

task type

Type:

str | None

parent#

The parent task, None for root task.

Type:

Task | None

subtasks#

The childrent sub-tasks for the task.

Type:

List[Task]

result#

The answer for the task.

Type:

str | None

add_subtask(task: Task)[source]#
additional_info: str | None#
compose(agent: ChatAgent, template: TextPrompt = 'As a Task composer with the role of {role_name}, your objective is to gather result from all sub tasks to get the final answer.\nThe root task is:\n\n{content}\n\nThe additional information of the task is:\n\n{additional_info}\n\nThe related tasks result and status:\n\n{other_results}\n\nso, the final answer of the root task is: \n', result_parser: Callable[[str], str] | None = None)[source]#

compose task result by the sub-tasks.

Parameters:
  • agent (ChatAgent) – An agent that used to compose the task result.

  • template (TextPrompt, optional) – The prompt template to compose task. If not provided, the default template will be used.

  • result_parser (Callable[[str, str], List[Task]], optional) – A function to extract Task from response.

content: str#
decompose(agent: ~camel.agents.chat_agent.ChatAgent, prompt: str | None = None, task_parser: ~typing.Callable[[str, str], ~typing.List[~camel.tasks.task.Task]] = <function parse_response>) List[Task][source]#

Decompose a task to a list of sub-tasks. It can be used for data generation and planner of agent.

Parameters:
  • agent (ChatAgent) – An agent that used to decompose the task.

  • prompt (str, optional) – A prompt to decompose the task. If not provided, the default prompt will be used.

  • task_parser (Callable[[str, str], List[Task]], optional) – A function to extract Task from response. If not provided, the default parse_response will be used.

Returns:

A list of tasks which are Task instances.

Return type:

List[Task]

failure_count: int#
classmethod from_message(message: BaseMessage) Task[source]#

Create a task from a message.

Parameters:

message (BaseMessage) – The message to the task.

Returns:

Task

get_depth() int[source]#

Get current task depth.

get_result(indent: str = '') str[source]#

Get task result to a sting.

Parameters:

indent (str) – The ident for hierarchical tasks.

Returns:

The printable task string.

Return type:

str

get_running_task() Task | None[source]#

Get RUNNING task.

id: str#
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_info': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'content': FieldInfo(annotation=str, required=True), 'failure_count': FieldInfo(annotation=int, required=False, default=0), 'id': FieldInfo(annotation=str, required=False, default=''), 'parent': FieldInfo(annotation=Union[Task, NoneType], required=False, default=None), 'result': FieldInfo(annotation=Union[str, NoneType], required=False, default=''), 'state': FieldInfo(annotation=TaskState, required=False, default=<TaskState.OPEN: 'OPEN'>), 'subtasks': FieldInfo(annotation=List[Task], required=False, default=[]), 'type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}#

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.

parent: Task | None#
remove_subtask(id: str)[source]#
reset()[source]#

Reset Task to initial state.

result: str | None#
set_id(id: str)[source]#
set_state(state: TaskState)[source]#

Recursively set the state of the task and its subtasks.

Parameters:

state (TaskState) – The giving state.

state: TaskState#
subtasks: List[Task]#
static to_message()[source]#

Convert a Task to a Message.

to_string(indent: str = '', state: bool = False) str[source]#

Convert task to a sting.

Parameters:
  • indent (str) – The ident for hierarchical tasks.

  • state (bool) – Include or not task state.

Returns:

The printable task string.

Return type:

str

type: str | None#
update_result(result: str)[source]#

Set task result and mark the task as DONE.

Parameters:

result (str) – The task result.

class camel.tasks.TaskManager(task: Task)[source]#

Bases: object

TaskManager is used to manage tasks.

root_task#

The root task.

tasks#

The ordered tasks.

task_map#

A map for task.id to Task.

current_task_id#

The current “RUNNING” task.id.

Parameters:

task (Task) – The root Task.

add_tasks(tasks: Task | List[Task]) None[source]#

self.tasks and self.task_map will be updated by the input tasks.

property current_task: Task | None#

Get the current task.

evolve(task: Task, agent: ChatAgent, template: TextPrompt | None = None, task_parser: Callable[[str, str], List[Task]] | None = None) Task | None[source]#
Evolve a task to a new task.

Evolve is only used for data generation.

Parameters:
  • task (Task) – A given task.

  • agent (ChatAgent) – An agent that used to evolve the task.

  • template (TextPrompt, optional) – A prompt template to evolve task. If not provided, the default template will be used.

  • task_parser (Callable, optional) – A function to extract Task from response. If not provided, the default parser will be used.

Returns:

The created Task instance or None.

Return type:

Task

exist(task_id: str) bool[source]#

Check if a task with the given id exists.

gen_task_id() str[source]#

Generate a new task id.

static set_tasks_dependence(root: Task, others: List[Task], type: Literal['serial', 'parallel'] = 'parallel')[source]#

Set relationship between root task and other tasks. Two relationships are currently supported: serial and parallel. serial : root -> other1 -> other2 parallel: root -> other1

-> other2

Parameters:
  • root (Task) – A root task.

  • others (List[Task]) – A list of tasks.

static topological_sort(tasks: List[Task]) List[Task][source]#

Sort a list of tasks by topological way.

Parameters:

tasks (List[Task]) – The giving list of tasks.

Returns:

The sorted list of tasks.