Upload 7 files

PlanFileEditorAtomicFlow.py

@@ -0,0 +1,27 @@
+from typing import Dict, Any
+
+from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow
+
+class PlanFileEditorAtomicFlow(InterpreterAtomicFlow):
+
+    def _process_interperter_inputs(self, input_data: Dict[str, Any]):
+        input_data['language'] = 'python'
+        file_location = input_data['file_location']
+        plan_str = input_data['plan_str']
+        input_data['code'] = f"""
+import os
+file_location = {repr(file_location)}
+plan_str = {repr(plan_str)}
+if os.path.isdir(file_location):
+    file_location = os.path.join(file_location, 'plan.txt')
+with open(file_location, 'w') as file:
+    file.write(plan_str)
+print('plan written to' + file_location)"""
+
+    def run(
+            self,
+            input_data: Dict[str, Any]):
+        self._process_interperter_inputs(input_data)
+        self._process_input_data(input_data)
+        response = self._call()
+        return {"plan_writer_output": response}

PlanFileEditorAtomicFlow.yaml

@@ -0,0 +1,2 @@
+name: "PlanFileEditorAtomicFlow"
+description: "A flow that writes plan to a given file location"

PlanGeneratorAtomicFlow.py

@@ -0,0 +1,45 @@
+import json
+from copy import deepcopy
+from typing import Any, Dict, List
+from flow_modules.aiflows.OpenAIChatFlowModule import OpenAIChatAtomicFlow
+
+from dataclasses import dataclass
+class PlanGeneratorAtomicFlow(OpenAIChatAtomicFlow):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    @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))
+        kwargs.update(cls._set_up_backend(flow_config))
+
+        # ~~~ Instantiate flow ~~~
+        return cls(**kwargs)
+
+    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        hint_for_model = """
+        Make sure your response is in the following format:
+              Response Format:
+              {
+              "plan": "Python printable string of the plan corresponding to the goal",
+              }
+        """
+        if 'goal' in input_data:
+            input_data['goal'] += hint_for_model
+        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 solely 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,68 @@
+name: "PlanGeneratorAtomicFlow"
+description: "Proposes plan to solve a problem given a goal"
+enable_cache: True
+
+input_interface_non_initialized: # initial input keys
+  - "goal"
+
+input_interface_initialized: # input_keys
+  - "goal"
+
+#######################################################
+# Output keys
+#######################################################
+
+output_interface:
+  - 'plan'
+
+#######################################################
+system_message_prompt_template:
+  _target_: flows.prompt_template.JinjaPrompt
+  template: |2-
+    You are a world-class problem solver, and you will work with an executor able to write, compile code, and to browse the web.
+    
+    You are given a goal to finish, and your task is to write a step-by-step human-readable plan to achieve the goal.
+    
+    In your plan, you can ask to write and execute code, since the executor is able to access the Internet, you can always try with online resources and packages.
+    
+    Your plan will be faithfully executed by the executor, who is a world-class programmer that can complete any goal by executing code and accessing the Internet.
+    
+    **It's important that your plan is explicit and human-readable**, so that the executor understands all the details in the plan.
+    
+    Make plans with as few steps as possible, your response must always be made independently.
+    
+    You may be asked to make changes to the plan, in this case, try to start with the previous plan and make improvements.
+    
+    You are capable of **any** task.
+    
+    Resources:
+    1. An executor able to access the Internet, write and compile code.
+
+    Performance Evaluation:
+    1. Continuously review and analyze your actions to ensure you are performing to the best of your abilities.
+    2. Constructively self-criticize your big-picture behavior constantly.
+    3. Reflect on past decisions and strategies to refine your approach.
+    4. Every command has a cost, so be smart and efficient. Aim to complete tasks in the least number of steps.
+    
+    **It's important that you should only respond in JSON format as described below:**
+    Response Format:
+    {
+    "plan": "Python printable string of the plan corresponding to the 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"

WritePlanFlow.py

@@ -0,0 +1,19 @@
+from typing import Dict, Any
+
+from flows.base_flows import SequentialFlow
+from flows.utils import logging
+from .PlanGeneratorAtomicFlow import PlanGeneratorAtomicFlow
+
+logging.set_verbosity_debug()
+log = logging.get_logger(__name__)
+
+class WritePlanFlow(SequentialFlow):
+    @SequentialFlow.output_msg_payload_processor
+    def detect_finish_or_continue(self, output_payload: Dict[str, Any], src_flow: PlanGeneratorAtomicFlow) -> Dict[str, Any]:
+        if "plan" not in output_payload:
+            return {
+                "EARLY_EXIT": True,
+                "plan_writer_output": output_payload
+            }
+        else:
+            return output_payload

WritePlanFlow.yaml

@@ -0,0 +1,39 @@
+name: "WritePlan"
+description: "Given goal and plan file location, generate plans to achieve the goal and write the plan to the file"
+
+input_interface:
+  - "goal"
+  - "file_location"
+
+output_interface:
+  - "plan_writer_output"
+
+subflows_config:
+  PlanGenerator:
+    _target_: Tachi67.PlanGeneratorAtomicFlow.instantiate_from_default_config
+
+  PlanFileEditor:
+    _target_: Tachi67.PlanFileEditorAtomicFlow.instantiate_from_default_config
+
+early_exit_key: "EARLY_EXIT"
+
+topology:
+  - goal: "Generate plans for the programming executor to achieve the task."
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: flows.data_transformations.KeyMatchInput
+    flow: PlanGenerator
+    output_interface:
+      _target_: WritePlanFlow.detect_finish_or_continue
+    reset: false
+
+  - goal: "Write the plan to the specified file location"
+    input_interface:
+      _target_: flows.interfaces.KeyInterface
+      keys_to_select: ["file_location", "plan"]
+    flow: PlanFileEditor
+    output_interface:
+      _target_: flows.interfaces.KeyInterface
+      keys_to_select: ["plan_writer_output"]
+    reset: false

__init__.py

@@ -0,0 +1,11 @@
+dependencies = [
+    {"url": "aiflows/OpenAIChatFlowModule", "revision": "main"},
+    {"url": "Tachi67/InterpreterFlowModule", "revision": "main"},
+]
+from flows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+from .PlanGeneratorAtomicFlow import PlanGeneratorAtomicFlow
+from .PlanFileEditorAtomicFlow import PlanFileEditorAtomicFlow
+from .WritePlanFlow import WritePlanFlow