log & small bug fixed

metagpt/roles/di/role_zero.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import inspect
 import json
 import traceback
-from typing import Literal
+from typing import Literal, Tuple
 
 from pydantic import model_validator
 
@@ -113,17 +113,7 @@ class RoleZero(Role):
         example = self._retrieve_experience()
 
         ### 2. Plan Status ###
-        plan_status = self.planner.plan.model_dump(include=["goal", "tasks"])
-        for task in plan_status["tasks"]:
-            task.pop("code")
-            task.pop("result")
-            task.pop("is_success")
-        # print(plan_status)
-        current_task = (
-            self.planner.plan.current_task.model_dump(exclude=["code", "result", "is_success"])
-            if self.planner.plan.current_task
-            else ""
-        )
+        plan_status, current_task = self._get_plan_status()
 
         ### 3. Tool/Command Info ###
         tools = await self.tool_recommender.recommend_tools()
@@ -205,7 +195,7 @@ class RoleZero(Role):
                     outputs.append(output)
                 except Exception as e:
                     tb = traceback.format_exc()
-                    print(e, tb)
+                    logger.exception(e + tb)
                     outputs.append(output + f": {tb}")
                     break  # Stop executing if any command fails
             else:
@@ -229,6 +219,20 @@ class RoleZero(Role):
 
         return is_special_cmd
 
+    def _get_plan_status(self) -> Tuple[str, str]:
+        plan_status = self.planner.plan.model_dump(include=["goal", "tasks"])
+        for task in plan_status["tasks"]:
+            task.pop("code")
+            task.pop("result")
+            task.pop("is_success")
+        # print(plan_status)
+        current_task = (
+            self.planner.plan.current_task.model_dump(exclude=["code", "result", "is_success"])
+            if self.planner.plan.current_task
+            else ""
+        )
+        return plan_status, current_task
+
     def _retrieve_experience(self) -> str:
         """Default implementation of experience retrieval. Can be overwritten in subclasses."""
         context = [str(msg) for msg in self.rc.memory.get(self.memory_k)]

metagpt/strategy/thinking_command.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel
 from metagpt.environment.mgx.mgx_env import MGXEnv
 from metagpt.memory import Memory
 from metagpt.roles import Role
-from metagpt.schema import Message, Task
+from metagpt.schema import Message
 
 
 class CommandDef(BaseModel):
@@ -92,17 +92,11 @@ async def run_env_command(role: Role, cmd: list[dict], role_memory: Memory = Non
 
 def run_plan_command(role: Role, cmd: list[dict]):
     if cmd["command_name"] == Command.APPEND_TASK.cmd_name:
-        role.planner.plan.append_task(Task(**cmd["args"]))
+        role.planner.plan.append_task(**cmd["args"])
     elif cmd["command_name"] == Command.RESET_TASK.cmd_name:
         role.planner.plan.reset_task(**cmd["args"])
     elif cmd["command_name"] == Command.REPLACE_TASK.cmd_name:
-        new_task = Task(
-            task_id=cmd["args"]["task_id"],
-            dependent_task_ids=cmd["args"]["new_dependent_task_ids"],
-            instruction=cmd["args"]["new_instruction"],
-            assignee=cmd["args"]["new_assignee"],
-        )
-        role.planner.plan.replace_task(new_task)
+        role.planner.plan.replace_task(**cmd["args"])
     elif cmd["command_name"] == Command.FINISH_CURRENT_TASK.cmd_name:
         if role.planner.plan.is_plan_finished():
             return