[CLEANUP][Tests][Non Vector DB Memory]

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2023 Eternal Reclaimer
+Copyright (c) 2023 The Galactic Swarm Corporation
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -1,5 +1,3 @@
-[![Multi-Modality](agorabanner.png)](https://discord.gg/qUtxnK2NMf)
-
 
 <div align="center">
   <a href="https://swarms.world">

examples/faiss_example.py

@@ -1,5 +1,5 @@
 from typing import List, Dict, Any
-from swarms_memory.faiss_wrapper import FAISSDB
+from swarms_memory.vector_dbs.faiss_wrapper import FAISSDB
 
 
 from transformers import AutoTokenizer, AutoModel

examples/pinecome_wrapper_example.py

@@ -1,5 +1,5 @@
 from typing import List, Dict, Any
-from swarms_memory.pinecone_wrapper import PineconeMemory
+from swarms_memory.vector_dbs.pinecone_wrapper import PineconeMemory
 
 
 # Example usage

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "swarms-memory"
-version = "0.0.5"
+version = "0.0.7"
 description = "Swarms Memory - Pytorch"
 license = "MIT"
 authors = ["Kye Gomez <[email protected]>"]
@@ -28,7 +28,9 @@ chromadb = "*"
 loguru = "*"
 sentence-transformers = "*"
 pinecone = "*"
-faiss = "*"
+faiss-cpu = "*"
+pydantic = "*"
+
 
 
 

requirements.txt

@@ -4,3 +4,4 @@ sentence-transformers
 pinecone
 faiss-cpu
 torch
+pydantic

swarms_memory/__init__.py

@@ -1,5 +1,2 @@
-from swarms_memory.chroma_db_wrapper import ChromaDB
-from swarms_memory.pinecone_wrapper import PineconeMemory
-from swarms_memory.faiss_wrapper import FAISSDB
-
-__all__ = ["ChromaDB", "PineconeMemory", "FAISSDB"]
+from swarms_memory.vector_dbs import * # noqa: F401, F403
+from swarms_memory.utils import * # noqa: F401, F403

swarms_memory/utils/__init__.py

@@ -0,0 +1,14 @@
+from swarms_memory.utils.action_subtask import ActionSubtaskEntry
+
+from swarms_memory.utils.dict_internal_memory import DictInternalMemory
+from swarms_memory.utils.dict_shared_memory import DictSharedMemory
+from swarms_memory.utils.short_term_memory import ShortTermMemory
+from swarms_memory.utils.visual_memory import VisualShortTermMemory
+
+__all__ = [
+    "ActionSubtaskEntry",
+    "DictInternalMemory",
+    "DictSharedMemory",
+    "ShortTermMemory",
+    "VisualShortTermMemory",
+]

swarms_memory/utils/action_subtask.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+
+
+class ActionSubtaskEntry(BaseModel):
+    """Used to store ActionSubtask data to preserve TaskMemory pointers and context in the form of thought and action.
+
+    Attributes:
+    thought: CoT thought string from the LLM.
+    action: ReAct action JSON string from the LLM.
+    answer: tool-generated and memory-processed response from Griptape.
+    """
+
+    thought: str
+    action: str
+    answer: str

swarms_memory/utils/dict_internal_memory.py

@@ -0,0 +1,86 @@
+import uuid
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Tuple
+
+
+class InternalMemoryBase(ABC):
+    """Abstract base class for internal memory of agents in the swarm."""
+
+    def __init__(self, n_entries):
+        """Initialize the internal memory. In the current architecture the memory always consists of a set of soltuions or evaluations.
+        During the operation, the agent should retrivie best solutions from it's internal memory based on the score.
+
+        Moreover, the project is designed around LLMs for the proof of concepts, so we treat all entry content as a string.
+        """
+        self.n_entries = n_entries
+
+    @abstractmethod
+    def add(self, score, entry):
+        """Add an entry to the internal memory."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_top_n(self, n):
+        """Get the top n entries from the internal memory."""
+        raise NotImplementedError
+
+
+class DictInternalMemory(InternalMemoryBase):
+    def __init__(self, n_entries: int):
+        """
+        Initialize the internal memory. In the current architecture the memory always consists of a set of solutions or evaluations.
+        Simple key-value store for now.
+
+        Args:
+            n_entries (int): The maximum number of entries to keep in the internal memory.
+        """
+        super().__init__(n_entries)
+        self.data: Dict[str, Dict[str, Any]] = {}
+
+    def add(self, score: float, content: Any) -> None:
+        """
+        Add an entry to the internal memory.
+
+        Args:
+            score (float): The score or fitness value associated with the entry.
+            content (Any): The content of the entry.
+
+        Returns:
+            None
+        """
+        random_key: str = str(uuid.uuid4())
+        self.data[random_key] = {"score": score, "content": content}
+
+        # keep only the best n entries
+        sorted_data: List[Tuple[str, Dict[str, Any]]] = sorted(
+            self.data.items(),
+            key=lambda x: x[1]["score"],
+            reverse=True,
+        )
+        self.data = dict(sorted_data[: self.n_entries])
+
+    def get_top_n(self, n: int) -> List[Tuple[str, Dict[str, Any]]]:
+        """
+        Get the top n entries from the internal memory.
+
+        Args:
+            n (int): The number of top entries to retrieve.
+
+        Returns:
+            List[Tuple[str, Dict[str, Any]]]: A list of tuples containing the random keys and corresponding entry data.
+        """
+        sorted_data: List[Tuple[str, Dict[str, Any]]] = sorted(
+            self.data.items(),
+            key=lambda x: x[1]["score"],
+            reverse=True,
+        )
+        return sorted_data[:n]
+
+    def len(self) -> int:
+        """
+        Get the number of entries in the internal memory.
+
+        Returns:
+            int: The number of entries in the internal memory.
+        """
+        return len(self.data)

swarms_memory/utils/dict_shared_memory.py

@@ -0,0 +1,96 @@
+import datetime
+import json
+import os
+import threading
+import uuid
+from pathlib import Path
+from typing import Any, Dict
+
+
+class DictSharedMemory:
+    """A class representing a shared memory that stores entries as a dictionary.
+
+    Attributes:
+        file_loc (Path): The file location where the memory is stored.
+        lock (threading.Lock): A lock used for thread synchronization.
+
+    Methods:
+        __init__(self, file_loc: str = None) -> None: Initializes the shared memory.
+        add_entry(self, score: float, agent_id: str, agent_cycle: int, entry: Any) -> bool: Adds an entry to the internal memory.
+        get_top_n(self, n: int) -> None: Gets the top n entries from the internal memory.
+        write_to_file(self, data: Dict[str, Dict[str, Any]]) -> bool: Writes the internal memory to a file.
+    """
+
+    def __init__(self, file_loc: str = None) -> None:
+        """Initialize the shared memory. In the current architecture the memory always consists of a set of soltuions or evaluations.
+        Moreover, the project is designed around LLMs for the proof of concepts, so we treat all entry content as a string.
+        """
+        if file_loc is not None:
+            self.file_loc = Path(file_loc)
+            if not self.file_loc.exists():
+                self.file_loc.touch()
+
+        self.lock = threading.Lock()
+
+    def add(
+        self,
+        score: float,
+        agent_id: str,
+        agent_cycle: int,
+        entry: Any,
+    ) -> bool:
+        """Add an entry to the internal memory."""
+        with self.lock:
+            entry_id = str(uuid.uuid4())
+            data = {}
+            epoch = datetime.datetime.utcfromtimestamp(0)
+            epoch = (datetime.datetime.utcnow() - epoch).total_seconds()
+            data[entry_id] = {
+                "agent": agent_id,
+                "epoch": epoch,
+                "score": score,
+                "cycle": agent_cycle,
+                "content": entry,
+            }
+            status = self.write_to_file(data)
+            self.plot_performance()
+            return status
+
+    def get_top_n(self, n: int) -> None:
+        """Get the top n entries from the internal memory."""
+        with self.lock:
+            with open(self.file_loc) as f:
+                try:
+                    file_data = json.load(f)
+                except Exception as e:
+                    file_data = {}
+                    raise e
+
+            sorted_data = dict(
+                sorted(
+                    file_data.items(),
+                    key=lambda item: item[1]["score"],
+                    reverse=True,
+                )
+            )
+            top_n = dict(list(sorted_data.items())[:n])
+            return top_n
+
+    def write_to_file(self, data: Dict[str, Dict[str, Any]]) -> bool:
+        """Write the internal memory to a file."""
+        if self.file_loc is not None:
+            with open(self.file_loc) as f:
+                try:
+                    file_data = json.load(f)
+                except Exception as e:
+                    file_data = {}
+                    raise e
+
+            file_data = file_data | data
+            with open(self.file_loc, "w") as f:
+                json.dump(file_data, f, indent=4)
+
+                f.flush()
+                os.fsync(f.fileno())
+
+            return True