0.2.15 - add delete_memories and agentlogger

README.md

@@ -338,6 +338,33 @@ id (str/int): The ID of the memory.
 >>> delete_memory("books", "1")
 ```
 
+#### `delete_memories(category, document=None, metadata=None)`
+
+Delete all memories in the category either by document, or by metadata, or by both.
+
+##### Arguments
+
+```
+# Required
+category (str): The category of the memory.
+
+# Optional
+document (str): Document text to match memories to delete. Defaults to None.
+metadata (dict): Metadata to match memories to delete. Defaults to None.
+```
+
+##### Returns
+
+```
+bool: True if memories were deleted, False otherwise.
+```
+
+##### Example
+
+```python
+>>> delete_memories("books", document="Harry Potter", metadata={"author": "J.K. Rowling"})
+```
+
 ## Check if a memory exists
 
 #### `memory_exists(category, id, includes_metadata=None)`

agentmemory/__init__.py

@@ -6,6 +6,7 @@
     get_memory,
     update_memory,
     delete_memory,
+    delete_memories,
     delete_similar_memories,
     count_memories,
     wipe_category,
@@ -37,6 +38,7 @@
     "get_memory",
     "update_memory",
     "delete_memory",
+    "delete_memories",
     "delete_similar_memories",
     "count_memories",
     "wipe_category",

agentmemory/helpers.py

@@ -1,27 +1,12 @@
 import json
 import os
 import dotenv
-from rich.panel import Panel
-from rich.console import Console
+from agentlogger import log
 
 dotenv.load_dotenv()
 
-console = Console()
-
 DEBUG = os.getenv("DEBUG", "false") == "true" or os.getenv("DEBUG", "false") == "True"
 
-DEFAULT_TYPE_COLORS = {
-    "unknown": "white",
-    "error": "red",
-    "warning": "yellow",
-    "info": "blue",
-    "prompt": "cyan",
-    "success": "green",
-    "critical": "red",
-    "system": "magenta",
-}
-
-
 def strip_embeddings(value):
     if isinstance(value, dict):
         value = value.copy()
@@ -42,26 +27,19 @@ def debug_log(
     content,
     input_dict=None,
     type="info",
-    color="blue",
-    type_colors=DEFAULT_TYPE_COLORS,
     panel=True,
     debug=DEBUG,
 ):
     if debug is not True:
         return
 
-    color = type_colors.get(type, color)
-
     if input_dict is not None:
         # traverse the dict and find any value called "embedding"
         # set "embedding" value to [] to avoid printing it
         new_dict = strip_embeddings(input_dict.copy())
         content = content + f"\n({json.dumps(new_dict, indent=4)})"
 
-    if panel:
-        console.print(Panel(content, title="agentmemory: " + type, style=color))
-    else:
-        console.print(content, style=color)
+    log(content, title="agentmemory", type=type, panel=panel)
 
 
 def chroma_collection_to_list(collection):

agentmemory/main.py

@@ -109,7 +109,7 @@ def search_memory(
     include_distances=True,
     max_distance=None,  # 0.0 - 1.0
     min_distance=None,  # 0.0 - 1.0
-    unique=False
+    unique=False,
 ):
     """
     Cearch a collection with given query texts.
@@ -243,7 +243,7 @@ def get_memories(
     filter_metadata=None,
     n_results=20,
     include_embeddings=True,
-    unique=False
+    unique=False,
 ):
     """
     Retrieve a list of memories from a given category, sorted by ID, with optional filtering.
@@ -286,7 +286,7 @@ def get_memories(
         ]
 
         filter_metadata = {"$and": filter_metadata}
-    
+
     if unique:
         if filter_metadata is None:
             filter_metadata = {}
@@ -392,6 +392,37 @@ def delete_memory(category, id):
     debug_log(f"Deleted memory {id} in category {category}")
 
 
+def delete_memories(category, document=None, metadata=None):
+    """
+    Delete all memories in the category either by document, or by metadata, or by both.
+
+    Arguments:
+        category (str): The category of the memories.
+        document (str, optional): The text of the memories to delete. Defaults to None.
+        metadata (dict, optional): The metadata of the memories to delete. Defaults to None.
+
+    Returns:
+        bool: True if memories are deleted, False otherwise.
+
+    Example:
+        >>> delete_memories("books", document="Harry Potter", metadata={"author": "J.K. Rowling"})
+    """
+    check_client_initialized()  # client is lazy loaded, so make sure it is is initialized
+
+    # Get or create the collection for the given category
+    memories = get_chroma_client().get_or_create_collection(category)
+
+    # Create a query to match either the document or the metadata
+    if document is not None:
+        memories.delete(where_document={"$contains": document})
+    if metadata is not None:
+        memories.delete(where=metadata)
+
+    debug_log(f"Deleted memories from category {category}")
+
+    return True
+
+
 def delete_similar_memories(category, content, similarity_threshold=0.95):
     """
     Search for memories that are similar to the item that contains the content and removes it.

agentmemory/tests/main.py

@@ -8,11 +8,13 @@
     count_memories,
     wipe_category,
     wipe_all_memories,
+    delete_memories,
 )
 from agentmemory.main import create_unique_memory, delete_similar_memories
 
 
 def test_memory_creation_and_retrieval():
+    wipe_category("test")
     # create 10 memories
     for i in range(20):
         create_memory(
@@ -28,9 +30,11 @@ def test_memory_creation_and_retrieval():
 
     # assert that the first memory is document 19
     assert memories[0]["document"] == "document 19"
+    wipe_category("test")
 
 
 def test_memory_deletion():
+    wipe_category("test")
     # Delete memory test
     create_memory("test", "delete memory test", metadata={"test": "test"})
     memories = get_memories("test")
@@ -39,18 +43,22 @@ def test_memory_deletion():
     # test delete_memory
     delete_memory("test", memory_id)
     assert count_memories("test") == num_memories - 1
+    wipe_category("test")
 
 
 def test_memory_update():
+    wipe_category("test")
     create_memory("test", "update memory test", metadata={"test": "test"})
     memories = get_memories("test")
     memory_id = memories[0]["id"]
 
     update_memory("test", memory_id, "doc 1 updated", metadata={"test": "test"})
     assert get_memory("test", memory_id)["document"] == "doc 1 updated"
+    wipe_category("test")
 
 
 def test_search_memory():
+    wipe_category("test")
     # rewrite as a for loop
     for i in range(5):
         create_memory(
@@ -68,6 +76,7 @@ def test_search_memory():
     )
 
     assert search_results[0]["document"] == "document 1"
+    wipe_category("test")
 
 
 def test_wipe_category():
@@ -77,26 +86,32 @@ def test_wipe_category():
 
 
 def test_count_memories():
+    wipe_category("test")
     for i in range(3):
         create_memory("test", "document " + str(i + 1), metadata={"test": "test"})
     assert count_memories("test") == 3
+    wipe_category("test")
 
 
 def test_memory_search_distance():
+    wipe_category("test")
     create_memory("test", "cinammon duck cakes")
     max_dist_limited_memories = search_memory(
         "test", "cinammon duck cakes", max_distance=0.1
     )
 
     assert len(max_dist_limited_memories) == 1
+    wipe_category("test")
 
 
 def test_wipe_all_memories():
+    create_memory("test", "test document")
     wipe_all_memories()
     assert count_memories("test") == 0
 
+
 def test_create_unique_memory():
-    wipe_all_memories()
+    wipe_category("test")
     # Test creating a unique memory
     create_unique_memory("test", "unique_memory_1")
     memories = get_memories("test")
@@ -108,20 +123,46 @@ def test_create_unique_memory():
     memories = get_memories("test")
     assert len(memories) == 2
     assert memories[0]["metadata"]["unique"] == "False"
+    wipe_category("test")
 
 
 def test_delete_similar_memories():
-    wipe_all_memories()
+    wipe_category("test")
     # Create a memory and a similar memory
     create_memory("test", "similar_memory_1")
     create_memory("test", "similar_memory_2")
 
     # Test deleting a similar memory
-    assert delete_similar_memories("test", "similar_memory_1", similarity_threshold=0.999999) is True
+    assert (
+        delete_similar_memories(
+            "test", "similar_memory_1", similarity_threshold=0.999999
+        )
+        is True
+    )
     memories = get_memories("test")
     assert len(memories) == 1
 
     # Test deleting a non-similar memory
     assert delete_similar_memories("test", "not_similar_memory") is False
     memories = get_memories("test")
     assert len(memories) == 1
+    wipe_category("test")
+
+
+def test_delete_memories():
+    wipe_category("books")
+    # create a memory to be deleted
+    create_memory("books", "Foundation", metadata={"author": "Isaac Asimov"}, id="1")
+    create_memory("books", "Foundation and Empire", metadata={"author": "Isaac Asimov"}, id="2")
+    create_memory("books", "Second Foundation", metadata={"author": "Isaac Asimov"}, id="3")
+
+    # assert the memory exists
+    assert get_memory("books", "1") is not None
+
+    # delete the memory
+    assert delete_memories("books", "Foundation")
+    assert delete_memories("books", metadata={"author": "Isaac Asimov"})
+
+    # assert the memory does not exist anymore
+    assert get_memory("books", "1") is None
+    wipe_category("books")

requirements.txt

@@ -1,2 +1,2 @@
-chromadb==0.4.0
-rich
+chromadb
+agentlogger

setup.py

@@ -13,7 +13,7 @@
 
 setup(
     name='agentmemory',
-    version='0.2.14',
+    version='0.2.15',
     description='Easy-to-use agent memory, powered by chromadb',
     long_description=long_description,  # added this line
     long_description_content_type="text/markdown",  # and this line
@@ -22,7 +22,7 @@
     author_email='[email protected]',
     license='MIT',
     packages=['agentmemory'],
-    install_requires=['chromadb', 'rich'],
+    install_requires=['chromadb', 'agentlogger'],
     readme = "README.md",
     classifiers=[
         'Development Status :: 4 - Beta',