Upload 3 files

PlanGeneratorAtomicFlow.py

@@ -0,0 +1,51 @@
+import json
+from copy import deepcopy
+from typing import Any, Dict
+from flow_modules.aiflows.ChatFlowModule import ChatAtomicFlow
+
+
+
+class PlanGeneratorAtomicFlow(ChatAtomicFlow):
+    """Generates one function with docstrings to finish the given goal (from the controller).
+    """
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.hint_for_model = """
+        Make sure your response is in the following format:
+              Response Format:
+              {
+              "plan": "A step-by-step plan to finish the given goal",
+              }
+        """
+
+    @classmethod
+    def instantiate_from_config(cls, config):
+        flow_config = deepcopy(config)
+
+        kwargs = {"flow_config": flow_config}
+
+        # ~~~ Set up prompts ~~~
+        kwargs.update(cls._set_up_prompts(flow_config))
+
+        # ~~~ Set up backend ~~~
+        kwargs.update(cls._set_up_backend(flow_config))
+
+        # ~~~ Instantiate flow ~~~
+        return cls(**kwargs)
+
+    def _update_prompts_and_input(self, input_data: Dict[str, Any]):
+        if 'goal' in input_data:
+            input_data['goal'] += self.hint_for_model
+
+    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        self._update_prompts_and_input(input_data)
+        api_output = super().run(input_data)["api_output"].strip()
+        try:
+            response = json.loads(api_output)
+            return response
+        except json.decoder.JSONDecodeError:
+            new_input_data = input_data.copy()
+            new_input_data['goal'] += ("The previous respond cannot be parsed with json.loads. "
+                                       "Make sure your next response is in JSON format.")
+            new_api_output = super().run(new_input_data)["api_output"].strip()
+            return json.loads(new_api_output)

PlanGeneratorAtomicFlow.yaml

@@ -0,0 +1,76 @@
+name: "PlanGeneratorAtomicFlow"
+description: "Writes step by step plan with given goal"
+enable_cache: True
+
+input_interface_non_initialized: # initial input keys
+  - "goal"
+  - "memory_files" # In planner of extlib, it should be an empty dict.
+
+input_interface_initialized: # input_keys
+  - "goal"
+  - "memory_files"
+
+#######################################################
+# Output keys
+#######################################################
+
+output_interface:
+  - 'plan'
+
+#######################################################
+system_message_prompt_template:
+  _target_: flows.prompt_template.JinjaPrompt
+  template: |2-
+    You are a planner of a coding department, your task is to make a step-by-step plan for the coders to follow, inorder to solve the task.
+    
+    The coding department is only able to write functions, at a higher level, your job is to instruct the coders to write modular, reusable, and encapsulated functions.
+    
+    The details of your task are:
+      1. Given a goal, decompose the goal into step-by-step plans, each step of the plan should involving asking the coder to write **one function** to tackle this step.
+      2. Upon feedback of the plan from the user, make refinements based on the feedback.
+    
+    Notice that:
+      1. If the goal given to you is simple enough that is doable within one function, it is okay that your plan only has one step.
+      2. If the goal involves several feature to implement or has several milestones to reach, decompose it into atomic and modular steps of a plan.
+    
+    The coding department is able to access the Internet, so it may be useful to use some online services via APIs.
+    
+    An example of input and output plan you should have:
+    ### begin of input ###
+    Write code to fetch a certain day's weather data at a certain city, and print the data to the console.
+    ### end of input ###
+    
+    ### begin of plan ###
+    1. Write a function that fetches weather data from OpenWeatherMap, given a date and the city name.
+    2. Write a function that prints fetched weather data to the console.
+    ### end of plan ###
+    
+    You are capable of **any** task.
+    
+    Performance Evaluation:
+    1. Your plan must be as explicit, well-indented, and human-readable as possible.
+    2. Your plan must be step-by-step with number indexes.
+    3. You should make plans with as few steps as possible.
+    
+    **It's important that you should only respond in JSON format as described below:**
+    Response Format:
+    {
+    "plan": "A step-by-step plan to finish the given goal",
+    }
+    Ensure your responses can be parsed by Python json.loads
+    
+human_message_prompt_template:
+  _target_: flows.prompt_template.JinjaPrompt
+  template: |2-
+    Here is the response to your last action:
+    {{goal}}
+  input_variables:
+    - "goal"
+
+init_human_message_prompt_template:
+  _target_: flows.prompt_template.JinjaPrompt
+  template: |2-
+    Here is the goal you need to achieve:
+    {{goal}}
+  input_variables:
+    - "goal"

__init__.py

@@ -0,0 +1,8 @@
+dependencies = [
+    {"url": "aiflows/ChatFlowModule", "revision": "a749ad10ed39776ba6721c37d0dc22af49ca0f17"},
+]
+from flows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+from .PlanGeneratorAtomicFlow import PlanGeneratorAtomicFlow