Update SaveCodeAtomicFlow.py
Browse files- SaveCodeAtomicFlow.py +21 -0
SaveCodeAtomicFlow.py
CHANGED
@@ -3,7 +3,20 @@ from aiflows.base_flows.atomic import AtomicFlow
|
|
3 |
|
4 |
|
5 |
class SaveCodeAtomicFlow(AtomicFlow):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
6 |
def _check_input(self, input_data: Dict[str, Any]):
|
|
|
|
|
|
|
|
|
7 |
assert "code" in input_data, "code is not passed to SaveCodeAtomicFlow"
|
8 |
assert "memory_files" in input_data, "memory_files is not passed to SaveCodeFlow"
|
9 |
assert "code_library" in input_data["memory_files"], "code_library not in memory_files"
|
@@ -13,6 +26,10 @@ class SaveCodeAtomicFlow(AtomicFlow):
|
|
13 |
pass
|
14 |
|
15 |
def _call(self, input_data: Dict[str, Any]):
|
|
|
|
|
|
|
|
|
16 |
try:
|
17 |
code_to_append = input_data["code"]
|
18 |
code_lib_loc = input_data["memory_files"]["code_library"]
|
@@ -33,5 +50,9 @@ class SaveCodeAtomicFlow(AtomicFlow):
|
|
33 |
self,
|
34 |
input_data: Dict[str, Any]
|
35 |
):
|
|
|
|
|
|
|
|
|
36 |
self._check_input(input_data)
|
37 |
return self._call(input_data)
|
|
|
3 |
|
4 |
|
5 |
class SaveCodeAtomicFlow(AtomicFlow):
|
6 |
+
"""This flow appends the code to the code library file.
|
7 |
+
*Input Interface*:
|
8 |
+
- `code` (str): the code to be appended to the code library
|
9 |
+
- `memory_files` (dict): the dictionary of memory files
|
10 |
+
|
11 |
+
*Output Interface*:
|
12 |
+
- `result` (str): the result of the flow, to be returned to the controller of the caller
|
13 |
+
- `summary` (str): the summary of the flow, to be appended to logs
|
14 |
+
"""
|
15 |
def _check_input(self, input_data: Dict[str, Any]):
|
16 |
+
"""Check if the input data is valid
|
17 |
+
:param input_data: the input data
|
18 |
+
:return: None
|
19 |
+
"""
|
20 |
assert "code" in input_data, "code is not passed to SaveCodeAtomicFlow"
|
21 |
assert "memory_files" in input_data, "memory_files is not passed to SaveCodeFlow"
|
22 |
assert "code_library" in input_data["memory_files"], "code_library not in memory_files"
|
|
|
26 |
pass
|
27 |
|
28 |
def _call(self, input_data: Dict[str, Any]):
|
29 |
+
"""The internal logic of the flow
|
30 |
+
:param input_data: the input data
|
31 |
+
:return: the output data
|
32 |
+
"""
|
33 |
try:
|
34 |
code_to_append = input_data["code"]
|
35 |
code_lib_loc = input_data["memory_files"]["code_library"]
|
|
|
50 |
self,
|
51 |
input_data: Dict[str, Any]
|
52 |
):
|
53 |
+
"""Run the flow
|
54 |
+
:param input_data: the input data
|
55 |
+
:return: the output data
|
56 |
+
"""
|
57 |
self._check_input(input_data)
|
58 |
return self._call(input_data)
|