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)

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)

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)

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]

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

Create a task from a message.

Parameters:

message (BaseMessage) – The message to the task.

Returns:

Task

get_depth() → int

Get current task depth.

get_result(indent: str = '') → str

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

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)

reset()

Reset Task to initial state.

result: str | None

set_id(id: str)

set_state(state: TaskState)

Recursively set the state of the task and its subtasks.

Parameters:

state (TaskState) – The giving state.

state: TaskState

subtasks: List[Task]

static to_message()

Convert a Task to a Message.

to_string(indent: str = '', state: bool = False) → str

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)

Set task result and mark the task as DONE.

Parameters:

result (str) – The task result.

class camel.tasks.task.TaskManager(task: Task)

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

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

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

Check if a task with the given id exists.

gen_task_id() → str

Generate a new task id.

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

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]

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)

Bases: str, Enum

An enumeration.

DELETED = 'DELETED'

DONE = 'DONE'

FAILED = 'FAILED'

OPEN = 'OPEN'

RUNNING = 'RUNNING'

classmethod states()

camel.tasks.task.parse_response(response: str, task_id: str | None = None) → List[Task]

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)

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)

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)

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]

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

Create a task from a message.

Parameters:

message (BaseMessage) – The message to the task.

Returns:

Task

get_depth() → int

Get current task depth.

get_result(indent: str = '') → str

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

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)

reset()

Reset Task to initial state.

result: str | None

set_id(id: str)

set_state(state: TaskState)

Recursively set the state of the task and its subtasks.

Parameters:

state (TaskState) – The giving state.

state: TaskState

subtasks: List[Task]

static to_message()

Convert a Task to a Message.

to_string(indent: str = '', state: bool = False) → str

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)

Set task result and mark the task as DONE.

Parameters:

result (str) – The task result.

class camel.tasks.TaskManager(task: Task)

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

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

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

Check if a task with the given id exists.

gen_task_id() → str

Generate a new task id.

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

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]

Sort a list of tasks by topological way.

Parameters:

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

Returns:

The sorted list of tasks.