| from faiss import IndexFlatL2,write_index,read_index |
| import numpy as np |
| from utils.convert_embedding import GetEmbedding |
|
|
|
|
| class VectorStore: |
| def __init__(self): |
| pass |
|
|
| def store_vectors(self,data:list,embedding_space_name:str = 'faiss_index.index',numpy_emb_space:str = 'embeddings.npy' ): |
| try: |
| embeddings = GetEmbedding(data=data).convert_emb() |
| diamension = embeddings.shape[1] |
| print("Diamension",diamension) |
| |
| index = IndexFlatL2(diamension) |
|
|
| index.add(embeddings) |
|
|
| write_index(index, embedding_space_name) |
|
|
| |
| np.save(numpy_emb_space, embeddings) |
| return True |
| except Exception as e: |
| print(e) |
| return False |
|
|
| def get_similary_search(self,query,embedding_space_name:str = 'faiss_index.index',numpy_emb_space:str = 'embeddings.npy',K:int= 1): |
| |
| index = read_index('faiss_index.index') |
|
|
| |
| embeddings_np = np.load('embeddings.npy') |
|
|
| |
| query = "What is photosynthesis?" |
| query_embedding = GetEmbedding([query]).convert_emb() |
| query_embedding = query_embedding.detach().numpy() |
| |
| |
| |
| |
| |
| distances, indices = index.search(query_embedding, k = K) |
|
|
| return indices |
|
|
|
|
|
|
|
|