update

examples/data_interpreter/di_agents.py

@@ -204,7 +204,7 @@ def reply(self, x: Optional[Union[Msg, Sequence[Msg]]] = None) -> Msg:
         - If you think code or tools are helpful for verification, use `execute_python_code` and/or other tools available to do verification.
         - Do not simply trust the claim in `result`. VERIFY IT.
         - If the information in `result` cannot solve `current_sub_task`, Do NOT attempt to fix it. Report it IMMEDIATELY. You job is just to do the verification.
-        - If the given result can succesfully solve `current_sub_task`, ALWAYS output 'True' at the very end of your response; otherwise, explain why the given result cannot succesfully solve `current_sub_task` and output 'False'.
+        - If the given result can succesfully solve `current_sub_task`, ALWAYS output 'True' at the very end of your response; otherwise, output why the given result cannot succesfully solve `current_sub_task` and followed by 'False'.
         - DO NOT call `finish` before the entire verification process is completed. After the entire verification is completed, use `finish` tool IMMEDIATELY."""
 
         msg = Msg(
@@ -350,3 +350,53 @@ def _replanning(self, task: str) -> List[Dict[str, Any]]:
         parser = MarkdownJsonObjectParser()
         parsed_response: List[Dict[str, Any]] = parser.parse(response)
         return parsed_response.parsed
+
+    def _decompose_task(
+        self,
+        task: str,
+        max_tasks: int = 5,
+    ) -> List[Dict[str, Any]]:
+        """
+        Decompose a complex subtask into smaller, more manageable subtasks.
+
+        Args:
+            task (str): The task to be decomposed
+            max_tasks (int, optional): Maximum number of subtasks allowed. Defaults to 5.
+
+        Returns:
+            List[Dict[str, Any]]: List of decomposed subtasks as dictionaries
+        """
+        message = [
+            {
+                "role": "user",
+                "content": f"""
+                    Task: {task}
+                    - Given the task above which was determined to be too complex, break it down into smaller, more manageable subtasks.
+                    - Every subtask should be solvable through either executing code or using tools. The information of all the tools available are here:
+                    {self.service_toolkit.tools_instruction}
+                    - The subtask should not be too simple. If a task can be solved with a single block of code in one go, it should not be broken down further.
+                    - Prioritize using other tools over `execute_python_code` and take the tools available into consideration when decomposing the task.
+                    - Provide a JSON structure with the following format for the decomposition:
+                        ```json
+                        [
+                            {{
+                                "task_id": str = "unique identifier for a task in plan, can be an ordinal",
+                                "dependent_task_ids": list[str] = "ids of tasks prerequisite to this task",
+                                "instruction": "what you should do in this task, one short phrase or sentence",
+                                "task_type": "type of this task, should be one of Available Task Types",
+                                "task_type": "type of this task, should be one of Available Task Types",
+                                "tool_info": "recommended tool(s)' name(s) for solving this task",
+                            }},
+                            ...
+                        ]
+                        ```
+                    - The maximum number of subtasks allowed is {max_tasks}.
+                    """,
+            },
+        ]
+
+        response_text: str = self.model(message).text.strip()
+        response = ModelResponse(text=response_text)
+        parser = MarkdownJsonObjectParser()
+        parsed_response: List[Dict[str, Any]] = parser.parse(response)
+        return parsed_response.parsed