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

Attributes: content: string content for task. id: An unique string identifier for the task. This should ideally be provided by the provider/model which created the task. state: The state which should be OPEN, RUNNING, DONE or DELETED. type: task type parent: The parent task, None for root task. subtasks: The childrent sub-tasks for the task. result: The answer for 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
):

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:

List[Task]: A list of tasks which are :obj:Task instances.

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

Attributes: 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.

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