Spaces:
Sleeping
Sleeping
File size: 6,696 Bytes
5102822 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
import os
import streamlit as st
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
from transformers import pipeline
from htmlTemplates import css, bot_template, user_template
from dotenv import load_dotenv
load_dotenv()
# Creating custom template to guide LLM model
custom_template = """Given the following conversation and a follow-up question, rephrase the follow-up question to be a standalone question, in its original language.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
CUSTOM_QUESTION_PROMPT = PromptTemplate.from_template(custom_template)
# Extracting text from .txt files
def get_text_files_content(folder):
text = ""
for filename in os.listdir(folder):
if filename.endswith('.txt'):
with open(os.path.join(folder, filename), 'r', encoding='utf-8') as file:
text += file.read() + "\n"
return text
# Converting text to chunks
def get_chunks(raw_text):
text_splitter = CharacterTextSplitter(
separator="\n",
chunk_size=1000,
chunk_overlap=200,
length_function=len
)
chunks = text_splitter.split_text(raw_text)
return chunks
# Using Hugging Face embeddings model and FAISS to create vectorstore
def get_vectorstore(chunks):
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2",
model_kwargs={'device': 'cpu'}
)
vectorstore = FAISS.from_texts(texts=chunks, embedding=embeddings)
return vectorstore
# Generating conversation chain with improved out-of-scope handling
def get_conversationchain(vectorstore):
# Use a Hugging Face model for question-answering
model_name = "distilbert-base-uncased-distilled-squad" # Pretrained QA model
qa_pipeline = pipeline("question-answering", model=model_name, tokenizer=model_name)
def qa_function(question, context):
response = qa_pipeline(question=question, context=context)
return response['answer'], response['score']
memory = ConversationBufferMemory(
memory_key='chat_history',
return_messages=True,
output_key='answer'
)
def conversation_chain(inputs):
question = inputs['question']
documents = vectorstore.similarity_search(question, k=5)
if not documents:
answer = "Sorry, I couldn't find relevant information in the document. Please ask a question related to the document."
memory.save_context({"user_input": question}, {"answer": answer})
return {"chat_history": memory.chat_memory.messages, "answer": answer}
context = "\n".join([doc.page_content for doc in documents])
answer, score = qa_function(question, context)
if score < 0.5:
answer = "Sorry, I couldn't find relevant information in the document. Please ask a question related to the document."
memory.save_context({"user_input": question}, {"answer": answer})
return {"chat_history": memory.chat_memory.messages, "answer": answer}
return conversation_chain
# Generating response from user queries and displaying them accordingly
def handle_question(question):
response = st.session_state.conversation({'question': question})
st.session_state.chat_history = response["chat_history"]
for i, msg in enumerate(st.session_state.chat_history):
if i % 2 == 0:
st.write(user_template.replace("{{MSG}}", msg.content), unsafe_allow_html=True)
else:
st.write(bot_template.replace("{{MSG}}", msg.content), unsafe_allow_html=True)
def main():
st.set_page_config(page_title="Chat with Notes and AI", page_icon=":books:", layout="wide")
st.write(css, unsafe_allow_html=True)
if "conversation" not in st.session_state:
st.session_state.conversation = None
if "chat_history" not in st.session_state:
st.session_state.chat_history = None
st.header("CSS Edge - Intelligent Document Chatbot with Notes :books:")
# Subject selection dropdown
subjects = [
"A Trumped World", "Agri Tax in Punjab", "Assad's Fall in Syria", "Elusive National Unity", "Europe and Trump 2.0",
"Going Down with Democracy", "Indonesia's Pancasila Philosophy", "Pakistan in Choppy Waters",
"Pakistan's Semiconductor Ambitions", "Preserving Pakistan's Cultural Heritage", "Tackling Informal Economy",
"Technical Education in Pakistan", "The Case for Solidarity Levies", "The Decline of the Sole Superpower",
"The Power of Big Oil", "Trump 2.0 and Pakistan's Emerging Foreign Policy", "Trump and the World 2.0",
"Trump vs BRICS", "US-China Trade War", "War on Humanity", "Women's Suppression in Afghanistan"
]
data_folder = "data"
preview_folder = "Preview"
subject_folders = {subject: os.path.join(data_folder, subject.replace(' ', '_')) for subject in subjects}
preview_folders = {subject: os.path.join(preview_folder, subject.replace(' ', '_')) for subject in subjects}
selected_subject = st.sidebar.selectbox("Select a Subject:", subjects)
st.sidebar.info(f"You have selected: {selected_subject}")
# Load preview content
preview_folder_path = preview_folders[selected_subject]
if os.path.exists(preview_folder_path):
preview_text = get_text_files_content(preview_folder_path)
st.subheader("Preview of Notes")
st.text_area("Preview Content:", preview_text, height=300, disabled=True)
else:
st.error(f"No preview available for {selected_subject}.")
# Process data folder for question answering
subject_folder_path = subject_folders[selected_subject]
if os.path.exists(subject_folder_path):
raw_text = get_text_files_content(subject_folder_path)
if raw_text:
text_chunks = get_chunks(raw_text)
vectorstore = get_vectorstore(text_chunks)
st.session_state.conversation = get_conversationchain(vectorstore)
else:
st.error("Could not load the content for question answering.")
else:
st.error(f"No data available for {selected_subject}.")
# Chat interface
question = st.text_input("Ask a question about your selected subject:")
if question and st.session_state.conversation:
st.write(f"**Subject:** {selected_subject}")
handle_question(question)
elif question:
st.warning("Please process a document before asking a question.")
if __name__ == '__main__':
main()
|