File size: 4,531 Bytes
923696f |
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 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
import os
import os.path
# Set a dedicated folder for file I/O
working_directory = "auto_gpt_workspace"
# Create the directory if it doesn't exist
if not os.path.exists(working_directory):
os.makedirs(working_directory)
def safe_join(base, *paths):
"""Join one or more path components intelligently."""
new_path = os.path.join(base, *paths)
norm_new_path = os.path.normpath(new_path)
if os.path.commonprefix([base, norm_new_path]) != base:
raise ValueError("Attempted to access outside of working directory.")
return norm_new_path
def split_file(content, max_length=4000, overlap=0):
"""
Split text into chunks of a specified maximum length with a specified overlap
between chunks.
:param text: The input text to be split into chunks
:param max_length: The maximum length of each chunk,
default is 4000 (about 1k token)
:param overlap: The number of overlapping characters between chunks,
default is no overlap
:return: A generator yielding chunks of text
"""
start = 0
content_length = len(content)
while start < content_length:
end = start + max_length
if end + overlap < content_length:
chunk = content[start : end + overlap]
else:
chunk = content[start:content_length]
yield chunk
start += max_length - overlap
def read_file(filename) -> str:
"""Read a file and return the contents"""
try:
filepath = safe_join(working_directory, filename)
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
return content
except Exception as e:
return f"Error: {str(e)}"
def ingest_file(filename, memory, max_length=4000, overlap=200):
"""
Ingest a file by reading its content, splitting it into chunks with a specified
maximum length and overlap, and adding the chunks to the memory storage.
:param filename: The name of the file to ingest
:param memory: An object with an add() method to store the chunks in memory
:param max_length: The maximum length of each chunk, default is 4000
:param overlap: The number of overlapping characters between chunks, default is 200
"""
try:
print(f"Working with file {filename}")
content = read_file(filename)
content_length = len(content)
print(f"File length: {content_length} characters")
chunks = list(split_file(content, max_length=max_length, overlap=overlap))
num_chunks = len(chunks)
for i, chunk in enumerate(chunks):
print(f"Ingesting chunk {i + 1} / {num_chunks} into memory")
memory_to_add = (
f"Filename: {filename}\n" f"Content part#{i + 1}/{num_chunks}: {chunk}"
)
memory.add(memory_to_add)
print(f"Done ingesting {num_chunks} chunks from {filename}.")
except Exception as e:
print(f"Error while ingesting file '{filename}': {str(e)}")
def write_to_file(filename, text):
"""Write text to a file"""
try:
filepath = safe_join(working_directory, filename)
directory = os.path.dirname(filepath)
if not os.path.exists(directory):
os.makedirs(directory)
with open(filepath, "w", encoding="utf-8") as f:
f.write(text)
return "File written to successfully."
except Exception as e:
return "Error: " + str(e)
def append_to_file(filename, text):
"""Append text to a file"""
try:
filepath = safe_join(working_directory, filename)
with open(filepath, "a") as f:
f.write(text)
return "Text appended successfully."
except Exception as e:
return "Error: " + str(e)
def delete_file(filename):
"""Delete a file"""
try:
filepath = safe_join(working_directory, filename)
os.remove(filepath)
return "File deleted successfully."
except Exception as e:
return "Error: " + str(e)
def search_files(directory):
found_files = []
if directory == "" or directory == "/":
search_directory = working_directory
else:
search_directory = safe_join(working_directory, directory)
for root, _, files in os.walk(search_directory):
for file in files:
if file.startswith("."):
continue
relative_path = os.path.relpath(os.path.join(root, file), working_directory)
found_files.append(relative_path)
return found_files
|