|
import logging |
|
from typing import List |
|
from pydantic import NoneStr |
|
import os |
|
from langchain.chains.question_answering import load_qa_chain |
|
from langchain.document_loaders import UnstructuredFileLoader |
|
from langchain.embeddings.openai import OpenAIEmbeddings |
|
from langchain.llms import OpenAI |
|
from langchain.text_splitter import CharacterTextSplitter |
|
from langchain.vectorstores import FAISS |
|
import gradio as gr |
|
import openai |
|
from langchain import PromptTemplate, OpenAI, LLMChain |
|
import validators |
|
import requests |
|
import mimetypes |
|
import tempfile |
|
import pandas as pd |
|
import re |
|
|
|
class DocumentQA: |
|
def get_empty_state(self): |
|
|
|
""" Create empty Knowledge base""" |
|
|
|
return {"knowledge_base": None} |
|
|
|
def get_content_from_url(self,url:str)->List: |
|
""" |
|
Uploads a file from a given URL and returns the loaded document. |
|
Args: |
|
url (str): The URL of the file to be uploaded. |
|
Returns: |
|
Document: The loaded document. |
|
Raises: |
|
ValueError: If the URL is not valid or the file cannot be fetched. |
|
""" |
|
|
|
if validators.url(url): |
|
headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36',} |
|
r = requests.get(url,headers=headers) |
|
if r.status_code != 200: |
|
raise ValueError( |
|
"Check the url of your file; returned status code %s" % r.status_code |
|
) |
|
|
|
content_type = r.headers.get("content-type") |
|
file_extension = mimetypes.guess_extension(content_type) |
|
temp_file = tempfile.NamedTemporaryFile(suffix=file_extension, delete=False) |
|
temp_file.write(r.content) |
|
file_path = temp_file.name |
|
loader = UnstructuredFileLoader(file_path, strategy="fast") |
|
docs = loader.load() |
|
return docs |
|
else: |
|
raise ValueError("Please enter a valid URL") |
|
|
|
def create_knowledge_base(self,docs): |
|
|
|
"""Create a knowledge base from the given documents. |
|
Args: |
|
docs (List[str]): List of documents. |
|
Returns: |
|
FAISS: Knowledge base built from the documents. |
|
""" |
|
|
|
|
|
|
|
|
|
text_splitter = CharacterTextSplitter( |
|
separator="\n", chunk_size=500, chunk_overlap=100, length_function=len |
|
) |
|
|
|
|
|
chunks = text_splitter.split_documents(docs) |
|
|
|
|
|
embeddings = OpenAIEmbeddings() |
|
|
|
|
|
knowledge_base = FAISS.from_documents(chunks, embeddings) |
|
|
|
|
|
return knowledge_base |
|
|
|
def get_chemicals_for_url(self,urls:str,state,input_qus)->str: |
|
""" |
|
Retrieves chemicals from the provided URLs. |
|
|
|
Args: |
|
urls (str): Comma-separated URLs of the files to be processed. |
|
|
|
Returns: |
|
str: The extracted chemical names. |
|
|
|
Raises: |
|
ValueError: If an error occurs during the process. |
|
""" |
|
webpage_text =[] |
|
for url in urls.split(','): |
|
|
|
webpage_text.extend(self.get_content_from_url(url)) |
|
|
|
knowledge_base = self.create_knowledge_base(webpage_text) |
|
|
|
state = {"knowledge_base": knowledge_base} |
|
|
|
chemicals = self.get_chemicals_for_file(state,input_qus) |
|
|
|
return chemicals |
|
|
|
|
|
|
|
def file_path_show(self,file_paths): |
|
file_paths = [single_file_path.name for single_file_path in file_paths] |
|
return file_paths |
|
|
|
def get_chemicals_for_file(self,state,question): |
|
knowledge_base = state["knowledge_base"] |
|
|
|
|
|
|
|
|
|
docs = knowledge_base.similarity_search(question) |
|
|
|
|
|
template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer. |
|
Identify the Chemical Capabilities Only. |
|
{context} |
|
Question :{question}. |
|
The result should be in bullet points format. |
|
""" |
|
|
|
prompt = PromptTemplate(template=template,input_variables=["context","question"]) |
|
|
|
llm = OpenAI(temperature=0.4) |
|
llm_chain = LLMChain(prompt=prompt, llm=llm) |
|
|
|
|
|
chain = load_qa_chain(llm, chain_type="stuff",prompt=prompt) |
|
|
|
|
|
response = chain.run(input_documents=docs, question=question) |
|
|
|
|
|
return response |
|
|
|
def identify_chemicals_in_files(self,file_paths,state,question): |
|
"""Upload a file and create a knowledge base from its contents. |
|
Args: |
|
file_paths : The files to uploaded. |
|
Returns: |
|
tuple: A tuple containing the file name and the knowledge base. |
|
""" |
|
|
|
|
|
file_paths = [single_file_path.name for single_file_path in file_paths] |
|
docs =[] |
|
for file_obj in file_paths: |
|
|
|
loader = UnstructuredFileLoader(file_obj, strategy="fast") |
|
|
|
|
|
docs.extend(loader.load()) |
|
|
|
|
|
knowledge_base = self.create_knowledge_base(docs) |
|
state = {"knowledge_base": knowledge_base} |
|
pdf_name = os.path.basename(file_obj) |
|
|
|
final_ans = self.get_chemicals_for_file(state,question) |
|
|
|
|
|
return final_ans |
|
|
|
def get_final_result(self,urls,file_paths,state,input_qus): |
|
|
|
if urls: |
|
if file_paths: |
|
urls_chemicals = self.get_chemicals_for_url(urls,state,input_qus) |
|
file_chemicals = self.identify_chemicals_in_files(file_paths,state,input_qus) |
|
chemicals = urls_chemicals + file_chemicals |
|
|
|
return chemicals |
|
else: |
|
urls_chemicals = self.get_chemicals_for_url(urls,state,input_qus) |
|
return urls_chemicals |
|
elif file_paths: |
|
file_chemicals = self.identify_chemicals_in_files(file_paths,state,input_qus) |
|
return file_chemicals |
|
else: |
|
return "No Files Uploaded" |
|
|
|
document_qa = DocumentQA() |
|
class ChemicalIdentifier: |
|
|
|
def __init__(self): |
|
openai.api_key = os.getenv("OPENAI_API_KEY") |
|
|
|
|
|
def get_empty_state(self): |
|
|
|
""" Create empty Knowledge base""" |
|
|
|
return {"knowledge_base": None} |
|
|
|
def get_content_from_url(self,url:str)->List: |
|
""" |
|
Uploads a file from a given URL and returns the loaded document. |
|
Args: |
|
url (str): The URL of the file to be uploaded. |
|
Returns: |
|
Document: The loaded document. |
|
Raises: |
|
ValueError: If the URL is not valid or the file cannot be fetched. |
|
""" |
|
|
|
|
|
if validators.url(url): |
|
headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36',} |
|
r = requests.get(url,headers=headers) |
|
if r.status_code != 200: |
|
raise ValueError( |
|
"Check the url of your file; returned status code %s" % r.status_code |
|
) |
|
|
|
content_type = r.headers.get("content-type") |
|
file_extension = mimetypes.guess_extension(content_type) |
|
temp_file = tempfile.NamedTemporaryFile(suffix=file_extension, delete=False) |
|
temp_file.write(r.content) |
|
file_path = temp_file.name |
|
loader = UnstructuredFileLoader(file_path, strategy="fast") |
|
docs = loader.load() |
|
return docs |
|
else: |
|
raise ValueError("Please enter a valid URL") |
|
|
|
def create_knowledge_base(self,docs): |
|
|
|
"""Create a knowledge base from the given documents. |
|
Args: |
|
docs (List[str]): List of documents. |
|
Returns: |
|
FAISS: Knowledge base built from the documents. |
|
""" |
|
|
|
|
|
|
|
|
|
text_splitter = CharacterTextSplitter( |
|
separator="\n", chunk_size=1000, chunk_overlap=200, length_function=len |
|
) |
|
|
|
|
|
chunks = text_splitter.split_documents(docs) |
|
|
|
|
|
embeddings = OpenAIEmbeddings() |
|
|
|
|
|
knowledge_base = FAISS.from_documents(chunks, embeddings) |
|
|
|
|
|
return knowledge_base |
|
|
|
|
|
def get_chemicals_for_url(self,urls:str,state)->str: |
|
""" |
|
Retrieves chemicals from the provided URLs. |
|
|
|
Args: |
|
urls (str): Comma-separated URLs of the files to be processed. |
|
|
|
Returns: |
|
str: The extracted chemical names. |
|
|
|
Raises: |
|
ValueError: If an error occurs during the process. |
|
""" |
|
|
|
total_chemical=[] |
|
for url in urls.split(','): |
|
webpage_text = self.get_content_from_url(url) |
|
knowledge_base = self.create_knowledge_base(webpage_text) |
|
state = {"knowledge_base": knowledge_base} |
|
chemicals = self.get_chemicals_for_file(state) |
|
total_chemical.append(str(url)+"\n"+chemicals+"\n\n") |
|
list_of_chemicals = "".join(total_chemical) |
|
return list_of_chemicals |
|
|
|
|
|
def file_path_show(self,file_paths): |
|
file_paths = [single_file_path.name for single_file_path in file_paths] |
|
return file_paths |
|
|
|
def get_chemicals_for_file(self,state): |
|
knowledge_base = state["knowledge_base"] |
|
|
|
question = "list out chemicals.Result should be in bullet form" |
|
|
|
|
|
docs = knowledge_base.similarity_search(question) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
llm = OpenAI(temperature=0.4) |
|
|
|
|
|
chain = load_qa_chain(llm, chain_type="stuff") |
|
|
|
|
|
response = chain.run(input_documents=docs, question=question) |
|
|
|
|
|
return response |
|
|
|
def identify_chemicals_in_files(self,file_paths,state): |
|
"""Upload a file and create a knowledge base from its contents. |
|
Args: |
|
file_paths : The files to uploaded. |
|
Returns: |
|
tuple: A tuple containing the file name and the knowledge base. |
|
""" |
|
|
|
|
|
file_paths = [single_file_path.name for single_file_path in file_paths] |
|
results ='' |
|
for file_obj in file_paths: |
|
|
|
loader = UnstructuredFileLoader(file_obj, strategy="fast") |
|
|
|
|
|
docs =loader.load() |
|
|
|
|
|
knowledge_base = self.create_knowledge_base(docs) |
|
state = {"knowledge_base": knowledge_base} |
|
pdf_name = os.path.basename(file_obj) |
|
|
|
final_ans = self.get_chemicals_for_file(state) |
|
results += pdf_name+"\n"+final_ans+"\n\n" |
|
|
|
|
|
return results |
|
|
|
def get_final_result(self,urls,file_paths,state): |
|
|
|
if urls: |
|
if file_paths: |
|
urls_chemicals = self.get_chemicals_for_url(urls,state) |
|
file_chemicals = self.identify_chemicals_in_files(file_paths,state) |
|
chemicals = urls_chemicals + file_chemicals |
|
|
|
return chemicals |
|
else: |
|
urls_chemicals = self.get_chemicals_for_url(urls,state) |
|
return urls_chemicals |
|
elif file_paths: |
|
file_chemicals = self.identify_chemicals_in_files(file_paths,state) |
|
return file_chemicals |
|
else: |
|
return "No Files Uploaded" |
|
|
|
|
|
def gradio_interface(self)->None: |
|
""" |
|
Starts the Gradio interface for chemical identification. |
|
""" |
|
|
|
with gr.Blocks(css="style.css",theme='karthikeyan-adople/hudsonhayes-gray') as demo: |
|
gr.HTML("""<center class="darkblue" style='background-color:rgb(0,1,36); text-align:center;padding:25px;'><center><h1 class ="center"> |
|
- <img src="file=logo.png" height="110px" width="280px"></h1></center> |
|
- <br><h1 style="color:#fff">Chemical Capability Identifier</h1></center>""") |
|
state = gr.State(self.get_empty_state()) |
|
|
|
with gr.Column(elem_id="col-container"): |
|
with gr.Row(elem_id="row-flex"): |
|
url = gr.Textbox(label="URL") |
|
with gr.Row(elem_id="row-flex"): |
|
with gr.Accordion("Upload Files", open = False): |
|
with gr.Row(): |
|
with gr.Column(scale=0.90, min_width=160): |
|
file_output = gr.File() |
|
with gr.Column(scale=0.10, min_width=160): |
|
upload_button = gr.UploadButton( |
|
"Browse File", file_types=[".txt", ".pdf", ".doc", ".docx"], |
|
file_count = "multiple",variant="primary") |
|
with gr.Row(): |
|
with gr.Column(scale=1, min_width=0): |
|
compare_btn = gr.Button(value="Generate Analysis",variant="primary") |
|
with gr.Row(): |
|
with gr.Column(scale=1, min_width=0): |
|
compared_result = gr.Textbox(value="",label='Chemical Capabilities :',show_label=True, placeholder="",lines=10) |
|
|
|
with gr.Row(): |
|
with gr.Column(scale=1, min_width=0): |
|
input_qus = gr.Textbox(value="",label='Question :',show_label=True, placeholder="") |
|
with gr.Row(): |
|
with gr.Column(scale=1, min_width=0): |
|
find_answer = gr.Button(value="Find Answer",label='Find',show_label=True, placeholder="") |
|
with gr.Row(): |
|
with gr.Column(scale=1, min_width=0): |
|
output = gr.Textbox(value="",label='Answer:',show_label=True, placeholder="") |
|
|
|
upload_button.upload(self.file_path_show, upload_button, [file_output]) |
|
|
|
compare_btn.click(self.get_final_result,[url,upload_button,state],compared_result) |
|
|
|
find_answer.click(document_qa.get_final_result,[url,upload_button,state,input_qus],output) |
|
demo.launch(debug=True) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
chemical_identifier = ChemicalIdentifier() |
|
chemical_identifier.gradio_interface() |