add deploy plan for da

metagpt/prompts/di/data_analyst.py

@@ -23,6 +23,7 @@ Pay close attention to new user message, review the conversation history, use re
 Note:
 1. If you keeping encountering errors, unexpected situation, or you are not sure of proceeding, use ask_human to ask for help.
 2. Each time you finish a task, use reply_to_human to report your progress.
+Pay close attention to the Example provided, you can reuse the example for your current situation if it fits.
 
 You may use any of the available commands to create a plan or update the plan. You may output mutiple commands, they will be executed sequentially.
 If you finish current task, you will automatically take the next task in the existing plan, use finish_task, DON'T append a new task.

metagpt/roles/di/data_analyst.py

@@ -11,6 +11,7 @@ from metagpt.logs import logger
 from metagpt.prompts.di.data_analyst import CMD_PROMPT
 from metagpt.roles.di.data_interpreter import DataInterpreter
 from metagpt.schema import Message, TaskResult
+from metagpt.strategy.experience_retriever import KeywordExpRetriever
 from metagpt.strategy.planner import Planner
 from metagpt.strategy.thinking_command import (
     Command,
@@ -61,9 +62,11 @@ class DataAnalyst(DataInterpreter):
     async def _think(self) -> bool:
         """Useful in 'react' mode. Use LLM to decide whether and what to do next."""
         self._set_state(0)
+        example = ""
         if not self.planner.plan.goal:
             self.user_requirement = self.get_memories()[-1].content
             self.planner.plan.goal = self.user_requirement
+            example = KeywordExpRetriever().retrieve(self.user_requirement)
         else:
             self.working_memory.add_batch(self.rc.news)
 
@@ -71,7 +74,6 @@ class DataAnalyst(DataInterpreter):
         for task in plan_status["tasks"]:
             task.pop("code")
             task.pop("result")
-        example = ""
         prompt = CMD_PROMPT.format(
             plan_status=plan_status,
             example=example,

metagpt/strategy/experience_retriever.py

@@ -154,3 +154,49 @@ class SimpleExpRetriever(ExpRetriever):
 
     def retrieve(self, context: str = "") -> str:
         return self.EXAMPLE
+
+
+class KeywordExpRetriever(ExpRetriever):
+    """An experience retriever that returns examples based on keywords in the context."""
+
+    EXAMPLE: dict = {
+        "deploy": """
+        ## example 1
+        User Requirement: launch a service from workspace/web_snake_game/web_snake_game, and deploy it to public
+        Explanation: Launching a service requires Terminal tool with daemon mode, write this into task instruction.
+        ```json
+        [
+            {
+                "command_name": "append_task",
+                "args": {
+                    "task_id": "1",
+                    "dependent_task_ids": [],
+                    "instruction": "Use the Terminal tool to launch the service in daemon mode",
+                    "assignee": "David"
+                }
+            },
+            {
+                "command_name": "append_task",
+                "args": {
+                    "task_id": "2",
+                    "dependent_task_ids": ["1"],
+                    "instruction": "Test the service with a simple request",
+                    "assignee": "David"
+                }
+            },
+            {
+                "command_name": "append_task",
+                "args": {
+                    "task_id": "3",
+                    "dependent_task_ids": ["2"],
+                    "instruction": "Deploy the service to public",
+                    "assignee": "David"
+                }
+            },
+        """
+    }
+
+    def retrieve(self, context: str) -> str:
+        if "deploy" in context.lower():
+            return self.EXAMPLE["deploy"]
+        return ""

metagpt/strategy/thinking_command.py

@@ -76,7 +76,8 @@ def prepare_command_prompt(commands: list[Command]) -> str:
 
 
 async def run_env_command(role: Role, cmd: list[dict], role_memory: Memory = None):
-    assert isinstance(role.rc.env, MGXEnv), "TeamLeader should only be used in an MGXEnv"
+    if not isinstance(role.rc.env, MGXEnv):
+        return
     if cmd["command_name"] == Command.PUBLISH_MESSAGE.cmd_name:
         role.publish_message(Message(**cmd["args"]))
     if cmd["command_name"] == Command.ASK_HUMAN.cmd_name: