CAMEL - Build Multi-Agent AI Systems

TaskValidationMode

class TaskValidationMode(Enum):

Validation modes for different use cases.

validate_task_content

def validate_task_content(
    content: str,
    task_id: str = 'unknown',
    min_length: int = 1,
    mode: TaskValidationMode = TaskValidationMode.INPUT,
    check_failure_patterns: bool = True
):

Unified validation for task content and results to avoid silent failures. Performs comprehensive checks to ensure content meets quality standards. Parameters:

content (str): The task content or result to validate.
task_id (str): Task ID for logging purposes. (default: :obj:"unknown")
min_length (int): Minimum content length after stripping whitespace. (default: :obj:1)
mode (TaskValidationMode): Validation mode - INPUT for task content, OUTPUT for task results. (default: :obj:TaskValidationMode.INPUT)
check_failure_patterns (bool): Whether to check for failure indicators in the content. Only effective in OUTPUT mode. (default: :obj:True)

Returns: bool: True if content passes validation, False otherwise.

is_task_result_insufficient

def is_task_result_insufficient(task: 'Task'):

Check if a task result is insufficient and should be treated as failed. This is a convenience wrapper around validate_task_content for backward compatibility and semantic clarity when checking task results. Parameters:

task (Task): The task to check.

Returns: bool: True if the result is insufficient, False otherwise.

parse_response

def parse_response(response: str, task_id: Optional[str] = None):

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: List[Task]: A list of tasks which is :obj:Task instance.

TaskState

class TaskState(str, Enum):

states

def states(cls):

Task

class Task(BaseModel):

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

content (str): string content for task.
id (str): An unique string identifier for the task. This should ideally be provided by the provider/model which created the task. (default: :obj:uuid.uuid4())
state (TaskState): The state which should be OPEN, RUNNING, DONE or DELETED. (default: :obj:TaskState.FAILED)
type (Optional[str]): task type. (default: :obj:None)
parent (Optional[Task]): The parent task, None for root task. (default: :obj:None)
subtasks (List[Task]): The childrent sub-tasks for the task. (default: :obj:[])
result (Optional[str]): The answer for the task. (default: :obj:"")
failure_count (int): The failure count for the task. (default: :obj:0)
assigned_worker_id (Optional[str]): The ID of the worker assigned to this task. (default: :obj:None)
dependencies (List[Task]): The dependencies for the task. (default: :obj:[])
additional_info (Optional[Dict[str, Any]]): Additional information for the task. (default: :obj:None)
image_list (Optional[List[Union[Image.Image, str]]]): Optional list of PIL Image objects or image URLs (strings) associated with the task. (default: :obj:None)
image_detail (Literal["auto", "low", "high"]): Detail level of the images associated with the task. (default: :obj:auto)
video_bytes (Optional[bytes]): Optional bytes of a video associated with the task. (default: :obj:None)
video_detail (Literal["auto", "low", "high"]): Detail level of the videos associated with the task. (default: :obj:auto)

repr

def __repr__(self):

Return a string representation of the task.

from_message

def from_message(cls, message: BaseMessage):

Create a task from a message. Parameters:

message (BaseMessage): The message to the task.

Returns: Task

to_message

def to_message():

Convert a Task to a Message.

reset

def reset(self):

Reset Task to initial state.

update_result

def update_result(self, result: str):

Set task result and mark the task as DONE. Parameters:

result (str): The task result.

set_id

def set_id(self, id: str):

Set the id of the task. Parameters:

id (str): The id of the task.

set_state

def set_state(self, state: TaskState):

Recursively set the state of the task and its subtasks. Parameters:

state (TaskState): The giving state.

add_subtask

def add_subtask(self, task: 'Task'):

Add a subtask to the current task. Parameters:

task (Task): The subtask to be added.

remove_subtask

def remove_subtask(self, id: str):

Remove a subtask from the current task. Parameters:

id (str): The id of the subtask to be removed.

get_running_task

def get_running_task(self):

Get RUNNING task.

to_string

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

Convert task to a string. Parameters:

indent (str): The ident for hierarchical tasks.
state (bool): Include or not task state.

Returns: str: The printable task string.

get_result

def get_result(self, indent: str = ''):

Get task result to a string. Parameters:

indent (str): The ident for hierarchical tasks.

Returns: str: The printable task string.

decompose

def decompose(
    self,
    agent: 'ChatAgent',
    prompt: Optional[str] = None,
    task_parser: Callable[[str, str], List['Task']] = parse_response,
    stream_callback: Optional[Callable[['ChatAgentResponse'], None]] = None
):

Decompose a task to a list of sub-tasks. Automatically detects streaming or non-streaming based on agent configuration. 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.
stream_callback (Callable[[ChatAgentResponse], None], optional): A callback function that receives each chunk (ChatAgentResponse) during streaming. This allows tracking the decomposition progress in real-time.

Returns: Union[List[Task], Generator[List[Task], None, None]]: If agent is configured for streaming, returns a generator that yields lists of new tasks as they are parsed. Otherwise returns a list of all tasks.

_decompose_streaming

def _decompose_streaming(
    self,
    response: Iterable,
    task_parser: Callable[[str, str], List['Task']],
    stream_callback: Optional[Callable[['ChatAgentResponse'], None]] = None
):

Handle streaming response for task decomposition. Parameters:

response: Streaming response from agent
task_parser: Function to parse tasks from response
stream_callback (Callable[[ChatAgentResponse], None], optional): A callback function that receives each chunk (ChatAgentResponse) during streaming.
Yields: List[Task]: New tasks as they are parsed from streaming response

_decompose_non_streaming

def _decompose_non_streaming(self, response, task_parser: Callable[[str, str], List['Task']]):

Handle non-streaming response for task decomposition. Parameters:

response: Regular response from agent
task_parser: Function to parse tasks from response

Returns: List[Task]: All parsed tasks

_parse_partial_tasks

def _parse_partial_tasks(self, response: str):

Parse tasks from potentially incomplete response. Parameters:

response: Partial response content

Returns: List[Task]: Tasks parsed from complete <task>``</task> blocks

compose

def compose(
    self,
    agent: 'ChatAgent',
    template: TextPrompt = TASK_COMPOSE_PROMPT,
    result_parser: Optional[Callable[[str], str]] = 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.

get_depth

def get_depth(self):

Get current task depth.

TaskManager

class TaskManager:

TaskManager is used to manage tasks. Parameters:

task (Task): The root Task.

init

def __init__(self, task: Task):

gen_task_id

def gen_task_id(self):

Generate a new task id.

exist

def exist(self, task_id: str):

Check if a task with the given id exists.

current_task

def current_task(self):

Get the current task.

topological_sort

def topological_sort(tasks: 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.

set_tasks_dependence

def 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.

add_tasks

def add_tasks(self, tasks: Union[Task, List[Task]]):

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

evolve

def evolve(
    self,
    task: Task,
    agent: 'ChatAgent',
    template: Optional[TextPrompt] = None,
    task_parser: Optional[Callable[[str, str], List[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: Task: The created :obj:Task instance or None.

​TaskValidationMode

​validate_task_content

​is_task_result_insufficient

​parse_response

​TaskState

​states

​Task

​repr

​from_message

​to_message

​reset

​update_result

​set_id

​set_state

​add_subtask

​remove_subtask

​get_running_task

​to_string

​get_result

​decompose

​_decompose_streaming

​_decompose_non_streaming

​_parse_partial_tasks

​compose

​get_depth

​TaskManager

​init

​gen_task_id

​exist

​current_task

​topological_sort

​set_tasks_dependence

​add_tasks

​evolve