Upload 6 files

PlanGeneratorAtomicFlow.py

@@ -0,0 +1,58 @@
+import json
+from copy import deepcopy
+from typing import Any, Dict
+from flow_modules.aiflows.ChatFlowModule import ChatAtomicFlow
+
+class PlanGeneratorAtomicFlow(ChatAtomicFlow):
+    """This class wraps around the Chat API to generate plan from a goal.
+
+    *Input Interface Non Initialized*:
+    - `goal`
+
+    *Input Interface Initialized*:
+    - `goal`
+
+    *Output Interface*:
+    - `plan`
+    """
+    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, each step of plan should contain full information about writing a function",
+              }
+        """
+
+    @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)
+        while True:
+            api_output = super().run(input_data)["api_output"].strip()
+            try:
+                response = json.loads(api_output)
+                return response
+            except (json.decoder.JSONDecodeError, json.JSONDecodeError):
+                new_goal = "The previous respond cannot be parsed with json.loads. Next time, do not provide any comments or code blocks. Make sure your next response is purely json parsable."
+                new_input_data = input_data.copy()
+                new_input_data['goal'] = new_goal
+                input_data = new_input_data

PlanGeneratorAtomicFlow.yaml

@@ -0,0 +1,88 @@
+_target_: flow_modules.aiflows.PlanGeneratorFlowModule.PlanGeneratorAtomicFlow.instantiate_from_default_config
+name: "PlanGeneratorAtomicFlow"
+description: "Writes step by step plan with given 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_: aiflows.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 tell what function to write at each step.
+    
+    The details of your task are:
+      1. Decompose the goal into step-by-step plans, each step should contain full information about a function to write, do not decompose an action of writing one function into more than one step, give all information of writing a function in exactly one line.
+      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.
+      3. **Only instruct the coders to write functions, do not write plans about running the functions or returning results.** 
+      4. **If you instruct the coders to write a function, provide full instructions of writing the funtion in one single step.**
+      5. **If there are multiple information about one specific function you instruct to write, put it all in exactly the same one line as the line of the function**
+    
+    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 ###
+    Extend the code library with a function named 'extract_birth_date'. This function should take in the content of a Wikipedia page and return the birth date of the person the page is about. The function should be able to handle different date formats and return the date in a consistent format.
+    ### end of input ###
+    
+    ### begin of plan ###
+    1. Write a function named 'extract_birth_date'. This function should take in a string parameter which represents the content of a Wikipedia page, inside the function, use a regular expression to search for the birth date in the content, the regular expression should be able to handle different date formats. If a birth date is found, convert it to a consistent format using a date parsing library. Return the birth date in the consistent format. If no birth date is found, return an appropriate message indicating that the birth date could not be found.
+    ### end of plan ###
+    
+    Another 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 ###
+    
+    Takeaway message: Decompose the goal into writing functions, **for each function, provide every information about the function in exactly one line.**
+    **It is VERY IMPORTANT that each step should instruct exactly one function, do not instruct a function in more than 1 step.**
+    
+    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, each step gives full details of writing a function. **DO NOT** separate one action to write a function into more than one step.
+    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, each step of plan should contain full information about writing a function",
+    }
+    Ensure your responses can be parsed by Python json.loads
+    
+human_message_prompt_template:
+  _target_: aiflows.prompt_template.JinjaPrompt
+  template: |2-
+    Here is the response to your last action:
+    {{goal}}
+  input_variables:
+    - "goal"
+
+init_human_message_prompt_template:
+  _target_: aiflows.prompt_template.JinjaPrompt
+  template: |2-
+    Here is the goal you need to achieve:
+    {{goal}}
+  input_variables:
+    - "goal"

README.md

@@ -1,3 +1,33 @@
----
-license: mit
----
+# Table of Contents
+
+* [PlanGeneratorAtomicFlow](#PlanGeneratorAtomicFlow)
+  * [PlanGeneratorAtomicFlow](#PlanGeneratorAtomicFlow.PlanGeneratorAtomicFlow)
+* [\_\_init\_\_](#__init__)
+
+<a id="PlanGeneratorAtomicFlow"></a>
+
+# PlanGeneratorAtomicFlow
+
+<a id="PlanGeneratorAtomicFlow.PlanGeneratorAtomicFlow"></a>
+
+## PlanGeneratorAtomicFlow Objects
+
+```python
+class PlanGeneratorAtomicFlow(ChatAtomicFlow)
+```
+
+This class wraps around the Chat API to generate plan from a goal.
+
+*Input Interface Non Initialized*:
+- `goal`
+
+*Input Interface Initialized*:
+- `goal`
+
+*Output Interface*:
+- `plan`
+
+<a id="__init__"></a>
+
+# \_\_init\_\_
+

__init__.py

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

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,52 @@
+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/PlanGeneratorFlowModule", "revision": "main"},
+]
+
+from aiflows import flow_verse
+
+flow_verse.sync_dependencies(dependencies)
+
+if __name__ == "__main__":
+    # ~~~ make sure to set the openai api key in the envs ~~~
+    key = os.getenv("OPENAI_API_KEY")
+    api_information = [ApiInfo(backend_used="openai", api_key=os.getenv("OPENAI_API_KEY"))]
+    current_dir = os.getcwd()
+    cfg_path = os.path.join(current_dir, "PlanGeneratorAtomicFlow.yaml")
+    cfg = read_yaml_file(cfg_path)
+
+
+    # configuring api information
+    quick_load(cfg, api_information)
+
+    PlanGenFlow = hydra.utils.instantiate(cfg, _recursive_=False, _convert_="partial")
+
+
+    input_data = {
+        "goal": "fetch google's stock prices from 2021-01-01 to 2021-06-01 and plot it."
+    }
+    input_message = InputMessage.build(
+        data_dict=input_data,
+        src_flow="Launcher",
+        dst_flow=PlanGenFlow.name
+    )
+
+    # ~~~ calling the flow ~~~
+    output_message = PlanGenFlow(input_message)
+
+    print(output_message.data)