src/kkms_kssw.py

import os

import src.constants as constants_utils
import src.langchain_utils as langchain_utils
import src.weather as weather_utils
import src.mandi_price as mandi_utils
import src.translator as translator_utils
import src.web_crawler as web_crawler_utils

import logging
logger = logging.getLogger(__name__)
logging.basicConfig(
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s", level=logging.INFO, datefmt="%Y-%m-%d %H:%M:%S"
)

import warnings
warnings.filterwarnings('ignore')



class KKMS_KSSW:
    def __init__(self):
        self.index_type = constants_utils.INDEX_TYPE
        self.load_from_existing_index_store = constants_utils.LOAD_FROM_EXISTING_INDEX_STORE

        # Instantiate langchain_utils class object
        self.langchain_utils_obj = langchain_utils.LANGCHAIN_UTILS(
            index_type=self.index_type,
            load_from_existing_index_store=self.load_from_existing_index_store
        )
        # Instantiate Mandi Price utils class object
        self.mandi_utils_obj = mandi_utils.MANDI_PRICE()
        # Instantiate Weather class object
        self.weather_utils_obj = weather_utils.WEATHER()
        # Instantiate translator_utils class object
        self.translator_utils_obj = translator_utils.TRANSLATOR()



    # Initialize index (vector store)
    def load_create_index(self):
        logger.info(f"Load/Create index")
        self.langchain_utils_obj.load_create_index()


    # Upload data and update the index
    def upload_data(
        self,
        doc_type,
        files_or_urls,
        index_category
    ):
        logger.info(f"Uploading data")
        self.langchain_utils_obj.upload_data(
            doc_type=doc_type,
            files_or_urls=files_or_urls,
            index_category=index_category
        )


    # Define query on index to retrieve the most relevant top K documents from the vector store
    def query(
        self,
        question,
        question_category
    ):
        '''
            Args:
                mode: can be any of [default, embedding]
                response_mode: can be any of [default, compact, tree_summarize]
        '''
        logger.info(f"Querying from index/vector store")
        
        return self.langchain_utils_obj.query(
            question=question,
            question_category=question_category
        )