app.py

import time
import openai
import os
import pandas as pd
import numpy as np
import gradio as gr
from llama_index import GPTSimpleVectorIndex
from openai.embeddings_utils import get_embedding, cosine_similarity

openai.api_key = os.getenv("OPENAI_API_KEY")
passwd = os.getenv("PASSWD_SECRET")

title = "Confidential forensics tool with ChatGPT"
examples = ["Who is Phillip Allen?", "What the project in Austin is about?", "Give me more details about the real estate project"]

index = GPTSimpleVectorIndex.load_from_disk('email.json')

df = pd.read_feather("emails.feather")
df2 = df["subject"].replace([None], "<no title>").head(10).to_frame()

def search_emails(opt, message, n=3):
 "Outputs the top n emails that match the most the pattern"
 if len(message.strip()) < 1:
 message = "Oops, it looks like your query was not valid. Please make sure you typed something in your text box and then try again."
 else:
 try:
 embedding = get_embedding(message)
 message = ""
 df['similarities'] = df.embedding.apply(func=(lambda x: cosine_similarity(x, embedding)))

 message_tmp = df.sort_values('similarities', ascending=False).head(n)
 message_tmp = [(row.file, row.body, row.similarities) for index, row in message_tmp.iterrows()]
 for msg in message_tmp:
 message += f"Mail ID: {msg[0]}\nContent: {msg[1].strip()}\nSimilarity score: {msg[2]}\n\n"
 except Exception as e:
 message = "An error occured when handling your query, please try again."
 print(e)
 return message, ""

def respond_upload(btn_upload, message, chat_history):
 time.sleep(2)
 message = "***File uploaded***"
 bot_message = "Your document has been uploaded and will be accounted for your queries."
 chat_history.append((message, bot_message))
 return btn_upload, "", chat_history

def respond2(message, chat_history, box, btn):
 if len(message.strip()) < 1:
 message = "***Empty***"
 bot_message = "Oops, it looks like your query was not valid. Please make sure you typed something in your text box and then try again."
 else:
 try:
 bot_message = str(index.query(message)).strip()
 except:
 bot_message = "An error occured when handling your query, please try again."
 chat_history.append((message, bot_message))
 return message, chat_history, box

def respond(message, chat_history):
 if len(message.strip()) < 1:
 message = "***Empty***"
 bot_message = "Oops, it looks like your query was not valid. Please make sure you typed something in your text box and then try again."
 else:
 try:
 bot_message = str(index.query(message)).strip()
 except:
 bot_message = "An error occured when handling your query, please try again."
 chat_history.append((message, bot_message))
 return "", chat_history

with gr.Blocks(title=title) as demo:
 gr.Markdown(
 """

 # """ + title + """
 """)
 dat = gr.Dataframe(
 value=df2,
 max_cols=1,
 max_rows=4,
 title="Documents loaded",
 overflow_row_behaviour="paginate",
 )
 btn_upload = gr.UploadButton("Upload a new document...", file_types=["text"])
 gr.Markdown(
 """
 ## Chatbot
 """)
 chatbot = gr.Chatbot().style(height=400)
 with gr.Row():
 with gr.Column(scale=0.85):
 msg = gr.Textbox(
 show_label=False,
 placeholder="Enter text and press enter, or click on Send.",
 ).style(container=False)
 with gr.Column(scale=0.15, min_width=0):
 btn_send = gr.Button("Send your query")
 with gr.Row():
 gr.Markdown(
 """
 Example of queries
 """)
 for ex in examples:
 btn = gr.Button(ex)
 btn.click(respond2, [btn, chatbot, msg], [btn, chatbot, msg])

 msg.submit(respond, [msg, chatbot], [msg, chatbot])
 btn_send.click(respond, [msg, chatbot], [msg, chatbot])
 btn_upload.upload(respond_upload, [btn_upload, msg, chatbot], [btn_upload, msg, chatbot])
 gr.Markdown(
 """

 ## Search the matching document
 """)
 opt = gr.Textbox(
 show_label=False,
 placeholder="The document matching with your query will be shown here.",
 interactive=False,
 lines=8
 )
 with gr.Row():
 with gr.Column(scale=0.85):
 msg2 = gr.Textbox(
 show_label=False,
 placeholder="Enter text and press enter, or click on Send.",
 ).style(container=False)
 with gr.Column(scale=0.15, min_width=0):
 btn_send2 = gr.Button("Send your query")

 btn_send2.click(search_emails, [opt, msg2], [opt, msg2])
 
if __name__ == "__main__":
 # demo.launch(auth=("mithril", passwd))
 demo.launch()