NeoGPT is an open-source, locally-run Language Model (LLM) 📚 with a focus on privacy and data security. Interact naturally with AI 🗣️ without sacrificing your personal data. Enjoy a private, customizable, and secure AI experience! 🔐🚀
import logging
import os
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor, as_completed
from langchain.docstore.document import Document
from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from typing import Optional, Iterator, List, Dict
from chromadb.config import Settings
import argparse
from vectorstore.chroma import ChromaStore
from vectorstore.faiss import FAISSStore
from config import (
    SOURCE_DIR,
    INGEST_THREADS,
    EMBEDDING_MODEL,
    DEVICE_TYPE,
    MODEL_DIRECTORY,
    DOCUMENT_EXTENSION,
)

def load_single_document(file_path: str) -> Document:
    # Loads a single document from a file path
    file_extension = os.path.splitext(file_path)[1]
    loader_class = DOCUMENT_EXTENSION.get(file_extension)
    if loader_class:
        loader = loader_class(file_path)
    else:
        raise ValueError("Document type is undefined")
    return loader.load()[0]


def load_document_batch(filepaths):
    logging.info("Loading document batch")
    # create a thread pool
    with ThreadPoolExecutor(len(filepaths)) as exe:
        # load files
        futures = [exe.submit(load_single_document, name) for name in filepaths]
        # collect data
        data_list = [future.result() for future in futures]
        # return data and file paths
        return (data_list, filepaths)


def load_documents(source_directory : str) -> list[Document]:
    doc_path = []


    for root,_,files in os.walk(source_directory):
        for file_name in files:
            file_extension = os.path.splitext(file_name)[1]
            source_file_path = os.path.join(root, file_name)
            if file_extension in DOCUMENT_EXTENSION.keys():
                doc_path.append(source_file_path)

    n_workers = min(INGEST_THREADS, max(len(doc_path), 1))
    chunk_size = round(len(doc_path) / n_workers)

    docs = []

    with ProcessPoolExecutor(n_workers) as executor:
        futures = []
        # split the load operations into chunks
        for i in range(0, len(doc_path), chunk_size):
            # select a chunk of filenames
            filepaths = doc_path[i : (i + chunk_size)]
            # submit the task
            future = executor.submit(load_document_batch, filepaths)
            futures.append(future)
        # process all results
        for future in as_completed(futures):
            # open the file and load the data
            contents, _ = future.result()
            docs.extend(contents)
    return docs


def builder(vectorstore: str = "Chroma"):
    logging.info(f"Loading Documents from {SOURCE_DIR}")
    documents = load_documents(SOURCE_DIR)
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200)

    texts = text_splitter.split_documents(documents)

    logging.info(f"Loaded {len(documents)} documents from {SOURCE_DIR}")
    logging.info(f"Split into {len(texts)} chunks of text")
    logging.info(f"Using {DEVICE_TYPE} device for embedding model")

    match vectorstore:
        case "Chroma":
            logging.info(f"Using Chroma for vectorstore")
            db = ChromaStore().from_documents(
                documents=texts,
            )
            logging.info(f"Loaded Documents to Chroma DB Successfully")
        case "FAISS":
            logging.info(f"Using FAISS for vectorstore")
            db = FAISSStore().from_documents(
                documents=texts,
            )
            logging.info(f"Loaded Documents to FAISS DB Successfully")
    
    logging.info(f"Builder👷🏻‍♀️ has built your VectorDB successfully!")
   
# def query():
#     embeddings = HuggingFaceInstructEmbeddings(
#         model_name=EMBEDDING_MODEL_NAME,
#         model_kwargs={"device": 'mps'},
#         cache_folder=os.path.join(os.path.dirname(__file__), "models"),
#     )
#     db = Chroma(
#         embedding_function=embeddings,
#         persist_directory=PERSIST_DIRECTORY,
#     )
#     while True:
#         query = input()
#         results= db.similarity_search(query)
#         print(results)

if __name__ == "__main__":
    logging.basicConfig(
        format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s - %(message)s", level=logging.INFO,
    )
    parser = argparse.ArgumentParser(description="NeoGPT CLI Interface")
    parser.add_argument(
        "--db",
        choices=["Chroma", "FAISS"],
        default="Chroma",
        help="Specify the vectorstore (Chroma, FAISS)",
    )
    args = parser.parse_args()
    builder(vectorstore=args.db)



builder.py
neokd/NeoGPT
import os 
import torch
from chromadb.config import Settings
from langchain.document_loaders import (
    PDFMinerLoader, 
    TextLoader,
    UnstructuredHTMLLoader,
    UnstructuredTSVLoader,
    CSVLoader,
    UnstructuredEmailLoader,
    UnstructuredEPubLoader,
    UnstructuredExcelLoader,
    UnstructuredPowerPointLoader,
    UnstructuredWordDocumentLoader,
    UnstructuredMarkdownLoader,
    JSONLoader
)

# Source Directory for Documents to Ingest
SOURCE_DIR = os.path.join(os.path.dirname(__file__), "documents")
# To store models from HuggingFace
MODEL_DIRECTORY = os.path.join(os.path.dirname(__file__), "models")
# PARENT DB DIRECTORY
PARENT_DB_DIRECTORY = os.path.join(os.path.dirname(__file__), "db")
# CHROMA DB DIRECTORY
CHROMA_PERSIST_DIRECTORY = os.path.join(PARENT_DB_DIRECTORY, "chroma")
# FAISS DB DIRECTORY
FAISS_PERSIST_DIRECTORY = os.path.join(PARENT_DB_DIRECTORY, "faiss")
# PINECONE DB DIRECTORY
PINECONE_PERSIST_DIRECTORY = os.path.join(PARENT_DB_DIRECTORY,"pinecone")

# GGUF MODELS (Recommended , Default and Fast)
MODEL_NAME = "TheBloke/Mistral-7B-Instruct-v0.1-GGUF"
MODEL_FILE = "mistral-7b-instruct-v0.1.Q4_K_M.gguf"
# HUGGING FACE MODEL (Not recommended for low RAM systems)
# MODEL_NAME = "microsoft/phi-1_5"
# MODEL_FILE = None
# DEFAULT EMBEDDING MODEL
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L12-v2"
# EMBEDDING MODEL CONFIG
INGEST_THREADS = 8 or os.cpu_count()

# MODEL CONFIG
MAX_TOKEN_LENGTH = 8192 # 8192 is the max for Mistral-7B
N_GPU_LAYERS = 40

# PYTORCH DEVICE COMPATIBILITY
if torch.cuda.is_available():
    DEVICE_TYPE = "cuda"
elif torch.backends.mps.is_available():
    DEVICE_TYPE = "mps"
else:
    DEVICE_TYPE = "cpu"

#  CHROMA DB SETTINGS
CHROMA_SETTINGS = Settings(
    anonymized_telemetry=False,
    is_persistent=True,
)

# PINECONE SETTINGS
EMBEDDING_DIMENSION = ""
INDEX_NAME = ""

# List of file supported for ingest 
DOCUMENT_EXTENSION = {
    '.pdf': PDFMinerLoader,
    '.txt': TextLoader,
    '.csv' :CSVLoader,
    '.html' :UnstructuredHTMLLoader, 
    '.tsv' :UnstructuredTSVLoader,
    '.eml' :UnstructuredEmailLoader,
    '.epub' :UnstructuredEPubLoader,
    '.xls' :UnstructuredExcelLoader,
    '.xlsx' :UnstructuredExcelLoader,
    '.pptx' :UnstructuredPowerPointLoader,
    '.ppt' :UnstructuredPowerPointLoader,
    '.docx' :UnstructuredWordDocumentLoader,
    '.doc' :UnstructuredWordDocumentLoader,
    '.md' :UnstructuredMarkdownLoader,
    '.json' :JSONLoader,
}

config.py
neokd/NeoGPT
import logging
import torch
from uuid import UUID
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.chains import RetrievalQA, RetrievalQAWithSourcesChain
from langchain.llms import LlamaCpp, HuggingFacePipeline
from langchain.callbacks.manager import CallbackManager
from langchain.schema.agent import AgentFinish
from langchain.schema.output import LLMResult
from huggingface_hub import hf_hub_download
from langchain.utilities import GoogleSearchAPIWrapper
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer, GenerationConfig, pipeline, AutoModelForCausalLM
from langchain.retrievers.web_research import WebResearchRetriever
from langchain.callbacks.base import BaseCallbackHandler
from vectorstore.chroma import ChromaStore
from vectorstore.faiss import FAISSStore
from langchain.vectorstores import Chroma
from pinecone import Pinecone,PineconeClient
import argparse
from dotenv import load_dotenv
import os
from prompts.prompt import get_prompt
load_dotenv()
from config import (
    MODEL_DIRECTORY,
    DEVICE_TYPE,
    MODEL_NAME,
    MODEL_FILE,
    N_GPU_LAYERS,
    MAX_TOKEN_LENGTH,
)
from typing import Any

# Define a custom callback handler class for token collection
class TokenCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        super().__init__()
        self._tokens = []

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        self._tokens.append(token)

    def on_llm_end(self, response: LLMResult, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any) -> Any:
        # Calculate the cost based on OpenAI's pricing
        cost_per_token = ((len(self._tokens) / 1000) * 0.002) * 83.33  # INR Cost per token
        total_tokens = len(self._tokens)
        print(f"\n\nTotal tokens generated: {total_tokens}")
        print(f"Total cost: {cost_per_token} INR")

# Function to load the LLM 
def load_model(device_type:str = DEVICE_TYPE, model_id:str = MODEL_NAME, model_basename:str = MODEL_FILE, LOGGING=logging):
    """
        input: device_type, model_id, model_basename, LOGGING
        output: Hugging Face model
        description:  The function loads the pre-trained model from Hugging Face and returns the loaded model.
    """
    if model_basename is not None and ".gguf" in model_basename.lower() :

        callback_manager = CallbackManager([StreamingStdOutCallbackHandler(),TokenCallbackHandler()])
        try:
        # Download the model checkpoint from the Hugging Face Hub
            model_path = hf_hub_download(
                repo_id=model_id,
                filename=model_basename,
                resume_download=True,
                cache_dir=MODEL_DIRECTORY,
            )
            # Model Parameters
            kwargs = {
                "model_path": model_path,
                "max_tokens": MAX_TOKEN_LENGTH,
                "n_ctx": MAX_TOKEN_LENGTH,
                "n_batch": 512,  
                "callback_manager": callback_manager,
                "verbose":False,
                "f16_kv":True,
                "streaming":True,
            }
            if device_type.lower() == "mps":
                kwargs["n_gpu_layers"] = 1 # only for MPS devices
            if device_type.lower() == "cuda":
                kwargs["n_gpu_layers"] = N_GPU_LAYERS  # set this based on your GPU
            # Create a LlamaCpp object (language model)
            llm =  LlamaCpp(**kwargs)
            LOGGING.info(f"Loaded {model_id} locally")
            return llm  # Returns a LlamaCpp object (language model)
        except Exception as e:
            LOGGING.info(f"Error {e}")

    else:
        try:
            # Load the Hugging Face model and tokenizer
            model = AutoModelForCausalLM.from_pretrained(
                model_id,
                device_map="auto",
                # torch_dtype=torch.float16,
                # low_cpu_mem_usage=True,
                # load_in_4bit=True,
                # bnb_4bit_quant_type="nf4",
                # bnb_4bit_compute_dtype=torch.float16,
                cache_dir=MODEL_DIRECTORY,
                trust_remote_code=True
            )
            # print(model)
            tokenizer = AutoTokenizer.from_pretrained(model_id,device_map="auto",trust_remote_code=True)
            # Move the model to the specified device (cuda or cpu)
            # model.to(device_type)
            # model.tie_weights()
            # torch.set_default_device(device_type)
            LOGGING.info(f"Loaded Hugging Face model: {model_id} successfully")
            generation_config = GenerationConfig.from_pretrained(model_id)
            pipe = pipeline(
                "text-generation",
                model=model,
                tokenizer=tokenizer,
                max_length=MAX_TOKEN_LENGTH,
                # temperature=0.2,
                # top_p=0.95,
                # do_sample=True,
                # repetition_penalty=1.15,
                generation_config=generation_config,
            )
            llm = HuggingFacePipeline(pipeline=pipe)
            llm("Hello World")
            return llm
        except Exception as e:
            LOGGING.info(f"Error {e}")

# Function to set up the retrieval-based question-answering system
def db_retriver(device_type:str = DEVICE_TYPE,vectorstore:str = "Chroma", LOGGING=logging):
    """
        input: device_type,vectorstore, LOGGING
        output: None
        description: The function is used to set up the retrieval-based question-answering system. It loads the LLM model, the Chroma DB, and the prompt and memory objects. It then creates a retrieval-based question-answering system using the LLM model and the Chroma DB.
    """
    match vectorstore:
        case "Chroma":
            # Load the Chroma DB with the embedding model
            db = ChromaStore()
            LOGGING.info(f"Loaded Chroma DB Successfully")
        case "FAISS":
            # Load the FAISS DB with the embedding model
            db = FAISSStore().load_local()
            LOGGING.info(f"Loaded FAISS DB Successfully")
        case "Pinecone":
            # Initialize Pinecone client
            # Load the Pinecone DB with the embedding model
            pinecone_api_key = "your_api_key"
            pinecone_environment = "your_environment_name"
            db= Pinecone(api_key=pinecone_api_key, environment=pinecone_environment)
            LOGGING.info(f"Initialized Pinecone DB Successfully")
            
    # Create a retriever object 
    retriever = db.as_retriever()
    # Load the LLM model
    llm = load_model(device_type, model_id=MODEL_NAME, model_basename=MODEL_FILE, LOGGING=logging)
    # Prompt Builder Function 
    prompt , memory = get_prompt()
    # Create a retrieval-based question-answering system using the LLM model and the Vector DB
    chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=retriever,
        chain_type="stuff",
        chain_type_kwargs={"prompt": prompt, "memory": memory},
    )
    # Run the retrieval-based question-answering system
    chain("Write linux command to copy file from one directory to another directory",return_only_outputs=True)

### TODO: Add the Web Search Retriever (In Progress)
def web_retriver(device_type:str = DEVICE_TYPE,vectorstore:str = "Chroma", LOGGING=logging):
    """
        input: device_type,vectorstore,LOGGING
        output: None
        description: The function is used to set up the retrieval-based question-answering system. It loads the LLM Model and searches on web for the answer backed up by the vectorstore.
        WARNING: The function is still in progress and is not yet complete.
    """
    # Import the env
    try:
        os.environ["GOOGLE_CSE_ID"] = os.environ.get("GOOGLE_CSE_ID")
        os.environ["GOOGLE_API_KEY"] = os.environ.get("GOOGLE_API_KEY")
        # Using Google Search API Wrapper
        print(os.environ.get("GOOGLE_CSE_ID"))
        
    except Exception as e:
        LOGGING.info(f"Error {e}")
    search = GoogleSearchAPIWrapper()
    # Load the Vector DB
    match vectorstore:
        case "Chroma":
            # Load the Chroma DB with the embedding model
            db = ChromaStore()
            LOGGING.info(f"Loaded Chroma DB Successfully")
        case "FAISS":
            # Load the FAISS DB with the embedding model
            db = FAISSStore().load_local()
            LOGGING.info(f"Loaded FAISS DB Successfully")
        case "Pinecone":
            # Initialize Pinecone client
            # Load the Pinecone DB with the embedding model
            pinecone_api_key = "your_api_key"
            pinecone_environment = "your_environment_name"
            db= Pinecone(api_key=pinecone_api_key, environment=pinecone_environment)
            LOGGING.info(f"Initialized Pinecone DB Successfully")
            
    llm = load_model(device_type, model_id=MODEL_NAME, model_basename=MODEL_FILE, LOGGING=logging)
          
    web_research_retriever = WebResearchRetriever.from_llm(
        vectorstore=db,
        llm=llm,
        search=search,
    )
    user_input = "What is Task Decomposition in LLM Powered Autonomous Agents?"
    qa_chain = RetrievalQAWithSourcesChain.from_chain_type(llm=llm ,retriever=web_research_retriever )
    result = qa_chain({"question": user_input})
    print(result)

if __name__ == '__main__':
    logging.basicConfig(
        format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s - %(message)s", level=logging.INFO,
    )
    # Parse the arguments
    parser = argparse.ArgumentParser(description="NeoGPT CLI Interface")
    parser.add_argument(
        "--device-type",
        choices=["cpu", "mps", "cuda"],
        default=DEVICE_TYPE,
        help="Specify the device type (cpu, mps, cuda)",
    )
    parser.add_argument(
        "--db",
        choices=["Chroma", "FAISS"],
        default="Chroma",
        help="Specify the vectorstore (Chroma, FAISS)",
    )
    args = parser.parse_args()
    db_retriver(device_type=args.device_type,vectorstore="Chroma", LOGGING=logging)
    # web_retriver(device_type=args.device_type,vectorstore="FAISS", LOGGING=logging)
    

main.py
neokd/NeoGPT
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferWindowMemory
from sentence_transformers import SentenceTransformer, util
import numpy as np

# Default System level prompt
SYSTEM_PROMPT = """
        You are a helpful assistant, you will use the provided context to answer user questions.
        Read the given context before answering questions and think step by step. If you can not answer a user  question based on the provided context, inform the user. Do not use any other information for answering user.
"""
# Default memory window size
DEFAULT_MEMORY_KEY = 2

def get_prompt(model_type:str = "mistral"):
    """
        input: model_type
        output: prompt, memory
        desciption: The function streamlines prompt generation for language models, mapping model_type to the appropriate format. It utilizes conversation memory, instruction tags, and model-specific templates, returning a tailored prompt and memory object. This utility enhances context-aware interactions with language models.
    """
    INSTRUCTION_TEMLATE = """
        Context: {history} \n {context}
        User: {question}
    """
    
    memory = ConversationBufferWindowMemory(k=DEFAULT_MEMORY_KEY,return_messages=True,input_key="question", memory_key="history")

    INSTRUCTION_BEGIN, INSTRUCTION_END = "[INST]", "[/INST]"
    SYSTEM_BEGIN, SYSTEM_END = "[SYS]", "[/SYS]"

    match model_type.lower():
        case "mistral":
            prompt_template = INSTRUCTION_BEGIN + SYSTEM_PROMPT + INSTRUCTION_TEMLATE + INSTRUCTION_END

    prompt = PromptTemplate(input_variables=["history", "context", "question"], template=prompt_template)
    
    return prompt, memory


# TASK AND RESPONSE PROMPTS
TASK_PROMPTS = {
    "rag": "You are a helpful assistant, you will use the provided context to answer user questions. Read the given context before answering questions and think step by step. If you can not answer a user  question based on the provided context, inform the user. Do not use any other information for answering user.",

    "summarization": "Your task is to summarize the text I give you in up to seven concise bullet points and start with a short, high-quality summary. Pick a suitable emoji for every bullet point. Your response should be in {{SELECTED_LANGUAGE}}. If the provided URL is functional and not a YouTube video, use the text from the {{URL}}. However, if the URL is not functional or is a YouTube video, use the following text: {{CONTENT}}.",

}


def hyper_prompt(model_type:str = "mistral",user_input:str = ""):
    try:
        model =  SentenceTransformer('/Users/kuldeep/Project/NeoGPT/models/sentence-transformers_all-MiniLM-L12-v2/')
    except:
        print("Please install sentence transformer model")
    
    user_embedding = model.encode(user_input, convert_to_tensor=True)

    # Calculate similarity scores between user input and each task's prompt
    similarity_scores = {}
    for task, prompt in TASK_PROMPTS.items():
        task_embedding = model.encode(prompt, convert_to_tensor=True)
        similarity = util.pytorch_cos_sim(user_embedding, task_embedding).item()
        similarity_scores[task] = similarity

    # Choose the task with the highest similarity score
    chosen_task = max(similarity_scores, key=similarity_scores.get)
    chosen_prompt = TASK_PROMPTS[chosen_task]
    print(chosen_prompt)
    return chosen_prompt

if __name__ == '__main__':
    hyper_prompt()





prompts/prompt.py
neokd/NeoGPT
from abc import ABC
from langchain.schema.document import Document
class VectorStore(ABC):
    def __init__(self) -> None:
        pass

    def from_documents(self, documents: list[Document]) -> Document:
        pass

    

vectorstore/base.py
neokd/NeoGPT
# --- Under Development ---
"""
    The Purpose of this file is to provide a wrapper around the ChromaDB
    to provide a simple interface for storing and retrieving documents
    from the database.
"""
from vectorstore.base import VectorStore
from langchain.schema.document import Document
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceInstructEmbeddings
from config import (
    CHROMA_PERSIST_DIRECTORY,
    CHROMA_SETTINGS,
    EMBEDDING_MODEL,
    DEVICE_TYPE,
    MODEL_DIRECTORY,
)
class ChromaStore(VectorStore):
    def __init__(self) -> None:
        self.embeddings = HuggingFaceInstructEmbeddings(
            model_name=EMBEDDING_MODEL,
            model_kwargs={"device": DEVICE_TYPE},
            cache_folder=MODEL_DIRECTORY,
        )
        self.chroma = Chroma(
            persist_directory=CHROMA_PERSIST_DIRECTORY,
            client_settings=CHROMA_SETTINGS,
            embedding_function=self.embeddings,
        )
    
    def from_documents(self, documents: list[Document]) -> Document:
        self.chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory,
            client_settings=CHROMA_SETTINGS,
        )
        return documents
    
    def as_retriever(self):
        return self.chroma.as_retriever()


    

vectorstore/chroma.py
neokd/NeoGPT
"""
    The Purpose of this file is to provide a wrapper around the FAISS from langchain
"""
from vectorstore.base import VectorStore
from langchain.schema.document import Document
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceInstructEmbeddings
from config import (
    FAISS_PERSIST_DIRECTORY,
    EMBEDDING_MODEL,
    DEVICE_TYPE,
    MODEL_DIRECTORY,
)

class FAISSStore(VectorStore):
    def __init__(self) -> None:
        self.embeddings = HuggingFaceInstructEmbeddings(
            model_name=EMBEDDING_MODEL,
            model_kwargs={"device": DEVICE_TYPE},
            cache_folder=MODEL_DIRECTORY,
        )
        self.faiss = FAISS(
            embedding_function=None,
            index=0,
            index_to_docstore_id={},
            docstore={},
        )
        self.docstore = None

    def from_documents(self, documents: list[Document]) -> Document:
        self.docstore = self.faiss.from_documents(
            documents=documents,
            embedding=self.embeddings,
        )
        self.docstore.save_local(FAISS_PERSIST_DIRECTORY) 
        # self.faiss.save_local(FAISS_PERSIST_DIRECTORY)

    def load_local(self):
        self.docstore = self.faiss.load_local(
            folder_path=FAISS_PERSIST_DIRECTORY,
            embeddings=self.embeddings,
        )
        return self.docstore

    def as_retriever(self):
        return self.docstore.as_retriever()
vectorstore/faiss.py
neokd/NeoGPT
"""
    The Purpose of this file is to provide a wrapper around the PINECONE from langchain
"""
from vectorstore.base import VectorStore
from langchain.schema.document import Document
from pinecone import Pinecone
from langchain.embeddings import HuggingFaceInstructEmbeddings
from config import (
    PINECONE_PERSIST_DIRECTORY,
    EMBEDDING_MODEL,
    DEVICE_TYPE,
    MODEL_DIRECTORY,
    EMBEDDING_DIMENSION,
    INDEX_NAME
)

class PineconeVectorStore(VectorStore):
     def __init__(self, api_key: str, environment: str) -> None:
        super().__init__()
        self.api_key = api_key
        self.environment = environment
        self.embeddings = HuggingFaceInstructEmbeddings(
            model_name=EMBEDDING_MODEL,
            model_kwargs={"device": DEVICE_TYPE},
            cache_folder=MODEL_DIRECTORY,
        )
        self.pinecone_client = Pinecone(
            api_key=api_key,
            environment=environment,
            persist_directory=PINECONE_PERSIST_DIRECTORY,
            embedding_dimension=EMBEDDING_DIMENSION, 
            index_name=INDEX_NAME,
            embedding_function=self.embeddings,
        )
     def from_documents(self, documents: list[Document]) -> Document:
        pinecone_data=[]
        for document in documents:
          pinecone_data.append({"fields_name":document.fields_name})

        self.pinecone_client.upsert(pinecone_data)

     def as_retriever(self):
       return self.pinecone_client.as_retriever()
       
api_key = "Your api key"
environment = "your environment name"
pinecone_store=PineconeVectoreStore(api_key,environment)
pinecone_store.from_documents(documents)

vectorstore/pinecone.py
neokd/NeoGPT