Upload 3 files

MemoryReadingAtomicFlow.py

@@ -0,0 +1,58 @@
+from typing import Dict, Any
+import os
+
+from flows.base_flows import AtomicFlow
+from flows.utils.general_helpers import get_pyfile_functions_metadata_from_file
+
+
+class MemoryReadingAtomicFlow(AtomicFlow):
+    """A flow to read memory from given files.
+
+    Any composite flow that uses this flow should have
+    memory_files: Dict[str, str] which maps memory name to its memory file location in the flow_state
+
+    *Input Interface*: - `memory_files` : name of the Dict which maps the memory name to its file location e.g.
+    {"plan": "examples/JARVIS/plan.txt"}
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.supported_mem_name = ["plan", "logs", "code_library"]
+
+    def _check_input_data(self, input_data: Dict[str, Any]):
+        """input data sanity check"""
+        for mem_name, mem_path in input_data.items():
+            assert mem_name in self.supported_mem_name, (f"{mem_name} is not supported in MemoryReadingAtomicFlow, "
+                                                         f"supported names are: {self.supported_mem_name}")
+            assert os.path.exists(mem_path), f"{mem_path} does not exist."
+            assert os.path.isfile(mem_path), f"{mem_path} is not a file."
+
+    def _read_text(self, file_location):
+        with open(file_location, 'r', encoding='utf-8') as file:
+            content = file.read()
+        return content
+
+    def _read_py_code_library(self, file_location):
+        metadata = get_pyfile_functions_metadata_from_file(file_location)
+        if len(metadata) == 0:
+            return "No functions yet."
+        descriptions = []
+        for function in metadata:
+            description = f"Function Name: {function['name']}\n"
+            description += f"Documentation: {function.get('docstring', 'No documentation')}\n"
+            description += f"Parameters: {', '.join(function.get('parameters', []))}\n"
+            description += f"Default Values: {function.get('defaults', 'None')}"
+            descriptions.append(description)
+        return "\n\n".join(descriptions)
+
+    def run(
+            self,
+            input_data: Dict[str, Any]):
+        self._check_input_data(input_data)
+        response = {}
+        for mem_name, mem_path in input_data.items():
+            if mem_name in ['plan', 'logs']:
+                response[mem_name] = self._read_text(mem_path)
+            elif mem_name == 'code_library' and mem_path.endswith('.py'):
+                response[mem_name] = self._read_py_code_library(mem_path)
+        return response

MemoryReadingAtomicFlow.yaml

@@ -0,0 +1,5 @@
+name: "MemoryReadingAtomicFlow"
+description: "A flow that reads memory from given file locations"
+
+input_interface:
+  - "memory_files"

__init__.py

@@ -0,0 +1 @@
+from .MemoryReadingAtomicFlow import MemoryReadingAtomicFlow