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 )