# Import required libraries
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import (
UnstructuredWordDocumentLoader,
PyMuPDFLoader,
UnstructuredFileLoader,
)
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.embeddings.huggingface import HuggingFaceEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.vectorstores import Pinecone, Chroma
from langchain.chains import ConversationalRetrievalChain, LLMChain
from langchain.chains.question_answering import load_qa_chain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
import os
import langchain
import pinecone
import streamlit as st
import shutil
import json
OPENAI_API_KEY = ''
PINECONE_API_KEY = ''
PINECONE_API_ENV = ''
gpt3p5 = 'gpt-3.5-turbo-1106'
gpt4 = 'gpt-4-1106-preview'
local_model_tuples = [
(0, 'mistral_7b', "TheBloke/OpenHermes-2-Mistral-7B-GGUF", "openhermes-2-mistral-7b.Q8_0.gguf", "mistral", "https://huggingface.co/TheBloke/OpenHermes-2-Mistral-7B-GGUF"),
(1, 'mistral_7b_inst_small', "TheBloke/Mistral-7B-Instruct-v0.1-GGUF", "mistral-7b-instruct-v0.1.Q2_K.gguf", "mistral", "https://huggingface.co/TheBloke/Mistral-7B-Instruct-v0.1-GGUF"),
(2, 'mistral_7b_inst_med', "TheBloke/Mistral-7B-Instruct-v0.1-GGUF", "mistral-7b-instruct-v0.1.Q8_0.gguf", "mistral", "https://huggingface.co/TheBloke/Mistral-7B-Instruct-v0.1-GGUF"),
(3, 'llama_13b_small', "TheBloke/Llama-2-13B-chat-GGUF", "llama-2-13b-chat.Q4_K_M.gguf", "llama", "https://huggingface.co/TheBloke/Llama-2-13B-chat-GGUF"),
(4, 'llama_13b_med', "TheBloke/Llama-2-13B-chat-GGUF", "llama-2-13b-chat.Q8_0.gguf", "llama", "https://huggingface.co/TheBloke/Llama-2-13B-chat-GGUF"),
(5, 'mixtral', "TheBloke/Mixtral-8x7B-v0.1-GGUF", "mixtral-8x7b-v0.1.Q8_0.gguf", "mixtral", "https://huggingface.co/TheBloke/Mixtral-8x7B-v0.1-GGUF"),
(6, 'mixtral_inst', "TheBloke/Mixtral-8x7B-Instruct-v0.1-GGUF", "mixtral-8x7b-instruct-v0.1.Q2_K.gguf", "mixtral", "https://huggingface.co/TheBloke/Mixtral-8x7B-Instruct-v0.1-GGUF"),
]
local_model_names = [t[1] for t in local_model_tuples]
langchain.verbose = False
@st.cache_data()
def init():
pinecone_index_name = ''
chroma_collection_name = ''
persist_directory = ''
docsearch_ready = False
directory_name = 'tmp_docs'
return pinecone_index_name, chroma_collection_name, persist_directory, docsearch_ready, directory_name
@st.cache_data()
def save_file(files):
# Remove existing files in the directory
if os.path.exists(directory_name):
for filename in os.listdir(directory_name):
file_path = os.path.join(directory_name, filename)
try:
if os.path.isfile(file_path):
os.remove(file_path)
except Exception as e:
print(f"Error: {e}")
# Save the new file with original filename
if files is not None:
for file in files:
file_name = file.name
file_path = os.path.join(directory_name, file_name)
with open(file_path, 'wb') as f:
shutil.copyfileobj(file, f)
def load_files():
all_texts = []
n_files = 0
n_char = 0
n_texts = 0
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=400, chunk_overlap=50
)
for filename in os.listdir(directory_name):
file = os.path.join(directory_name, filename)
if os.path.isfile(file):
if file.endswith(".docx"):
loader = UnstructuredWordDocumentLoader(file)
elif file.endswith(".pdf"):
loader = PyMuPDFLoader(file)
else: # assume a pure text format and attempt to load it
loader = UnstructuredFileLoader(file)
data = loader.load()
texts = text_splitter.split_documents(data)
n_files += 1
n_char += len(data[0].page_content)
n_texts += len(texts)
all_texts.extend(texts)
st.write(
f"Loaded {n_files} file(s) with {n_char} characters, and split into {n_texts} split-documents."
)
return all_texts, n_texts
@st.cache_resource()
def ingest(_all_texts, use_pinecone, _embeddings, pinecone_index_name, chroma_collection_name, persist_directory):
if use_pinecone:
docsearch = Pinecone.from_texts(
[t.page_content for t in _all_texts], _embeddings, index_name=pinecone_index_name) # add namespace=pinecone_namespace if provided
else:
docsearch = Chroma.from_documents(
_all_texts, _embeddings, collection_name=chroma_collection_name, persist_directory=persist_directory)
return docsearch
def setup_retriever(docsearch, k):
retriever = docsearch.as_retriever(
search_type="similarity", search_kwargs={"k": k}, include_metadata=True)
return retriever
def setup_docsearch(use_pinecone, pinecone_index_name, embeddings, chroma_collection_name, persist_directory):
docsearch = []
n_texts = 0
if use_pinecone:
# Load the pre-created Pinecone index.
# The index which has already be stored in pinecone.io as long-term memory
if pinecone_index_name in pinecone.list_indexes():
docsearch = Pinecone.from_existing_index(
pinecone_index_name, embeddings) # add namespace=pinecone_namespace if provided
index_client = pinecone.Index(pinecone_index_name)
# Get the index information
index_info = index_client.describe_index_stats()
# namespace_name = ''
# if index_info is not None:
# print(index_info['namespaces'][namespace_name]['vector_count'])
# else:
# print("Index information is not available.")
# n_texts = index_info['namespaces'][namespace_name]['vector_count']
n_texts = index_info['total_vector_count']
else:
raise ValueError('''Cannot find the specified Pinecone index.
Create one in pinecone.io or using, e.g.,
pinecone.create_index(
name=index_name, dimension=1536, metric="cosine", shards=1)''')
else:
docsearch = Chroma(persist_directory=persist_directory, embedding_function=embeddings,
collection_name=chroma_collection_name)
n_texts = docsearch._collection.count()
return docsearch, n_texts
def get_response(query, chat_history, CRqa):
result = CRqa({"question": query, "chat_history": chat_history})
return result['answer'], result['source_documents']
@st.cache_resource()
def use_local_llm(r_llm, local_llm_path, temperature):
from langchain.llms import LlamaCpp
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from huggingface_hub import hf_hub_download
callback_manager = CallbackManager([StreamingStdOutCallbackHandler()])
entry = local_model_names.index(r_llm)
model_id, local_model_name, model_name, model_file, model_type, model_link = local_model_tuples[entry]
model_path = os.path.join( local_llm_path, model_name, model_file )
model_path = os.path.normpath( model_path )
model_dir = os.path.join( local_llm_path, model_name )
model_dir = os.path.normpath( model_dir )
if not os.path.exists(model_path):
print("model not existing at ", model_path, "\n")
model_path = hf_hub_download(repo_id=model_name, filename=model_file, repo_type="model",
#cache_dir=local_llm_path,
#local_dir=local_llm_path,
local_dir=model_dir,
local_dir_use_symlinks=False)
print("\n model downloaded at path=",model_path)
else:
print("model existing at ", model_path)
llm = LlamaCpp(
model_path=model_path,
temperature=temperature,
# n_batch=300,
n_ctx=4000,
max_tokens=2000,
# n_gpu_layers=10,
# n_threads=12,
# top_p=1,
# repeat_penalty=1.15,
# verbose=False,
# callback_manager=callback_manager,
# streaming=True,
# chat_format="llama-2",
# verbose=True, # Verbose is required to pass to the callback manager
)
return llm
def setup_prompt(r_llm, usage):
B_INST, E_INST = "[INST]", "[/INST]"
B_SYS_LLAMA, E_SYS_LLAMA = "<>\n", "\n<>\n\n"
B_SYS_MIS, E_SYS_MIS = " ", " "
B_SYS_MIXTRAL, E_SYS_MIXTRAL = "[INST]", "[/INST][INST]"
system_prompt_rag = """Answer the question in your own words as truthfully as possible from the context given to you.
Supply sufficient information, evidence, reasoning, source from the context, etc., to justify your answer with details and logic.
Think step by step and do not jump to conclusion during your reasoning at the beginning.
Sometimes user's question may appear to be directly related to the context but may still be indirectly related,
so try your best to understand the question based on the context and chat history.
If questions are asked where there is no relevant context available,
respond using out-of-context knowledge with
"This question does not seem to be relevant to the documents. I am trying to explore knowledge outside the context." """
system_prompt_chat = """Answer the question in your own words.
Supply sufficient information, evidence, reasoning, source from the context, etc., to justify your answer with details and logic.
Think step by step and do not jump to conclusion during your reasoning at the beginning.
"""
system_prompt_task = """You will be given a task, and you are an expert in that task.
Perform the task for the given context, and output the result. Do not include extra descriptions. Just output the desired result defined by the task.
Example: You are a professional translator and are given a translation task. Then you translate the text in the context and output only the translated text.
Example: You are a professional proofreader and are given a proofreading task. Then you proofread the text in the context and output only the translated text.
"""
if usage == 'RAG':
system_prompt = system_prompt_rag
instruction = """
Context: {context}
Chat history: {chat_history}
User: {question}
Bot: answer """
elif usage == 'Chat':
system_prompt = system_prompt_chat
instruction = """
Chat history: {chat_history}
User: {question}
Bot: answer """
elif usage == 'Task':
system_prompt = system_prompt_task
instruction = """
Context: {context}
User: {question}
Bot: answer """
if r_llm == gpt3p5 or r_llm == gpt4:
template = system_prompt + instruction
else:
entry = local_model_names.index(r_llm)
if local_model_tuples[entry][4] == 'llama':
template = B_INST + B_SYS_LLAMA + system_prompt + E_SYS_LLAMA + instruction + E_INST
elif local_model_tuples[entry][4] == 'mistral':
template = B_SYS_MIS + B_INST + system_prompt + E_INST + E_SYS_MIS + B_INST + instruction + E_INST
elif local_model_tuples[entry][4] == 'mixtral':
template = B_SYS_MIXTRAL + system_prompt + E_SYS_MIXTRAL + B_INST + instruction + E_INST
else:
# Handle other models or raise an exception
pass
if usage == 'RAG':
prompt = PromptTemplate(
input_variables=["context", "chat_history", "question"], template=template
)
elif usage == 'Chat':
prompt = PromptTemplate(
input_variables=["chat_history", "question"], template=template
)
elif usage == 'Task':
prompt = PromptTemplate(
input_variables=["context", "question"], template=template
)
return prompt
def setup_em_llm(OPENAI_API_KEY, temperature, r_llm, local_llm_path, usage):
if (r_llm == gpt3p5 or r_llm == gpt4) and OPENAI_API_KEY:
# Set up OpenAI embeddings
embeddings = OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY)
# Use Open AI LLM with gpt-3.5-turbo or gpt-4.
# Set the temperature to be 0 if you do not want it to make up things
llm = ChatOpenAI(temperature=temperature, model_name=r_llm, streaming=True,
openai_api_key=OPENAI_API_KEY)
else:
if usage == 'RAG':
em_model_name='sentence-transformers/all-mpnet-base-v2'
embeddings = HuggingFaceEmbeddings(model_name=em_model_name)
else:
embeddings = []
llm = use_local_llm(r_llm, local_llm_path, temperature)
return embeddings, llm
def load_chat_history(CHAT_HISTORY_FILENAME):
try:
with open(CHAT_HISTORY_FILENAME, 'r') as f:
chat_history = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
chat_history = []
return chat_history
def save_chat_history(chat_history, CHAT_HISTORY_FILENAME):
with open(CHAT_HISTORY_FILENAME, 'w') as f:
json.dump(chat_history, f)
pinecone_index_name, chroma_collection_name, persist_directory, docsearch_ready, directory_name = init()
def main(pinecone_index_name, chroma_collection_name, persist_directory, docsearch_ready, directory_name):
docsearch_ready = False
chat_history = []
latest_chats = []
reply = ''
source = ''
LLMs = [gpt3p5, gpt4] + local_model_names
usage = 'RAG'
local_llm_path = './models/'
user_llm_path = ''
# Get user input of whether to use Pinecone or not
col1, col2, col3 = st.columns([1, 1, 1])
# create the radio buttons and text input fields
with col1:
usage = st.radio('Usage: RAG for ingested files, chat (no files), or task (for all ingested texts)', ('RAG', 'Chat', 'Task'))
temperature = st.slider('Temperature', 0.0, 1.0, 0.1)
if usage == 'RAG':
r_pinecone = st.radio('Vector store:', ('Pinecone (online)', 'Chroma (local)'))
k_sources = st.slider('# source(s) to print out', 0, 20, 2)
r_ingest = st.radio('Ingest file(s)?', ('Yes', 'No'))
if r_pinecone == 'Pinecone (online)':
use_pinecone = True
else:
use_pinecone = False
if usage == 'Task':
r_ingest = 'Yes'
with col2:
r_llm = st.radio(label='LLM:', options=LLMs)
if r_llm == gpt3p5 or r_llm == gpt4:
use_openai = True
else:
use_openai = False
if use_openai == True:
OPENAI_API_KEY = st.text_input(
"OpenAI API key:", type="password")
else:
OPENAI_API_KEY = ''
if usage == 'RAG' and use_pinecone == True:
st.write('Local GPT model (and local embedding model) is selected. Online vector store is selected.')
elif usage == 'RAG' and use_pinecone == False:
st.write('Local GPT model (and local embedding model) and local vector store are selected. All info remains local.')
else:
st.write('Local GPT model is selected. All info remains local.')
with col3:
if usage == 'RAG':
if use_pinecone == True:
PINECONE_API_KEY = st.text_input(
"Pinecone API key:", type="password")
PINECONE_API_ENV = st.text_input(
"Pinecone API env:", type="password")
pinecone_index_name = st.text_input('Pinecone index:')
pinecone.init(api_key=PINECONE_API_KEY,
environment=PINECONE_API_ENV)
else:
chroma_collection_name = st.text_input(
'''Chroma collection name of 3-63 characters:''')
persist_directory = "./vectorstore"
else:
hist_fn = st.text_input('Chat history filename')
if use_openai == False:
user_llm_path = st.text_input(
"Path for local model (TO BE DOWNLOADED IF NOT EXISTING), type 'default' to use default path:",
placeholder="default")
if 'default' in user_llm_path:
user_llm_path = local_llm_path
if ( (pinecone_index_name or chroma_collection_name or usage == 'Task' or usage == 'Chat')
and ( (use_openai and OPENAI_API_KEY) or (not use_openai and user_llm_path) ) ):
embeddings, llm = setup_em_llm(OPENAI_API_KEY, temperature, r_llm, user_llm_path, usage)
#if ( pinecone_index_name or chroma_collection_name ) and embeddings and llm:
session_name = pinecone_index_name + chroma_collection_name + hist_fn
if usage != 'Chat':
if r_ingest.lower() == 'yes':
files = st.file_uploader(
'Upload Files', accept_multiple_files=True)
if files:
save_file(files)
all_texts, n_texts = load_files()
if usage == 'RAG':
docsearch = ingest(all_texts, use_pinecone, embeddings, pinecone_index_name,
chroma_collection_name, persist_directory)
docsearch_ready = True
else:
st.write(
'No data is to be ingested. Make sure the Pinecone index or Chroma collection name you provided contains data.')
docsearch, n_texts = setup_docsearch(use_pinecone, pinecone_index_name,
embeddings, chroma_collection_name, persist_directory)
docsearch_ready = True
else:
docsearch_ready = True
if docsearch_ready:
prompt = setup_prompt(r_llm, usage)
#if usage == 'RAG' or usage == 'Chat':
memory = ConversationBufferMemory(memory_key='chat_history', return_messages=True, output_key='answer')
if usage == 'RAG':
# number of sources (split-documents when ingesting files); default is 4
k = min([20, n_texts])
retriever = setup_retriever(docsearch, k)
CRqa = ConversationalRetrievalChain.from_llm(
llm,
chain_type="stuff",
retriever=retriever,
memory=memory,
return_source_documents=True,
combine_docs_chain_kwargs={'prompt': prompt},
)
elif usage == 'Chat':
CRqa = LLMChain(
llm=llm,
prompt=prompt,
)
elif usage == 'Task':
CRqa = load_qa_chain(
llm=llm,
chain_type="stuff",
prompt=prompt
)
st.title(':blue[Chatbot]')
# Get user input
query = st.text_area('Enter your question:', height=10,
placeholder='''Summarize the context.
\nAfter typing your question, click on SUBMIT to send it to the bot.''')
submitted = st.button('SUBMIT')
CHAT_HISTORY_FILENAME = f"chat_history/{session_name}_chat_hist.json"
chat_history = load_chat_history(CHAT_HISTORY_FILENAME)
st.markdown('', unsafe_allow_html=True)
if query and submitted:
# Generate a reply based on the user input and chat history
chat_history = [(user, bot)
for user, bot in chat_history]
if usage == 'RAG':
reply, source = get_response(query, chat_history, CRqa)
elif usage == 'Chat':
reply = CRqa({"question": query, "chat_history": chat_history, "return_only_outputs": True})
reply = reply['text']
elif usage == 'Task':
reply = []
for a_text in all_texts:
output_text = CRqa.run(input_documents=[a_text], question=query )
reply.append ( output_text )
# Update the chat history with the user input and system response
chat_history.append(('User', query))
chat_history.append(('Bot', reply))
save_chat_history(chat_history, CHAT_HISTORY_FILENAME)
c = chat_history[-4:]
if len(chat_history) >= 4:
latest_chats = [c[2],c[3],c[0],c[1]]
else:
latest_chats = c
if latest_chats:
chat_history_str1 = '
'.join([f'{x[0]}: {x[1]}' for x in latest_chats])
st.markdown(f'{chat_history_str1}
', unsafe_allow_html=True)
if usage == 'RAG' and reply and source:
# Display sources
for i, source_i in enumerate(source):
if i < k_sources:
if len(source_i.page_content) > 400:
page_content = source_i.page_content[:400]
else:
page_content = source_i.page_content
if source_i.metadata:
metadata_source = source_i.metadata['source']
st.markdown(f"Source {i+1}: {metadata_source}
{page_content}", unsafe_allow_html=True)
else:
st.markdown(f"Source {i+1}:
{page_content}", unsafe_allow_html=True)
all_chats = chat_history
all_chat_history_str = '\n'.join(
[f'{x[0]}: {x[1]}' for x in all_chats])
st.title(':blue[All chat records]')
st.text_area('Chat records in ascending order:', value=all_chat_history_str, height=250, label_visibility='collapsed')
if __name__ == '__main__':
main(pinecone_index_name, chroma_collection_name, persist_directory,
docsearch_ready, directory_name)