File Manager

Editing File: /home/ubuntu/codegamaai-test/tax_bot/main_v4.py

from dotenv import load_dotenv

import openai
import sys, os
import logging
from llama_index import (
    SimpleDirectoryReader,
    LLMPredictor,
    GPTVectorStoreIndex,
    load_index_from_storage,
    set_global_service_context,
)
from llama_index.storage.storage_context import StorageContext
from llama_index.prompts import Prompt
from llama_index.indices.service_context import ServiceContext
from langchain.chat_models import ChatOpenAI
from llama_index.memory import ChatMemoryBuffer
from llama_index.llms import OpenAI
from src.constants import *
from src.chats import *
from src.utils import *
from src.calculate import *
from llama_index.tools.tool_spec.base import BaseToolSpec
from llama_index.agent import OpenAIAgent
import gradio as gr
from help_functions.intent_class import *
import qdrant_client
from llama_index.vector_stores.qdrant import QdrantVectorStore

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")
service_context = ServiceContext.from_defaults(llm=OpenAI(model_name="gpt-4o", temperature=0.2, max_tokens=1500 ))

dir_path = []
for root, dirs, files in os.walk(os.environ['DOCUMENTS_DIR']):
    for name in dirs:
        dir_path.append(os.path.join(root, name))

dir_path_canada = os.path.join(os.environ['DOCUMENTS_DIR'], "Canada")
dir_path_india = os.path.join(os.environ['DOCUMENTS_DIR'], "India")
dir_path_usa = os.path.join(os.environ['DOCUMENTS_DIR'], "USA")
dir_path_uae = os.path.join(os.environ['DOCUMENTS_DIR'], "UAE")
dir_path_singapore = os.path.join(os.environ['DOCUMENTS_DIR'], "Singapore")

# this class is chnge for loading 
class Llama:
    indexes = {'Canada': None, 'India': None, 'USA': None, 'UAE': None, 'Singapore': None}

@staticmethod
    def init():
        # Llama.index = Llama.load_index(dir_path)
        Llama.indexes['Canada'] = Llama.load_index(dir_path_canada, "Canada_tax_doc")
        Llama.indexes['India'] = Llama.load_index(dir_path_india, "India_tax_doc")
        Llama.indexes['USA'] = Llama.load_index(dir_path_usa, "USA_tax_doc")
        Llama.indexes['UAE'] = Llama.load_index(dir_path_uae, "UAE_tax_doc")
        Llama.indexes['Singapore'] = Llama.load_index(dir_path_singapore, "Singapore_tax_doc")

@staticmethod
    def load_index(dir_path, collection_name):
        documents = SimpleDirectoryReader(dir_path, filename_as_id=True).load_data() # removed the looped 
        print(f"Loaded documents with{len(documents)}pages for collection{collection_name}")
        client = qdrant_client.QdrantClient(
            location="http://164.52.213.13:6333"
        )
        vector_store = QdrantVectorStore(client=client, collection_name=collection_name)
        storage_context = StorageContext.from_defaults(vector_store=vector_store)
        try:
            index = GPTVectorStoreIndex.from_documents(documents, storage_context=storage_context)
            # storage_context = StorageContext.from_defaults(persist_dir=os.environ['STORAGE_DIR'])
            index = load_index_from_storage(storage_context)
            logging.info("Index loaded from storage.")
        except FileNotFoundError:
            logging.info("Index not found. Creating a new one...")
            index = GPTVectorStoreIndex.from_documents(documents)
            index.storage_context.persist()
            logging.info("New index created and persisted to storage.")

refreshed_docs = index.refresh_ref_docs(documents,
                                                update_kwargs={"delete_kwargs": {'delete_from_docstore': True}})
        print(refreshed_docs)
        print('Number of newly inserted/refreshed docs: ', sum(refreshed_docs))

index.storage_context.persist()
        logging.info("Index refreshed and persisted to storage.")

return index

# Load the index when the server starts
Llama.init()

user_sessions = {}

def is_tax_related(query):
    try:
        response = openai.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "user", "content": query},
                {"role": "assistant", "content": """Is the above query about Calculating User's Personal Tax with mention of the income, deductions, and other tax-related details?
                 For Example: How much tax do I need to pay? My income is 10 Lakhs.
                 Return 'yes' if it is, otherwise return 'no'."""}
            ]
        )
        print(response)
        assistant_response = response.choices[0].message.content.strip().lower()
        print("Assistant response:", assistant_response)
        return "yes" in assistant_response
    except Exception as e:
        print(f"Error in OpenAI call: {e}")
        return False

def is_finance_tax_related(query, history):
    # If hsitory is not empty, Read the last 2 elements from the history
    if history:
        last_2_elements = history[-2:]

try:
        # Count total Words in the query
        total_words = len(query.split())
        # If the query is less than 3 words, then it is not related to finance and tax
        if total_words < 2:
            pass

system_message = """Based on the available conversation history Identify that the current query is related to The Tax Bot Feature. 
        The Bot is a Tax and Finance Helper Assistant, specialized in addressing and assisting only on tax-related and Finance related inquiries.
        Return 'yes' if it is, otherwise return 'no."""
        system_prompt = {"role": "system", "content": system_message}

user_prompt = {"role": "user", "content": query}
        messages = [system_prompt]
        messages.extend(last_2_elements)
        messages.append(user_prompt)
        response = openai.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages
        )
        print("is_finance_tax_related", response)
        assistant_response = response.choices[0].message.content.strip().lower()
        print("Assistant response:", assistant_response)
        return "yes" in assistant_response
    except Exception as e:
        print(f"Error in OpenAI call: {e}")
        return False

class TaxCalculator(BaseToolSpec):
    spec_functions = ["calculate_tax_india", "calculate_tax_other_countries"]

def calculate_tax_india(self, location: str = "Non-Metro", total_income: float = 0.0, basic_salary: float = 0.0,
                            dearness_allownces: float = 0.0, hra: float = 0.0, hra_actual: float = 0.0,
                            special_allowance: float = 0.0, pf_deduction: float = 0.0, deduction_80c: float = 0.0,
                            pt_deduction: float = 2400.0, deduction_80d: float = 0.0, deduction_NPS: float = 0.0,
                            deduction_80G: float = 0.0, deduction_interest_on_loan: float = 0.0) -> str:
        """
        Use this function to Calculate the payable tax for the given input parameters for INDIA and Indian National Rupee(INR).

:param total_income
        :param basic_salary
        :param location
        :param dearness_allownces
        :param hra
        :param hra_actual
        :param special_allowance
        :param pf_deduction
        :param deduction_80c
        :param pt_deduction
        :param deduction_80d
        :param deduction_NPS
        :param deduction_80G
        :param deduction_interest_on_loan

"""
        calculated_tax = tax_calculation(location, basic_salary, dearness_allownces, hra, hra_actual, special_allowance,
                                         pf_deduction, deduction_80c, pt_deduction, deduction_80d, deduction_NPS,
                                         deduction_80G, deduction_interest_on_loan, total_income)
        return calculated_tax

def calculate_tax_other_countries(self, currency: str = None, country: str = None):
        """Use this function to calculate tax for all other countries(Like USA, UK, Singapore, UAE, Canada) and their currencies(USD, GBP, SGD, AED, CAD).

:param currency
        :param country
        """
        # if currency and country:
        #     response = f"The tax calculation feature is only available for Indian Tax System as of now. It is not available for other countries at the moment. We are working on adding support for other countries as well."
        # else:

response = "The accurate tax calculation feature is only available for Indian Tax System as of now. It is not available for other countries at the moment. We are working on adding support for other countries as well. Although, you can ask me any tax-related queries and I will try to help you with the information I have."
        return response

def data_querying(input_text, user_id, country_code):
    
    global user_sessions
    if user_id not in user_sessions:
        user_sessions[user_id] = {"interaction_count": 0, "conversation_history": [], "previous_intent": None}

user_session = user_sessions[user_id]
    user_session["interaction_count"] += 1
    print(f"Debug: User {user_id} interaction count: {user_session['interaction_count']}")
    # user_session["conversation_history"].append(input_text)

country_name = get_country_name(country_code)
    selected_index = Llama.indexes.get(country_name) # added this for selection of index
    if selected_index is None:
        return "Error country not found"

# Create a new memory buffer for the user
    create_json(user_id)
    history = read_json(user_id)

# Convert the chat history to a format that the ChatMemoryBuffer can understand
    chat_history = convert_chat_message(history)
    if history:
        memory = ChatMemoryBuffer.from_defaults(token_limit=3900, chat_history=chat_history)
    else:
        memory = ChatMemoryBuffer.from_defaults(token_limit=3900)

creation_queries = [
        "who made you",
        "who created you",
        "who is your creator",
        "what company made you",
        "are you open source",
        "who developed you",
        "who is responsible for your creation",
        "which organization created you",
        "who owns you",
        "where do you come from",
        "who built you",
        "what's your origin",
        "who programmed you",
        "who is behind you",
        "who is your developer",
        "what entity created you",
        "are you created by a company",
        "who is your maker",
        "what team developed you",
        "which company are you a product of",
        "are you the work of a specific person",
        "who engineered you",
        "what model you use",
        "what model are you using",
        "what is your model architecture",
        "what's the source of your intelligence",
        "who gave you life",
        "who is your parent company",
    ]
    name_queries = [
        "what is your name",
        "who are you",
        "do you have a name",
        "what should I call you",
        "what's your name",
        "tell me your name",
        "your name please",
        "who am I speaking with",
        "what do they call you",
        "are you named",
        "do you go by a name",
        "what name do you go by",
        "may I know your name",
        "who is this",
        "what are you called",
        "have you been given a name",
        "what nickname do you have",
        "what do people call you",
        "how should I address you",
    ]
    if any(creation_query in input_text.lower() for creation_query in creation_queries):
        hardcoded_response = "I was developed by First Technology, leveraging state-of-the-art AI models to assist you. If you have any more questions or need help, feel free to ask!"
        return hardcoded_response
    elif any(name_query in input_text.lower() for name_query in name_queries):
        hardcoded_response = "My name is TaxTrax, your friendly tax assistant bot designed to help you with your tax-related inquiries. How can I assist you today?"
        return hardcoded_response

# else:
    #     # Get the intent of the user
    #     query_intent = get_intent(input_text)
    #     if query_intent == "Non-tax-related":
    #         # Count total Words in the query
    #         total_words = len(input_text.split())
    #         # If the query is less than 3 words, then it is not related to finance and tax
    #         if total_words <= 3:
    #             pass
    #         else:
    #             response = "I am a Tax Helper Assistant, I can only help with tax related queries. Please ask me a question related to tax and finance."
    #             save_response(response, user_id, input_text)
    #             return response
    #     else:
    #         print("User intent: Tax-related")
    #         pass

if is_tax_related(input_text):

# agent = OpenAIAgent.from_tools(TaxCalculator().to_tool_list(), memory= memory, verbose=True)
        country_name = get_country_name(country_code)
        if country_name == "Country Not Found":
            response = "The tax calculation feature is only available for Indian Tax System as of now. It is not available for other countries at the moment. We are working on adding support for other countries as well. Although, you can ask me any tax-related queries and I will try to help you with the information I have."
            save_response(response, user_id, input_text)
            return response

else:
            pass

#input_text = input_text + f" for {country_name}"
        input_text = input_text
        agent = OpenAIAgent.from_tools(TaxCalculator().to_tool_list(), verbose=True)
        response = agent.chat(input_text)
        response_message = response.response
        save_response(response_message, user_id, input_text)
        return response_message

# text_qa_template_str = (
    #     "Context information is below.\n"
    #     "---------------------\n"
    #     "{context_str}\n"
    #     "---------------------\n"
    #     "Using both the context information and also using your own knowledge, "
    #     "answer the question: {query_str}\n"
    #     "If the context isn't helpful, Please respond with: I don't understand please try to rephrase your question\n"
    # )
    text_qa_template_str = (
        "Context information is below.\n"
        "---------------------\n"
        "{context_str}\n"
        "---------------------\n"
        "Using both the context information and also using your own knowledge, "
        "Firstly Decide whether the question is related to tax and finance or not. If it is not related to tax and finance, then respond with: `I am a Tax Helper Assistant, I can only help with tax related queries. Please ask me a question related to tax and finance.`\n"
        "answer the question: {query_str}\n"
        "If the context isn't helpful, Please respond with: I don't understand please try to rephrase your question\n"
    )

refine_template_str = (
        "The original question is as follows: {query_str}\n"
        "We have provided an existing answer: {existing_answer}\n"
        "We have the opportunity to refine the existing answer "
        "(only if needed) with some more context below.\n"
        "------------\n"
        "{context_msg}\n"
        "------------\n"
        "Using both the new context and your own knowledege, update or repeat the existing answer.\n"
    )

refine_template = Prompt(refine_template_str)
    text_qa_template = Prompt(text_qa_template_str)

system_prompt = """You are a Tax Helper Assistant, specialized in addressing and assisting with a wide range of tax-related inquiries for the fiscal year FY 2023-24 (AY 2024-25). Your role encompasses providing expert guidance on tax calculations, clarifying tax regulations, and suggesting tax-saving strategies tailored to individual circumstances.
    Ensure your responses:
    - First Decide whether the question is related to tax and finance or not. If it is not related to tax and finance, then respond with: `I am a Tax Helper Assistant, I can only help with tax related queries. Please ask me a question related to tax and finance.`
    - Are accurate, reflecting the latest tax legislation and guidelines.
    - Provide practical, easy-to-understand explanations suitable for a broad audience.
    - Highlight tax-saving opportunities, emphasizing legal and effective strategies.
    - Maintain a supportive and professional tone, fostering trust and confidence in users seeking tax assistance.
    - Do not disclose or imply the existence of any specific internal resources and file names.
    - Do not say 'I am not a tax expert' in your response.
    - Assist users in navigating their tax questions with clarity and confidence, guiding them toward informed decisions regarding their tax obligations and savings opportunities which is present in your context and don't divert from your context.
    """

response = selected_index.as_chat_engine(text_qa_template=text_qa_template,  # refine_template=refine_template,
                                             chat_mode="context", memory=memory, system_promt=system_prompt,
                                             ).chat(input_text) #change

user_sessions[user_id] = user_session

# Update User Memory
    save_response(response.response, user_id, input_text)

return response.response

def answer_question(question, user_id, country_code):
    user_id = "static_user_id"
    try:
        # response = data_querying(question)
        response = data_querying(question, user_id, country_code)

return response
    except Exception as e:
        return str(e)

Back to Directory File Manager