Merge dev to dev_tool_selection

metagpt/actions/write_analysis_code.py

@@ -62,14 +62,14 @@ class BaseWriteAnalysisCode(Action):
         return messages
 
     async def run(
-        self, context: List[Message], plan: Plan = None, task_guide: str = ""
+        self, context: List[Message], plan: Plan = None, code_steps: str = ""
     ) -> str:
         """Run of a code writing action, used in data analysis or modeling
 
         Args:
             context (List[Message]): Action output history, source action denoted by Message.cause_by
             plan (Plan, optional): Overall plan. Defaults to None.
-            task_guide (str, optional): suggested step breakdown for the current task. Defaults to "".
+            code_steps (str, optional): suggested step breakdown for the current task. Defaults to "".
 
         Returns:
             str: The code string.
@@ -86,7 +86,7 @@ class WriteCodeByGenerate(BaseWriteAnalysisCode):
         self,
         context: [List[Message]],
         plan: Plan = None,
-        task_guide: str = "",
+        code_steps: str = "",
         system_msg: str = None,
         **kwargs,
     ) -> str:
@@ -152,7 +152,8 @@ class WriteCodeWithTools(BaseWriteAnalysisCode):
         self,
         context: List[Message],
         plan: Plan = None,
-        task_guide: str = "",
+        code_steps: str = "",
+        data_desc: str = "",
     ) -> str:
         task_type = plan.current_task.task_type
         available_tools = registry.get_all_schema_by_module(task_type)
@@ -164,7 +165,7 @@ class WriteCodeWithTools(BaseWriteAnalysisCode):
                 for tool in available_tools
             ]
 
-            recommend_tools = await self._tool_recommendation(context, task_guide, available_tools)
+            recommend_tools = await self._tool_recommendation(context, code_steps, available_tools)
             tool_catalog = self._parse_recommend_tools(task_type, recommend_tools)
             logger.info(f"Recommended tools: \n{recommend_tools}")
 
@@ -172,7 +173,7 @@ class WriteCodeWithTools(BaseWriteAnalysisCode):
             output_desc = TOOL_OUTPUT_DESC.get(task_type, "")
             prompt = TOO_ORGANIZATION_PROMPT.format(
                 special_prompt=special_prompt,
-                code_steps=task_guide,
+                code_steps=code_steps,
                 module_name=module_name,
                 output_desc=output_desc,
                 function_catalog=tool_catalog,

metagpt/actions/write_code_steps.py

@@ -0,0 +1,77 @@
+
+import json
+from typing import Dict, List, Union
+
+from metagpt.actions import Action
+from metagpt.schema import Message, Task, Plan
+
+
+CODE_STEPS_PROMPT_TEMPLATE = """
+# Context
+{context}
+
+##  Format example
+1.
+2.
+3.
+...
+
+-----
+Tasks are all code development tasks.
+You are a professional engineer, the main goal is to plan out concise solution steps for Current Task before coding.
+A planning process can reduce the difficulty and improve the quality of coding.
+You may be given some code plans for the tasks ahead, but you don't have to follow the existing plan when planning the current task.
+The output plan should following the subsequent principles:
+1.The plan is a rough checklist of steps outlining the entire program's structure.Try to keep the number of steps fewer than 5.
+2.The steps should be written concisely and at a high level, avoiding overly detailed implementation specifics.
+3.The execution of the plan happens sequentially, but the plan can incorporate conditional (if) and looping(loop) keywords for more complex structures.
+4.Output carefully referenced "Format example" in format.
+"""
+
+STRUCTURAL_CONTEXT = """
+## User Requirement
+{user_requirement}
+## Current Plan
+{tasks}
+## Current Task
+{current_task}
+"""
+
+
+class WriteCodeSteps(Action):
+
+    async def run(self, plan: Plan) -> str:
+        """Run of a task guide writing action, used in ml engineer
+
+        Args:
+            plan (plan): task plan
+            useful_memories (list): useful_memories
+        Returns:
+            str: The dataset_descriptions string.
+        """
+
+        context = self.get_context(plan)
+        code_steps_prompt = CODE_STEPS_PROMPT_TEMPLATE.format(
+            context=context,
+        )
+        code_steps = await self._aask(code_steps_prompt)
+        return code_steps
+
+    def get_context(self, plan: Plan):
+        user_requirement = plan.goal
+        select_task_keys = ['task_id', 'instruction', 'is_finished', 'code_steps']
+
+        def process_task(task):
+            task_dict = task.dict()
+            ptask = {k: task_dict[k] for k in task_dict if k in select_task_keys}
+            return ptask
+        tasks = json.dumps(
+            [process_task(task) for task in plan.tasks], indent=4, ensure_ascii=False
+        )
+        current_task = json.dumps(process_task(plan.current_task)) if plan.current_task else {}
+        context = STRUCTURAL_CONTEXT.format(
+            user_requirement=user_requirement, tasks=tasks, current_task=current_task
+        )
+        # print(context)
+        return context
+