rm fixed sop logic from mgx_env to simplify code

metagpt/environment/mgx/mgx_env.py

@@ -1,27 +1,16 @@
 from __future__ import annotations
 
-from metagpt.actions import (
-    UserRequirement,
-    WriteDesign,
-    WritePRD,
-    WriteTasks,
-    WriteTest,
-)
-from metagpt.actions.summarize_code import SummarizeCode
 from metagpt.const import AGENT, IMAGES, MESSAGE_ROUTE_TO_ALL, TEAMLEADER_NAME
 from metagpt.environment.base_env import Environment
 from metagpt.logs import get_human_input
-from metagpt.roles import Architect, ProductManager, ProjectManager, Role
+from metagpt.roles import Role
 from metagpt.schema import Message, SerializationMixin
-from metagpt.utils.common import any_to_str, any_to_str_set, extract_and_encode_images
+from metagpt.utils.common import extract_and_encode_images
 
 
 class MGXEnv(Environment, SerializationMixin):
     """MGX Environment"""
 
-    # If True, fixed software sop bypassing TL is allowed, otherwise, TL will fully take over the routing
-    allow_bypass_team_leader: bool = False
-
     direct_chat_roles: set[str] = set()  # record direct chat: @role_name
 
     is_public_chat: bool = True
@@ -54,29 +43,6 @@ class MGXEnv(Environment, SerializationMixin):
             # direct chat response from a certain role to human user, team leader and other roles in the env should not be involved, no need to publish
             self.direct_chat_roles.remove(message.sent_from)
 
-        elif (
-            self.allow_bypass_team_leader
-            and self.message_within_software_sop(message)
-            and not self.has_user_requirement()
-        ):
-            # Quick routing for messages within software SOP, bypassing TL.
-            # Use rules to check for user intervention and to finish task.
-            # NOTE: This escapes TL's supervision and has pitfalls such as routing obsolete messages even if TL has acquired a new user requirement.
-            #       In addition, we should not determine the status of a task based on message cause_by.
-            #       Consider replacing this in the future.
-            self._publish_message(message)
-            if self.is_software_task_finished(message):
-                tl.rc.memory.add(self.move_message_info_to_content(message))
-                from metagpt.utils.report import CURRENT_ROLE
-
-                role = CURRENT_ROLE.get(None)
-                if role:
-                    CURRENT_ROLE.set(tl)
-                    tl.finish_current_task()
-                    CURRENT_ROLE.set(role)
-                else:
-                    tl.finish_current_task()
-
         elif publicer == tl.profile:
             if message.send_to == {"no one"}:
                 # skip the dummy message from team leader
@@ -102,21 +68,6 @@ class MGXEnv(Environment, SerializationMixin):
         # NOTE: Can be overwritten in remote setting
         return "SUCCESS, human has received your reply. Refrain from resending duplicate messages.  If you no longer need to take action, use the command ‘end’ to stop."
 
-    def message_within_software_sop(self, message: Message) -> bool:
-        # Engineer, QaEngineer can be end of the SOP. Their msg requires routing outside.
-        members_concerned = [ProductManager, Architect, ProjectManager]
-        return message.sent_from in any_to_str_set(members_concerned)
-
-    def has_user_requirement(self, k=1) -> bool:
-        """A heuristics to check if there is a recent user intervention"""
-        return any_to_str(UserRequirement) in [msg.cause_by for msg in self.history.get(k)]
-
-    def is_software_task_finished(self, message: Message) -> bool:
-        """Use a hard-coded rule to check if one software task is finished"""
-        return message.cause_by in any_to_str_set([WritePRD, WriteDesign, WriteTasks, SummarizeCode]) or (
-            message.cause_by == any_to_str(WriteTest) and "Exceeding" in message.content
-        )
-
     def move_message_info_to_content(self, message: Message) -> Message:
         """Two things here:
         1. Convert role, since role field must be reserved for LLM API, and is limited to, for example, one of ["user", "assistant", "system"]

tests/metagpt/environment/mgx_env/run_mgx_env.py

@@ -18,7 +18,7 @@ async def main(requirement="", enable_human_input=False, use_fixed_sop=False, al
     else:
         engineer = Engineer2()
 
-    env = MGXEnv(allow_bypass_team_leader=use_fixed_sop)
+    env = MGXEnv()
     env.add_roles(
         [
             TeamLeader(),