| import os |
| from dotenv import load_dotenv |
|
|
| from .ConnectorStrategy import ConnectorStrategy |
|
|
| from pinecone import Pinecone, ServerlessSpec |
| from langchain_openai import OpenAIEmbeddings |
| from langchain_pinecone import PineconeVectorStore |
| from langchain_core.documents import Document |
|
|
| import unicodedata |
| import time |
|
|
| class PineconeConnector(ConnectorStrategy): |
| def __init__(self): |
|
|
| load_dotenv() |
|
|
| pinecone_api_key = os.environ.get("PINECONE_API_KEY") |
|
|
| self.index_name = os.environ.get("PINECONE_INDEX_NAME") |
| self.namespace = os.environ.get("PINECONE_NAMESPACE") |
|
|
|
|
| pc = Pinecone(api_key=pinecone_api_key) |
|
|
| existing_indexes = [index_info["name"] for index_info in pc.list_indexes()] |
|
|
| if self.index_name not in existing_indexes: |
| pc.create_index( |
| name=self.index_name, |
| dimension=3072, |
| metric="cosine", |
| spec=ServerlessSpec(cloud="aws", region="us-east-1"), |
| ) |
| while not pc.describe_index(self.index_name).status["ready"]: |
| time.sleep(1) |
|
|
| self.index = pc.Index(self.index_name) |
|
|
| |
| def getDocs(self): |
| |
| |
| docs_names = [] |
| for ids in self.index.list(namespace=self.namespace): |
| for id in ids: |
| name_doc = "_".join(id.split("_")[:-1]) |
| if name_doc not in docs_names: |
| docs_names.append(name_doc) |
|
|
| return docs_names |
| |
| |
| def addDoc(self, filename, text_chunks, embedding): |
| try: |
| vector_store = PineconeVectorStore(index=self.index, embedding=embedding,namespace=self.namespace) |
|
|
| file_name = filename.split(".")[0].replace(" ","_").replace("-","_").replace(".","_").replace("/","_").replace("\\","_").strip() |
|
|
| documents = [] |
| uuids = [] |
|
|
| for i, chunk in enumerate(text_chunks): |
| clean_filename = remove_non_standard_ascii(file_name) |
| uuid = f"{clean_filename}_{i}" |
|
|
| document = Document( |
| page_content=chunk, |
| metadata={ "filename":filename, "chunk_id":uuid }, |
| ) |
|
|
| uuids.append(uuid) |
| documents.append(document) |
| |
|
|
| vector_store.add_documents(documents=documents, ids=uuids) |
|
|
| return {"filename_id":clean_filename} |
| |
| except Exception as e: |
| print(e) |
| return False |
| |
| def retriever(self, query, embedding): |
|
|
| vector_store = PineconeVectorStore(index=self.index, embedding=embedding,namespace=self.namespace) |
|
|
| retriever = vector_store.as_retriever( |
| search_type="similarity_score_threshold", |
| search_kwargs={"k": 3, "score_threshold": 0.6}, |
| ) |
|
|
| return retriever.invoke(query) |
| |
|
|
| def remove_non_standard_ascii(input_string: str) -> str: |
| normalized_string = unicodedata.normalize('NFKD', input_string) |
| return ''.join(char for char in normalized_string if 'a' <= char <= 'z' or 'A' <= char <= 'Z' or char.isdigit() or char in ' .,!?') |
|
|
|
|
