improve details

metagpt/actions/ci/execute_nb_code.py

@@ -39,10 +39,9 @@ class ExecuteNbCode(Action):
 
     def __init__(
         self,
-        nb=None,
+        nb=nbformat.v4.new_notebook(),
         timeout=600,
     ):
-        nb = nb or nbformat.v4.new_notebook()
         super().__init__(
             nb=nb,
             nb_client=NotebookClient(nb, timeout=timeout),
@@ -199,17 +198,10 @@ class ExecuteNbCode(Action):
 
 def truncate(result: str, keep_len: int = 2000, is_success: bool = True):
     """对于超出keep_len个字符的result: 执行失败的代码, 展示result后keep_len个字符; 执行成功的代码, 展示result前keep_len个字符。"""
-    desc = f"Executed code {'successfully. ' if is_success else 'failed, please reflect the cause of bug and then debug. '}"
-    is_same_desc = False
-
     if is_success:
-        desc += f"Truncated to show only first {keep_len} characters\n"
+        desc = f"Executed code successfully. Truncated to show only first {keep_len} characters\n"
     else:
-        desc += f"Truncated to show only last {keep_len} characters\n"
-
-    if result.startswith(desc):
-        result = result[len(desc) :]
-        is_same_desc = True
+        desc = f"Executed code failed, please reflect the cause of bug and then debug. Truncated to show only last {keep_len} characters\n"
 
     if result.strip().startswith("<coroutine object"):
         result = "Executed code failed, you need use key word 'await' to run a async code."
@@ -219,7 +211,7 @@ def truncate(result: str, keep_len: int = 2000, is_success: bool = True):
         result = result[-keep_len:] if not is_success else result[:keep_len]
         return desc + result, is_success
 
-    return result if not is_same_desc else desc + result, is_success
+    return result, is_success
 
 
 def remove_escape_and_color_codes(input_str: str):

metagpt/actions/ci/write_analysis_code.py

@@ -56,7 +56,7 @@ class WriteCodeWithoutTools(BaseWriteAnalysisCode):
 class WriteCodeWithTools(BaseWriteAnalysisCode):
     """Write code with help of local available tools. Choose tools first, then generate code to use the tools"""
 
-    # selected tools to choose from, listed by their names. En empty list means selection from all tools.
+    # selected tools to choose from, listed by their names. An empty list means selection from all tools.
     selected_tools: list[str] = []
 
     def _get_tools_by_type(self, tool_type: str) -> dict:
@@ -71,18 +71,15 @@ class WriteCodeWithTools(BaseWriteAnalysisCode):
         """
         candidate_tools = TOOL_REGISTRY.get_tools_by_type(tool_type)
         if self.selected_tools:
-            candidate_tools = {
-                tool_name: candidate_tools[tool_name]
-                for tool_name in self.selected_tools
-                if tool_name in candidate_tools
-            }
+            candidate_tool_names = set(self.selected_tools) & candidate_tools.keys()
+            candidate_tools = {tool_name: candidate_tools[tool_name] for tool_name in candidate_tool_names}
         return candidate_tools
 
     async def _recommend_tool(
         self,
         task: str,
         available_tools: dict,
-    ) -> list:
+    ) -> dict:
         """
         Recommend tools for the specified task.
 

metagpt/actions/ci/write_plan.py

@@ -49,19 +49,19 @@ class WritePlan(Action):
             tasks (list[dict]): tasks to be assigned task type
 
         Returns:
-            list[dict]: tasks with task type assigned
+            str: tasks with task type assigned in a json string
         """
-        task_list = "\n".join([f"Task {task['task_id']}: {task['instruction']}" for task in tasks])
+        task_info = "\n".join([f"Task {task['task_id']}: {task['instruction']}" for task in tasks])
         task_type_desc = "\n".join(
             [f"- **{tool_type.name}**: {tool_type.desc}" for tool_type in TOOL_REGISTRY.get_tool_types().values()]
         )  # task type are binded with tool type now, should be improved in the future
         prompt = ASSIGN_TASK_TYPE_PROMPT.format(
-            task_list=task_list, task_type_desc=task_type_desc
+            task_info=task_info, task_type_desc=task_type_desc
         )  # task types are set to be the same as tool types, for now
         tool_config = create_func_call_config(ASSIGN_TASK_TYPE_CONFIG)
         rsp = await self.llm.aask_code(prompt, **tool_config)
         task_type_list = rsp["task_type"]
-        print(f"assigned task types: {task_type_list}")
+        logger.info(f"assigned task types: {task_type_list}")
         for task, task_type in zip(tasks, task_type_list):
             task["task_type"] = task_type
         return json.dumps(tasks)