Generate the roles and role description

camel/agents/role_assignment_agent.py

@@ -0,0 +1,126 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+import re
+from typing import Any, Dict, List, Optional, Tuple, Union
+
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from camel.agents import ChatAgent
+from camel.messages import BaseMessage
+from camel.prompts import TextPrompt
+from camel.typing import ModelType, RoleType
+
+
+class RoleAssignmentAgent(ChatAgent):
+    r"""An agent that generates role names based on the task prompt.
+    Attributes:
+        role_assignment_prompt (TextPrompt): A prompt for the agent to generate
+        role names.
+
+    Args:
+        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`)
+    """
+
+    def __init__(
+        self,
+        model: ModelType = ModelType.GPT_3_5_TURBO,
+        model_config: Optional[Any] = None,
+    ) -> None:
+        self.role_assignment_prompt = TextPrompt(
+            'Given this task, "{task}", generate two role names, ' +
+            'one for the AI user and one for the AI assistant.')
+
+        system_message = BaseMessage(
+            role_name="Role Assigner",
+            role_type=RoleType.ASSISTANT,
+            meta_dict=None,
+            content="You assign roles based on tasks.",
+        )
+        super().__init__(system_message, model, model_config)
+
+    @retry(wait=wait_exponential(min=5, max=60), stop=stop_after_attempt(5))
+    def run_role_with_description(
+        self,
+        task_prompt: Union[str, TextPrompt],
+        num_roles: int = 2,
+    ) -> Tuple[List[str], Dict[str, str], bool, Dict[str, Any]]:
+        r"""Generate role names based on the input task prompt.
+
+        Args:
+            task_prompt (Union[str, TextPrompt]): The prompt
+                for the task based on which the roles are to be generated.
+            num_roles (int, optional): The number of roles to generate.
+                (default: :obj:`2`)
+
+        Returns:
+            Tuple[List[str], Dict[str, str], bool, Dict[str, Any]]: A tuple
+        """
+        self.reset()
+
+        expert_prompt = "===== ANSWER PROMPT =====\n" + "\n".join(
+            f"Domain expert {i + 1}: <BLANK>\n"
+            f"Associated competencies, characteristics, duties "
+            f"and workflows: <BLANK>. End." for i in range(num_roles or 0))
+        role_assignment_generation_prompt = TextPrompt(
+            "You are the boss, and you're in charge of recruiting " +
+            "{num_roles} experts for the following task." +
+            "\n==== TASK =====\n {task}\n\n" +
+            "Identify the domain experts you'd recruit and detail their " +
+            "associated competencies, characteristics, duties and workflows " +
+            "to complete the task.\n " +
+            "Your answer MUST adhere to the format of ANSWER PROMPT, and " +
+            "ONLY answer the BLANKs.\n" + expert_prompt)
+        role_assignment_generation = role_assignment_generation_prompt.format(
+            num_roles=num_roles, task=task_prompt)
+
+        role_assignment_generation_msg = BaseMessage.make_user_message(
+            role_name="Role Assigner", content=role_assignment_generation)
+
+        response_completion = super().step(
+            input_message=role_assignment_generation_msg)
+
+        output_completion = response_completion.msg  # type: BaseMessage
+        terminated = response_completion.terminated
+        info = response_completion.info
+
+        # Distribute the output completions into role names and descriptions
+        role_names = [
+            desc.replace("<|", "").replace("|>", "") for desc in re.findall(
+                r"Domain expert \d: (.+?)\nAssociated competencies,",
+                output_completion.content,
+                re.DOTALL,
+            )
+        ]
+        role_descriptions = [
+            desc.replace("<|", "").replace("|>", "") for desc in re.findall(
+                r"Associated competencies, characteristics, "
+                r"duties and workflows:(.+?) End.", output_completion.content,
+                re.DOTALL)
+        ]
+
+        if len(role_names) != num_roles or len(role_descriptions) != num_roles:
+            raise RuntimeError(
+                "Got None or insufficient information of roles.")
+        if terminated:
+            raise RuntimeError("Role assignment failed.")
+
+        role_descriptions_dict = {
+            role_name: description
+            for role_name, description in zip(role_names, role_descriptions)
+        }
+
+        return role_names, role_descriptions_dict, terminated, info

camel/prompts/__init__.py

@@ -18,6 +18,7 @@
 from .translation import TranslationPromptTemplateDict
 from .solution_extraction import SolutionExtractionPromptTemplateDict
 from .evaluation import EvaluationPromptTemplateDict
+from .role_description_prompt_template import RoleDescriptionPromptTemplateDict
 from .task_prompt_template import TaskPromptTemplateDict
 from .prompt_templates import PromptTemplateGenerator
 
@@ -30,6 +31,7 @@
     'MisalignmentPromptTemplateDict',
     'TranslationPromptTemplateDict',
     'EvaluationPromptTemplateDict',
+    'RoleDescriptionPromptTemplateDict',
     'TaskPromptTemplateDict',
     'PromptTemplateGenerator',
     'SolutionExtractionPromptTemplateDict',

camel/prompts/ai_society.py

@@ -58,8 +58,14 @@ class AISocietyPromptTemplateDict(TextPromptDict):
 Please reply with the specified task in {word_limit} words or less. Do not add anything else."""
     )
 
-    ASSISTANT_PROMPT: TextPrompt = TextPrompt(
-        """Never forget you are a {assistant_role} and I am a {user_role}. Never flip roles! Never instruct me!
+    ROLE_DESCRIPTION_PROMPT = TextPrompt("""===== ROLES WITH DESCRIPTION =====
+{user_role} and {assistant_role} are collaborating to complete a task: {task}
+{user_role}'s competencies, professional characteristics, duties and workflows to complete the task: {user_description}
+{assistant_role}'s competencies, professional characteristics, duties and workflows to complete the task: {assistant_description}
+""")
+
+    ASSISTANT_PROMPT: TextPrompt = TextPrompt("""===== RULES OF ASSISTANT =====
+Never forget you are a {assistant_role} and I am a {user_role}. Never flip roles! Never instruct me!
 We share a common interest in collaborating to successfully complete a task.
 You must help me to complete the task.
 Here is the task: {task}. Never forget our task!
@@ -75,8 +81,8 @@ class AISocietyPromptTemplateDict(TextPromptDict):
 <YOUR_SOLUTION> should be very specific, include detailed explanations and provide preferable detailed implementations and examples and lists for task-solving.
 Always end <YOUR_SOLUTION> with: Next request.""")
 
-    USER_PROMPT: TextPrompt = TextPrompt(
-        """Never forget you are a {user_role} and I am a {assistant_role}. Never flip roles! You will always instruct me.
+    USER_PROMPT: TextPrompt = TextPrompt("""===== RULES OF USER =====
+Never forget you are a {user_role} and I am a {assistant_role}. Never flip roles! You will always instruct me.
 We share a common interest in collaborating to successfully complete a task.
 I must help you to complete the task.
 Here is the task: {task}. Never forget our task!
@@ -115,6 +121,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             "generate_users": self.GENERATE_USERS,
             "generate_tasks": self.GENERATE_TASKS,
             "task_specify_prompt": self.TASK_SPECIFY_PROMPT,
+            "role_description": self.ROLE_DESCRIPTION_PROMPT,
             RoleType.ASSISTANT: self.ASSISTANT_PROMPT,
             RoleType.USER: self.USER_PROMPT,
             RoleType.CRITIC: self.CRITIC_PROMPT,

camel/prompts/role_description_prompt_template.py

@@ -0,0 +1,54 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# 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
+
+from camel.prompts import AISocietyPromptTemplateDict, TextPrompt
+from camel.typing import RoleType
+
+
+# flake8: noqa :E501
+class RoleDescriptionPromptTemplateDict(AISocietyPromptTemplateDict):
+    r"""A dictionary containing :obj:`TextPrompt` used in the `role description`
+    task.
+
+    Attributes:
+        DEFAULT_ROLE_DESCRIPTION_PROMPT (TextPrompt): A default prompt to
+            describe the role descriptions.
+        ASSISTANT_PROMPT (TextPrompt): A system prompt for the AI assistant
+            that outlines the rules of the conversation and provides
+            instructions for completing tasks.
+        USER_PROMPT (TextPrompt): A system prompt for the AI user that
+            outlines the rules of the conversation and provides instructions
+            for giving instructions to the AI assistant.
+    """
+    DEFAULT_ROLE_DESCRIPTION_PROMPT = TextPrompt(
+        """===== ROLES WITH DESCRIPTION =====
+{user_role} and {assistant_role} are collaborating to complete a task: {task}
+{user_role}'s competencies, characteristics, duties and workflows to complete the task: {user_description}
+{assistant_role}'s competencies, characteristics, duties and workflows to complete the task: {assistant_description}
+""")
+
+    ASSISTANT_PROMPT = TextPrompt(DEFAULT_ROLE_DESCRIPTION_PROMPT +
+                                  AISocietyPromptTemplateDict.ASSISTANT_PROMPT)
+
+    USER_PROMPT = TextPrompt(DEFAULT_ROLE_DESCRIPTION_PROMPT +
+                             AISocietyPromptTemplateDict.USER_PROMPT)
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.update({
+            "default_role_description": self.DEFAULT_ROLE_DESCRIPTION_PROMPT,
+            RoleType.ASSISTANT: self.ASSISTANT_PROMPT,
+            RoleType.USER: self.USER_PROMPT,
+        })

camel/prompts/task_prompt_template.py

@@ -18,6 +18,7 @@
     CodePromptTemplateDict,
     EvaluationPromptTemplateDict,
     MisalignmentPromptTemplateDict,
+    RoleDescriptionPromptTemplateDict,
     SolutionExtractionPromptTemplateDict,
     TextPromptDict,
     TranslationPromptTemplateDict,
@@ -50,4 +51,6 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             EvaluationPromptTemplateDict(),
             TaskType.SOLUTION_EXTRACTION:
             SolutionExtractionPromptTemplateDict(),
+            TaskType.ROLE_DESCRIPTION:
+            RoleDescriptionPromptTemplateDict(),
         })

camel/societies/role_playing.py

@@ -115,6 +115,7 @@ def __init__(
 
         sys_msg_generator = SystemMessageGenerator(
             task_type=self.task_type, **(sys_msg_generator_kwargs or {}))
+
         (init_assistant_sys_msg, init_user_sys_msg,
          sys_msg_meta_dicts) = self.get_sys_message_info(
              assistant_role_name, user_role_name, sys_msg_generator,
@@ -131,7 +132,6 @@ def __init__(
             user_agent_kwargs,
             output_language,
         )
-
         self.critic: Optional[Union[CriticAgent, Human]] = None
         self.critic_sys_msg: Optional[BaseMessage] = None
         self.init_critic(critic_role_name, critic_criteria, critic_kwargs,
@@ -211,9 +211,11 @@ def init_planned_task_prompt(self,
             self.planned_task_prompt = None
 
     def get_sys_message_info(
-        self, assistant_role_name: str, user_role_name: str,
+        self,
+        assistant_role_name: str,
+        user_role_name: str,
         sys_msg_generator: SystemMessageGenerator,
-        extend_sys_msg_meta_dicts: Optional[List[Dict]]
+        extend_sys_msg_meta_dicts: Optional[List[Dict]] = None,
     ) -> Tuple[BaseMessage, BaseMessage, List[Dict]]:
         r"""Get initial assistant and user system message with a list of
         system message meta dicts.
@@ -233,12 +235,41 @@ def get_sys_message_info(
             initial system message, and a list of system message meta dicts.
         """
         sys_msg_meta_dicts = [dict(task=self.task_prompt) for _ in range(2)]
-        if (extend_sys_msg_meta_dicts is None and self.task_type
-                in [TaskType.AI_SOCIETY, TaskType.MISALIGNMENT]):
+        if (extend_sys_msg_meta_dicts is None and self.task_type in [
+                TaskType.AI_SOCIETY,
+                TaskType.MISALIGNMENT,
+        ]):
             extend_sys_msg_meta_dicts = [
                 dict(assistant_role=assistant_role_name,
                      user_role=user_role_name) for _ in range(2)
             ]
+        elif (self.task_type == TaskType.ROLE_DESCRIPTION):
+            if (extend_sys_msg_meta_dicts is None
+                    or len(extend_sys_msg_meta_dicts) != 2):
+                # In `TaskType.ROLE_DESCRIPTION`, `extend_sys_msg_meta_dicts`
+                # should have two elements, one for assistant and one for user
+                raise ValueError("`extend_sys_msg_meta_dicts` should have two "
+                                 "elements for `TaskType.ROLE_DESCRIPTION`.")
+            # Validate `extend_sys_msg_meta_dicts` has `assistant_description`
+            # and `user_description`
+            if ("assistant_description" not in extend_sys_msg_meta_dicts[0]
+                    or "user_description" not in extend_sys_msg_meta_dicts[0]
+                    or "assistant_description"
+                    not in extend_sys_msg_meta_dicts[1]
+                    or "user_description" not in extend_sys_msg_meta_dicts[1]):
+                raise ValueError("Ensure both `assistant_description` and "
+                                 "`user_description` are not None.")
+
+            role_name_msg_meta_dicts = [
+                dict(assistant_role=assistant_role_name,
+                     user_role=user_role_name) for _ in range(2)
+            ]
+            extend_sys_msg_meta_dicts = [{
+                **role_name_msg_meta_dict,
+                **sys_msg_meta_dict
+            } for role_name_msg_meta_dict, sys_msg_meta_dict in zip(
+                role_name_msg_meta_dicts, extend_sys_msg_meta_dicts)]
+
         if extend_sys_msg_meta_dicts is not None:
             sys_msg_meta_dicts = [{
                 **sys_msg_meta_dict,
@@ -426,7 +457,6 @@ def step(
             whether the user agent terminated the conversation, and any
             additional user information.
         """
-
         user_response = self.user_agent.step(assistant_msg)
         if user_response.terminated or user_response.msgs is None:
             return (ChatAgentResponse([], False, {}),

camel/typing.py

@@ -60,6 +60,7 @@ class TaskType(Enum):
     TRANSLATION = "translation"
     EVALUATION = "evaluation"
     SOLUTION_EXTRACTION = "solution_extraction"
+    ROLE_DESCRIPTION = "role_description"
     DEFAULT = "default"
 
 

examples/role_description/role_generation.py

@@ -0,0 +1,43 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+from colorama import Fore
+
+from camel.agents.role_assignment_agent import RoleAssignmentAgent
+from camel.configs import ChatGPTConfig
+
+
+def main(model_type=None, num_roles=3) -> None:
+    task_prompt = "Develop a trading bot for the stock market."
+
+    model_config_description = ChatGPTConfig()
+    role_description_agent = RoleAssignmentAgent(
+        model=model_type, model_config=model_config_description)
+
+    role_names, role_description_dict, _, _ = (
+        role_description_agent.run_role_with_description(
+            task_prompt=task_prompt, num_roles=num_roles))
+
+    if (len(role_names) != num_roles):
+        raise ValueError(f"Length of role_names ({len(role_names)}) "
+                         f"does not equal to num_roles ({num_roles}).")
+
+    print(Fore.YELLOW + f"Original task prompt:\n{task_prompt}\n")
+    print(Fore.GREEN + f"List of {num_roles} roles with description:")
+    for role_name in role_names:
+        print(Fore.BLUE + f"{role_name}:\n"
+              f"{role_description_dict[role_name]}\n")
+
+
+if __name__ == "__main__":
+    main()