base software company roles on RoleZero

metagpt/roles/architect.py

@@ -5,13 +5,12 @@
 @Author  : alexanderwu
 @File    : architect.py
 """
-
 from metagpt.actions import WritePRD
 from metagpt.actions.design_api import WriteDesign
-from metagpt.roles.role import Role
+from metagpt.roles.di.role_zero import RoleZero
 
 
-class Architect(Role):
+class Architect(RoleZero):
     """
     Represents an Architect role in a software development process.
 
@@ -30,11 +29,26 @@ class Architect(Role):
         "libraries. Use same language as user requirement"
     )
 
+    instruction: str = """Use WriteDesign tool to write a system design document"""
+    max_react_loop: int = 1  # FIXME: Read and edit files requires more steps, consider later
+    tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WriteDesign"]
+
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
+
+        # NOTE: The following init setting will only be effective when self.use_fixed_sop is changed to True
         self.enable_memory = False
         # Initialize actions specific to the Architect role
         self.set_actions([WriteDesign])
 
         # Set events or actions the Architect should watch or be aware of
         self._watch({WritePRD})
+
+    def _update_tool_execution(self):
+        wd = WriteDesign()
+        self.tool_execution_map.update(
+            {
+                "WriteDesign.run": wd.run,
+                "WriteDesign": wd.run,  # alias
+            }
+        )

metagpt/roles/di/engineer2.py

@@ -1,10 +1,7 @@
 from __future__ import annotations
 
-from pydantic import model_validator
-
 from metagpt.prompts.di.engineer2 import ENGINEER2_INSTRUCTION
 from metagpt.roles.di.role_zero import RoleZero
-from metagpt.tools.libs.editor import Editor
 
 
 class Engineer2(RoleZero):
@@ -14,18 +11,3 @@ class Engineer2(RoleZero):
     instruction: str = ENGINEER2_INSTRUCTION
 
     tools: str = ["Plan", "Editor:write,read,write_content", "RoleZero"]
-    editor: Editor = Editor()
-
-    @model_validator(mode="after")
-    def set_tool_execution(self) -> "RoleZero":
-        self.tool_execution_map = {
-            "Plan.append_task": self.planner.plan.append_task,
-            "Plan.reset_task": self.planner.plan.reset_task,
-            "Plan.replace_task": self.planner.plan.replace_task,
-            "Editor.write": self.editor.write,
-            "Editor.write_content": self.editor.write_content,
-            "Editor.read": self.editor.read,
-            "RoleZero.ask_human": self.ask_human,
-            "RoleZero.reply_to_human": self.reply_to_human,
-        }
-        return self

metagpt/roles/di/role_zero.py

@@ -9,13 +9,14 @@ from pydantic import model_validator
 
 from metagpt.actions import Action
 from metagpt.actions.di.run_command import RunCommand
-from metagpt.environment.mgx.mgx_env import MGXEnv
 from metagpt.logs import logger
 from metagpt.prompts.di.role_zero import CMD_PROMPT, ROLE_INSTRUCTION
 from metagpt.roles import Role
 from metagpt.schema import AIMessage, Message, UserMessage
 from metagpt.strategy.experience_retriever import DummyExpRetriever, ExpRetriever
 from metagpt.strategy.planner import Planner
+from metagpt.tools.libs.browser import Browser
+from metagpt.tools.libs.editor import Editor
 from metagpt.tools.tool_recommend import BM25ToolRecommender, ToolRecommender
 from metagpt.tools.tool_registry import register_tool
 from metagpt.utils.common import CodeParser
@@ -43,6 +44,10 @@ class RoleZero(Role):
     tool_recommender: ToolRecommender = None
     tool_execution_map: dict[str, callable] = {}
     special_tool_commands: list[str] = ["Plan.finish_current_task", "end"]
+    # Equipped with three basic tools by default for optional use
+    editor: Editor = Editor()
+    browser: Browser = Browser()
+    # terminal: Terminal = Terminal()  # FIXME: TypeError: cannot pickle '_thread.lock' object
 
     # Experience
     experience_retriever: ExpRetriever = DummyExpRetriever()
@@ -64,7 +69,6 @@ class RoleZero(Role):
         if self.tools and not self.tool_recommender:
             self.tool_recommender = BM25ToolRecommender(tools=self.tools, force=True)
         self.set_actions([RunCommand])
-        self._set_state(0)
 
         # HACK: Init Planner, control it through dynamic thinking; Consider formalizing as a react mode
         self.planner = Planner(goal="", working_memory=self.rc.working_memory, auto_run=True)
@@ -73,7 +77,23 @@ class RoleZero(Role):
 
     @model_validator(mode="after")
     def set_tool_execution(self) -> "RoleZero":
-        raise NotImplementedError
+        # default map
+        self.tool_execution_map = {
+            "Plan.append_task": self.planner.plan.append_task,
+            "Plan.reset_task": self.planner.plan.reset_task,
+            "Plan.replace_task": self.planner.plan.replace_task,
+            "Editor.write": self.editor.write,
+            "Editor.write_content": self.editor.write_content,
+            "Editor.read": self.editor.read,
+            "RoleZero.ask_human": self.ask_human,
+            "RoleZero.reply_to_human": self.reply_to_human,
+        }
+        # can be updated by subclass
+        self._update_tool_execution()
+        return self
+
+    def _update_tool_execution(self):
+        pass
 
     async def _think(self) -> bool:
         """Useful in 'react' mode. Use LLM to decide whether and what to do next."""
@@ -82,9 +102,9 @@ class RoleZero(Role):
             return await super()._think()
 
         ### 0. Preparation ###
-        if not self.rc.todo and not self.rc.news:
+        if not self.rc.todo:
             return False
-        self._set_state(0)
+
         if not self.planner.plan.goal:
             self.user_requirement = self.get_memories()[-1].content
             self.planner.plan.goal = self.user_requirement
@@ -118,7 +138,7 @@ class RoleZero(Role):
             instruction=self.instruction.strip(),
         )
         context = self.llm.format_msg(self.rc.memory.get(self.memory_k) + [UserMessage(content=prompt)])
-        print(*context, sep="\n" + "*" * 5 + "\n")
+        # print(*context, sep="\n" + "*" * 5 + "\n")
         async with ThoughtReporter():
             self.command_rsp = await self.llm.aask(context, system_msgs=self.system_msg)
         self.rc.memory.add(AIMessage(content=self.command_rsp))
@@ -146,11 +166,15 @@ class RoleZero(Role):
         )
 
     async def _react(self) -> Message:
+        # NOTE: Diff 1: Each time landing here means observing news, set todo to allow news processing in _think
+        self._set_state(0)
+
         actions_taken = 0
         rsp = AIMessage(content="No actions taken yet", cause_by=Action)  # will be overwritten after Role _act
         while actions_taken < self.rc.max_react_loop:
-            # NOTE: difference here, keep observing within react
+            # NOTE: Diff 2: Keep observing within _react, news will go into memory, allowing adapting to new info
             await self._observe()
+
             # think
             has_todo = await self._think()
             if not has_todo:
@@ -215,6 +239,8 @@ class RoleZero(Role):
     async def ask_human(self, question: str) -> str:
         """Use this when you fail the current task or if you are unsure of the situation encountered. Your response should contain a brief summary of your situation, ended with a clear and concise question."""
         # NOTE: Can be overwritten in remote setting
+        from metagpt.environment.mgx.mgx_env import MGXEnv  # avoid circular import
+
         if not isinstance(self.rc.env, MGXEnv):
             return "Not in MGXEnv, command will not be executed."
         return await self.rc.env.get_human_input(question, sent_from=self)
@@ -222,6 +248,8 @@ class RoleZero(Role):
     async def reply_to_human(self, content: str) -> str:
         """Reply to human user with the content provided. Use this when you have a clear answer or solution to the user's question."""
         # NOTE: Can be overwritten in remote setting
+        from metagpt.environment.mgx.mgx_env import MGXEnv  # avoid circular import
+
         if not isinstance(self.rc.env, MGXEnv):
             return "Not in MGXEnv, command will not be executed."
         return await self.rc.env.reply_to_human(content, sent_from=self)

metagpt/roles/di/team_leader.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from pydantic import model_validator
-
 from metagpt.actions.di.run_command import RunCommand
 from metagpt.prompts.di.team_leader import (
     FINISH_CURRENT_TASK_CMD,
@@ -26,17 +24,13 @@ class TeamLeader(RoleZero):
 
     experience_retriever: ExpRetriever = SimpleExpRetriever()
 
-    @model_validator(mode="after")
-    def set_tool_execution(self) -> "RoleZero":
-        self.tool_execution_map = {
-            "Plan.append_task": self.planner.plan.append_task,
-            "Plan.reset_task": self.planner.plan.reset_task,
-            "Plan.replace_task": self.planner.plan.replace_task,
-            "RoleZero.ask_human": self.ask_human,
-            "RoleZero.reply_to_human": self.reply_to_human,
-            "TeamLeader.publish_team_message": self.publish_team_message,
-        }
-        return self
+    def _update_tool_execution(self):
+        self.tool_execution_map.update(
+            {
+                "TeamLeader.publish_team_message": self.publish_team_message,
+                "TeamLeader.publish_message": self.publish_team_message,  # alias
+            }
+        )
 
     def set_instruction(self):
         team_info = ""

metagpt/roles/product_manager.py

@@ -7,15 +7,15 @@
 @Modified By: mashenquan, 2023/11/27. Add `PrepareDocuments` action according to Section 2.2.3.5.1 of RFC 135.
 """
 
-
 from metagpt.actions import UserRequirement, WritePRD
 from metagpt.actions.prepare_documents import PrepareDocuments
-from metagpt.roles.role import Role, RoleReactMode
+from metagpt.roles.di.role_zero import RoleZero
+from metagpt.roles.role import RoleReactMode
 from metagpt.utils.common import any_to_name, any_to_str
 from metagpt.utils.git_repository import GitRepository
 
 
-class ProductManager(Role):
+class ProductManager(RoleZero):
     """
     Represents a Product Manager role responsible for product development and management.
 
@@ -30,18 +30,35 @@ class ProductManager(Role):
     profile: str = "Product Manager"
     goal: str = "efficiently create a successful product that meets market demands and user expectations"
     constraints: str = "utilize the same language as the user requirements for seamless communication"
-    todo_action: str = ""
+    todo_action: str = any_to_name(WritePRD)
+
+    instruction: str = """Use WritePRD tool to write PRD"""
+    max_react_loop: int = 1  # FIXME: Read and edit files requires more steps, consider later
+    tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WritePRD"]
 
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
+        # NOTE: The following init setting will only be effective when self.use_fixed_sop is changed to True
         self.enable_memory = False
         self.set_actions([PrepareDocuments(send_to=any_to_str(self)), WritePRD])
         self._watch([UserRequirement, PrepareDocuments])
-        self.rc.react_mode = RoleReactMode.BY_ORDER
-        self.todo_action = any_to_name(WritePRD)
+        if self.use_fixed_sop:
+            self.rc.react_mode = RoleReactMode.BY_ORDER
+
+    def _update_tool_execution(self):
+        wp = WritePRD()
+        self.tool_execution_map.update(
+            {
+                "WritePRD.run": wp.run,
+                "WritePRD": wp.run,  # alias
+            }
+        )
 
     async def _think(self) -> bool:
         """Decide what to do"""
+        if not self.use_fixed_sop:
+            return await super()._think()
+
         if GitRepository.is_git_dir(self.config.project_path) and not self.config.git_reinit:
             self._set_state(1)
         else:

metagpt/roles/project_manager.py

@@ -5,13 +5,12 @@
 @Author  : alexanderwu
 @File    : project_manager.py
 """
-
 from metagpt.actions import WriteTasks
 from metagpt.actions.design_api import WriteDesign
-from metagpt.roles.role import Role
+from metagpt.roles.di.role_zero import RoleZero
 
 
-class ProjectManager(Role):
+class ProjectManager(RoleZero):
     """
     Represents a Project Manager role responsible for overseeing project execution and team efficiency.
 
@@ -30,8 +29,22 @@ class ProjectManager(Role):
     )
     constraints: str = "use same language as user requirement"
 
+    instruction: str = """Use WriteTasks tool to write a project task list"""
+    max_react_loop: int = 1  # FIXME: Read and edit files requires more steps, consider later
+    tools: list[str] = ["Editor:write,read,write_content", "RoleZero", "WriteTasks"]
+
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
+        # NOTE: The following init setting will only be effective when self.use_fixed_sop is changed to True
         self.enable_memory = False
         self.set_actions([WriteTasks])
         self._watch([WriteDesign])
+
+    def _update_tool_execution(self):
+        wt = WriteTasks()
+        self.tool_execution_map.update(
+            {
+                "WriteTasks.run": wt.run,
+                "WriteTasks": wt.run,  # alias
+            }
+        )