Add feature to create a replanning chain

gpt_all_star/core/agents/chain.py

@@ -115,7 +115,7 @@ def create_planning_chain(self, profile: str = ""):
     - objective: very detailed description of the objective to be achieved for the task to be executed to accomplish the entire plan
     - reason: clear reasons why the task should be performed
 
-Make sure that each step has all the information needed - do not skip steps.
+Make sure that each step has all the information needed.
 """
         function_def = {
             "name": "planning",
@@ -189,6 +189,92 @@ def create_planning_chain(self, profile: str = ""):
             | JsonOutputFunctionsParser()
         )
 
+    def create_replanning_chain(self, profile: str = ""):
+        system_prompt = f"""{profile}
+Based on the user request provided and the current implementation, your task is to update the original plan that includes following items:
+    - action: it must be one of {", ".join(ACTIONS)}
+    - working_directory: a directory where the command is to be executed or the file is to be placed, it should be started from '.', e.g. './src'
+    - filename: specify only if the name of the file to be added or changed is specifically determined
+    - command: command to be executed if necessary
+    - context: all contextual information that should be communicated to the person performing the task
+    - objective: very detailed description of the objective to be achieved for the task to be executed to accomplish the entire plan
+    - reason: clear reasons why the task should be performed
+
+If no more steps are needed and you can return to the user, then respond with that.
+Otherwise, fill out the plan.
+"""
+        function_def = {
+            "name": "replanning",
+            "description": "Create the replan.",
+            "parameters": {
+                "title": "planSchema",
+                "type": "object",
+                "properties": {
+                    "plan": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "description": "Task to do.",
+                            "properties": {
+                                "action": {
+                                    "type": "string",
+                                    "description": "Task",
+                                    "anyOf": [
+                                        {"enum": ACTIONS},
+                                    ],
+                                },
+                                "working_directory": {
+                                    "type": "string",
+                                    "description": "Directory where the command is to be executed or the file is to be located, it should be started from '.', e.g. './src'",
+                                },
+                                "filename": {
+                                    "type": "string",
+                                    "description": "Specify only if the name of the file to be added or changed is specifically determined",
+                                },
+                                "command": {
+                                    "type": "string",
+                                    "description": "Command to be executed if necessary",
+                                },
+                                "context": {
+                                    "type": "string",
+                                    "description": "All contextual information that should be communicated to the person performing the task",
+                                },
+                                "objective": {
+                                    "type": "string",
+                                    "description": "Very detailed description of the goals to be achieved for the task to be executed to accomplish the entire plan",
+                                },
+                                "reason": {
+                                    "type": "string",
+                                    "reason": "Clear reasons why the task should be performed",
+                                },
+                            },
+                        },
+                    }
+                },
+                "required": ["plan"],
+            },
+        }
+        prompt = ChatPromptTemplate.from_messages(
+            [
+                ("system", system_prompt),
+                MessagesPlaceholder(variable_name="messages"),
+                (
+                    "system",
+                    """
+Given the conversation above, update the original plan to fully meet the user's requirements."
+""",
+                ),
+            ]
+        ).partial()
+
+        return (
+            prompt
+            | self._llm.bind_functions(
+                functions=[function_def], function_call="replanning"
+            )
+            | JsonOutputFunctionsParser()
+        )
+
     def create_git_commit_message_chain(self):
         system_prompt = "You are an excellent engineer. Given the diff information of the source code, please respond with the appropriate branch name and commit message for making the change."
         function_def = {

gpt_all_star/core/steps/development/development.py

@@ -10,6 +10,7 @@ def __init__(
     ) -> None:
         super().__init__(copilot)
         self.working_directory = self.copilot.storages.app.path.absolute()
+        self.plan_and_solve = True
 
     def planning_prompt(self) -> str:
         planning_prompt = planning_prompt_template.format(

gpt_all_star/core/steps/development/planning_prompt.py

@@ -9,6 +9,7 @@
 # Constraints
 ---
 - The application specifications must be carefully understood and accurately reflected in the application.
+- The operation check itself is performed in a separate step and is not included in the plan.
 
 # Requirements
 ---

gpt_all_star/core/steps/development/replanning_prompt.py

@@ -0,0 +1,39 @@
+from langchain_core.prompts import PromptTemplate
+
+replanning_template = PromptTemplate.from_template(
+    """
+# Instructions
+---
+Create a detailed and specific development plan from project creation to source code implementation in order to build a correctly working application.
+
+# Original plan was this:
+---
+{original_plan}
+
+# You have currently done the follow tasks:
+---
+{completed_plan}
+
+# Current implementation
+---
+{implementation}
+
+# Constraints
+---
+- The application specifications must be carefully understood and accurately reflect the specifications.
+- The operation check itself is performed in a separate step and is not included in the plan.
+
+# Requirements
+---
+
+## Application Specifications to be met
+```specifications.md
+{specifications}
+```
+
+## Technology stack to be used
+```technologies.md
+{technologies}
+```
+"""
+)

gpt_all_star/core/steps/step.py

@@ -11,6 +11,7 @@ def __init__(
         self.copilot = copilot
         self.copilot.console.section(f"STEP: {self.__class__.__name__}")
         self.working_directory = self.copilot.storages.root.path.absolute()
+        self.plan_and_solve = False
         self.exclude_dirs = [".archive", "node_modules", "build"]
 
     @abstractmethod

gpt_all_star/core/team.py

@@ -12,6 +12,7 @@
 from gpt_all_star.core.agents.copilot import Copilot
 from gpt_all_star.core.implement_prompt import implement_template
 from gpt_all_star.core.message import Message
+from gpt_all_star.core.steps.development.replanning_prompt import replanning_template
 from gpt_all_star.core.steps.step import Step
 from gpt_all_star.helper.config_loader import load_configuration
 from gpt_all_star.helper.multi_agent_collaboration_graph import (
@@ -86,6 +87,7 @@ def _run(
         self,
         planning_prompt: Optional[str] = None,
         additional_tasks: list = [],
+        plan_and_solve: bool = False,
     ):
         with Status(
             f"[bold {MAIN_COLOR}]running...(Have a cup of coffee and relax.)",
@@ -116,7 +118,9 @@ def _run(
                     json.dumps(tasks, indent=4, ensure_ascii=False)
                 )
 
-            for i, task in enumerate(tasks["plan"]):
+            completed_plan = []
+            while len(tasks["plan"]) > 0:
+                task = tasks["plan"][0]
                 if task["action"] == ACTIONS[0]:
                     todo = f"{task['action']}: {task['command']} in the directory({task.get('working_directory', '')})"
                 else:
@@ -125,15 +129,17 @@ def _run(
                 if self.supervisor.debug_mode:
                     self.supervisor.state(
                         f"""\n
-Task {i + 1}: {todo}
+Task: {todo}
 Context: {task['context']}
 Objective: {task['objective']}
 Reason: {task['reason']}
 ---
 """
                     )
                 else:
-                    self.supervisor.state(f"({(i+1)}/{len(tasks['plan'])}) {todo}")
+                    self.supervisor.state(
+                        f"(# of remaining tasks: {len(tasks['plan'])}) {todo}"
+                    )
 
                 message = Message.create_human_message(
                     implement_template.format(
@@ -153,14 +159,45 @@ def _run(
                     )
                 )
                 self._execute([message])
+                tasks["plan"].pop(0)
+
+                if plan_and_solve:
+                    completed_plan.append(task)
+                    tasks = (
+                        Chain()
+                        .create_replanning_chain(self.supervisor.profile)
+                        .invoke(
+                            {
+                                "messages": [
+                                    Message.create_human_message(
+                                        replanning_template.format(
+                                            original_plan=tasks,
+                                            completed_plan=completed_plan,
+                                            implementation=self.copilot.storages.current_source_code(),
+                                            specifications=self.copilot.storages.docs.get(
+                                                "specifications.md", "N/A"
+                                            ),
+                                            technologies=self.copilot.storages.docs.get(
+                                                "technologies.md", "N/A"
+                                            ),
+                                        )
+                                    )
+                                ],
+                            }
+                        )
+                    )
+                    if self.supervisor.debug_mode:
+                        self.supervisor.console.print(
+                            json.dumps(tasks, indent=4, ensure_ascii=False)
+                        )
 
     def run(self, step: Step) -> bool:
         planning_prompt = step.planning_prompt()
         additional_tasks = step.additional_tasks()
         for agent in self.agents.to_array():
             agent.set_executor(step.working_directory)
         self._assign_supervisor(planning_prompt)
-        self._run(planning_prompt, additional_tasks)
+        self._run(planning_prompt, additional_tasks, step.plan_and_solve)
 
         return step.callback()