Upload 5 files

PlanFileEditAtomicFlow.py

@@ -0,0 +1,94 @@
+from typing import Dict, Any
+from aiflows.base_flows.atomic import AtomicFlow
+import os
+
+
+class PlanFileEditAtomicFlow(AtomicFlow):
+    """This class is used to write plan to a temp code file, with commented instructions to give information
+    to the user.
+
+    *Input Interface*:
+    - `plan`: str
+    - `plan_file_location`: str
+
+    *Output Interface*:
+    - `plan_editor_output`: str
+    - `temp_plan_file_location`: str
+    """
+    def _generate_content(self, plan_file_location, plan_str) -> str:
+        content = (
+                "The below plan will be written to " +
+                plan_file_location + "\n"
+                "Edit the plan directly or provide your thoughts down below if you have any suggestions.\n"
+                "When you are done editing plan and providing feedback, save file and close the current VSCode session to continue.\n"
+                "###########\n"
+                "Plan:\n" +
+                plan_str +
+                "\n############\n"
+                "Thoughts:"
+        )
+        return content
+
+    def _generate_temp_file_location(self, plan_file_location):
+        directory = os.path.dirname(plan_file_location)
+        ret = os.path.join(directory, 'temp_plan.txt')
+        return ret
+
+    def _write_plan_content_to_file(self, file_location, content: str):
+        try:
+            with open(file_location, "w") as file:
+                file.write(content)
+
+            return True, f"Plan written to {file_location}", file_location
+
+        except Exception as e:
+            return False, str(e), file_location
+
+    def _check_input(self, input_data: Dict[str, Any]):
+        assert any(item in input_data for item in ["plan", "new_plan"]), "plan or new_plan is not passed to PlanFileEditAtomicFlow"
+        assert "plan_file_location" in input_data, "plan_file_location not passed to PlanFileEditAtomicFlow"
+        plan_file_loc = input_data["plan_file_location"]
+        assert os.path.exists(plan_file_loc), f"{plan_file_loc} does not exist"
+        assert os.path.isfile(plan_file_loc), f"{plan_file_loc} is not a file"
+
+    def _generate_input_to_writer(self, input_data: Dict[str, Any]):
+        """
+        sometimes the plan generator will return an arrary of indexed plans, like
+        [
+            "1. Extend the code library with a function named 'import_libraries'. This function should import necessary libraries for the task...",
+            "2. Extend the code library with a function named 'fetch_stock_prices'. This function should take two inputs: 'company_code' and 'duration'...",
+            "3. Investigate the issue with importing the 'fetch_stock_prices' function from the library..."
+        ]
+        In this case we need to prase this format accorrdingly.
+        """
+        
+        plan = input_data['plan'] if "plan" in input_data else input_data['new_plan']
+        if isinstance(plan, str):
+            plan_str = plan
+        elif isinstance(plan, list):
+            plan_str = "\n".join(plan)
+        else:
+             raise TypeError("plan is neither a string nor a list")
+        plan_file_location = input_data["plan_file_location"]
+        content_to_write = self._generate_content(plan_file_location, plan_str)
+        file_location_to_write = self._generate_temp_file_location(plan_file_location)
+        return content_to_write, file_location_to_write
+
+    def run(
+            self,
+            input_data: Dict[str, Any]
+    ):
+        self._check_input(input_data)
+
+        # ~~~ Getting input data to the file editor ~~~
+        content_to_write, file_location_to_write = self._generate_input_to_writer(input_data)
+
+        # ~~~ Calling the writer function ~~~
+        result, plan_editor_output, temp_file_location = self._write_plan_content_to_file(
+            file_location_to_write, content_to_write)
+
+        # ~~~ Generating return variables ~~~
+        response = {}
+        response["plan_editor_output"] = plan_editor_output
+        response["temp_plan_file_location"] = temp_file_location
+        return response

PlanFileEditAtomicFlow.yaml

@@ -0,0 +1,10 @@
+name: "PlanFileEditAtomicFlow"
+description: "A flow that writes plan to a temp file"
+
+input_interface:
+  - "plan"
+  - "plan_file_location"
+
+output_interface:
+  - "plan_editor_output"
+  - "temp_plan_file_location"

README.md

@@ -1,3 +1,33 @@
----
-license: mit
----
+# Table of Contents
+
+* [PlanFileEditAtomicFlow](#PlanFileEditAtomicFlow)
+  * [PlanFileEditAtomicFlow](#PlanFileEditAtomicFlow.PlanFileEditAtomicFlow)
+* [\_\_init\_\_](#__init__)
+
+<a id="PlanFileEditAtomicFlow"></a>
+
+# PlanFileEditAtomicFlow
+
+<a id="PlanFileEditAtomicFlow.PlanFileEditAtomicFlow"></a>
+
+## PlanFileEditAtomicFlow Objects
+
+```python
+class PlanFileEditAtomicFlow(AtomicFlow)
+```
+
+This class is used to write plan to a temp code file, with commented instructions to give information
+to the user.
+
+*Input Interface*:
+- `plan`: str
+- `plan_file_location`: str
+
+*Output Interface*:
+- `plan_editor_output`: str
+- `temp_plan_file_location`: str
+
+<a id="__init__"></a>
+
+# \_\_init\_\_
+

__init__.py

@@ -0,0 +1 @@
+from .PlanFileEditAtomicFlow import PlanFileEditAtomicFlow

pip_requirements.txt

@@ -0,0 +1,15 @@
+colorama==0.4.6
+pytest==7.3.1
+pytest-cov==4.1.0
+hydra-core==1.3.2
+hydra-colorlog==1.1.0
+wrapt-timeout-decorator==1.3.12.2
+diskcache==5.6.1
+openai==1.0.0
+huggingface_hub==0.19.4
+jsonlines==3.1.0
+jinja2==3.1.2
+mock==5.0.2
+rich==12.6.0
+litellm==1.0.0
+aiflows