Systems Ready!

.gitignore

@@ -0,0 +1,5 @@
+/db
+/DB
+/venv
+/.idea
+/__pycache__/

app.py

@@ -0,0 +1,304 @@
+import os
+import time
+from pathlib import Path
+from textwrap import dedent
+from types import SimpleNamespace
+
+import gradio as gr
+from charset_normalizer import detect
+from chromadb.config import Settings
+from epub2txt import epub2txt
+from langchain.chains import RetrievalQA
+from langchain.docstore.document import Document
+from langchain.document_loaders import (
+    CSVLoader,
+    Docx2txtLoader,
+    PDFMinerLoader,
+    TextLoader,
+)
+
+# from constants import CHROMA_SETTINGS, SOURCE_DIRECTORY, PERSIST_DIRECTORY
+from langchain.embeddings import HuggingFaceInstructEmbeddings
+from langchain.llms import HuggingFacePipeline
+from langchain.text_splitter import (
+    RecursiveCharacterTextSplitter,
+)
+import torch
+
+# FAISS instead of PineCone
+from langchain.vectorstores import  Chroma
+from loguru import logger
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+import argparse
+
+parser = argparse.ArgumentParser('LocalGPT falcon', add_help=False)
+parser.add_argument('--device_type', type=str, default="cuda", choices=["cpu", "mps", "cuda"], help='device type', )
+args = parser.parse_args()
+
+
+ROOT_DIRECTORY = Path(__file__).parent
+PERSIST_DIRECTORY = f"{ROOT_DIRECTORY}/db"
+
+# Define the Chroma settings
+CHROMA_SETTINGS = Settings(
+    chroma_db_impl="duckdb+parquet",
+    persist_directory=PERSIST_DIRECTORY,
+    anonymized_telemetry=False,
+)
+ns = SimpleNamespace(qa=None)
+
+# INSTRUCTORS_EMBEDDINGS_MODEL = "hkunlp/instructor-xl"
+INSTRUCTORS_EMBEDDINGS_MODEL = "hkunlp/instructor-large"
+# INSTRUCTORS_EMBEDDINGS_MODEL = "hkunlp/instructor-large"
+# INSTRUCTORS_EMBEDDINGS_MODEL = "hkunlp/instructor-base"
+
+def load_single_document(file_path: str or Path) -> Document:
+    """ingest.py"""
+    # Loads a single document from a file path
+    # encoding = detect(open(file_path, "rb").read()).get("encoding", "utf-8")
+    encoding = detect(Path(file_path).read_bytes()).get("encoding", "utf-8")
+    if file_path.endswith(".txt"):
+        if encoding is None:
+            logger.warning(
+                f" {file_path}'s encoding is None "
+                "Something is fishy, return empty str "
+            )
+            return Document(page_content="", metadata={"source": file_path})
+
+        try:
+            loader = TextLoader(file_path, encoding=encoding)
+        except Exception as exc:
+            logger.warning(f" {exc}, return dummy ")
+            return Document(page_content="", metadata={"source": file_path})
+
+    elif file_path.endswith(".pdf"):
+        loader = PDFMinerLoader(file_path)
+    elif file_path.endswith(".csv"):
+        loader = CSVLoader(file_path)
+    elif Path(file_path).suffix in [".docx"]:
+        try:
+            loader = Docx2txtLoader(file_path)
+        except Exception as exc:
+            logger.error(f" {file_path} errors: {exc}")
+            return Document(page_content="", metadata={"source": file_path})
+    elif Path(file_path).suffix in [".epub"]:  # for epub? epub2txt unstructured
+        try:
+            _ = epub2txt(file_path)
+        except Exception as exc:
+            logger.error(f" {file_path} errors: {exc}")
+            return Document(page_content="", metadata={"source": file_path})
+        return Document(page_content=_, metadata={"source": file_path})
+    else:
+        if encoding is None:
+            logger.warning(
+                f" {file_path}'s encoding is None "
+                "Likely binary files, return empty str "
+            )
+            return Document(page_content="", metadata={"source": file_path})
+        try:
+            loader = TextLoader(file_path)
+        except Exception as exc:
+            logger.error(f" {exc}, returnning empty string")
+            return Document(page_content="", metadata={"source": file_path})
+
+    return loader.load()[0]
+
+def greet(name):
+    """Test."""
+    logger.debug(f" name: [{name}] ")
+    return "Hello " + name + "!!"
+
+
+def upload_files(files):
+    """Upload files."""
+    try:
+        file_paths = [file.name for file in files]
+    except:
+        file_paths = [files]
+    logger.info(file_paths)
+
+    res = ingest(file_paths)
+    logger.info("Processed:\n{res}")
+    del res
+
+    ns.qa = load_qa()
+
+    return file_paths
+
+
+def ingest(
+    file_paths: list
+):
+    """Gen Chroma db.
+    torch.cuda.is_available()
+    file_paths =
+    []
+    """
+    logger.info("Doing ingest...")
+
+    documents = []
+    for file_path in file_paths:
+        documents.append(load_single_document(f"{file_path}"))
+
+    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
+    texts = text_splitter.split_documents(documents)
+
+    logger.info(f"Loaded {len(documents)} documents ")
+    logger.info(f"Split into {len(texts)} chunks of text")
+
+    # Create embeddings
+    logger.info(f"Load InstructEmbeddings model: {INSTRUCTORS_EMBEDDINGS_MODEL}")
+    embeddings = HuggingFaceInstructEmbeddings(
+        model_name=INSTRUCTORS_EMBEDDINGS_MODEL, model_kwargs={"device": args.device_type}
+    )
+
+    db = Chroma.from_documents(
+        texts,
+        embeddings,
+        persist_directory=PERSIST_DIRECTORY,
+        client_settings=CHROMA_SETTINGS,
+    )
+    db.persist()
+    db = None
+    logger.info("Done ingest")
+
+    return [
+        [Path(doc.metadata.get("source")).name, len(doc.page_content)]
+        for doc in documents
+    ]
+
+
+# https://huggingface.co/tiiuae/falcon-7b-instruct
+def gen_local_llm():
+    """Gen a local llm.
+    localgpt run_localgpt
+    """
+    model = "tiiuae/falcon-7b-instruct"
+
+    if args.device_type == "cuda":
+        tokenizer = AutoTokenizer.from_pretrained(model)
+    else: # cpu
+        tokenizer=AutoTokenizer.from_pretrained(model)
+        model=AutoModelForCausalLM.from_pretrained(model, trust_remote_code=True)
+    
+    pipe = pipeline(
+        "text-generation",
+        model=model,
+        tokenizer=tokenizer,
+        torch_dtype=torch.float32 if args.device_type =="cpu" else torch.bfloat16,
+        trust_remote_code=True,
+        device_map="cpu" if args.device_type =="cpu" else "auto",
+        max_length=2048,
+        temperature=0,
+        top_p=0.95,
+        top_k=10,
+        repetition_penalty=1.15,
+        num_return_sequences=1,
+        pad_token_id=tokenizer.eos_token_id
+    )
+
+    local_llm = HuggingFacePipeline(pipeline=pipe)
+
+    return local_llm
+
+
+def load_qa():
+    """Gen qa."""
+    logger.info("Doing qa")
+
+    embeddings = HuggingFaceInstructEmbeddings(
+        model_name=INSTRUCTORS_EMBEDDINGS_MODEL, model_kwargs={"device": args.device_type}
+    )
+    # xl 4.96G, large 3.5G,
+    db = Chroma(
+        persist_directory=PERSIST_DIRECTORY,
+        embedding_function=embeddings,
+        client_settings=CHROMA_SETTINGS,
+    )
+    retriever = db.as_retriever()
+
+    llm = gen_local_llm()  # "tiiuae/falcon-7b-instruct"
+
+    qa = RetrievalQA.from_chain_type(
+        llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True
+    )
+
+    logger.info("Done qa")
+
+    return qa
+
+
+def main1():
+    """Lump codes"""
+    with gr.Blocks() as demo:
+        iface = gr.Interface(fn=greet, inputs="text", outputs="text")
+        iface.launch()
+
+    demo.launch()
+
+
+def main():
+    """Do blocks."""
+    logger.info(f"ROOT_DIRECTORY: {ROOT_DIRECTORY}")
+
+    with gr.Blocks(theme=gr.themes.Soft()) as demo:
+        with gr.Accordion("Info", open=False):
+            _ = """
+                Talk to your docs (.pdf, .docx, .csv, .txt .md). It
+                takes quite a while to ingest docs (10-30 min. depending
+                on net, RAM, CPU etc.).
+                """
+            gr.Markdown(dedent(_))
+        title = """
+            <div style="text-align: center;">
+                <h1>LocalGPT with Falcon</h1>
+                <p style="text-align: center;">Upload your docs (.pdf, .docx, .csv, .txt .md) by clicking the "Load docs to LangChain" and wait until the upload is complete, <br />
+                when everything is ready, you can start asking questions about the docs <br />
+            </div>
+        """
+        gr.HTML(title)
+        with gr.Tab("Upload files"):
+            # Upload files and generate embeddings database
+            file_output = gr.File()
+            upload_button = gr.UploadButton(
+                "Load docs to LangChain",
+                file_count="multiple",
+            )
+            upload_button.upload(upload_files, upload_button, file_output)
+
+            chatbot = gr.Chatbot()
+            msg = gr.Textbox(label="Query")
+            clear = gr.Button("Clear")
+
+            def respond(message, chat_history):
+                if ns.qa is None:  # no files processed yet
+                    bot_message = "Provide some file(s) for processsing first."
+                    chat_history.append((message, bot_message))
+                    return "", chat_history
+                try:
+                    res = ns.qa(message)
+                    answer, docs = res["result"], res["source_documents"]
+                    bot_message = f"{answer}"
+                except Exception as exc:
+                    logger.error(exc)
+                    bot_message = f"bummer! {exc}"
+
+                chat_history.append((message, bot_message))
+
+                return "", chat_history
+
+            msg.submit(respond, [msg, chatbot], [msg, chatbot])
+            clear.click(lambda: None, None, chatbot, queue=False)
+
+    try:
+        from google import colab
+
+        share = True  # start share when in colab
+    except Exception:
+        share = False
+
+    demo.launch(share=share)
+
+
+if __name__ == "__main__":
+    main()

constants.py

@@ -0,0 +1,18 @@
+import os
+# from dotenv import load_dotenv
+from chromadb.config import Settings
+
+# load_dotenv()
+ROOT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
+
+# Define the folder for storing database
+SOURCE_DIRECTORY = f"{ROOT_DIRECTORY}/SOURCE_DOCUMENTS"
+
+PERSIST_DIRECTORY = f"{ROOT_DIRECTORY}/DB"
+
+# Define the Chroma settings
+CHROMA_SETTINGS = Settings(
+        chroma_db_impl='duckdb+parquet',
+        persist_directory=PERSIST_DIRECTORY,
+        anonymized_telemetry=False
+)

ingest.py

@@ -0,0 +1,78 @@
+import os
+import click
+from typing import List
+from utils import xlxs_to_csv
+from langchain.document_loaders import TextLoader, PDFMinerLoader, CSVLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.vectorstores import Chroma
+from langchain.docstore.document import Document
+from constants import CHROMA_SETTINGS, SOURCE_DIRECTORY, PERSIST_DIRECTORY
+from langchain.embeddings import HuggingFaceInstructEmbeddings
+
+
+def load_single_document(file_path: str) -> Document:
+    # Loads a single document from a file path
+    if file_path.endswith(".txt"):
+        loader = TextLoader(file_path, encoding="utf8")
+    elif file_path.endswith(".pdf"):
+        loader = PDFMinerLoader(file_path)
+    elif file_path.endswith(".csv"):
+        loader = CSVLoader(file_path)
+    return loader.load()[0]
+
+
+def load_documents(source_dir: str) -> List[Document]:
+    # Loads all documents from source documents directory
+    all_files = os.listdir(source_dir)
+    docs = []
+    for file_path in all_files:
+        if file_path[-4:] == 'xlsx':
+            for doc in xlxs_to_csv(f"{source_dir}/{file_path}"):
+                docs.append(load_single_document(doc))
+        elif file_path[-4:] in ['.txt', '.pdf', '.csv']:
+            docs.append(load_single_document(f"{source_dir}/{file_path}"))
+    return docs
+    # return [load_single_document(f"{source_dir}/{file_path}") for file_path in all_files if
+    #         file_path[-4:] in ['.txt', '.pdf', '.csv']]
+
+
+# @click.command()
+# @click.option('--device_type', default='gpu', help='device to run on, select gpu or cpu')
+# def main(device_type, ):
+#     # load the instructorEmbeddings
+#     if device_type in ['cpu', 'CPU']:
+#         device='cpu'
+#     else:
+#         device='cuda'
+
+
+@click.command()
+@click.option('--device_type', default='cuda', help='device to run on, select gpu, cpu or mps')
+def main(device_type, ):
+    # load the instructorEmbeddings
+    if device_type in ['cpu', 'CPU']:
+        device='cpu'
+    elif device_type in ['mps', 'MPS']:
+        device='mps'
+    else:
+        device='cuda'
+
+    #  Load documents and split in chunks
+    print(f"Loading documents from {SOURCE_DIRECTORY}")
+    documents = load_documents(SOURCE_DIRECTORY)
+    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
+    texts = text_splitter.split_documents(documents)
+    print(f"Loaded {len(documents)} documents from {SOURCE_DIRECTORY}")
+    print(f"Split into {len(texts)} chunks of text")
+
+    # Create embeddings
+    embeddings = HuggingFaceInstructEmbeddings(model_name="hkunlp/instructor-base",
+                                               model_kwargs={"device": device})
+
+    db = Chroma.from_documents(texts, embeddings, persist_directory=PERSIST_DIRECTORY, client_settings=CHROMA_SETTINGS)
+    db.persist()
+    db = None
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,23 @@
+langchain==0.0.166
+chromadb==0.3.22
+llama-cpp-python==0.1.48
+urllib3==1.26.6
+pdfminer.six==20221105
+InstructorEmbedding
+sentence-transformers
+faiss-cpu
+huggingface_hub
+transformers
+protobuf==3.20.0
+accelerate
+bitsandbytes
+click
+openpyxl
+einops
+xformers
+gradio
+charset-normalizer
+PyPDF2
+epub2txt
+docx2txt
+loguru

run_localGPT.py

@@ -0,0 +1,108 @@
+from langchain.chains import RetrievalQA
+# from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
+from langchain.vectorstores import Chroma
+from langchain.embeddings import HuggingFaceInstructEmbeddings
+from langchain.llms import HuggingFacePipeline
+from constants import CHROMA_SETTINGS, PERSIST_DIRECTORY
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+import click
+import torch
+from constants import CHROMA_SETTINGS
+
+
+def load_model(device):
+    """
+    Select a model on huggingface.
+    If you are running this for the first time, it will download a model for you.
+    subsequent runs will use the model from the disk.
+    """
+    model = "tiiuae/falcon-7b-instruct"
+
+    if device == "cuda":
+        tokenizer = AutoTokenizer.from_pretrained(model)
+    else: # cpu
+        tokenizer=AutoTokenizer.from_pretrained(model)
+        model=AutoModelForCausalLM.from_pretrained(model, trust_remote_code=True)
+    
+    pipe = pipeline(
+        "text-generation",
+        model=model,
+        tokenizer=tokenizer,
+        torch_dtype=torch.float32 if device =="cpu" else torch.bfloat16,
+        trust_remote_code=True,
+        device_map=device if device =="cpu" else "auto",
+        max_length=2048,
+        temperature=0,
+        top_p=0.95,
+        top_k=10,
+        repetition_penalty=1.15,
+        num_return_sequences=1,
+        pad_token_id=tokenizer.eos_token_id
+    )
+
+    local_llm = HuggingFacePipeline(pipeline=pipe)
+
+    return local_llm
+
+
+# @click.command()
+# @click.option('--device_type', default='gpu', help='device to run on, select gpu or cpu')
+# def main(device_type, ):
+#     # load the instructorEmbeddings
+#     if device_type in ['cpu', 'CPU']:
+#         device='cpu'
+#     else:
+#         device='cuda'
+ 
+    
+ ## for M1/M2 users:
+
+@click.command()
+@click.option('--device_type', default='cuda', help='device to run on, select gpu, cpu or mps')
+def main(device_type, ):
+    # load the instructorEmbeddings
+    if device_type in ['cpu', 'CPU']:
+        device='cpu'
+    elif device_type in ['mps', 'MPS']:
+        device='mps'
+    else:
+        device='cuda'
+
+    print(f"Running on: {device}")
+
+    embeddings = HuggingFaceInstructEmbeddings(model_name="hkunlp/instructor-base",
+                                               model_kwargs={"device": device})
+    # load the vectorstore
+    db = Chroma(persist_directory=PERSIST_DIRECTORY, embedding_function=embeddings, client_settings=CHROMA_SETTINGS)
+    retriever = db.as_retriever()
+    # Prepare the LLM
+    # callbacks = [StreamingStdOutCallbackHandler()]
+    # load the LLM for generating Natural Language responses. 
+    llm = load_model(device)
+    qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True)
+    # Interactive questions and answers
+    while True:
+        query = input("\nEnter a query: ")
+        if query == "exit":
+            break
+
+        # Get the answer from the chain
+        res = qa(query)
+        answer, docs = res['result'], res['source_documents']
+
+        # Print the result
+        print("\n\n> Question:")
+        print(query)
+        print("\n> Answer:")
+        print(answer)
+
+        # Print the relevant sources used for the answer
+        print("----------------------------------SOURCE DOCUMENTS---------------------------")
+        for document in docs:
+            print("\n> " + document.metadata["source"] + ":")
+            print(document.page_content)
+        print("----------------------------------SOURCE DOCUMENTS---------------------------")
+
+
+if __name__ == "__main__":
+    main()

utils.py

@@ -0,0 +1,37 @@
+import openpyxl
+import csv
+import tempfile
+
+
+def xlxs_to_csv(file_path: str, sheet_name: str = None) -> list[str]:
+    """
+    Convert a workbook into a list of csv files
+    :param file_path: the path to the workbook
+    :param sheet_name: the name of the sheet to convert
+    :return: a list of temporary file names
+    """
+    # Load the workbook and select the active worksheet
+    wb = openpyxl.load_workbook(file_path)
+    # ws = wb.active
+    #
+    # # Create a new temporary file and write the contents of the worksheet to it
+    # with tempfile.NamedTemporaryFile(mode='w+', newline='', delete=False) as f:
+    #     c = csv.writer(f)
+    #     for r in ws.rows:
+    #         c.writerow([cell.value for cell in r])
+    # return f.name
+    # load all sheets if sheet_name is None
+    wb = wb if sheet_name is None else [wb[sheet_name]]
+    temp_file_name = []
+    # Iterate over the worksheets in the workbook
+    for ws in wb:
+        # Create a new temporary file and write the contents of the worksheet to it
+        with tempfile.NamedTemporaryFile(mode='w+', newline='', suffix='.csv', delete=False) as f:
+            c = csv.writer(f)
+            for r in ws.rows:
+                c.writerow([cell.value for cell in r])
+            temp_file_name.append(f.name)
+    # print(f'all Sheets are saved to temporary file {temp_file_name}')
+    return temp_file_name
+
+