File size: 4,034 Bytes
faf6c9a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 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 |