camel.tasks 包

子模块

camel.tasks.task 模块

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

基类：BaseModel

任务是可以传递给代理的具体工作。

content

任务字符串内容。

Type:

字符串

id

任务的唯一字符串标识符。这应该

Type:

字符串

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

state

状态应为 OPEN、RUNNING、DONE 或 DELETED。

Type:

camel.tasks.task.TaskState

type

任务类型

Type:

字符串 | 无

parent

父任务，根任务为None。

Type:

camel.tasks.task.Task | 无

subtasks

该任务的子任务。

Type:

列表[camel.tasks.task.Task]

result

任务的答案。

Type:

字符串 | 无

add_subtask(task: 任务)

在当前任务中添加一个子任务。

Parameters:

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)

根据子任务组合任务结果。

Parameters:

• agent (ChatAgent) - 用于组合任务结果的代理。

• template (TextPrompt, optional) - 用于组合任务的提示模板。如果未提供，将使用默认模板。

• result_parser (Callable[[str, str], List[Task]], optional) – 一个用于从响应中提取Task的函数。

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[任务]

将一个任务分解为一系列子任务。可用于数据生成和智能体规划。

Parameters:

• agent (ChatAgent) - 用于分解任务的代理。

• prompt (str, optional) – 用于分解任务的提示词。如果未提供，将使用默认提示词。

• task_parser (Callable[[str, str], List[Task]], optional) – 一个用于从响应中提取Task的函数。如果未提供，将使用默认的parse_response。

Returns:

一个任务列表，这些任务是Task实例。

Return type:

列表[任务]

failure_count: int

classmethod from_message(message: BaseMessage) → 任务

从消息创建任务。

Parameters:

message (BaseMessage) – 发送给任务的消息。

Returns:

任务

get_depth() → int

获取当前任务深度。

get_result(indent: str = '') → str

获取任务结果到字符串。

Parameters:

indent (str) – 用于分层任务的缩进标识。

Returns:

可打印的任务字符串。

Return type:

字符串

get_running_task() → 任务 | None

获取正在运行的任务。

id: str

model_config: ClassVar[ConfigDict] = {}

模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

parent: 任务 | None

remove_subtask(id: str)

从当前任务中移除一个子任务。

Parameters:

id (str) – 要删除的子任务ID。

reset()

将任务重置为初始状态。

result: str | None

set_id(id: str)

设置任务的ID。

Parameters:

id (str) – 任务的ID。

set_state(state: TaskState)

递归设置任务及其子任务的状态。

Parameters:

state (TaskState) – 给定的状态。

state: TaskState

subtasks: List[任务]

static to_message()

将任务转换为消息。

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

将任务转换为字符串。

Parameters:

• indent (str) - 用于分层任务的缩进标识。

• state (bool) - 是否包含任务状态。

Returns:

可打印的任务字符串。

Return type:

字符串

type: str | None

update_result(result: str)

设置任务结果并将任务标记为完成。

Parameters:

result (str) – 任务结果。

class camel.tasks.task.TaskManager(task: 任务)

基类：object

TaskManager 用于管理任务。

root_task

根任务。

tasks

已排序的任务。

task_map

一个从task.id到Task的映射表。

current_task_id

当前“RUNNING”任务的ID。

Parameters:

任务 (Task) – 根任务。

add_tasks(tasks: 任务 | List[任务]) → None

self.tasks 和 self.task_map 将会被输入的任务更新。

property current_task: 任务 | None

获取当前任务。

evolve(task: 任务, agent: ChatAgent, template: TextPrompt | None = None, task_parser: Callable[[str, str], List[任务]] | None = None) → 任务 | None

Evolve a task to a new task.

Evolve仅用于数据生成。

Parameters:

• task (Task) – 给定的任务。

• agent (ChatAgent) – 用于演进任务的代理。

• template (TextPrompt, optional) – 用于演化任务的提示模板。 如果未提供，将使用默认模板。

• task_parser (Callable, optional) – 用于从响应中提取Task的函数。如果未提供，将使用默认解析器。

Returns:

创建的 Task 实例或 None。

Return type:

任务

exist(task_id: str) → bool

检查是否存在具有给定ID的任务。

gen_task_id() → str

生成一个新的任务ID。

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

设置根任务与其他任务之间的关系。 目前支持两种关系：串行和并行。 serial : 根任务 -> 其他任务1 -> 其他任务2 parallel: 根任务 -> 其他任务1

-> 其他2

Parameters:

• root (Task) - 根任务。

• 其他 (列表[Task]) - 任务列表。

static topological_sort(tasks: List[任务]) → List[任务]

按拓扑顺序对任务列表进行排序。

Parameters:

tasks (List[Task]) – 给定的任务列表。

Returns:

已排序的任务列表。

class camel.tasks.task.TaskState(value)

基类: str, Enum

一个枚举。

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[任务]

从响应中解析任务。

Parameters:

• response (str) - 模型响应。

• task_id (str, optional) – 父任务ID，默认值为"0"

Returns:

一个任务列表，其中每个任务都是Task的实例。

Return type:

列表[任务]

camel.tasks.task_prompt 模块

模块内容

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

基类：BaseModel

任务是可以传递给代理的具体工作。

content

任务字符串内容。

Type:

字符串

id

任务的唯一字符串标识符。这应该

Type:

字符串

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

state

状态应为 OPEN、RUNNING、DONE 或 DELETED。

Type:

camel.tasks.task.TaskState

type

任务类型

Type:

字符串 | 无

parent

父任务，根任务为None。

Type:

Task | 无

subtasks

该任务的子任务。

Type:

列表[任务]

result

任务的答案。

Type:

字符串 | 无

add_subtask(task: 任务)

在当前任务中添加一个子任务。

Parameters:

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)

根据子任务组合任务结果。

Parameters:

• agent (ChatAgent) - 用于组合任务结果的代理。

• template (TextPrompt, optional) – 用于组合任务的提示模板。如果未提供，将使用默认模板。

• result_parser (Callable[[str, str], List[Task]], optional) – 一个从响应中提取Task的函数。

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[任务]

将一个任务分解为一系列子任务。可用于数据生成和智能体规划。

Parameters:

• agent (ChatAgent) - 用于分解任务的代理。

• prompt (str, optional) – 用于分解任务的提示词。如果未提供，将使用默认提示词。

• task_parser (Callable[[str, str], List[Task]], optional) – 一个用于从响应中提取Task的函数。如果未提供，将使用默认的parse_response。

Returns:

一个任务列表，这些任务是Task实例。

Return type:

列表[任务]

failure_count: int

classmethod from_message(message: BaseMessage) → 任务

从消息创建任务。

Parameters:

message (BaseMessage) – 发送给任务的消息。

Returns:

任务

get_depth() → int

获取当前任务深度。

get_result(indent: str = '') → str

获取任务结果到字符串。

Parameters:

indent (str) – 用于分层任务的缩进标识。

Returns:

可打印的任务字符串。

Return type:

字符串

get_running_task() → 任务 | None

获取正在运行的任务。

id: str

model_config: ClassVar[ConfigDict] = {}

模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

parent: 任务 | None

remove_subtask(id: str)

从当前任务中移除一个子任务。

Parameters:

id (str) – 要删除的子任务ID。

reset()

将任务重置为初始状态。

result: str | None

set_id(id: str)

设置任务的ID。

Parameters:

id (str) – 任务的ID。

set_state(state: TaskState)

递归设置任务及其子任务的状态。

Parameters:

state (TaskState) – 给定的状态。

state: TaskState

subtasks: List[任务]

static to_message()

将任务转换为消息。

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

将任务转换为字符串。

Parameters:

• indent (str) - 用于分层任务的缩进标识。

• state (bool) - 是否包含任务状态。

Returns:

可打印的任务字符串。

Return type:

字符串

type: str | None

update_result(result: str)

设置任务结果并将任务标记为完成。

Parameters:

result (str) – 任务结果。

class camel.tasks.TaskManager(task: 任务)

基类：object

TaskManager 用于管理任务。

root_task

根任务。

tasks

已排序的任务。

task_map

一个从task.id到Task的映射表。

current_task_id

当前“RUNNING”任务的ID。

Parameters:

任务 (Task) – 根任务。

add_tasks(tasks: 任务 | List[任务]) → None

self.tasks 和 self.task_map 将会被输入的任务更新。

property current_task: 任务 | None

获取当前任务。

evolve(task: 任务, agent: ChatAgent, template: TextPrompt | None = None, task_parser: Callable[[str, str], List[任务]] | None = None) → 任务 | None

Evolve a task to a new task.

Evolve仅用于数据生成。

Parameters:

• task (Task) – 给定的任务。

• agent (ChatAgent) – 用于演进任务的代理。

• template (TextPrompt, optional) – 用于演化任务的提示模板。 如果未提供，将使用默认模板。

• task_parser (Callable, optional) – 用于从响应中提取Task的函数。如果未提供，将使用默认解析器。

Returns:

创建的 Task 实例或 None。

Return type:

任务

exist(task_id: str) → bool

检查是否存在具有给定ID的任务。

gen_task_id() → str

生成一个新的任务ID。

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

设置根任务与其他任务之间的关系。 目前支持两种关系：串行和并行。 serial : 根任务 -> 其他任务1 -> 其他任务2 parallel: 根任务 -> 其他任务1

-> 其他2

Parameters:

• root (Task) - 根任务。

• 其他 (列表[Task]) - 任务列表。

static topological_sort(tasks: List[任务]) → List[任务]

按拓扑顺序对任务列表进行排序。

Parameters:

tasks (List[Task]) – 给定的任务列表。

Returns:

已排序的任务列表。