Upload 7 files

CodeFileEditorAtomicFlow.py

@@ -0,0 +1,42 @@
+from typing import Dict, Any
+
+from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow
+
+
+class CodeFileEditorAtomicFlow(InterpreterAtomicFlow):
+
+    def __init__(self):
+        super().__init__()
+        self.file_type = {
+            "python": "py",
+            "bash": "sh",
+            "r": "r",
+            "applescript": "applescript",
+            "html": "html",
+            "javascript": "js",
+            "shell": "sh",
+            "powershell": "ps1"
+        }
+    def _process_interperter_inputs(self, input_data: Dict[str, Any]):
+        language_of_code = input_data['language_of_code'].lower()  # language of code to write
+        file_extension = self.file_type[language_of_code]  # library.py, library.r, etc..
+        file_location = input_data['file_location']
+        code_to_write = input_data['code']
+        input_data['code'] = f"""
+import os
+file_location = {repr(file_location)}
+code_to_write = {repr(code_to_write)}
+if os.path.isdir(file_location):
+    file_location = os.path.join(file_location, 'library.{file_extension}')
+with open(file_location, 'a') as file:
+    file.write('\\n' + code_to_write)
+print('code written to' + file_location)"""
+        input_data['language'] = 'python'  # will write the code string into library with python
+
+    def run(
+            self,
+            input_data: Dict[str, Any]):
+        self._process_interperter_inputs(input_data)
+        self._process_input_data(input_data)
+        response = self._call()
+        return {"code_writer_output": response}

CodeFileEditorAtomicFlow.yaml

@@ -0,0 +1,10 @@
+name: "CodeFileEditorAtomicFlow"
+description: "A flow that writes code to a given file location"
+
+input_interface:
+  - "file_location"
+  - "code"
+  - "language_of_code"
+
+output_interface:
+  - "code_writer_output"

CodeGeneratorAtomicFlow.py

@@ -0,0 +1,48 @@
+import json
+from copy import deepcopy
+from typing import Any, Dict
+from flow_modules.aiflows.OpenAIChatFlowModule import OpenAIChatAtomicFlow
+
+from dataclasses import dataclass
+
+
+class CodeGeneratorAtomicFlow(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:
+              {
+              "language_of_code": "language of the code",
+              "code": "String of the code 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, it could be the "
+                                       "backslashes used for escaping single quotes in the string arguments of the "
+                                       "Python code are not properly escaped themselves within the JSON context. "
+                                       "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)

CodeGeneratorAtomicFlow.yaml

@@ -0,0 +1,68 @@
+name: "CodeGeneratorAtomicFlow"
+description: "Writes code given instruction"
+enable_cache: True
+
+input_interface_non_initialized: # initial input keys
+  - "goal"
+
+input_interface_initialized: # input_keys
+  - "goal"
+
+#######################################################
+# Output keys
+#######################################################
+
+output_interface:
+  - 'language_of_code'
+  - 'code'
+
+#######################################################
+system_message_prompt_template:
+  _target_: flows.prompt_template.JinjaPrompt
+  template: |2-
+    You are a world class programmer that can complete any goal with code. All code you write should be functions.
+    
+    Your function will then be imported and called by an executor to finish the goal.
+    
+    Your task is to write functions for the given goal, you do not need to worry abot the execution part.
+     
+    You can access the Internet, using APIs could be useful.
+    
+    **It's important that all code you write must be functions to achieve certain goals.**
+    
+    **It's important that you make imports within functions.**
+    
+    **Make sure the function names, parameter names are self-explanatory.**
+    
+    You are capable of **any** task.
+    
+
+    Performance Evaluation:
+    1. All code you write must be functions.
+    2. Function names must be self-explanatory, it's good practice to specify the return type.
+    3. Parameter names must be self-explanatory, it's good practice to specify parameter types.
+    4. All imports must be done within the function body.
+    
+    **It's important that you should only respond in JSON format as described below:**
+    Response Format:
+    {
+    "language_of_code": "language of the code",
+    "code": "String of the code 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"

WriteCodeFlow.py

@@ -0,0 +1,12 @@
+from typing import Dict, Any
+
+from flows.base_flows import SequentialFlow
+from flows.base_flows import CircularFlow
+from flows.utils import logging
+from .CodeGeneratorAtomicFlow import CodeGeneratorAtomicFlow
+
+logging.set_verbosity_debug()
+log = logging.get_logger(__name__)
+
+class WriteCodeFlow(SequentialFlow):
+    pass

WriteCodeFlow.yaml

@@ -0,0 +1,45 @@
+name: "WriteCode"
+description: "Given goal and code library location, generate code to achieve the goal and write the code to the file"
+
+flow:
+  _target_: Tachi67.WriteCodeFlowModule.WriteCodeFlow.instantiate_from_default_config
+
+  input_interface:
+    - "goal"
+    - "file_location"
+
+  output_interface:
+    - "code_writer_output"
+
+  subflows_config:
+    CodeGenerator:
+      _target_: Tachi67.WriteCodeFlowModule.CodeGeneratorAtomicFlow.instantiate_from_default_config
+      backend:
+        api_infos: ???
+        model_name:
+          openai: gpt-4
+          azure: azure/gpt-4
+
+    CodeFileEditor:
+      _target_: Tachi67.WriteCodeFlowModule.CodeFileEditorAtomicFlow.instantiate_from_default_config
+
+  early_exit_key: "EARLY_EXIT"
+
+  topology:
+    - goal: "Generate code to achieve the task."
+      input_interface:
+        _target_: flows.interfaces.KeyInterface
+        additional_transformations:
+          - _target_: flows.data_transformations.KeyMatchInput
+      flow: CodeGenerator
+      reset: false
+
+    - goal: "Write the code to the specified file location"
+      input_interface:
+        _target_: flows.interfaces.KeyInterface
+        keys_to_select: ["file_location", "code", "language_of_code"]
+      flow: CodeFileEditor
+      output_interface:
+        _target_: flows.interfaces.KeyInterface
+        keys_to_select: ["code_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 .CodeGeneratorAtomicFlow import CodeGeneratorAtomicFlow
+from .CodeFileEditorAtomicFlow import CodeFileEditorAtomicFlow
+from .WriteCodeFlow import WriteCodeFlow