Upload 10 files

CodeTestingAtomicFlow.py

@@ -0,0 +1,49 @@
+import os
+import time
+import subprocess
+from typing import Dict, Any
+from flow_modules.aiflows.InterpreterFlowModule import InterpreterAtomicFlow
+
+
+class CodeTestingAtomicFlow(InterpreterAtomicFlow):
+    """This class inherits from InterpreterAtomicFlow and is used to test code.
+    
+    *Input Interface*:
+    - `temp_code_file_location`: Location of the file containing the code to be tested.
+    
+    *Output Interface*:
+    - `feedback`: Feedback from the test (i.e. test results).
+    """
+    def _open_file_and_wait_for_file_update(self, file_location, check_interval=1):
+        process = subprocess.Popen(["code", "--wait", file_location])
+        while True:
+            if process.poll() is not None:
+                break
+            time.sleep(check_interval)
+    def _prepare_code_and_lang(self, input_data: Dict[str, Any]):
+        file_location = input_data["temp_code_file_location"]
+        input_data["language"] = "python"
+        with open(file_location, "r") as file:
+            code_str = file.read()
+        input_data["code"] = code_str
+
+    def _check_input(self, input_data: Dict[str, Any]):
+        assert "temp_code_file_location" in input_data, "temp_code_file_location not passed to CodeTestingAtomicFlow"
+
+    def _delete_file(self, file_location):
+        if os.path.exists(file_location):
+            os.remove(file_location)
+
+    def run(
+            self,
+            input_data: Dict[str, Any]):
+        self._check_input(input_data)
+        file_loc = input_data["temp_code_file_location"]
+        self._open_file_and_wait_for_file_update(file_loc)
+        self._prepare_code_and_lang(input_data)
+        self._process_input_data(input_data)
+        response = self._call()
+        self._delete_file(file_loc)
+        if response.strip() == '':
+            response = "test passed"
+        return {"feedback": response}

CodeTestingAtomicFlow.yaml

@@ -0,0 +1,10 @@
+_target_: flow_modules.aiflows.TestCodeFlowModule.CodeTestingAtomicFlow.instantiate_from_default_config
+name: "CodeTestingAtomicFlow"
+description: "A flow that runs test code"
+
+
+input_interface:
+  - "temp_code_file_location"
+
+output_interface:
+  - "feedback"

README.md

@@ -1,3 +1,101 @@
----
-license: mit
----
+### Structure of TestCodeFlow
+
+```
+               code
+                |
+                v
+        +-------------------+
+        |  TestCodeFileEdit | Edit a temp code file with the code to be tested and necessary imports (manually added): https://huggingface.co/Tachi67/TestCodeFlowModule/blob/main/TestCodeFileEditAtomicFlow.py
+        +-------------------+                        
+                |                                
+                | (temp_code_file_location)        
+                |                                
+                v                                
+        +------------------+                     
+        |   CodeTesting    |  Opens up the temp file until user closes the file, run the test code.
+        +------------------+                     
+                |                                
+                | (feedback)                      
+                |                                
+                v                                            
+             feedback     
+                
+
+```
+Memory_files: 
+- library.py
+
+# Table of Contents
+
+* [TestCodeFlow](#TestCodeFlow)
+  * [TestCodeFlow](#TestCodeFlow.TestCodeFlow)
+* [TestCodeFileEditAtomicFlow](#TestCodeFileEditAtomicFlow)
+  * [TestCodeFileEditAtomicFlow](#TestCodeFileEditAtomicFlow.TestCodeFileEditAtomicFlow)
+* [\_\_init\_\_](#__init__)
+* [CodeTestingAtomicFlow](#CodeTestingAtomicFlow)
+  * [CodeTestingAtomicFlow](#CodeTestingAtomicFlow.CodeTestingAtomicFlow)
+
+<a id="TestCodeFlow"></a>
+
+# TestCodeFlow
+
+<a id="TestCodeFlow.TestCodeFlow"></a>
+
+## TestCodeFlow Objects
+
+```python
+class TestCodeFlow(SequentialFlow)
+```
+
+This class is used to test code. It is a sequential flow that runs the following steps:
+1. Prepares the code to be tested, it is composed of the code to be tested and necessary import statements manually added.
+2. Opens the code in VSCode and waits for the user to clode the vscode session. The user is able to add tests.
+3. The following will be tested:
+    a. (Default & Compulsory) Code syntax;
+    b. (Added by user) Any other tests.
+4. Runs the test and returns the output.
+
+*Input Interface*:
+- `code` (str): The code to be tested.
+
+*Output Interface*:
+- `feedback` (str): The test results.
+
+<a id="TestCodeFileEditAtomicFlow"></a>
+
+# TestCodeFileEditAtomicFlow
+
+<a id="TestCodeFileEditAtomicFlow.TestCodeFileEditAtomicFlow"></a>
+
+## TestCodeFileEditAtomicFlow Objects
+
+```python
+class TestCodeFileEditAtomicFlow(CodeFileEditAtomicFlow)
+```
+
+Refer to: https://huggingface.co/Tachi67/CodeFileEditFlowModule/tree/main
+
+<a id="__init__"></a>
+
+# \_\_init\_\_
+
+<a id="CodeTestingAtomicFlow"></a>
+
+# CodeTestingAtomicFlow
+
+<a id="CodeTestingAtomicFlow.CodeTestingAtomicFlow"></a>
+
+## CodeTestingAtomicFlow Objects
+
+```python
+class CodeTestingAtomicFlow(InterpreterAtomicFlow)
+```
+
+This class inherits from InterpreterAtomicFlow and is used to test code.
+
+*Input Interface*:
+- `temp_code_file_location`: Location of the file containing the code to be tested.
+
+*Output Interface*:
+- `feedback`: Feedback from the test (i.e. test results).
+

TestCodeFileEditAtomicFlow.py

@@ -0,0 +1,43 @@
+import os
+from typing import Dict, Any
+from flow_modules.aiflows.CodeFileEditFlowModule import CodeFileEditAtomicFlow
+
+class TestCodeFileEditAtomicFlow(CodeFileEditAtomicFlow):
+    """Refer to: https://huggingface.co/Tachi67/CodeFileEditFlowModule/tree/main"""
+    def _generate_import_statement(self, code_lib_location):
+        module_dir = os.path.dirname(code_lib_location)
+        module_name = os.path.splitext(os.path.basename(code_lib_location))[0]
+
+        import_code = (
+            f"import sys\n"
+            f"sys.path.insert(0, '{module_dir}')\n"
+            f"from {module_name} import *\n"
+        )
+        return import_code
+
+    def _generate_content(self, code_lib_location, code_str) -> str:
+        import_code_lib_str = self._generate_import_statement(code_lib_location)
+        content = (
+                "# Don't touch this import statement \n"
+                + import_code_lib_str + "\n"
+                "# Here is the code just generated \n" +
+                code_str + "\n"
+                "# Below, please provide code to test it.\n"
+                "# The simplest form could be just calling it with appropriate parameters. \n"
+                "# You could also assert the output, anyway, the test results will be informed to JARVIS. \n"
+                "# If you do not write anything, JARVIS just checks if the syntax is alright. \n"
+                "###########\n"
+                "# Test Code:\n" +
+                "\n############\n"
+        )
+        return content
+
+    def _generate_temp_file_location(self, code_lib_location):
+        directory = os.path.dirname(code_lib_location)
+        ret = os.path.join(directory, 'temp_tests.py')
+        return ret
+
+    def _check_input(self, input_data: Dict[str, Any]):
+        assert "code" in input_data, "code is not passed to TestCodeFileEditAtomicFlow"
+        assert "memory_files" in input_data, "memory_files is not passed to TestCodeFileEditAtomicFlow"
+

TestCodeFileEditAtomicFlow.yaml

@@ -0,0 +1,11 @@
+_target_: flow_modules.aiflows.TestCodeFlowModule.TestCodeFileEditAtomicFlow.instantiate_from_default_config
+name: "TestCodeFileEditorAtomicFlow"
+description: "A flow that writes tests code to a temp file"
+
+input_interface:
+  - "code"
+  - "memory_files"
+
+output_interface:
+  - "code_editor_output"
+  - "temp_code_file_location"

TestCodeFlow.py

@@ -0,0 +1,66 @@
+from copy import deepcopy
+from typing import Any, Dict
+
+from aiflows.base_flows import SequentialFlow
+from aiflows.utils import logging
+
+logging.set_verbosity_debug()
+log = logging.get_logger(__name__)
+
+class TestCodeFlow(SequentialFlow):
+    """This class is used to test code. It is a sequential flow that runs the following steps:
+    1. Prepares the code to be tested, it is composed of the code to be tested and necessary import statements manually added.
+    2. Opens the code in VSCode and waits for the user to clode the vscode session. The user is able to add tests.
+    3. The following will be tested:
+        a. (Default & Compulsory) Code syntax;
+        b. (Added by user) Any other tests.
+    4. Runs the test and returns the output.
+    
+    *Input Interface*:
+    - `code` (str): The code to be tested.
+    
+    *Output Interface*:
+    - `feedback` (str): The test results.
+    """
+    REQUIRED_KEYS_CONFIG = ["max_rounds", "early_exit_key", "topology", "memory_files"]
+
+    def __init__(
+            self,
+            memory_files: Dict[str, Any],
+            **kwargs
+    ):
+        super().__init__(**kwargs)
+        self.memory_files = memory_files
+
+    @classmethod
+    def instantiate_from_config(cls, config):
+        flow_config = deepcopy(config)
+
+        kwargs = {"flow_config": flow_config}
+
+        # ~~~ Set up memory file ~~~
+        memory_files = flow_config["memory_files"]
+        kwargs.update({"memory_files": memory_files})
+
+        # ~~~ Set up subflows ~~~
+        kwargs.update({"subflows": cls._set_up_subflows(flow_config)})
+
+        # ~~~ Instantiate flow ~~~
+        return cls(**kwargs)
+
+    def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        # ~~~ sets the input_data in the flow_state dict ~~~
+        self._state_update_dict(update_data=input_data)
+
+        # ~~~ set the memory file to the flow state ~~~
+        self._state_update_dict(update_data={"memory_files": self.memory_files})
+
+        max_rounds = self.flow_config.get("max_rounds", 1)
+        if max_rounds is None:
+            log.info(f"Running {self.flow_config['name']} without `max_rounds` until the early exit condition is met.")
+
+        self._sequential_run(max_rounds=max_rounds)
+
+        output = self._get_output_from_state()
+
+        return output

TestCodeFlow.yaml

@@ -0,0 +1,38 @@
+name: "TestCodeFlow"
+description: "Test the code generated by InteractiveCodeGenFlow"
+
+_target_: flow_modules.aiflows.TestCodeFlowModule.TestCodeFlow.instantiate_from_default_config
+
+memory_files: ???
+
+input_interface:
+  - "code"
+
+output_interface:
+  - "feedback"
+
+subflows_config:
+  TestCodeFileEdit:
+    _target_: flow_modules.aiflows.TestCodeFlowModule.TestCodeFileEditAtomicFlow.instantiate_from_default_config
+
+  CodeTesting:
+    _target_: flow_modules.aiflows.TestCodeFlowModule.CodeTestingAtomicFlow.instantiate_from_default_config
+
+early_exit_key: "EARLY_EXIT"
+
+topology:
+  - goal: "Write the code & instructions for testing to a temp file"
+    input_interface:
+      _target_: aiflows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: aiflows.data_transformations.KeyMatchInput
+    flow: TestCodeFileEdit
+    reset: false
+
+  - goal: "Test the code"
+    input_interface:
+      _target_: aiflows.interfaces.KeyInterface
+      additional_transformations:
+        - _target_: aiflows.data_transformations.KeyMatchInput
+    flow: CodeTesting
+    reset: false

__init__.py

@@ -0,0 +1,12 @@
+dependencies = [
+    {"url": "aiflows/CodeFileEditFlowModule", "revision": "main"},
+    {"url": "aiflows/InterpreterFlowModule", "revision": "main"},
+]
+
+from aiflows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+from .CodeTestingAtomicFlow import CodeTestingAtomicFlow
+from .TestCodeFileEditAtomicFlow import TestCodeFileEditAtomicFlow
+from .TestCodeFlow import TestCodeFlow

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

run.py

@@ -0,0 +1,50 @@
+import os
+
+import hydra
+
+from aiflows.backends.api_info import ApiInfo
+from aiflows.messages import InputMessage
+from aiflows.utils.general_helpers import read_yaml_file, quick_load
+
+from aiflows import logging
+from aiflows.flow_cache import CACHING_PARAMETERS, clear_cache
+
+CACHING_PARAMETERS.do_caching = False  # Set to True in order to disable caching
+
+logging.set_verbosity_debug()
+logging.auto_set_dir()
+
+dependencies = [
+    {"url": "aiflows/TestCodeFlowModule", "revision": "main"},
+]
+
+from aiflows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+if __name__ == "__main__":
+    current_dir = os.getcwd()
+    cfg_path = os.path.join(current_dir, "TestCodeFlow.yaml")
+    cfg = read_yaml_file(cfg_path)
+    with open(os.path.join(current_dir, "example_library.py"), 'w') as f:
+        pass
+    memory_files = {"code_library": os.path.join(current_dir, "example_library.py")}
+    cfg["memory_files"] = memory_files
+
+
+    TestCodeFlow = hydra.utils.instantiate(cfg, _recursive_=False, _convert_="partial")
+
+
+    input_data = {
+        "code": "def test():\n    print('hello world')\n'",
+    }
+    input_message = InputMessage.build(
+        data_dict=input_data,
+        src_flow="Launcher",
+        dst_flow=TestCodeFlow.name
+    )
+
+    # ~~~ calling the flow ~~~
+    output_message = TestCodeFlow(input_message)
+
+    print(output_message.data)