Create an abstract MemoryProviderSingleton class. Pass config instead of instantiating a new one where used.

scripts/memory/pinecone.py

@@ -0,0 +1,51 @@
+
+import pinecone
+
+from memory.base import MemoryProviderSingleton, get_ada_embedding
+
+
+class PineconeMemory(MemoryProviderSingleton):
+    def __init__(self, cfg):
+        pinecone_api_key = cfg.pinecone_api_key
+        pinecone_region = cfg.pinecone_region
+        pinecone.init(api_key=pinecone_api_key, environment=pinecone_region)
+        dimension = 1536
+        metric = "cosine"
+        pod_type = "p1"
+        table_name = "auto-gpt"
+        # this assumes we don't start with memory.
+        # for now this works.
+        # we'll need a more complicated and robust system if we want to start with memory.
+        self.vec_num = 0
+        if table_name not in pinecone.list_indexes():
+            pinecone.create_index(table_name, dimension=dimension, metric=metric, pod_type=pod_type)
+        self.index = pinecone.Index(table_name)
+
+    def add(self, data):
+        vector = get_ada_embedding(data)
+        # no metadata here. We may wish to change that long term.
+        resp = self.index.upsert([(str(self.vec_num), vector, {"raw_text": data})])
+        _text = f"Inserting data into memory at index: {self.vec_num}:\n data: {data}"
+        self.vec_num += 1
+        return _text
+
+    def get(self, data):
+        return self.get_relevant(data, 1)
+
+    def clear(self):
+        self.index.delete(deleteAll=True)
+        return "Obliviated"
+
+    def get_relevant(self, data, num_relevant=5):
+        """
+        Returns all the data in the memory that is relevant to the given data.
+        :param data: The data to compare to.
+        :param num_relevant: The number of relevant data to return. Defaults to 5
+        """
+        query_embedding = get_ada_embedding(data)
+        results = self.index.query(query_embedding, top_k=num_relevant, include_metadata=True)
+        sorted_results = sorted(results.matches, key=lambda x: x.score)
+        return [str(item['metadata']["raw_text"]) for item in sorted_results]
+
+    def get_stats(self):
+        return self.index.describe_index_stats()