import pickle import openai import re import numpy as np import tensorflow_hub as hub import openai import os import gradio as gr import pandas as pd import io import fitz from sklearn.neighbors import NearestNeighbors import warnings warnings.filterwarnings(action='ignore') indi_pkl_file_path = "individual_list.pkl" # Specify the file path from which to read the list busi_pkl_file_path = "business_list.pkl" # Specify the file path from which to read the list indi_loaded_array_file_path= "individual_numpy.npy" busi_loaded_array_file_path= "business_numpy.npy" with open(indi_pkl_file_path, "rb") as f: indi_texts = pickle.load(f) with open(busi_pkl_file_path, "rb") as f: busi_texts = pickle.load(f) #df = pd.read_excel('ABS_Statewise_Tax_Revenue_data.xlsx') #ndf= df.copy() global indi_loaded_array global busi_loaded_array indi_loaded_array = np.load(indi_loaded_array_file_path) busi_loaded_array = np.load(busi_loaded_array_file_path) openAI_key = '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' # key1: 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 start_page=1 def text_to_chunks(texts, word_length=150, start_page=1): text_toks = [t.split(' ') for t in texts] # length of text_toks is equal to number of pages in PDF #print("text_toks: ",text_toks) page_nums = [] chunks = [] for idx, words in enumerate(text_toks): #print(f'idx: {idx}\nwords: {words}') for i in range(0, len(words), word_length): #print("i: ",i) chunk = words[i:i+word_length] #print("chunk: ",chunk) if (i+word_length) > len(words) and (len(chunk) < word_length) and ( len(text_toks) != (idx+1)): text_toks[idx+1] = chunk + text_toks[idx+1] continue chunk = ' '.join(chunk).strip() chunk = f'[{idx+start_page}]' + ' ' + '"' + chunk + '"' #print(chunk) chunks.append(chunk) return chunks class SemanticSearch: def __init__(self,data,loaded_array, batch=1000, n_neighbors=5) : self.use = hub.load('https://tfhub.dev/google/universal-sentence-encoder/4') self.fitted = False self.data = data self.embeddings = loaded_array n_neighbors = min(n_neighbors, len(self.embeddings)) self.nn = NearestNeighbors(n_neighbors=n_neighbors) self.nn.fit(self.embeddings) self.fitted = True def __call__(self, text, return_data=True): inp_emb = self.use([text]) neighbors = self.nn.kneighbors(inp_emb, return_distance=False)[0] if return_data: return [self.data[i] for i in neighbors] else: return neighbors def generate_text(openAI_key,prompt, engine="text-davinci-003"): openai.api_key = openAI_key completions = openai.Completion.create( engine=engine, prompt=prompt, max_tokens=512, n=1, stop=None, temperature=0, ) message = completions.choices[0].text return message def generate_answer(question,openAI_key): topn_chunks = recommender(question) prompt = "" prompt += 'search results:\n\n' for c in topn_chunks: prompt += c + '\n\n' prompt += "Instructions: Compose a comprehensive reply to the query using the search results given. "\ "Consider the provided information from Australian Taxation Office (ATO) and answer only when context is clear to you. For example : Tax, Australia etc."\ "Keep The tone conversational "\ "Provide example from the text to make the answer understandable."\ "If there are multiple points, use bullets to write those points."\ "If there is any related hyperlink, include that in the answer."\ "If the search results mention multiple subjects with the same name, create separate answers for each. "\ "Only include information found in the results and don't add any additional information."\ "Make sure the answer is correct and don't output false content. "\ "If the text does not relate to the query, simply state 'Are you sure this is a taxation related query? If not kindly ask taxation related queries'." \ "Ignore outlier search results which has nothing to do with the question. Only answer what is asked."\ "The answer should be short and concise. Answer step-by-step. \n\nQuery: {question}\nAnswer:"\ prompt += f"Query: {question}\nAnswer:" answer = generate_text(openAI_key, prompt,"text-davinci-003") return answer def classify_chatgpt(question, openAI_key, engine='gpt-3.5-turbo-0301'): openai.api_key = openAI_key message = [{'role': 'system', 'content': '''We have a user query related to taxes from Australia and your task is to classify the query into one of the category from the following given two categories only: 1. Business Taxation Query 2. Individual Taxation Query Context: Individual Taxation Query are those queries which are related to the individual taxation of the residents of a country (here from Australia).\n Business Taxation Query are those queries which are related from the Business perspective. Instructions: 1. Understand the query very well.\n 2. Return the result only as a string. \n 3. Do not put any punctuation mark after classifying it. \n 4. Do not write any extra information and just do the Classification from the mentioned two categories.'''}, {'role': 'user', 'content': question}] completions = openai.ChatCompletion.create( model=engine, messages=message, max_tokens=512, n=1, stop=None, temperature=0.7, ) message = completions.choices[0].message['content'] #print(message) return message def question_answer(question,openAI_key): return generate_answer(question,openAI_key) indi_chunks = text_to_chunks(indi_texts, start_page=start_page) busi_chunks = text_to_chunks(busi_texts, start_page=start_page) def output_generation(typed_ques): type_of_query = 'dummy' if typed_ques == "Hello" : out_pu = "Do you want to continue with myGovID or as Guest." return out_pu, type_of_query elif typed_ques == "myGovID": out_pu = "Hi User, Please Enter Your Username" return out_pu, type_of_query elif len(typed_ques.split('.')) == 2: out_pu = "Please enter your password" return out_pu, type_of_query elif typed_ques.isdigit() or typed_ques == "Guest": out_pu = "Welcome to Tax GenAI, Please ask your question" return out_pu, type_of_query else: type_of_query= classify_chatgpt(typed_ques, openAI_key) global recommender #print(type_of_query) if type_of_query in ['Individual Taxation Query','Individual Taxation Query.']: recommender = SemanticSearch(data=indi_chunks,loaded_array = indi_loaded_array) out_pu = question_answer(question= typed_ques, openAI_key= openAI_key) #print(out_pu) return out_pu, type_of_query elif type_of_query in ['Business Taxation Query','Business Taxation Query.']: recommender = SemanticSearch(data=busi_chunks, loaded_array=busi_loaded_array) out_pu = question_answer(question= typed_ques, openAI_key= openAI_key) #print(out_pu) return out_pu, type_of_query title = 'CDI Citizen Intelligence 360 Tool: Tax Advisory' #description = """ Citizen Intelligence 360 sources data from various government agencies and makes it accessible to citizens in a user-friendly format. This can help citizens better understand how their local government works and stay informed about important initiatives and changes in their communities. Citizen Intelligence 360 is a citizen intelligence tool that uses mapping technology to provide citizens with access to government data, performance report and other civic information. It helps to increase transparency and accountability in local government, and empowers citizens to make informed decisions and participate in the democratic process.""" description = """Welcome to Citizen Intelligence 360 Tool, your intelligent tax companion. Harnessing the power of advanced GEN AI, the tax advisory tool is here to revolutionize your tax experience. With its unrivaled expertise and personalized guidance, the tool simplifies complex tax matters, providing accurate answers to your questions instantly. Say goodbye to confusing endless research and experience the future of tax advisory with Citizen Intelligence 360 Tool - the intelligent solution for your tax needs.""" img_ = '''''' def gradio_chatbox(input,history): history = history or [] typed_ques = input answe, query_type = output_generation(typed_ques) ccc = "{0}".format(answe) output = ccc history.append((input,output)) #print(history) return history, history block = gr.Blocks() with block: gr.Markdown(f'

{title}

') gr.Markdown(f'

{img_}

') gr.Markdown(description) chatbot = gr.Chatbot(label="Tax GenAI") message = gr.Textbox(label='Please ask your question',placeholder = "Welcome! This is Tax GenAI. Type ""Hello"" to continue.") state = gr.State() submit = gr.Button("SEND") #submit1 = gr.Button("SEND") #submit2 = gr.Button("SEND") submit.style(full_width=None, size='lg' ) #submit1.style(full_width=None, size='lg') #submit2.style(full_width=None, size='lg') submit.click(gradio_chatbox, inputs=[message, state], outputs=[chatbot, state]) block.launch()