Implement babyagi_agent and integrate it into multimodel role playing

camel/agents/__init__.py

@@ -13,7 +13,12 @@
 # =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
 from .base import BaseAgent
 from .chat_agent import ChatAgent, ChatAgentResponse
-from .task_agent import TaskPlannerAgent, TaskSpecifyAgent
+from .task_agent import (
+    TaskSpecifyAgent,
+    TaskPlannerAgent,
+    TaskCreationAgent,
+    TaskPrioritizationAgent,
+)
 from .critic_agent import CriticAgent
 from .tool_agents.base import BaseToolAgent
 from .tool_agents.hugging_face_tool_agent import HuggingFaceToolAgent
@@ -25,6 +30,8 @@
     'ChatAgentResponse',
     'TaskSpecifyAgent',
     'TaskPlannerAgent',
+    'TaskCreationAgent',
+    'TaskPrioritizationAgent',
     'CriticAgent',
     'BaseToolAgent',
     'HuggingFaceToolAgent',

camel/agents/task_agent.py

@@ -11,13 +11,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from camel.agents import ChatAgent
 from camel.configs import ChatGPTConfig
 from camel.messages import BaseMessage
 from camel.prompts import PromptTemplateGenerator, TextPrompt
 from camel.typing import ModelType, RoleType, TaskType
+from camel.utils import get_task_list
 
 
 class TaskSpecifyAgent(ChatAgent):
@@ -29,15 +30,15 @@ class TaskSpecifyAgent(ChatAgent):
         task_specify_prompt (TextPrompt): The prompt for specifying the task.
 
     Args:
-        model (ModelType): The type of model to use for the agent.
+        model (ModelType, optional): The type of model to use for the agent.
             (default: :obj:`ModelType.GPT_3_5_TURBO`)
-        task_type (TaskType): The type of task for which to generate a prompt.
-            (default: :obj:`TaskType.AI_SOCIETY`)
-        model_config (Any): The configuration for the model.
+        task_type (TaskType, optional): The type of task for which to generate
+            a prompt. (default: :obj:`TaskType.AI_SOCIETY`)
+        model_config (Any, optional): The configuration for the model.
             (default: :obj:`None`)
-        task_specify_prompt (Optional[TextPrompt]): The prompt for specifying
-            the task. (default: :obj:`None`)
-        word_limit (int): The word limit for the task prompt.
+        task_specify_prompt (Union[str, TextPrompt], optional): The prompt for
+            specifying the task. (default: :obj:`None`)
+        word_limit (int, optional): The word limit for the task prompt.
             (default: :obj:`50`)
         output_language (str, optional): The language to be output by the
         agent. (default: :obj:`None`)
@@ -86,26 +87,23 @@ def run(
         Args:
             task_prompt (Union[str, TextPrompt]): The original task
                 prompt.
-            meta_dict (Optional[Dict[str, Any]]): A dictionary containing
+            meta_dict (Dict[str, Any], optional): A dictionary containing
                 additional information to include in the prompt.
                 (default: :obj:`None`)
 
         Returns:
             TextPrompt: The specified task prompt.
         """
         self.reset()
-        self.task_specify_prompt = self.task_specify_prompt.format(
-            task=task_prompt)
+        task_specify_prompt = self.task_specify_prompt.format(task=task_prompt)
 
         if meta_dict is not None:
-            self.task_specify_prompt = (self.task_specify_prompt.format(
-                **meta_dict))
+            task_specify_prompt = (task_specify_prompt.format(**meta_dict))
 
-        task_msg = BaseMessage.make_user_message(
-            role_name="Task Specifier", content=self.task_specify_prompt)
+        task_msg = BaseMessage.make_user_message(role_name="Task Specifier",
+                                                 content=task_specify_prompt)
         specifier_response = self.step(task_msg)
-        if (specifier_response.msgs is None
-                or len(specifier_response.msgs) == 0):
+        if len(specifier_response.msgs) == 0:
             raise RuntimeError("Task specification failed.")
         specified_task_msg = specifier_response.msgs[0]
 
@@ -124,9 +122,9 @@ class TaskPlannerAgent(ChatAgent):
             the task into subtasks.
 
     Args:
-        model (ModelType): The type of model to use for the agent.
+        model (ModelType, optional: The type of model to use for the agent.
             (default: :obj:`ModelType.GPT_3_5_TURBO`)
-        model_config (Any): The configuration for the model.
+        model_config (Any, optional): The configuration for the model.
             (default: :obj:`None`)
         output_language (str, optional): The language to be output by the
         agent. (default: :obj:`None`)
@@ -135,13 +133,12 @@ class TaskPlannerAgent(ChatAgent):
     def __init__(
         self,
         model: Optional[ModelType] = None,
-        model_config: Any = None,
+        model_config: Optional[Any] = None,
         output_language: Optional[str] = None,
     ) -> None:
 
         self.task_planner_prompt = TextPrompt(
             "Divide this task into subtasks: {task}. Be concise.")
-
         system_message = BaseMessage(
             role_name="Task Planner",
             role_type=RoleType.ASSISTANT,
@@ -167,17 +164,214 @@ def run(
         """
         # TODO: Maybe include roles information.
         self.reset()
-        self.task_planner_prompt = self.task_planner_prompt.format(
-            task=task_prompt)
+        task_planner_prompt = self.task_planner_prompt.format(task=task_prompt)
+
+        task_msg = BaseMessage.make_user_message(role_name="Task Planner",
+                                                 content=task_planner_prompt)
 
-        task_msg = BaseMessage.make_user_message(
-            role_name="Task Planner", content=self.task_planner_prompt)
         task_response = self.step(task_msg)
 
-        if task_response.msgs is None:
-            raise RuntimeError("Got None Subtasks messages.")
+        if len(task_response.msgs) == 0:
+            raise RuntimeError("Got no task planning messages.")
         if task_response.terminated:
             raise RuntimeError("Task planning failed.")
 
         sub_tasks_msg = task_response.msgs[0]
         return TextPrompt(sub_tasks_msg.content)
+
+
+class TaskCreationAgent(ChatAgent):
+    r"""An agent that helps create new tasks based on the objective
+    and last completed task. Compared to :obj:`TaskPlannerAgent`,
+    it's still a task planner, but it has more context information
+    like last task and incomplete task list. Modified from
+    `BabyAGI <https://github.com/yoheinakajima/babyagi>`_.
+
+    Attributes:
+        task_creation_prompt (TextPrompt): A prompt for the agent to create
+            new tasks.
+
+    Args:
+        objective (Union[str, TextPrompt]): The objective of the Agent to
+            perform the task.
+        model (ModelType, optional): The type of model to use for the agent.
+            (default: :obj:`ModelType.GPT_3_5_TURBO`)
+        model_config (Any, optional): The configuration for the model.
+            (default: :obj:`None`)
+        output_language (str, optional): The language to be output by the
+            agent. (default: :obj:`None`)
+    """
+
+    def __init__(
+        self,
+        objective: Union[str, TextPrompt],
+        model: Optional[ModelType] = None,
+        model_config: Optional[Any] = None,
+        output_language: Optional[str] = None,
+    ) -> None:
+
+        task_creation_prompt = TextPrompt(
+            """Create new tasks with the following objective: {objective}.
+The last completed task has the result:
+{task_result}.
+This result was based on this task: {task}.
+Based on the result, return a list of tasks to be completed in order to meet \
+the objective.
+
+{unsolved_tasks}
+
+Return one task per line in your response.
+The result must be a numbered list in the format:
+
+1. First task
+2. Second task
+
+The number of each entry must be followed by a period. Be concise.""")
+
+        self.task_creation_prompt = task_creation_prompt.format(
+            objective=objective)
+        self.objective = objective
+
+        system_message = BaseMessage(
+            role_name="Task Creator",
+            role_type=RoleType.ASSISTANT,
+            meta_dict=None,
+            content="You are a helpful task creator.",
+        )
+
+        super().__init__(system_message, model, model_config,
+                         output_language=output_language)
+
+    def run(
+        self,
+        previous_task: Union[str, TextPrompt],
+        task_result: Union[str, TextPrompt],
+        task_list: Optional[List[str]] = None,
+    ) -> List[str]:
+        r"""Generate subtasks based on the previous task results and
+        incomplete task list.
+
+        Args:
+            previous_task (Union[str, TextPrompt]): The last completed task to
+                be used to create future plans.
+            task_result (Union[str, TextPrompt]): The result of last completed
+                task to be used to create future plans.
+            task_list (List[str], optional): The incomplete task list
+                which should not overlap with new created tasks.
+                (default: :obj:`None`)
+        Returns:
+            List[str]: The new task list generated by the Agent.
+        """
+        self.reset()
+        task_creation_prompt = self.task_creation_prompt.format(
+            task=previous_task, task_result=task_result)
+        if task_list is not None:
+            unsolved = (
+                f"These are unsolved tasks: {task_list}.\n"
+                "These new tasks must not overlap with incomplete tasks.")
+
+            task_creation_prompt = task_creation_prompt.format(
+                unsolved_tasks=unsolved)
+        else:
+            task_creation_prompt = task_creation_prompt.format(
+                unsolved_tasks="")
+
+        task_msg = BaseMessage.make_user_message(role_name="Task Creator",
+                                                 content=task_creation_prompt)
+
+        task_response = self.step(task_msg)
+
+        if len(task_response.msgs) == 0:
+            raise RuntimeError("Got no task creation messages.")
+        if task_response.terminated:
+            raise RuntimeError("Task creation failed.")
+
+        sub_tasks_msg = task_response.msgs[0]
+        return get_task_list(sub_tasks_msg.content)
+
+
+class TaskPrioritizationAgent(ChatAgent):
+    r"""An agent that helps re-prioritize the task list and
+    returns numbered prioritized list. Modified from
+    `BabyAGI <https://github.com/yoheinakajima/babyagi>`_.
+
+    Attributes:
+        task_prioritization_prompt (TextPrompt): A prompt for the agent to
+            prioritize tasks.
+
+    Args:
+        objective (Union[str, TextPrompt]): The objective of the Agent to
+            perform the task.
+        model (ModelType, optional): The type of model to use for the agent.
+            (default: :obj:`ModelType.GPT_3_5_TURBO`)
+        model_config (Any, optional): The configuration for the model.
+            (default: :obj:`None`)
+        output_language (str, optional): The language to be output by the
+            agent. (default: :obj:`None`)
+    """
+
+    def __init__(
+        self,
+        objective: Union[str, TextPrompt],
+        model: Optional[ModelType] = None,
+        model_config: Optional[Any] = None,
+        output_language: Optional[str] = None,
+    ) -> None:
+
+        task_prioritization_prompt = TextPrompt(
+            """Prioritize the following tasks : {task_list}.
+Consider the ultimate objective of you: {objective}.
+Tasks should be sorted from highest to lowest priority,
+where higher-priority tasks are those that act as pre-requisites \
+or are more essential for meeting the objective.
+Return one task per line in your response. Do not remove any tasks.
+The result must be a numbered list in the format:
+
+1. First task
+2. Second task
+
+The entries must be consecutively numbered, starting with 1.
+The number of each entry must be followed by a period.
+Be concise.""")
+
+        self.task_prioritization_prompt = task_prioritization_prompt.format(
+            objective=objective)
+        self.objective = objective
+
+        system_message = BaseMessage(
+            role_name="Task Prioritizer",
+            role_type=RoleType.ASSISTANT,
+            meta_dict=None,
+            content="You are a helpful task prioritizer.",
+        )
+
+        super().__init__(system_message, model, model_config,
+                         output_language=output_language)
+
+    def run(
+        self,
+        task_list: List[str],
+    ) -> List[str]:
+        r"""Prioritize the task list given the agent objective.
+
+        Args:
+            task_list (List[str]): The un-prioritized task list of agent.
+        Returns:
+            List[str]: The new prioritized task list generated by the Agent.
+        """
+        self.reset()
+        task_prioritization_prompt = self.task_prioritization_prompt.format(
+            task_list=task_list)
+
+        task_msg = BaseMessage.make_user_message(
+            role_name="Task Prioritizer", content=task_prioritization_prompt)
+
+        task_response = self.step(task_msg)
+
+        if len(task_response.msgs) == 0:
+            raise RuntimeError("Got no task prioritization messages.")
+        if task_response.terminated:
+            raise RuntimeError("Task prioritization failed.")
+
+        sub_tasks_msg = task_response.msgs[0]
+        return get_task_list(sub_tasks_msg.content)

camel/utils/__init__.py

@@ -22,6 +22,7 @@
     get_first_int,
     download_tasks,
     parse_doc,
+    get_task_list,
 )
 
 __all__ = [
@@ -35,4 +36,5 @@
     'download_tasks',
     'PythonInterpreter',
     'parse_doc',
+    'get_task_list',
 ]

camel/utils/functions.py

@@ -314,3 +314,26 @@ def parse_doc(func: Callable) -> Dict[str, Any]:
     }
 
     return function_dict
+
+
+def get_task_list(task_response: str) -> List[str]:
+    r"""Parse the response of the Agent and return task list.
+
+    Args:
+        task_response (str): The string response of the Agent.
+
+    Returns:
+        List[str]: A list of the string tasks.
+    """
+
+    new_tasks_list = []
+    task_string_list = task_response.strip().split('\n')
+    # each task starts with #.
+    for task_string in task_string_list:
+        task_parts = task_string.strip().split(".", 1)
+        if len(task_parts) == 2:
+            task_id = ''.join(s for s in task_parts[0] if s.isnumeric())
+            task_name = re.sub(r'[^\w\s_]+', '', task_parts[1]).strip()
+            if task_name.strip() and task_id.isnumeric():
+                new_tasks_list.append(task_name)
+    return new_tasks_list