app files

src/FunctionTools.py

@@ -0,0 +1,249 @@
+import json
+import chardet
+import streamlit as st
+import pandas as pd
+from llama_index.core.tools.tool_spec.base import BaseToolSpec
+
+class ScriptureDescriptionToolSpec(BaseToolSpec):
+    '''
+    Purpose: Obtains the description or summary about vedas, mandalas, kandas, shuktas, archakah, adhyaya, and other scriptural elements.
+    Returns: A dictionary containing the description or basic information about the specified scriptural element.
+    Sample query:
+    1. Describe the first kandah, second shukta from Atharvaveda?
+    2. Summarize ShuklaYajurVeda?
+    3. What is the difference between ShuklaYajurVeda and KrishnaYajurVeda?
+    '''
+    # Define the functions that we export to the LLM
+    spec_functions = ["get_description"]
+
+    with open("Data/scripture_descriptions.csv", 'rb') as f:
+        result = chardet.detect(f.read())
+
+    encoding = result['encoding']
+    df = pd.read_csv("Data/scripture_descriptions.csv", encoding=encoding)
+
+    @st.cache_data
+    def get_description(_self, level_0, level_1:int=None, level_2:int=None, level_3:int=None):
+        """
+        To get the description or basic information about vedas/mandalas/kandas/shukatas/archakah/adhyaya and others.
+        """
+        try:
+            if level_3 is not None:
+                # Case with Level-2 specified
+                result = _self.df[(_self.df['scripture_name'].str.lower() == level_0.lower())
+                                  & (_self.df['level_1'] == str(level_1))
+                                  & (_self.df['level_2'] == str(level_2)) & (_self.df['level_3'] == str(level_3))]
+            elif level_2 is not None:
+                # Case with Level-2 specified
+                result = _self.df[(_self.df['scripture_name'].str.lower() == level_0.lower())
+                                  & (_self.df['level_1'] == str(level_1)) & (_self.df['level_2'] == str(level_2))]
+            elif level_1 is not None:
+                # Case with Level-1 specified
+                result = _self.df[(_self.df['scripture_name'].str.lower() == level_0.lower())
+                                  & (_self.df['level_1'] == str(level_1))]
+            else:
+                # Case with only Level-0 specified
+                result = _self.df[_self.df['scripture_name'].str.lower() == level_0.lower()]
+
+            return result.iloc[0].to_dict()
+        except IndexError as e:
+            return json.dumps({"error": f"Failed to get scripture description. {e}"})
+
+class MantraToolSpec(BaseToolSpec):
+    '''
+    To obtain translations or meaning of vedamantras from RigVeda and AtharvaVeda using the function `get_translation`.
+    The mantra details such as vedamantra, padapatha, rishi, chandah, devata, and swarah from the vedas accessible through the function `get_vedamantra_details`.
+    The mantra summary like anvaya, adhibautic, ahyatmic, adhidaivic meaning of vedamantra accessible using the function 'get_vedamantra_summary'
+    Sample Query:
+    1. What is the vedamantra of the mantra from Rigveda, first mandala, first shukta, and first mantra?
+    2. What is the devata of the vedamantra from Rigveda, first mandala, first shukta, and first mantra?
+    3. What is the meaning of the vedamantra from Rigveda, first mandala, first shukta, and first mantra written by Tulsi Ram?
+    4. What is the (adhibautic) meaning of the vedamantra from RigVeda, first mandala, first shukta, and first mantra?
+    '''
+    spec_functions = ["get_translation", "get_vedamantra_details", "get_vedamantra_summary"]
+
+    TRANSLATION_CSV_PATH = 'Data/trans_Rig_Ath_index_v2.csv'
+    VEDAMANTRA_CSV_PATH = "Data/veda_content_modified_v3.csv"
+
+    def __init__(self):
+        super().__init__()
+        self.df_translation = pd.read_csv(self.TRANSLATION_CSV_PATH, encoding='utf-8')
+        self.df_vedamantra = pd.read_csv(self.VEDAMANTRA_CSV_PATH, encoding='utf-8')
+
+    @st.cache_data
+    def get_translation(_self, mantraid=None, scripture_name=None, MahatmaName=None, KandahNumber=None,
+                        MandalaNumber=None, ArchikahNumber=None, ShuktaNumber=None,
+                        AnvayaNumber=None, PrapatakNumber=None, MantraNumber=None,
+                        AnuvakNumber=None, AdhyayaNumber=None):
+        """
+        Get the translation of mantras from RigVeda and AtharvaVeda.
+        Sample Query:
+        1. What is the translation of Tulsi Ram of the vedamantra from Rigveda, first mandala, first shukta, and first mantra?
+        2. What is the translation or adhibautic meaning of the vedamantra from RigVeda, first mandala, first shukta, and first mantra?
+        3. What is the subject of the mantra 1.1.84.1?
+        """
+        try:
+            if mantraid is None:
+                scripture_name_lower = scripture_name.lower() if scripture_name is not None else False
+                if scripture_name_lower == 'rigveda':
+                    details = _self.df_translation[
+                        (_self.df_translation['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_translation['MandalaNumber'] == MandalaNumber)
+                        & (_self.df_translation['ShuktaNumber'] == int(ShuktaNumber))
+                        & (_self.df_translation['MantraNumber'] == int(MantraNumber))
+                    ].to_dict(orient='records')
+                elif scripture_name_lower == 'atharvaveda':
+                    details = _self.df_translation[
+                        (_self.df_translation['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_translation['KandahNumber'] == KandahNumber)
+                        & (_self.df_translation['ShuktaNumber'] == ShuktaNumber)
+                        & (_self.df_translation['MantraNumber'] == MantraNumber)].to_dict(orient='records')
+                elif scripture_name_lower == 'samaveda':
+                    details = _self.df_translation[
+                        (_self.df_translation['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_translation['ArchikahNumber'] == ArchikahNumber)
+                        & (_self.df_translation['ShuktaNumber'] == ShuktaNumber)
+                        & (_self.df_translation['MantraNumber'] == MantraNumber)].to_dict(orient='records')
+                elif scripture_name_lower == 'krishnayajurveda':
+                    details = _self.df_translation[
+                        (_self.df_translation['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_translation['PrapatakNumber'] == PrapatakNumber)
+                        & (_self.df_translation['AnuvakNumber'] == AnuvakNumber)
+                        & (_self.df_translation['MantraNumber'] == MantraNumber)].to_dict(orient='records')
+                else:
+                    details = _self.df_translation[
+                        (_self.df_translation['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_translation['AdhyayaNumber'] == AdhyayaNumber)
+                        & (_self.df_translation['MantraNumber'] == MantraNumber)
+                    ].to_dict(orient='records')
+            else:
+                details = _self.df_translation[_self.df_translation['mantra_id'] == mantraid].to_dict(orient='records')
+
+            if MahatmaName is not None:
+                for item in details:
+                    if item['MahatmaName'] == MahatmaName:
+                        return item
+            else:
+                return details
+        except Exception as e:
+            return json.dumps({"error": f"Failed to get translation. {e}"})
+
+    @st.cache_data
+    def get_vedamantra_details(_self, mantraid=None, scripture_name=None, KandahNumber=None,
+                               MandalaNumber=None, ArchikahNumber=None, ShuktaNumber=None,
+                               AnvayaNumber=None, PrapatakNumber=None, MantraNumber=None,
+                               AnuvakNumber=None, AdhyayaNumber=None):
+        """
+        To obtain the vedamantra details such as vedamantra, padapata, devata, rishi, swarah, and chandah.
+        1. What is the vedamantra of the mantra from Rigveda, first mandala, first shukta, and first mantra?
+        2. What is the devata of the vedamantra from Rigveda, first mandala, first shukta, and first mantra?
+        """
+        try:
+            if mantraid is None:
+                scripture_name_lower = scripture_name.lower() if scripture_name is not None else False
+
+                if scripture_name_lower == 'rigveda':
+                    conditions = (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower) & \
+                                  (_self.df_vedamantra['MandalaNumber'] == MandalaNumber) & \
+                                  (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber) & \
+                                  (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    details = _self.df_vedamantra[conditions]['mantra_json'].values
+                    vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]['mandala']['shukta']['mantra']
+
+                elif scripture_name_lower == 'atharvaveda':
+                    conditions = (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower) & \
+                                  (_self.df_vedamantra['KandahNumber'] == KandahNumber) & \
+                                  (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber) & \
+                                  (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    details = _self.df_vedamantra[conditions]['mantra_json'].values
+                    vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]['kandah']['shukta']['mantra']
+                elif scripture_name_lower == 'samaveda':
+                    conditions = (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower) & \
+                                  (_self.df_vedamantra['ArchikahNumber'] == ArchikahNumber) & \
+                                  (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber) & \
+                                  (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    details = _self.df_vedamantra[conditions]['mantra_json'].values
+                    vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]['archikah']
+                elif scripture_name_lower == 'krishnayajurveda':
+                    conditions = (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower) & \
+                                  (_self.df_vedamantra['PrapatakNumber'] == PrapatakNumber) & \
+                                  (_self.df_vedamantra['AnuvakNumber'] == AnuvakNumber) & \
+                                  (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    details = _self.df_vedamantra[conditions]['mantra_json'].values
+                    vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]['kandah']['prapatak']['anuvak']
+                else:
+                    conditions = (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower) & \
+                                  (_self.df_vedamantra['AdhyayaNumber'] == AdhyayaNumber) & \
+                                  (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    details = _self.df_vedamantra[conditions]['mantra_json'].values
+                    vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]['adhyaya']['mantra']
+
+            else:
+                # Handle case when mantraid is provided
+                details = _self.df_vedamantra[_self.df_vedamantra['mantra_number'] == mantraid]['mantra_json'].values
+                vedamantra_details = json.loads(details[0])['mantraHeader']['language'][1]
+
+            return vedamantra_details
+        except Exception as e:
+            return json.dumps({"error": f"Failed to get vedamantra details. {str(e)}"})
+
+    @st.cache_data
+    def get_vedamantra_summary(_self, mantraid=None, scripture_name=None, KandahNumber=None,
+                               MandalaNumber=None, ArchikahNumber=None, ShuktaNumber=None,
+                               AnvayaNumber=None, PrapatakNumber=None, MantraNumber=None,
+                               AnuvakNumber=None, AdhyayaNumber=None):
+        '''
+        To obtain the vedamantra summary like anvaya, translation, adhibautic, adhyatmic, adhidaivic meaning of the mantra.
+        1. What is the adhibautic meaning of the mantra from AtharvaVeda, first kandah, first shukta, and first mantra?
+        2. What is the anvaya of the vedamantra from Rigveda, first mandala, first shukta, and first mantra?
+        '''
+        try:
+            if mantraid is None:
+                scripture_name_lower = scripture_name.lower() if scripture_name is not None else False
+                if scripture_name_lower == 'rigveda':
+                    details = _self.df_vedamantra[
+                        (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_vedamantra['MandalaNumber'] == MandalaNumber)
+                        & (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber)
+                        & (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    ]['mantra_json'].values
+                elif scripture_name_lower == 'atharvaveda':
+                    details = _self.df_vedamantra[
+                        (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_vedamantra['KandahNumber'] == KandahNumber)
+                        & (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber)
+                        & (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    ]['mantra_json'].values
+                elif scripture_name_lower == 'samaveda':
+                    details = _self.df_vedamantra[
+                        (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_vedamantra['ArchikahNumber'] == ArchikahNumber)
+                        & (_self.df_vedamantra['ShuktaNumber'] == ShuktaNumber)
+                        & (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    ]['mantra_json'].values
+                elif scripture_name_lower == 'krishnayajurveda':
+                    details = _self.df_vedamantra[
+                        (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_vedamantra['PrapatakNumber'] == PrapatakNumber)
+                        & (_self.df_vedamantra['AnuvakNumber'] == AnuvakNumber)
+                        & (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    ]['mantra_json'].values
+                else:
+                    details = _self.df_vedamantra[
+                        (_self.df_vedamantra['scripture_name'].str.lower() == scripture_name_lower)
+                        & (_self.df_vedamantra['AdhyayaNumber'] == AdhyayaNumber)
+                        & (_self.df_vedamantra['MantraNumber'] == str(MantraNumber))
+                    ]['mantra_json'].values
+            else:
+                details = _self.df_vedamantra[_self.df_vedamantra['mantra_number'] == mantraid]['mantra_json'].values
+
+            jsonDict = json.loads(details[0])
+            mantraSummary = jsonDict['mantraSummary']['language']
+            mantraSummary_IAST = jsonDict['mantraSummary']['language'][1]
+            vedamantra_summary = {"Roman-IAST summary of vedamantra": mantraSummary_IAST}
+            for item in mantraSummary:
+                if item['languageName'] == 'English':
+                    vedamantra_summary.update({"English summary of vedamantra": item})
+            return vedamantra_summary
+        except Exception as e:
+            return json.dumps({"error": f"Failed to get vedamantra summary. {e}"})

src/app.py

@@ -0,0 +1,149 @@
+import streamlit as st
+st.set_page_config(
+    page_title="SVARUPA AI",
+    layout="centered",  # or "wide"
+    initial_sidebar_state="auto"  # or "expanded" or "collapsed"
+    )
+from llama_index.core import VectorStoreIndex, StorageContext, Document
+from llama_index.llms.openai import OpenAI
+import os
+import pandas as pd
+from llama_index.core import Settings
+from llama_index.vector_stores.pinecone import PineconeVectorStore
+import pinecone
+from pinecone import Pinecone, PodSpec
+from llama_index.core.query_engine import PandasQueryEngine
+from llama_index.core.agent import ReActAgent
+from llama_index.core.memory import ChatMemoryBuffer
+from sentence_transformers import SentenceTransformer
+from llama_index.embeddings.huggingface import HuggingFaceEmbedding
+#from llama_index.indices.postprocessor import SimilarityPostprocessor
+#from llama_index.postprocessor import SentenceTransformerRerank
+import tiktoken
+from llama_index.core.callbacks import CallbackManager, TokenCountingHandler
+from llama_index.core.tools import QueryEngineTool, ToolMetadata
+from FunctionTools import ScriptureDescriptionToolSpec, MantraToolSpec
+
+
+#load keys
+openai_api_key = st.secrets["OPENAI_APIKEY_CS"]
+pinecone_api_key = st.secrets["PINECONE_API_KEY_SAM"]
+
+#llm
+llm_AI4 = OpenAI(temperature=0, model="gpt-4-1106-preview",api_key=openai_api_key, max_tokens=512)
+token_counter = TokenCountingHandler(
+    tokenizer=tiktoken.encoding_for_model("gpt-4-1106-preview").encode
+    )
+
+# global settings
+Settings.embed_model = HuggingFaceEmbedding(
+    model_name="BAAI/bge-large-en-v1.5",
+    embed_batch_size=8
+)
+Settings.llm = llm_AI4 
+Settings.chunk_size = 512
+Settings.chunk_overlap = 50
+Settings.callback_manager = CallbackManager([token_counter])
+#memory for bot
+memory = ChatMemoryBuffer.from_defaults(token_limit=3900)
+
+#load vector database
+pc = Pinecone(api_key=pinecone_api_key)
+pinecone_index = pc.Index("pod-index")
+vector_store_pine = PineconeVectorStore(pinecone_index=pinecone_index)
+storage_context_pine = StorageContext.from_defaults(vector_store=vector_store_pine)
+index_store = VectorStoreIndex.from_vector_store(vector_store_pine,storage_context=storage_context_pine)
+query_engine_vector = index_store.as_query_engine(similarity_top_k=5,vector_store_query_mode ='hybrid',alpha=0.6)
+#pandas Engine
+df_veda_details = pd.read_csv("Data/veda_content_details.csv",encoding='utf-8')
+query_engine_pandas = PandasQueryEngine(df=df_veda_details)
+
+# Query Engine Tools
+query_engine_tools = [
+    QueryEngineTool(
+        query_engine=query_engine_vector,
+        metadata=ToolMetadata(
+            name="vector_engine",
+            description=(
+                '''Helpful to get semantic information from the documents. These documents containing comprehensive information about the Vedas.\
+                They also covers various aspects, including general details about the Vedas, fundamental terminology associated with Vedic literature, \
+                and detailed information about Vedamantras for each Veda. The Vedamantra details encompass essential elements such as padapatha, rishi, chandah,\
+                devata, and swarah.This tool is very useful to answer general questions related to vedas.\
+                Sample Query:\
+                1. What is the meaning of devata ?\
+                2. What are the different Brahmanas associated with SamaVeda?\
+                3. What is the difference between Shruti and Smriti.
+               '''
+            ),
+        ),
+    ),
+    QueryEngineTool(
+        query_engine=query_engine_pandas,
+        metadata=ToolMetadata(
+            name="pandas_engine",
+            description=(
+                '''Helpful to answer the queries related to count from the documents. This document is a .csv file with different columns containing comprehensive information about the Vedas.\
+                The column names as follows:\
+                'mantra_id', 'scripture_name', 'KandahNumber', 'PrapatakNumber','AnuvakNumber', 'MantraNumber', 'DevataName', 'RishiName', 'SwarahName', 'ChandaName',\
+                'padapatha', 'vedamantra', 'AdhyayaNumber', 'ArchikahNumber', 'ArchikahName', 'ShuktaNumber', 'keyShukta', 'ParyayaNumber', 'MandalaNumber'
+                ''This tool is very useful to answer questions related to vedas on.\
+                Sample Query:\
+                1. How many mantras are there in RigVeda whose swarah is gāndhāraḥ?\
+                2. How many different devata present in rigveda?\
+                3. Which Kandah has the maximum number of in KrishnaYajurVeda?
+                4. How many mantras are there in RigVeda?
+               '''
+            ),
+        ),
+    )
+    ]
+
+# tools
+mantra_tools = MantraToolSpec().to_tool_list()
+description_tools = ScriptureDescriptionToolSpec().to_tool_list()
+tools = [*mantra_tools,*description_tools,*query_engine_tools]
+
+# context
+context = """
+  You are an expert on Vedas and related scriptures.\
+  Your role is to respond to questions about vedic scriptures and associated information based on available sources.\
+  For every query, you must use either any one of the tool or use available history/context.
+  Please provide well-informed answers. Don't use prior knowledge.
+"""
+
+# Function to create ReActAgent instance (change it based on your initialization logic)
+@st.cache_resource(show_spinner=False)  # Set allow_output_mutation to True for mutable objects like instances
+def create_react_agent():
+    return ReActAgent.from_tools(tools, llm=llm_AI4, context=context, verbose=True)
+
+# Example usage
+react_agent_instance = create_react_agent()
+
+# Streamlit Components Initialization
+st.title("Svarupa Bot ")
+
+if "messages" not in st.session_state.keys():
+    st.session_state.messages = [
+        {"role": "assistant", "content": "Hi. I am Svarupa AI Assistant. Ask me a question about Vedas!"}
+    ]
+
+if "chat_engine" not in st.session_state.keys():
+    # Using st.cache_resource for caching the unserializable react_agent
+    st.session_state.chat_engine = create_react_agent()
+
+if prompt := st.chat_input("Your question"):
+    st.session_state.messages.append({"role": "user", "content": prompt})
+
+for message in st.session_state.messages:
+    with st.chat_message(message["role"]):
+        st.write(message["content"])
+
+if st.session_state.messages[-1]["role"] != "assistant":
+    with st.chat_message("assistant"):
+        with st.spinner("Thinking..."):
+            # Using the cached chat_engine
+            response = st.session_state.chat_engine.chat(prompt)
+            st.write(response.response)
+            message = {"role": "assistant", "content": response.response}
+            st.session_state.messages.append(message)
+