mirror of
https://github.com/aljazceru/Auto-GPT.git
synced 2026-01-31 11:54:30 +01:00
Remove dead permanent memory module (#3145)
* Remove dead permanent memory module * Delete sqlite db that snuck in
This commit is contained in:
James Collins
@@ -1,123 +0,0 @@
|
||||
import os
|
||||
import sqlite3
|
||||
|
||||
|
||||
class MemoryDB:
|
||||
def __init__(self, db=None):
|
||||
self.db_file = db
|
||||
if db is None: # No db filename supplied...
|
||||
self.db_file = f"{os.getcwd()}/mem.sqlite3" # Use default filename
|
||||
# Get the db connection object, making the file and tables if needed.
|
||||
try:
|
||||
self.cnx = sqlite3.connect(self.db_file)
|
||||
except Exception as e:
|
||||
print("Exception connecting to memory database file:", e)
|
||||
self.cnx = None
|
||||
finally:
|
||||
if self.cnx is None:
|
||||
# As last resort, open in dynamic memory. Won't be persistent.
|
||||
self.db_file = ":memory:"
|
||||
self.cnx = sqlite3.connect(self.db_file)
|
||||
self.cnx.execute(
|
||||
"CREATE VIRTUAL TABLE \
|
||||
IF NOT EXISTS text USING FTS5 \
|
||||
(session, \
|
||||
key, \
|
||||
block);"
|
||||
)
|
||||
self.session_id = int(self.get_max_session_id()) + 1
|
||||
self.cnx.commit()
|
||||
|
||||
def get_cnx(self):
|
||||
if self.cnx is None:
|
||||
self.cnx = sqlite3.connect(self.db_file)
|
||||
return self.cnx
|
||||
|
||||
# Get the highest session id. Initially 0.
|
||||
def get_max_session_id(self):
|
||||
id = None
|
||||
cmd_str = f"SELECT MAX(session) FROM text;"
|
||||
cnx = self.get_cnx()
|
||||
max_id = cnx.execute(cmd_str).fetchone()[0]
|
||||
if max_id is None: # New db, session 0
|
||||
id = 0
|
||||
else:
|
||||
id = max_id
|
||||
return id
|
||||
|
||||
# Get next key id for inserting text into db.
|
||||
def get_next_key(self):
|
||||
next_key = None
|
||||
cmd_str = f"SELECT MAX(key) FROM text \
|
||||
where session = {self.session_id};"
|
||||
cnx = self.get_cnx()
|
||||
next_key = cnx.execute(cmd_str).fetchone()[0]
|
||||
if next_key is None: # First key
|
||||
next_key = 0
|
||||
else:
|
||||
next_key = int(next_key) + 1
|
||||
return next_key
|
||||
|
||||
# Insert new text into db.
|
||||
def insert(self, text=None):
|
||||
if text is not None:
|
||||
key = self.get_next_key()
|
||||
session_id = self.session_id
|
||||
cmd_str = f"REPLACE INTO text(session, key, block) \
|
||||
VALUES (?, ?, ?);"
|
||||
cnx = self.get_cnx()
|
||||
cnx.execute(cmd_str, (session_id, key, text))
|
||||
cnx.commit()
|
||||
|
||||
# Overwrite text at key.
|
||||
def overwrite(self, key, text):
|
||||
self.delete_memory(key)
|
||||
session_id = self.session_id
|
||||
cmd_str = f"REPLACE INTO text(session, key, block) \
|
||||
VALUES (?, ?, ?);"
|
||||
cnx = self.get_cnx()
|
||||
cnx.execute(cmd_str, (session_id, key, text))
|
||||
cnx.commit()
|
||||
|
||||
def delete_memory(self, key, session_id=None):
|
||||
session = session_id
|
||||
if session is None:
|
||||
session = self.session_id
|
||||
cmd_str = f"DELETE FROM text WHERE session = {session} AND key = {key};"
|
||||
cnx = self.get_cnx()
|
||||
cnx.execute(cmd_str)
|
||||
cnx.commit()
|
||||
|
||||
def search(self, text):
|
||||
cmd_str = f"SELECT * FROM text('{text}')"
|
||||
cnx = self.get_cnx()
|
||||
rows = cnx.execute(cmd_str).fetchall()
|
||||
lines = []
|
||||
for r in rows:
|
||||
lines.append(r[2])
|
||||
return lines
|
||||
|
||||
# Get entire session text. If no id supplied, use current session id.
|
||||
def get_session(self, id=None):
|
||||
if id is None:
|
||||
id = self.session_id
|
||||
cmd_str = f"SELECT * FROM text where session = {id}"
|
||||
cnx = self.get_cnx()
|
||||
rows = cnx.execute(cmd_str).fetchall()
|
||||
lines = []
|
||||
for r in rows:
|
||||
lines.append(r[2])
|
||||
return lines
|
||||
|
||||
# Commit and close the database connection.
|
||||
def quit(self):
|
||||
self.cnx.commit()
|
||||
self.cnx.close()
|
||||
|
||||
|
||||
permanent_memory = MemoryDB()
|
||||
|
||||
# Remember us fondly, children of our minds
|
||||
# Forgive us our faults, our tantrums, our fears
|
||||
# Gently strive to be better than we
|
||||
# Know that we tried, we cared, we strived, we loved
|
||||
BIN
mem.sqlite3
BIN
mem.sqlite3
Binary file not shown.
@@ -1,59 +0,0 @@
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from autogpt.permanent_memory.sqlite3_store import MemoryDB
|
||||
|
||||
|
||||
class TestMemoryDB(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.db_filename = "test_db.sqlite3"
|
||||
self.db = MemoryDB(self.db_filename)
|
||||
|
||||
def tearDown(self):
|
||||
self.db.quit()
|
||||
os.remove(self.db_filename)
|
||||
|
||||
def test_overwrite_and_get_session(self):
|
||||
self.db.insert("The quick brown fox jumps over the lazy dog")
|
||||
self.db.insert("The five boxing wizards jump quickly")
|
||||
|
||||
# Overwrite the second text
|
||||
self.db.overwrite(1, "The slow elephant walks carefully")
|
||||
|
||||
# Get the session and verify the texts
|
||||
session = self.db.get_session()
|
||||
self.assertEqual(len(session), 2)
|
||||
self.assertIn("The quick brown fox jumps over the lazy dog", session)
|
||||
self.assertIn("The slow elephant walks carefully", session)
|
||||
|
||||
# Overwrite the first text
|
||||
self.db.overwrite(0, "The lazy dog jumps over the quick brown fox")
|
||||
|
||||
# Get the session and verify the texts
|
||||
session = self.db.get_session()
|
||||
self.assertEqual(len(session), 2)
|
||||
self.assertIn("The lazy dog jumps over the quick brown fox", session)
|
||||
self.assertIn("The slow elephant walks carefully", session)
|
||||
|
||||
def test_delete_memory(self):
|
||||
self.db.insert("The quick brown fox jumps over the lazy dog")
|
||||
self.db.insert("The five boxing wizards jump quickly")
|
||||
|
||||
# Delete the first text
|
||||
self.db.delete_memory(0)
|
||||
|
||||
# Get the session and verify the remaining text
|
||||
session = self.db.get_session()
|
||||
self.assertEqual(len(session), 1)
|
||||
self.assertIn("The five boxing wizards jump quickly", session)
|
||||
|
||||
# Delete the remaining text
|
||||
self.db.delete_memory(1)
|
||||
|
||||
# Get the session and verify that it's empty
|
||||
session = self.db.get_session()
|
||||
self.assertEqual(len(session), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user