Merge branch 'main' of https://github.com/endomorphosis/ipfs_datasets

Author: endomorphosis

ipfs_datasets_py/ipfs_datasets.py

@@ -0,0 +1,3 @@
+from .ipfs_embeddings import ipfs_embeddings_py
+from .ipfs_only_hash import ipfs_only_hash_py
+from .ipfs_multiformats import ipfs_multiformats_py

ipfs_datasets_py/ipfs_embeddings_py/__init__.py

@@ -0,0 +1,180 @@
+from .ipfs_multiformats import *
+from .ipfs_only_hash import *
+import asyncio
+import subprocess
+import os
+import datasets
+
+class ipfs_embeddings_py:
+    def __init__(self, resources, metedata):
+        self.multiformats = ipfs_multiformats_py(resources, metedata)
+        self.ipfs_only_hash = ipfs_only_hash_py(resources, metedata)
+        self.tei_https_endpoints = {}
+        self.libp2p_endpoints = {}
+        self.cid_queue = iter([])
+        self.knn_queue = iter([])
+        self.cid_index = {}
+        self.knn_index = {}
+        self.endpoint_status = {}
+        return None
+    
+    def load_index(self, index):
+        self.index = index
+        return None 
+    
+    def add_tei_https_endpoint(self, model, endpoint, batch_size):
+        if model not in self.tei_https_endpoints:
+            self.tei_https_endpoints[model] = {}
+        if endpoint not in self.tei_https_endpoints[model]:  
+            self.tei_https_endpoints[model][endpoint] = batch_size
+        return None
+    
+    def add_libp2p_endpoint(self, model, endpoint, batch_size):
+        if model not in self.libp2p_endpoints:
+            self.libp2p_endpoints[model] = {}
+        if endpoint not in self.libp2p_endpoints[model]:  
+            self.libp2p_endpoints[model][endpoint] = batch_size
+        return None
+    
+    def rm_tei_https_endpoint(self, model, endpoint):
+        if model in self.tei_https_endpoints and endpoint in self.tei_https_endpoints[model]:
+            del self.tei_https_endpoints[model][endpoint]
+            del self.endpoint_status[endpoint]
+        return None
+    
+    def rm_libp2p_endpoint(self, model, endpoint):
+        if model in self.libp2p_endpoints and endpoint in self.libp2p_endpoints[model]:
+            del self.libp2p_endpoints[model][endpoint]
+            del self.endpoint_status[endpoint]
+        return None
+    
+    def test_tei_https_endpoint(self, model, endpoint):
+        if model in self.tei_https_endpoints and endpoint in self.tei_https_endpoints[model]:
+            return True
+        return False
+
+    def test_libp2p_endpoint(self, model, endpoint):
+        if model in self.libp2p_endpoints and endpoint in self.libp2p_endpoints[model]:
+            return True
+        return False
+
+    def get_tei_https_endpoint(self, model):
+        if model in self.tei_https_endpoints:
+            return self.tei_https_endpoints[model]
+        return None
+
+    def request_tei_https_endpoint(self, model, batch_size):
+        if model in self.tei_https_endpoints:
+            for endpoint in self.tei_https_endpoints[model]:
+                if self.endpoint_status[endpoint] == 1:
+                    return endpoint
+        return None
+
+    def index_ipfs(self, samples):
+        if type(samples) is None:
+            raise ValueError("samples must be a list")
+        if type(samples) is str:
+            samples = [samples]
+        if type(samples) is iter:
+            for this_sample in samples:
+                this_sample_cid = self.multiformats.get_cid(this_sample)
+                self.cid_index[this_sample_cid] = this_sample
+            pass
+        if type(samples) is list:
+            for this_sample in samples:
+                this_sample_cid = self.multiformats.get_cid(this_sample)
+                self.cid_index[this_sample_cid] = this_sample
+        return None
+    
+    def index_knn(self, samples):
+        if type(samples) is None:
+            raise ValueError("samples must be a list")
+        if type(samples) is str:
+            samples = [samples]
+        if type(samples) is iter:
+            for this_sample in samples:
+                this_sample_cid = self.multiformats.get_cid(this_sample)
+                self.knn_index[this_sample_cid] = this_sample
+            pass
+        if type(samples) is list:
+            for this_sample in samples:
+                this_sample_cid = self.multiformats.get_cid(this_sample)
+                self.knn_index[this_sample_cid] = this_sample
+        return None
+    
+    def queue_index_cid(self, samples):
+        if type(samples) is None:
+            raise ValueError("samples must be a list")
+        if type(samples) is str:
+            samples = [samples]
+        if type(samples) is iter:
+            for this_sample in samples:
+                self.cid_queue.append(this_sample)
+            pass
+        if type(samples) is list:
+            for this_sample in samples:
+                self.cid_queue.append(this_sample)
+
+        return None
+    
+    def choose_endpoint(self):
+        filtered_endpoints = {}
+        filtered_endpoints = {k: v for k, v in self.endpoint_status.items() if v == 1}
+        if len(filtered_endpoints) == 0:
+            return None
+        else:
+            return filtered_endpoints
+        
+    def https_index_cid(self, samples, endpoint):
+        endpoint_chunk_size = self.tei_https_endpoints[endpoint]
+        all_chunk = []
+        this_chunk = []
+        for i in range(samples):
+            self
+            ## request endpoint
+            pass
+        return None
+    
+    def pop_https_index_cid(self, samples):
+
+        choose_endpoint = self.choose_endpoint()
+        endpoint_chunk_size = self.tei_https_endpoints[choose_endpoint]
+        all_chunk = []
+        this_chunk = []
+        for i in range(samples):
+            this_chunk.append(self.cid_queue.pop())
+            if i % endpoint_chunk_size == 0:
+                all_chunk.append(this_chunk)
+                this_chunk = []
+
+
+    def test(self):
+        self.add_tei_https_endpoint("BAAI/bge-m3", "62.146.169.111:80/embed",1)
+        self.add_tei_https_endpoint("BAAI/bge-m3", "62.146.169.111:8080/embed",1)
+        self.add_tei_https_endpoint("BAAI/bge-m3", "62.146.168.111:8081/embed",1)
+        test_knn_index = {}
+        test_cid_index = {}
+        test_data = {
+            "test1", "test2", "test3"
+        }
+
+        for data in test_data:
+            test_cid_index = self.index_ipfs(data)
+            test_knn_index = self.index_knn(data)
+        
+
+        print("test")
+
+    def status(self):
+        return self.endpointStatus
+    
+    def setStatus(self,endpoint , status):
+        self.endpointStatus[endpoint] = status
+        return None
+
+if __name__ == '__main__':
+    resources = {}
+    metedata = {}
+    ipfs_embeddings = ipfs_embeddings_py(resources, metedata)
+    ipfs_embeddings.test()
+    print("test")

ipfs_datasets_py/ipfs_embeddings_py/ipfs_embeddings.py

@@ -0,0 +1,34 @@
+import hashlib
+from multiformats import CID, multihash
+
+class ipfs_multiformats_py:
+    def __init__(self, resources, metadata): 
+        self.multihash = multihash
+        return None
+    
+    # Step 1: Hash the file content with SHA-256
+    def get_file_sha256(self, file_path):
+        hasher = hashlib.sha256()
+        with open(file_path, 'rb') as f:
+            while chunk := f.read(8192):
+                hasher.update(chunk)
+        return hasher.digest()
+
+    # Step 2: Wrap the hash in Multihash format
+    def get_multihash_sha256(self, file_content_hash):
+        mh = self.multihash.wrap(file_content_hash, 'sha2-256')
+        return mh
+
+    # Step 3: Generate CID from Multihash (CIDv1)
+    def get_cid(self, file_path):
+        file_content_hash = self.get_file_sha256(file_path)
+        mh = self.get_multihash_sha256(file_content_hash)
+        cid = CID('base32', 'raw', mh)
+        return str(cid)
+    
+
+if __name__ == '__main__':
+    ipfs_multiformats = ipfs_multiformats_py()
+    file_path = 'path_to_your_file'
+    cid = ipfs_multiformats.get_cid(file_path)
+    print(f"CID: {cid}")

ipfs_datasets_py/ipfs_embeddings_py/ipfs_multiformats.py

@@ -0,0 +1,20 @@
+import os
+import subprocess
+
+class ipfs_only_hash_py:
+    def __init__(self, resources, metadata):
+
+        return None
+    
+    def __call__(self, file_path):
+        absolute_path = os.path.abspath(file_path)
+        ipfs_hash_cmd = "bash -c 'npx ipfs-only-hash " + absolute_path 
+        ipfs_hash = subprocess.check_output(ipfs_hash_cmd, shell=True).decode('utf-8').strip()
+        return ipfs_hash
+    
+    def __test__(self):
+        test_file_path = "test.txt"
+        test_ipfs_hash = self(test_file_path)
+        print(test_ipfs_hash)
+        return None
+        

ipfs_datasets_py/ipfs_embeddings_py/ipfs_only_hash.py

@@ -0,0 +1,36 @@
+
+from datasets import load_dataset
+import ipfs_embeddings_py.ipfs_embeddings as ipfs_embeddings
+
+class test_ipfs_embeddings:
+    def __init__(self):
+        resources = {}
+        metadata = {}
+        self.dataset = {}
+        self.ipfs_embeddings = ipfs_embeddings.ipfs_embeddings_py(resources, metadata)
+        return None
+    
+    def process(self, dataset, output):
+        num_rows = dataset.num_rows['data']
+        processed_data = {}
+        for i in range(num_rows):
+            row = dataset['data'][i]
+            data = row['data']
+            processed_data[row] = self.ipfs_embeddings.add_tei_https_queue(data, self.callback())
+        return None
+
+    def callback(self, data):
+        return None
+
+
+    def test(self):
+        load_these_datasets = ["laion/Wikipedia-X", "laion/Wikipedia-X-Full", "laion/Wikipedia-X-Concat", "laion/Wikipedia-X-M3"]
+        self.dataset = load_dataset(load_these_datasets[0])
+        print(len(self.dataset))
+        self.ipfs_embeddings
+        return None
+    
+if __name__ == '__main__':
+    test = test_ipfs_embeddings()
+    test.test()
+    print("Test passed")

ipfs_datasets_py/ipfs_embeddings_py/test/test.py

@@ -0,0 +1,107 @@
+import os
+import torch.nn.functional as F
+from torch import inference_mode, float16, Tensor
+from transformers import AutoTokenizer, AutoModelForCausalLM, StoppingCriteriaList
+from transformers.generation.streamers import TextStreamer
+from cloudkit_worker import dispatch_result
+from sentence_transformers import SentenceTransformer
+from InstructorEmbedding import INSTRUCTOR
+from FlagEmbedding import FlagModel
+import json
+
+embedding_models = [
+	"text-embedding-ada-002",
+	"gte-large",
+	"gte-base",
+	"gte-small",
+	"gte-tiny",
+	"bge-small-en-v1.5",
+	"bge-base-en-v1.5",
+	"bge-large-en-v1.5",
+	"instructor-base",
+	"instructor-large",
+	"instructor-xl",
+	"UAE-Large-V1"
+]
+
+class hf_embed:
+
+	def __init__(self, resources, meta):
+		self.modelName = meta['modelName']
+		self.hf_embed = self.embed
+		self.instruct_embed = self.embed
+		if  "gte" in resources['checkpoint']:
+			self.tokenizer = AutoTokenizer.from_pretrained(resources['checkpoint'])
+		if "instructor" in resources['checkpoint']:
+			self.model = INSTRUCTOR(resources['checkpoint'])
+		elif "gte" in resources['checkpoint']:
+			self.model = SentenceTransformer(
+				resources['checkpoint']
+				)
+		elif "bge" in resources['checkpoint']:
+			self.model = None
+		
+			
+	def __call__(self, method, **kwargs):
+		if method == 'hf_embed':
+			return self.embed(**kwargs)
+		elif method == 'instruct_embed':
+			return self.embed(**kwargs)
+		else:
+			raise Exception('unknown method: %s' % method)
+	
+	def embed(self, instruction, text , **kwargs):
+		self.input = text
+		self.method = 'embed'
+		embeddings = None
+		if "instructor" in self.modelName:
+			embeddings = self.model.encode([[instruction,self.input]])
+			print(embeddings)
+		if "gte" in self.modelName:
+			embeddings = self.model.encode([self.input])
+			print(embeddings)
+		if "bge" in self.modelName:
+			if self.model == None:
+				self.model = FlagModel(
+					'BAAI/'+self.modelName, query_instruction_for_retrieval=instruction,
+					use_fp16=True
+				)
+			embeddings = self.model.encode(str(self.input))
+			print(embeddings)
+
+		if type(embeddings) != str:
+			embeddings = json.dumps(embeddings.tolist())
+
+		return {
+			'text': embeddings, 
+			'done': True
+		}
+		
+	def average_pool(last_hidden_states: Tensor, attention_mask: Tensor) -> Tensor:
+		last_hidden = last_hidden_states.masked_fill(~attention_mask[..., None].bool(), 0.0)
+		return last_hidden.sum(dim=1) / attention_mask.sum(dim=1)[..., None]
+
+
+
+def test():
+	cwd = os.getcwd()
+	dir = os.path.dirname(__file__)
+	grandparent = os.path.dirname(dir)
+	models = os.path.join(grandparent, "models")
+	checkpoint = 'bge-base-en-v1.5'
+	resources = {}
+	resources['checkpoint'] = models + "/" + checkpoint + "@hf"
+	
+	print(resources["checkpoint"])
+	meta = {"modelName":"bge-base-en-v1.5"}
+	text = "sample text to embed"
+	model = "bge-base-en-v1.5"
+	instruction = "Represent this sentence for searching relevant passages:"
+	embed = hf_embed(resources, meta)
+	results = embed.embed(instruction, text)
+	print(results)
+	return results
+
+if __name__ == '__main__':
+	test()
+	# pass