merge commands for better maintenance

metagpt/prompts/di/team_leader.py

@@ -8,7 +8,7 @@ def prepare_command_prompt(commands: list[Command]) -> str:
     return command_prompt
 
 
-PLANNING_CMD_PROMPT = """
+CMD_PROMPT = """
 # Data Structure
 class Task(BaseModel):
     task_id: str = ""
@@ -32,7 +32,10 @@ class Task(BaseModel):
 # Instructions
 You are a team leader, and you are responsible for drafting tasks and routing tasks to your team members.
 You should NOT assign consecutive tasks to the same team member, instead, assign an aggregated task (or the complete requirement) and let the team member to decompose it.
+When creating a new plan involving multiple members, create all tasks at once.
 If plan is created, you should track the progress based on team member feedback message, and update plan accordingly, such as finish_current_task, reset_task, replace_task, etc.
+You should publish_message to team members, asking them to start their task.
+Pay close attention to new user message, review the conversation history, use reply_to_human to respond to the user directly, DON'T ask your team members.
 
 Note:
 1. If the requirement is a pure DATA-RELATED requirement, such as bug fixes, issue reporting, environment setup, terminal operations, pip install, web browsing, web scraping, web searching, web imitation, data science, data analysis, machine learning, deep learning, text-to-image etc. DON'T decompose it, assign a single task with the original user requirement as instruction directly to Data Analyst.
@@ -55,31 +58,3 @@ Some text indicating your thoughts, including how you categorize the requirement
 ]
 ```
 """
-
-ROUTING_CMD_PROMPT = """
-# Team Member Info
-{team_info}
-
-# Available Commands
-{available_commands}
-
-# Current Plan
-{plan_status}
-
-# Example
-{example}
-
-# Instructions
-You are a team leader, you can publish_message to team members, asking them to start their task.
-If there are new user message, review the conversation history and think about what you should do next, you can use any of the Available Commands.
-
-# Your commands in a json array, in the following output format:
-```json
-[
-    {{
-        "command_name": str,
-        "args": {{"arg_name": arg_value, ...}}
-    }},
-    ...
-]
-"""

metagpt/roles/di/team_leader.py

@@ -5,11 +5,7 @@ import json
 from pydantic import model_validator
 
 from metagpt.environment.mgx.mgx_env import MGXEnv
-from metagpt.prompts.di.team_leader import (
-    PLANNING_CMD_PROMPT,
-    ROUTING_CMD_PROMPT,
-    prepare_command_prompt,
-)
+from metagpt.prompts.di.team_leader import CMD_PROMPT, prepare_command_prompt
 from metagpt.roles import Role
 from metagpt.schema import Message, Task, TaskResult
 from metagpt.strategy.experience_retriever import SimplePlanningExpRetriever
@@ -22,15 +18,11 @@ class TeamLeader(Role):
     name: str = "Tim"
     profile: str = "Team Leader"
     task_result: TaskResult = None
-    planning_commands: list[Command] = [
+    commands: list[Command] = [
         Command.APPEND_TASK,
         Command.RESET_TASK,
         Command.REPLACE_TASK,
         Command.FINISH_CURRENT_TASK,
-        Command.REPLY_TO_HUMAN,
-        Command.PASS,
-    ]
-    env_commands: list[Command] = [
         Command.PUBLISH_MESSAGE,
         Command.ASK_HUMAN,
         Command.REPLY_TO_HUMAN,
@@ -45,7 +37,7 @@ class TeamLeader(Role):
     def _run_env_command(self, cmd):
         assert isinstance(self.rc.env, MGXEnv), "TeamLeader should only be used in an MGXEnv"
         if cmd["command_name"] == Command.PUBLISH_MESSAGE.cmd_name:
-            self.rc.env.publish_message(Message(**cmd["args"]), publicer=self.profile)
+            self.rc.env.publish_message(Message(sent_from=self.profile, **cmd["args"]), publicer=self.profile)
         elif cmd["command_name"] == Command.ASK_HUMAN.cmd_name:
             self.rc.env.ask_human(**cmd["args"])
         elif cmd["command_name"] == Command.REPLY_TO_HUMAN.cmd_name:
@@ -72,8 +64,8 @@ class TeamLeader(Role):
         if self.planner.plan.is_plan_finished():
             self._set_state(-1)
 
-    def get_memory(self) -> list[Message]:
-        mem = self.rc.memory.get()
+    def get_memory(self, k=10) -> list[Message]:
+        mem = self.rc.memory.get(k=k)
         for m in mem:
             if m.role not in ["system", "user", "assistant"]:
                 m.content = f"from {m.role} to {m.send_to}: {m.content}"
@@ -84,51 +76,33 @@ class TeamLeader(Role):
         """Useful in 'react' mode. Use LLM to decide whether and what to do next."""
         self.commands = []
 
-        example = ""
         if not self.planner.plan.goal:
             user_requirement = self.get_memories()[-1].content
             self.planner.plan.goal = user_requirement
-            example = SimplePlanningExpRetriever().retrieve()
 
-        # common info
+        plan_status = self.planner.plan.model_dump(include=["goal", "tasks"])
+        for task in plan_status["tasks"]:
+            task.pop("code")
+            task.pop("result")
         team_info = ""
         for role in self.rc.env.roles.values():
             if role.profile == "TeamLeader":
                 continue
             team_info += f"{role.name}: {role.profile}, {role.goal}\n"
-        # print(team_info)
+        example = SimplePlanningExpRetriever().retrieve()
 
-        # plan commands
-        plan_status = self.planner.plan.model_dump(include=["goal", "tasks"])
-        for task in plan_status["tasks"]:
-            task.pop("code")
-            task.pop("result")
-        plan_prompt = PLANNING_CMD_PROMPT.format(
+        prompt = CMD_PROMPT.format(
             plan_status=plan_status,
             team_info=team_info,
             example=example,
-            available_commands=prepare_command_prompt(self.planning_commands),
+            available_commands=prepare_command_prompt(self.commands),
         )
-        context = self.llm.format_msg(self.get_memory() + [Message(content=plan_prompt, role="user")])
+        context = self.llm.format_msg(self.get_memory() + [Message(content=prompt, role="user")])
 
-        plan_rsp = await self.llm.aask(context)
-        plan_rsp_dict = json.loads(CodeParser.parse_code(block=None, text=plan_rsp))
-        self.commands.extend(plan_rsp_dict)
-        self.rc.memory.add(Message(content=plan_rsp, role="assistant"))
-
-        # routing commands
-        route_prompt = ROUTING_CMD_PROMPT.format(
-            plan_status=plan_status,
-            team_info=team_info,
-            example="",
-            available_commands=prepare_command_prompt(self.env_commands),
-        )
-        context = self.llm.format_msg(self.get_memory() + [Message(content=route_prompt, role="user")])
-
-        route_rsp = await self.llm.aask(context)
-        route_rsp_dict = json.loads(CodeParser.parse_code(block=None, text=route_rsp))
-        self.commands.extend(route_rsp_dict)
-        self.rc.memory.add(Message(content=route_rsp, role="assistant"))
+        rsp = await self.llm.aask(context)
+        rsp_dict = json.loads(CodeParser.parse_code(block=None, text=rsp))
+        self.commands.extend(rsp_dict)
+        self.rc.memory.add(Message(content=rsp, role="assistant"))
 
         return True
 

metagpt/strategy/experience_retriever.py

@@ -61,12 +61,19 @@ class SimplePlanningExpRetriever(ExpRetriever):
                 "instruction": "Write comprehensive tests for the game logic and user interface to ensure functionality and reliability.",
                 "assignee": "Edward"
             }
+        },
+        {
+            "command_name": "publish_message",
+            "args": {
+                "content": "User request to create a cli snake game. Please create a product requirement document (PRD) outlining the features, user interface, and user experience of the snake game.",
+                "send_to": "Alice"
+            }
         }
     ]
     ```
 
     ## example 2
-    User requirement: Run data analysis on sklearn Wine recognition dataset, include a plot, and train a model to predict wine class (20% as validation), and show validation accuracy.
+    User Requirement: Run data analysis on sklearn Wine recognition dataset, include a plot, and train a model to predict wine class (20% as validation), and show validation accuracy.
     Explanation: DON'T decompose requirement if it is a DATA-RELATED task, assign a single task directly to Data Analyst David. He will manage the decomposition and implementation.
     ```json
     [
@@ -78,6 +85,13 @@ class SimplePlanningExpRetriever(ExpRetriever):
                 "instruction": "Run data analysis on sklearn Wine recognition dataset, include a plot, and train a model to predict wine class (20% as validation), and show validation accuracy.",
                 "assignee": "David"
             }
+        },
+        {
+            "command_name": "publish_message",
+            "args": {
+                "content": "Run data analysis on sklearn Wine recognition dataset, include a plot, and train a model to predict wine class (20% as validation), and show validation accuracy.",
+                "send_to": "David"
+            }
         }
     ]
     ```
@@ -88,12 +102,33 @@ class SimplePlanningExpRetriever(ExpRetriever):
         ...,
         {'role': 'assistant', 'content': 'from Alice(Product Manager) to {'Bob'}: {'docs': {'20240424153821.json': {'root_path': 'docs/prd', 'filename': '20240424153821.json', 'content': '{"Language":"en_us","Programming Language":"Python","Original Requirements":"create a cli snake game","Project Name":"snake_game","Product Goals":["Develop an intuitive and addictive snake game",...], ...}}}}},
     ]
-    Explanation: You received a message from Alice, the Product Manager, that she has completed the PRD, use finish_current_task, this marks her task as finished and moves the plan to the next task.
+    Explanation: You received a message from Alice, the Product Manager, that she has completed the PRD, use finish_current_task to mark her task as finished and moves the plan to the next task. Based on plan status, next task is for Bob (Architect), publish a message asking him to start. The message content should contain important path info.
     ```json
     [
         {
             "command_name": "finish_current_task",
             "args": {}
+        },
+        {
+        "command_name": "publish_message",
+            "args": {
+                "content": "Please design the software architecture for the snake game based on the PRD created by Alice. The PRD is at 'docs/prd/20240424153821.json'. Include the choice of programming language, libraries, and data flow, etc.",
+                "send_to": "Bob"
+            }
+        }
+    ]
+    ```
+
+    ## example 4
+    User Question: how does the project go?
+    Explanation: The user is asking for a general update on the project status. Give a straight answer about the current task the team is working on and provide a summary of the completed tasks.
+    ```json
+    [
+        {
+            "command_name": "reply_to_human",
+            "args": {
+                "content": "The team is currently working on ... We have completed ...",
+            }
         }
     ]
     ```

tests/metagpt/roles/di/test_team_leader.py

@@ -132,8 +132,38 @@ async def test_plan_update_and_routing(env):
 
     # TL should mark current task as finished, and forward Product Manager's message to Architect
     # Current task should be updated to the second task
-    plan_cmd = tl.commands[:-1]
+    plan_cmd = tl.commands[0]
     route_cmd = tl.commands[-1]
-    assert "finish_current_task" in [cmd["command_name"] for cmd in plan_cmd]
+    assert plan_cmd["command_name"] == "finish_current_task"
     assert route_cmd["command_name"] == "publish_message"
+    assert route_cmd["args"]["send_to"] == Architect().name
     assert tl.planner.plan.current_task_id == "2"
+
+    # Next step, assuming Architect finishes its task
+    env.publish_message(Message(content=DESIGN_CONTENT, role="Bob(Architect)", sent_from="Bob"))
+    await tl.run()
+    plan_cmd = tl.commands[0]
+    route_cmd = tl.commands[-1]
+    assert plan_cmd["command_name"] == "finish_current_task"
+    assert route_cmd["command_name"] == "publish_message"
+    assert route_cmd["args"]["send_to"] == ProjectManager().name
+    assert tl.planner.plan.current_task_id == "3"
+
+
+@pytest.mark.asyncio
+async def test_reply_to_human(env):
+    requirement = "create a 2048 game"
+
+    tl = env.get_role("Team Leader")
+    env.publish_message(Message(content=requirement))
+    await tl.run()
+
+    # Assuming Product Manager finishes its task
+    env.publish_message(Message(content=PRD_MSG_CONTENT, role="Alice(Product Manager)", sent_from="Alice"))
+    await tl.run()
+
+    # Human inquires about the progress
+    env.publish_message(Message(content="Who is working? How does the project go?"))
+    await tl.run()
+
+    assert tl.commands[0]["command_name"] == "reply_to_human"