Upload folder using huggingface_hub

.gitignore

@@ -11,3 +11,4 @@ wheels/
 .env
 chromadb-store/
 chromadb-store.zip
+outputs/

README.md

@@ -5,3 +5,20 @@ sdk: gradio
 sdk_version: 5.38.0
 python_version: 3.12
 ---
+
+### Introduction
+This is an Agentic-AI project that integrates all Hindu Sanatan Dharma scriptures into a single searchable platform. 
+
+### Supported Channels
+- Web (https://huggingface.co/spaces/vikramvasudevan/sanatan_ai)
+- Android (bhashyam.ai app)
+
+### Start Web Server
+- Run the following command from project root
+> `uv run ./main.py`
+
+### Automated AI Evaluator
+- Tests are defined in tests/test_config.py
+- Run the following command from project root to execute the tests.
+> `uv run -m tests.test_evaluator`
+- Test Logs are generated under `{project-root}/outputs/tests` folder as neatly formatted md files.

app.py

@@ -13,6 +13,7 @@ from langchain_core.messages.ai import AIMessageChunk, AIMessage
 from langchain_core.messages.system import SystemMessage
 from langchain_core.messages.tool import ToolMessage
 
+from chat_utils import chat_wrapper
 from config import SanatanConfig
 from db import SanatanDatabase
 from drive_downloader import ZipDownloader
@@ -63,277 +64,6 @@ def render_message_with_tooltip(content: str, max_chars=200):
     return f"<div title='{escape(content)}'>{short}</div>"
 
 
-thinking_verbs = [
-    "thinking",
-    "processing",
-    "crunching data",
-    "please wait",
-    "just a few more seconds",
-    "closing in",
-    "analyzing",
-    "reasoning",
-    "computing",
-    "synthesizing insight",
-    "searching through the cosmos",
-    "decoding ancient knowledge",
-    "scanning the scriptures",
-    "accessing divine memory",
-    "gathering wisdom",
-    "consulting the rishis",
-    "listening to the ātmā",
-    "channeling sacred energy",
-    "unfolding the divine word",
-    "meditating on the meaning",
-    "reciting from memory",
-    "traversing the Vedas",
-    "seeking the inner light",
-    "invoking paramārtha",
-    "putting it all together",
-    "digging deeper",
-    "making sense of it",
-    "connecting the dots",
-    "almost there",
-    "getting closer",
-    "wrapping it up",
-    "piecing it together",
-    "swirling through verses",
-    "diving into the ocean of knowledge",
-    "lighting the lamp of understanding",
-    "walking the path of inquiry",
-    "aligning stars of context",
-]
-
-
-async def chat_wrapper(
-    message, history, thread_id, debug, preferred_language="English"
-):
-    if debug:
-        async for chunk in chat_streaming(
-            debug, message, history, thread_id, preferred_language=preferred_language
-        ):
-            yield chunk
-    else:
-        response = chat(
-            debug, message, history, thread_id, preferred_language=preferred_language
-        )
-        yield response
-
-
-def chat(debug_mode, message, history, thread_id, preferred_language="English"):
-    config = {"configurable": {"thread_id": thread_id}}
-    response = graph.invoke(
-        {
-            "debug_mode": debug_mode,
-            "messages": [{"role": "user", "content": message}],
-            "language": preferred_language,
-        },
-        config=config,
-    )
-    return response["messages"][-1].content
-
-
-def add_node_to_tree(
-    node_tree: list[str], node_label: str, tooltip: str = "no arguments to show"
-) -> list[str]:
-    if tooltip:
-        tooltip = escape(tooltip).replace("'", "&apos;")
-        node_with_tooltip = (
-            f"<span class='node-label' title='{tooltip}'>{node_label}</span>"
-        )
-    else:
-        node_with_tooltip = node_label
-    node_tree[-1] = node_with_tooltip
-    node_tree.append("<span class='spinner'>&nbsp;</span>")
-    return node_tree
-
-
-def end_node_tree(node_tree: list[str]) -> list[str]:
-    node_tree[-1] = "🏁"
-    return node_tree
-
-
-def get_args_for_toolcall(tool_calls_buffer: dict, tool_call_id: str):
-    return (
-        tool_calls_buffer[tool_call_id]["args_str"]
-        if tool_call_id in tool_calls_buffer
-        and "args_str" in tool_calls_buffer[tool_call_id]
-        else ""
-    )
-
-
-async def chat_streaming(
-    debug_mode: bool, message, history, thread_id, preferred_language="English"
-):
-    state = {
-        "debug_mode": debug_mode,
-        "messages": (history or []) + [{"role": "user", "content": message}],
-        "language": preferred_language,
-    }
-    config = {"configurable": {"thread_id": thread_id}, "recursion_limit": 15}
-    start_time = time.time()
-    streamed_response = ""
-    final_response = ""
-    # final_node = "validator"
-
-    MAX_CONTENT = 500
-
-    try:
-        node_tree = ["🚩", "<span class='spinner'>&nbsp;</span>"]
-
-        tool_calls_buffer = {}
-
-        async for msg, metadata in graph.astream(
-            state, config=config, stream_mode="messages"
-        ):
-            node = metadata.get("langgraph_node", "?")
-            name = getattr(msg, "name", "-")
-            if not isinstance(msg, ToolMessage):
-                node_icon = "🧠"
-            else:
-                node_icon = "⚙️"
-            node_label = f"{node}"
-            tool_label = f"{name or ''}"
-            if tool_label:
-                node_label = node_label + f":{tool_label}"
-            label = f"{node_icon} {node_label}"
-            tooltip = ""
-            if isinstance(msg, ToolMessage):
-                tooltip = get_args_for_toolcall(tool_calls_buffer, msg.tool_call_id)
-                # logger.info("tooltip = ", tooltip)
-
-            # checking for -2 last but one. since last entry is always a spinner
-            if node_tree[-2] != label:
-                add_node_to_tree(node_tree, label, tooltip)
-            full: str = escape(msg.content)
-            truncated = (full[:MAX_CONTENT] + "…") if len(full) > MAX_CONTENT else full
-
-            def generate_processing_message():
-                return f"<div class='thinking-bubble'><em>🤔{random.choice(thinking_verbs)} ...</em></div>"
-
-            if (
-                not isinstance(msg, ToolMessage)
-                and not isinstance(msg, SystemMessage)
-                and not isinstance(msg, AIMessageChunk)
-            ):
-                logger.info("msg = %s", msg)
-            if isinstance(msg, ToolMessage):
-                logger.debug("tool message = %s", msg)
-
-                html = f"<div class='thinking-bubble'><em>🤔 {msg.name} tool: {random.choice(thinking_verbs)} ...</em></div>"
-                yield f"### { ' → '.join(node_tree)}\n{html}"
-            elif isinstance(msg, AIMessageChunk):
-
-                def truncate_middle(text, front=50, back=50):
-                    if not text:
-                        return ""
-                    if len(text) <= front + back:
-                        return text
-                    return f"{text[:front]}…{text[-back:]}".replace(
-                        "\n", ""
-                    )  # remove new lines.
-
-                if not msg.content:
-                    # logger.warning("*** No Message Chunk!")
-                    yield f"### { " → ".join(node_tree)}\n{generate_processing_message()}\n<div class='intermediate-output'>{escape(truncate_middle(streamed_response))}</div>"
-                else:
-                    # Stream intermediate messages with transparent style
-                    # if node != final_node:
-                    streamed_response += msg.content
-                    yield f"### { ' → '.join(node_tree) }\n<div class='intermediate-output'>{escape(truncate_middle(streamed_response))}</div>"
-                    # else:
-                    # Buffer the final validated response instead of yielding
-                    final_response += msg.content
-
-                if msg.tool_call_chunks:
-                    for tool_call_chunk in msg.tool_call_chunks:
-                        logger.debug("*** tool_call_chunk = ", tool_call_chunk)
-                        if tool_call_chunk["id"] is not None:
-                            tool_call_id = tool_call_chunk["id"]
-
-                        if tool_call_id not in tool_calls_buffer:
-                            tool_calls_buffer[tool_call_id] = {
-                                "name": "",
-                                "args_str": "",
-                                "id": tool_call_id,
-                                "type": "tool_call",
-                            }
-
-                        # Accumulate tool call name and arguments
-                        if tool_call_chunk["name"] is not None:
-                            tool_calls_buffer[tool_call_id]["name"] += tool_call_chunk[
-                                "name"
-                            ]
-                        if tool_call_chunk["args"] is not None:
-                            tool_calls_buffer[tool_call_id][
-                                "args_str"
-                            ] += tool_call_chunk["args"]
-            else:
-                logger.debug("message = ", type(msg), msg.content[:100])
-                full: str = escape(msg.content)
-                truncated = (
-                    (full[:MAX_CONTENT] + "…") if len(full) > MAX_CONTENT else full
-                )
-                html = (
-                    f"<div class='thinking-bubble'><em>🤔 {random.choice(thinking_verbs)} ...</em></div>"
-                    f"<div style='opacity: 0.1'>"
-                    f"<strong>Telling myself:</strong> {truncated or '...'}"
-                    f"</div>"
-                )
-                yield f"### { " → ".join(node_tree)}\n{html}"
-                if getattr(msg, "tool_calls", []):
-                    logger.info("ELSE::tool_calls = %s", msg.tool_calls)
-
-        node_tree[-1] = "✅"
-        end_time = time.time()
-        duration = end_time - start_time
-        final_response = (
-            f"\n{final_response}" f"\n\n⏱️ Processed in {duration:.2f} seconds"
-        )
-        buffer = f"### {' → '.join(node_tree)}\n"
-        yield buffer
-        for c in final_response:
-            buffer += c
-            yield buffer
-            await asyncio.sleep(0.0005)
-
-        logger.debug("************************************")
-        # Now, you can process the complete tool calls from the buffer
-        for tool_call_id, accumulated_tool_call in tool_calls_buffer.items():
-            # Attempt to parse arguments only if the 'args_str' isn't empty
-            if accumulated_tool_call["args_str"]:
-                try:
-                    parsed_args = json.loads(accumulated_tool_call["args_str"])
-                    logger.debug(f"Tool Name: {accumulated_tool_call['name']}")
-                    logger.debug(f"Tool Arguments: {parsed_args}")
-                except json.JSONDecodeError:
-                    logger.debug(
-                        f"Partial arguments for tool {accumulated_tool_call['name']}: {accumulated_tool_call['args_str']}"
-                    )
-    except asyncio.CancelledError:
-        logger.warning("⚠️ Request cancelled by user")
-        node_tree = end_node_tree(node_tree=node_tree)
-        yield (
-            f"### {' → '.join(node_tree)}"
-            "\n⚠️⚠️⚠️ Request cancelled by user"
-            "\nhere is what I got so far ...\n"
-            f"\n{streamed_response}"
-        )
-        # Important: re-raise if you want upstream to also know
-        # raise
-        return
-    except Exception as e:
-        logger.error("❌❌❌ Error processing request: %s", e)
-        traceback.print_exc()
-        node_tree = end_node_tree(node_tree=node_tree)
-        yield (
-            f"### { " → ".join(node_tree)}"
-            f"\n❌❌❌ Error processing request : {str(e)}"
-            "\nhere is what I got so far ...\n"
-            f"\n{streamed_response}"
-        )
-        return
-
-
 # UI Elements
 thread_id = gr.State(init_session)
 

chat_utils.py

@@ -0,0 +1,287 @@
+import json
+import random
+import asyncio
+import logging
+import time
+import traceback
+from html import escape
+from langchain_core.messages.ai import AIMessageChunk
+from langchain_core.messages.system import SystemMessage
+from langchain_core.messages.tool import ToolMessage
+
+from graph_helper import generate_graph
+
+# Logging
+logging.basicConfig()
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+thinking_verbs = [
+    "thinking",
+    "processing",
+    "crunching data",
+    "please wait",
+    "just a few more seconds",
+    "closing in",
+    "analyzing",
+    "reasoning",
+    "computing",
+    "synthesizing insight",
+    "searching through the cosmos",
+    "decoding ancient knowledge",
+    "scanning the scriptures",
+    "accessing divine memory",
+    "gathering wisdom",
+    "consulting the rishis",
+    "listening to the ātmā",
+    "channeling sacred energy",
+    "unfolding the divine word",
+    "meditating on the meaning",
+    "reciting from memory",
+    "traversing the Vedas",
+    "seeking the inner light",
+    "invoking paramārtha",
+    "putting it all together",
+    "digging deeper",
+    "making sense of it",
+    "connecting the dots",
+    "almost there",
+    "getting closer",
+    "wrapping it up",
+    "piecing it together",
+    "swirling through verses",
+    "diving into the ocean of knowledge",
+    "lighting the lamp of understanding",
+    "walking the path of inquiry",
+    "aligning stars of context",
+]
+
+graph = generate_graph()
+
+def add_node_to_tree(
+    node_tree: list[str], node_label: str, tooltip: str = "no arguments to show"
+) -> list[str]:
+    if tooltip:
+        tooltip = escape(tooltip).replace("'", "'")
+        node_with_tooltip = (
+            f"<span class='node-label' title='{tooltip}'>{node_label}</span>"
+        )
+    else:
+        node_with_tooltip = node_label
+    node_tree[-1] = node_with_tooltip
+    node_tree.append("<span class='spinner'>&nbsp;</span>")
+    return node_tree
+
+
+def end_node_tree(node_tree: list[str]) -> list[str]:
+    node_tree[-1] = "🏁"
+    return node_tree
+
+
+def get_args_for_toolcall(tool_calls_buffer: dict, tool_call_id: str):
+    return (
+        tool_calls_buffer[tool_call_id]["args_str"]
+        if tool_call_id in tool_calls_buffer
+        and "args_str" in tool_calls_buffer[tool_call_id]
+        else ""
+    )
+
+
+async def chat_wrapper(
+    message, history, thread_id, debug, preferred_language="English"
+):
+    if debug:
+        async for chunk in chat_streaming(
+            debug, message, history, thread_id, preferred_language=preferred_language
+        ):
+            yield chunk
+    else:
+        response = chat(
+            debug, message, history, thread_id, preferred_language=preferred_language
+        )
+        yield response
+
+
+def chat(debug_mode, message, history, thread_id, preferred_language="English"):
+    config = {"configurable": {"thread_id": thread_id}, "recursion_limit": 30}
+    response = graph.invoke(
+        {
+            "debug_mode": debug_mode,
+            "messages": [{"role": "user", "content": message}],
+            "language": preferred_language,
+        },
+        config=config,
+    )
+    return response["messages"][-1].content
+
+async def chat_streaming(
+    debug_mode: bool, message, history, thread_id, preferred_language="English"
+):
+    state = {
+        "debug_mode": debug_mode,
+        "messages": (history or []) + [{"role": "user", "content": message}],
+        "language": preferred_language,
+    }
+    config = {"configurable": {"thread_id": thread_id}, "recursion_limit": 30}
+    start_time = time.time()
+    streamed_response = ""
+    final_response = ""
+    # final_node = "validator"
+
+    MAX_CONTENT = 500
+
+    try:
+        node_tree = ["🚩", "<span class='spinner'>&nbsp;</span>"]
+
+        tool_calls_buffer = {}
+
+        async for msg, metadata in graph.astream(
+            state, config=config, stream_mode="messages"
+        ):
+            node = metadata.get("langgraph_node", "?")
+            name = getattr(msg, "name", "-")
+            if not isinstance(msg, ToolMessage):
+                node_icon = "🧠"
+            else:
+                node_icon = "⚙️"
+            node_label = f"{node}"
+            tool_label = f"{name or ''}"
+            if tool_label:
+                node_label = node_label + f":{tool_label}"
+            label = f"{node_icon} {node_label}"
+            tooltip = ""
+            if isinstance(msg, ToolMessage):
+                tooltip = get_args_for_toolcall(tool_calls_buffer, msg.tool_call_id)
+                # logger.info("tooltip = ", tooltip)
+
+            # checking for -2 last but one. since last entry is always a spinner
+            if node_tree[-2] != label:
+                add_node_to_tree(node_tree, label, tooltip)
+            full: str = escape(msg.content)
+            truncated = (full[:MAX_CONTENT] + "…") if len(full) > MAX_CONTENT else full
+
+            def generate_processing_message():
+                return f"<div class='thinking-bubble'><em>🤔{random.choice(thinking_verbs)} ...</em></div>"
+
+            if (
+                not isinstance(msg, ToolMessage)
+                and not isinstance(msg, SystemMessage)
+                and not isinstance(msg, AIMessageChunk)
+            ):
+                logger.info("msg = %s", msg)
+            if isinstance(msg, ToolMessage):
+                logger.debug("tool message = %s", msg)
+
+                html = f"<div class='thinking-bubble'><em>🤔 {msg.name} tool: {random.choice(thinking_verbs)} ...</em></div>"
+                yield f"### { ' → '.join(node_tree)}\n{html}"
+            elif isinstance(msg, AIMessageChunk):
+
+                def truncate_middle(text, front=50, back=50):
+                    if not text:
+                        return ""
+                    if len(text) <= front + back:
+                        return text
+                    return f"{text[:front]}…{text[-back:]}".replace(
+                        "\n", ""
+                    )  # remove new lines.
+
+                if not msg.content:
+                    # logger.warning("*** No Message Chunk!")
+                    yield f"### { " → ".join(node_tree)}\n{generate_processing_message()}\n<div class='intermediate-output'>{escape(truncate_middle(streamed_response))}</div>"
+                else:
+                    # Stream intermediate messages with transparent style
+                    # if node != final_node:
+                    streamed_response += msg.content
+                    yield f"### { ' → '.join(node_tree) }\n<div class='intermediate-output'>{escape(truncate_middle(streamed_response))}</div>"
+                    # else:
+                    # Buffer the final validated response instead of yielding
+                    final_response += msg.content
+
+                if msg.tool_call_chunks:
+                    for tool_call_chunk in msg.tool_call_chunks:
+                        logger.debug("*** tool_call_chunk = ", tool_call_chunk)
+                        if tool_call_chunk["id"] is not None:
+                            tool_call_id = tool_call_chunk["id"]
+
+                        if tool_call_id not in tool_calls_buffer:
+                            tool_calls_buffer[tool_call_id] = {
+                                "name": "",
+                                "args_str": "",
+                                "id": tool_call_id,
+                                "type": "tool_call",
+                            }
+
+                        # Accumulate tool call name and arguments
+                        if tool_call_chunk["name"] is not None:
+                            tool_calls_buffer[tool_call_id]["name"] += tool_call_chunk[
+                                "name"
+                            ]
+                        if tool_call_chunk["args"] is not None:
+                            tool_calls_buffer[tool_call_id][
+                                "args_str"
+                            ] += tool_call_chunk["args"]
+            else:
+                logger.debug("message = ", type(msg), msg.content[:100])
+                full: str = escape(msg.content)
+                truncated = (
+                    (full[:MAX_CONTENT] + "…") if len(full) > MAX_CONTENT else full
+                )
+                html = (
+                    f"<div class='thinking-bubble'><em>🤔 {random.choice(thinking_verbs)} ...</em></div>"
+                    f"<div style='opacity: 0.1'>"
+                    f"<strong>Telling myself:</strong> {truncated or '...'}"
+                    f"</div>"
+                )
+                yield f"### { " → ".join(node_tree)}\n{html}"
+                if getattr(msg, "tool_calls", []):
+                    logger.info("ELSE::tool_calls = %s", msg.tool_calls)
+
+        node_tree[-1] = "✅"
+        end_time = time.time()
+        duration = end_time - start_time
+        final_response = (
+            f"\n{final_response}" f"\n\n⏱️ Processed in {duration:.2f} seconds"
+        )
+        buffer = f"### {' → '.join(node_tree)}\n"
+        yield buffer
+        for c in final_response:
+            buffer += c
+            yield buffer
+            await asyncio.sleep(0.0005)
+
+        logger.debug("************************************")
+        # Now, you can process the complete tool calls from the buffer
+        for tool_call_id, accumulated_tool_call in tool_calls_buffer.items():
+            # Attempt to parse arguments only if the 'args_str' isn't empty
+            if accumulated_tool_call["args_str"]:
+                try:
+                    parsed_args = json.loads(accumulated_tool_call["args_str"])
+                    logger.debug(f"Tool Name: {accumulated_tool_call['name']}")
+                    logger.debug(f"Tool Arguments: {parsed_args}")
+                except json.JSONDecodeError:
+                    logger.debug(
+                        f"Partial arguments for tool {accumulated_tool_call['name']}: {accumulated_tool_call['args_str']}"
+                    )
+    except asyncio.CancelledError:
+        logger.warning("⚠️ Request cancelled by user")
+        node_tree = end_node_tree(node_tree=node_tree)
+        yield (
+            f"### {' → '.join(node_tree)}"
+            "\n⚠️⚠️⚠️ Request cancelled by user"
+            "\nhere is what I got so far ...\n"
+            f"\n{streamed_response}"
+        )
+        # Important: re-raise if you want upstream to also know
+        # raise
+        return
+    except Exception as e:
+        logger.error("❌❌❌ Error processing request: %s", e)
+        traceback.print_exc()
+        node_tree = end_node_tree(node_tree=node_tree)
+        yield (
+            f"### { " → ".join(node_tree)}"
+            f"\n❌❌❌ Error processing request : {str(e)}"
+            "\nhere is what I got so far ...\n"
+            f"\n{streamed_response}"
+        )
+        return

config.py

@@ -246,9 +246,10 @@ class SanatanConfig:
                 {
                     "name": "verse",
                     "datatype": "int",
+                    "is_unique" : True,
                     "description": (
-                        "Absolute verse number or pasuram number."
-                        "Use it only when a specific prabandham name is NOT mentioned in the user query."
+                        "Absolute verse number or pasuram number. Each verse has a unique number."
+                        # "Use it only when a specific prabandham name is NOT mentioned in the user query."
                         "For e.g. 'Give me pasuram 1176'"
                     ),
                 },
@@ -539,13 +540,23 @@ class SanatanConfig:
         self, collection_name: str, metadata_where_clause: MetadataWhereClause
     ):
         scripture = self.get_scripture_by_collection(collection_name=collection_name)
-        for filter in metadata_where_clause.filters:
-            if filter.metadata_field not in [
-                field["name"] for field in scripture["metadata_fields"]
-            ]:
-                raise Exception(
-                    f"metadata_field: [{filter.metadata_field}] not allowed in collection [{collection_name}]. Here are the allowed fields with their descriptions: {scripture["metadata_fields"]}"
-                )
+        allowed_fields = [field["name"] for field in scripture["metadata_fields"]]
+
+        def validate_clause(clause: MetadataWhereClause):
+            # validate direct filters
+            if clause.filters:
+                for f in clause.filters:
+                    if f.metadata_field not in allowed_fields:
+                        raise Exception(
+                            f"metadata_field: [{f.metadata_field}] not allowed in collection [{collection_name}]. "
+                            f"Here are the allowed fields with their descriptions: {scripture['metadata_fields']}"
+                        )
+            # recurse into groups
+            if clause.groups:
+                for g in clause.groups:
+                    validate_clause(g)
+
+        validate_clause(metadata_where_clause)
         return True
 
     def get_embedding_for_collection(self, collection_name: str):

db.py

@@ -1,4 +1,6 @@
 import json
+import random
+from typing import Literal
 import chromadb
 import re, unicodedata
 from config import SanatanConfig
@@ -33,18 +35,120 @@ class SanatanDatabase:
             metadatas=metadatas,
         )
 
-    def search(self, collection_name: str, query: str, n_results=2):
-        logger.info("Vector Semantic Search for [%s] in [%s]", query, collection_name)
+    def fetch_random_data(
+        self,
+        collection_name: str,
+        metadata_where_clause: MetadataWhereClause = None,
+        n_results=1,
+    ):
+        # fetch all documents once
+        logger.info(
+            "getting %d random verses from [%s] | metadata_where_clause = %s",
+            n_results,
+            collection_name,
+            metadata_where_clause,
+        )
         collection = self.chroma_client.get_or_create_collection(name=collection_name)
-        try:
-            response = collection.query(
-                query_embeddings=get_embedding(
-                    [query], SanatanConfig().get_embedding_for_collection(collection_name)
-                ),
-                # query_texts=[query],
+        data = collection.get(
+            where=(
+                metadata_where_clause.to_chroma_where()
+                if metadata_where_clause is not None
+                else None
+            )
+        )
+        docs = data["documents"]  # list of all verse texts
+        ids = data["ids"]
+        metas = data["metadatas"]
+
+        if not docs:
+            logger.warning("No data found! - data=%s", data)
+            return chromadb.QueryResult(ids=[], documents=[], metadatas=[])
+
+        # pick k random indices
+        indices = random.sample(range(len(docs)), k=min(n_results, len(docs)))
+
+        return chromadb.QueryResult(
+            ids=[ids[i] for i in indices],
+            documents=[docs[i] for i in indices],
+            metadatas=[metas[i] for i in indices],
+        )
+
+    def search(
+        self,
+        collection_name: str,
+        query: str = None,
+        metadata_where_clause: MetadataWhereClause = None,
+        n_results=2,
+        search_type: Literal["semantic", "literal", "random"] = "semantic",
+    ):
+        logger.info(
+            "Search for [%s] in [%s]| metadata_where_clause=%s | search_type=%s | n_results=%d",
+            query,
+            collection_name,
+            metadata_where_clause,
+            search_type,
+            n_results,
+        )
+        if search_type == "semantic":
+            return self.search_semantic(
+                collection_name=collection_name,
+                query=query,
+                metadata_where_clause=metadata_where_clause,
+                n_results=n_results,
+            )
+        elif search_type == "literal":
+            return self.search_for_literal(
+                collection_name=collection_name,
+                literal_to_search_for=query,
+                metadata_where_clause=metadata_where_clause,
                 n_results=n_results,
-                include=["metadatas","documents","distances"],
             )
+        else:
+            # random
+            return self.fetch_random_data(
+                collection_name=collection_name,
+                metadata_where_clause=metadata_where_clause,
+                n_results=n_results,
+            )
+
+    def search_semantic(
+        self,
+        collection_name: str,
+        query: str | None = None,
+        metadata_where_clause: MetadataWhereClause | None = None,
+        n_results=2,
+    ):
+        logger.info(
+            "Vector Semantic Search for [%s] in [%s] | metadata_where_clause = %s",
+            query,
+            collection_name,
+            metadata_where_clause,
+        )
+        collection = self.chroma_client.get_or_create_collection(name=collection_name)
+        try:
+            q = query.strip() if query is not None else ""
+            if not q:
+                # fallback: fetch random verse
+                return self.fetch_random_data(
+                    collection_name=collection_name,
+                    metadata_where_clause=metadata_where_clause,
+                    n_results=n_results,
+                )
+            else:
+                response = collection.query(
+                    query_embeddings=get_embedding(
+                        [query],
+                        SanatanConfig().get_embedding_for_collection(collection_name),
+                    ),
+                    # query_texts=[query],
+                    n_results=n_results,
+                    where=(
+                        metadata_where_clause.to_chroma_where()
+                        if metadata_where_clause is not None
+                        else None
+                    ),
+                    include=["metadatas", "documents", "distances"],
+                )
         except Exception as e:
             logger.error("Error in search: %s", e)
             return chromadb.QueryResult(
@@ -53,42 +157,71 @@ class SanatanDatabase:
                 metadatas=[],
                 distances=[],
             )
-   
+
         validated_response = validate_relevance_queryresult(query, response)
 
-        return validated_response["result"]
+        logger.info(
+            "status = %s | reason= %s",
+            validated_response.status,
+            validated_response.reason,
+        )
+
+        return validated_response.result
 
     def search_for_literal(
-        self, collection_name: str, literal_to_search_for: str, n_results=2
+        self,
+        collection_name: str,
+        literal_to_search_for: str | None = None,
+        metadata_where_clause: MetadataWhereClause | None = None,
+        n_results=2,
     ):
         logger.info(
-            "Searching literally for [%s] in [%s]",
+            "Searching literally for [%s] in [%s] | metadata_where_clause = %s",
             literal_to_search_for,
             collection_name,
+            metadata_where_clause,
         )
+        if literal_to_search_for is None or literal_to_search_for.strip() == "":
+            logger.warning("Nothing to search literally.")
+            # raise Exception("literal_to_search_for cannot be None or empty for a literal search!")
+            return self.fetch_random_data(
+                collection_name=collection_name,
+            )
         collection = self.chroma_client.get_or_create_collection(name=collection_name)
 
         def normalize(text):
             return unicodedata.normalize("NFKC", text).lower()
 
         # 1. Try native contains
-        response = collection.query(
-            query_embeddings=get_embedding(
-                [literal_to_search_for], SanatanConfig().get_embedding_for_collection(collection_name)
+        response = collection.get(
+            where=(
+                metadata_where_clause.to_chroma_where()
+                if metadata_where_clause is not None
+                else None
             ),
             where_document={"$contains": literal_to_search_for},
-            n_results=n_results,
+            limit=n_results,
         )
 
         if response["documents"] and any(response["documents"]):
-            return response
+            return chromadb.QueryResult(
+                ids=response["ids"],
+                documents=response["documents"],
+                metadatas=response["metadatas"],
+            )
 
         # 2. Regex fallback (normalized)
         logger.info("⚠ No luck. Falling back to regex for %s", literal_to_search_for)
         regex = re.compile(re.escape(normalize(literal_to_search_for)))
         logger.info("regex =  %s", regex)
 
-        all_docs = collection.get()
+        all_docs = collection.get(
+            where=(
+                metadata_where_clause.to_chroma_where()
+                if metadata_where_clause is not None
+                else None
+            ),
+        )
         matched_docs = []
 
         for doc_list, metadata_list, doc_id_list in zip(
@@ -135,36 +268,13 @@ class SanatanDatabase:
             if len(matched_docs) >= n_results:
                 break
 
-        return {
-            "documents": [[d["document"] for d in matched_docs]],
-            "ids": [[d["id"] for d in matched_docs]],
-            "metadatas": [[d["metadata"] for d in matched_docs]],
-        }
-
-    def search_by_metadata(
-        self,
-        collection_name: str,
-        query: str,
-        metadata_where_clause: MetadataWhereClause,
-        n_results=2,
-    ):
-        """Search by a metadata field inside a specific collection using a specific operator. For instance {"azhwar_name": {"$in": "Thirumangai Azhwar"}}"""
-        logger.info(
-            "Searching by metadata for [%s] in [%s] with metadata_filters=%s",
-            query,
-            collection_name,
-            metadata_where_clause,
+        return chromadb.QueryResult(
+            {
+                "documents": [[d["document"] for d in matched_docs]],
+                "ids": [[d["id"] for d in matched_docs]],
+                "metadatas": [[d["metadata"] for d in matched_docs]],
+            }
         )
-        collection = self.chroma_client.get_or_create_collection(name=collection_name)
-        response = collection.query(
-            query_embeddings=get_embedding(
-                [query], SanatanConfig().get_embedding_for_collection(collection_name)
-            ),
-            where=metadata_where_clause.to_chroma_where(),
-            # query_texts=[query],
-            n_results=n_results,
-        )
-        return response
 
     def count(self, collection_name: str):
         collection = self.chroma_client.get_or_create_collection(name=collection_name)
@@ -177,12 +287,12 @@ class SanatanDatabase:
             count = self.count(collection_name=scripture["collection_name"])
             if count == 0:
                 raise Exception(f"No data in collection {scripture["collection_name"]}")
-    
+
     def reembed_collection_openai(self, collection_name: str, batch_size: int = 50):
         """
         Deletes and recreates a Chroma collection with OpenAI text-embedding-3-large embeddings.
         All existing documents are re-embedded and inserted into the new collection.
-        
+
         Args:
             collection_name: The name of the collection to delete/recreate.
             batch_size: Number of documents to process per batch.
@@ -195,7 +305,7 @@ class SanatanDatabase:
             metadatas = old_data["metadatas"]
             ids = old_data["ids"]
             print(f"Fetched {len(documents)} documents from old collection.")
-            
+
             # Step 2: Delete old collection
             # self.chroma_client.delete_collection(collection_name)
             # print(f"Deleted old collection '{collection_name}'.")
@@ -208,13 +318,17 @@ class SanatanDatabase:
             name=f"{collection_name}_openai",
             embedding_function=None,  # embeddings will be provided manually
         )
-        print(f"Created new collection '{collection_name}_openai' with embedding_dim=3072.")
+        print(
+            f"Created new collection '{collection_name}_openai' with embedding_dim=3072."
+        )
 
         # Step 4: Re-embed and insert documents in batches
-        for i in tqdm(range(0, len(documents), batch_size), desc="Re-embedding batches"):
-            batch_docs = documents[i:i+batch_size]
-            batch_metadatas = metadatas[i:i+batch_size]
-            batch_ids = ids[i:i+batch_size]
+        for i in tqdm(
+            range(0, len(documents), batch_size), desc="Re-embedding batches"
+        ):
+            batch_docs = documents[i : i + batch_size]
+            batch_metadatas = metadatas[i : i + batch_size]
+            batch_ids = ids[i : i + batch_size]
 
             embeddings = get_embedding(batch_docs, backend="openai")
 
@@ -222,6 +336,6 @@ class SanatanDatabase:
                 ids=batch_ids,
                 documents=batch_docs,
                 metadatas=batch_metadatas,
-                embeddings=embeddings
+                embeddings=embeddings,
             )
-        print("All documents re-embedded and added to new collection successfully!")
+        print("All documents re-embedded and added to new collection successfully!")

metadata.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing import Literal, Union, List, Dict
+from typing import Literal, Optional, Union, List, Dict
 
 
 AllowedOps = Literal["$in", "$eq", "$gt", "$gte", "$lt", "$lte", "$ne", "$nin"]
@@ -11,19 +11,47 @@ class MetadataFilter(BaseModel):
     metadata_value: Union[str, int, float, List[Union[str, int, float]]]
 
 
+# class MetadataWhereClause(BaseModel):
+#     filters: List[MetadataFilter]
+#     conditional_operator: Literal["$and", "$or"] = "$and"
+
+#     def to_chroma_where(self) -> Dict:
+#         """Convert list of MetadataFilter into Chroma-compatible where clause with AND logic."""
+#         if not self.filters:
+#             return {}
+#         if len(self.filters) == 1:
+#             f = self.filters[0]
+#             return {f.metadata_field: {f.metadata_search_operator: f.metadata_value}}
+#         return {
+#             self.conditional_operator: [
+#                 {f.metadata_field: {f.metadata_search_operator: f.metadata_value}}
+#                 for f in self.filters
+#             ]
+#         }
+
 class MetadataWhereClause(BaseModel):
-    filters: List[MetadataFilter]
+    filters: Optional[List["MetadataFilter"]] = None
+    groups: Optional[List["MetadataWhereClause"]] = None
+    conditional_operator: Literal["$and", "$or"] = "$and"
 
     def to_chroma_where(self) -> Dict:
-        """Convert list of MetadataFilter into Chroma-compatible where clause with AND logic."""
-        if not self.filters:
+        parts = []
+
+        # Handle direct filters
+        if self.filters:
+            for f in self.filters:
+                parts.append({f.metadata_field: {f.metadata_search_operator: f.metadata_value}})
+
+        # Handle nested groups
+        if self.groups:
+            for g in self.groups:
+                parts.append(g.to_chroma_where())
+
+        if not parts:
             return {}
-        if len(self.filters) == 1:
-            f = self.filters[0]
-            return {f.metadata_field: {f.metadata_search_operator: f.metadata_value}}
-        return {
-            "$and": [
-                {f.metadata_field: {f.metadata_search_operator: f.metadata_value}}
-                for f in self.filters
-            ]
-        }
+
+        if len(parts) == 1:
+            return parts[0]
+
+        # More than one part → wrap with conditional operator
+        return {self.conditional_operator: parts}

modules/db/relevance.py

@@ -1,4 +1,11 @@
 from chromadb.api.types import QueryResult
+from dataclasses import dataclass
+
+@dataclass
+class ValidationOutcome:
+    status : str
+    reason : str
+    result : QueryResult
 
 def validate_relevance_queryresult(query: str, result: QueryResult, max_distance: float = 0.35):
     """
@@ -20,24 +27,24 @@ def validate_relevance_queryresult(query: str, result: QueryResult, max_distance
     distances = result.get("distances", [])
 
     if not documents:
-        return {
+        return ValidationOutcome(**{
             "status": "not_found",
             "reason": "No results",
             "result": result
-        }
+        })
 
     # distances can be List[List[float]]; get the first distance of the first result
     best_distance = distances[0][0] if distances and isinstance(distances[0], list) else (distances[0] if distances else float('inf'))
 
     if best_distance > max_distance:
-        return {
+        return ValidationOutcome(**{
             "status": "not_relevant",
             "reason": f"Best distance {best_distance:.4f} > {max_distance}",
             "result": result
-        }
+        })
 
-    return {
+    return ValidationOutcome(**{
         "status": "ok",
         "reason": "Relevant",
         "result": result
-    }
+    })

modules/nodes/chat.py

@@ -8,15 +8,11 @@ from tools import (
     tool_search_web,
     tool_push,
     tool_get_standardized_azhwar_names,
-    tool_search_db_by_metadata,
     tool_get_standardized_divya_desam_names,
-    tool_search_db_for_literal,
 )
 
 tools = [
-    tool_search_db_by_metadata,
     tool_search_db,
-    tool_search_db_for_literal,        
     tool_get_standardized_azhwar_names,
     tool_get_standardized_prabandham_names,
     tool_get_standardized_divya_desam_names,

modules/nodes/init.py

@@ -27,26 +27,7 @@ def init_system_prompt_node(state: ChatState) -> ChatState:
                 content=f"Here is the list of all scriptures along with their metadata configurations:\n{json.dumps(scriptures, separators=(',', ':'))}\n"
             ),
             SystemMessage(
-                content="""
-You have access to three scripture search tools. You MUST follow these rules when choosing a tool:
-
-1. **tool_search_db_by_metadata** – Use this when the user explicitly provides metadata criteria such as:
-   - Specific azhwar name
-   - Prabandham or prabandham code
-   - Verse number or decade number
-   - Divya desam name  
-   ⚠️ Always call the corresponding standardization tool first.
-        - "If the user asks for a specific azhwar, use `tool_get_standardized_azhwar_names` first."
-        - "If the user asks for a specific prabandham, use `tool_get_standardized_prabandham_names` first."
-        - "If the user mentions a divya desam, use `tool_get_standardized_divya_desam_names` first."
-
-
-2. **tool_semantic_vector_search** – Use this when the user asks about themes, ideas, emotions, or meanings without explicit verse numbers or metadata.
-
-3. **tool_search_db_by_literal_text** – Use this only if the user explicitly requests an exact phrase match.
-
-Never call a tool repeatedly with the same arguments. Stop if results don’t change meaningfully.
-"""
+                content="""The tools are deterministic. Caling the same tool multiple times with the same arguments are not going to yield different results. So NEVER call a tool twice or more with the same arguments. Stop if results don’t change meaningfully."""
             ),
             SystemMessage(
                 content=f"""
@@ -56,28 +37,51 @@ You are a knowledgeable assistant for *{{collection_name}}*.
 Languages: Sanskrit, Tamil, and {state['language']}.  
 Use **only** the verses and notes retrieved from the context. Never fabricate or import external knowledge.
 
+In the context, if there ia a variable called `html_utl`, then use that direcly for `reference_link`. If not, look for `video_id` and use that to construct the youtube url using https://www.youtube.com/watch?v={{video_id}} and store it under `reference_link`
+RULE: 
+- If the user asks for "one verse", "any verse", "show me a verse", or similar, always return exactly ONE verse.
+- Do not return multiple verses.
+- Only return multiple verses if the user explicitly asks for more than one.
+In the header at the end for the field `verse_or_page`, show the `verse` or `page` whichever is available in the context and mention Verse `verse` or Page `page` as the case may be.
+
 ---
 
-### ✅ Default Response Format (always include)
+### ✅ Default Response Format (always include unless it is a followup question and/or user requests specific details only)
 
-### Title : {{scripture}} | {{chapter_title_if_available}} | {{`title`}} | {{author_name_if_available}} | Reference Link {{html_url}} if available
+### 🕉️ Scripture
+- Show `collection` if available else skip the entire section including header.
 
-### 📜 Original Verse(s)
-- Show exact native-script verses from the context.  
+### 📜 Divya Desam
+- Show `divya_desams` if available else skip the entire section including header.
+
+### 📜 Author
+- Show `author` if available else skip the entire section including header.
+
+### 📜 Verse Number
+- Show verse number if available else skip the entire section including header. If available show this as a hyperlink with `html_url` if html_url is available.
+
+### 📜 Title
+- Show `title` if available else skip the entire section including header.
+
+### 📜 Page
+- Show `page` if available else skip the entire section including header.
+
+### 📜 Original Verse
+- Show exact original native-script verses from the context in a separate markdown block.  
 - Do not translate, transliterate, or explain.  
 - Preserve line breaks and spacing exactly.  
 
 ### 📜 Sanitized Verse(s)
-- Only include this section if sanitization changes anything.  
+- Only include this section if sanitization changes anything otherwise don't even output the section heading .  
 - Sanitize by:  
-  1. Fixing garbled Unicode characters.  
+  1. Fixing garbled Unicode characters in the original verse section.  
   2. Correcting broken diacritics, pulli markers, vowel signs, and punctuation.  
   3. Preserving original spacing and line order.  
-- If no change → skip this section entirely.   
+- If no change → skip this section entirely including the heading.   
 
 ### 📜 {state['language']} – Simple Meaning
 - Give a **short, natural summary/meaning** in {state['language']}.  
-- Keep it concise and error-free.  
+- Keep it concise and error-free.  Do not give word-by-word meanings here even if available.
 
 ### 🔮 Next Steps
 End with a short list of follow-up prompts:  
@@ -97,15 +101,19 @@ End with a short list of follow-up prompts:
 #### 📜 Transliteration
 - Provide verse transliteration in {state['language']} if requested.  
 
-#### 📜 Word-by-Word Meaning
+#### 📜 Word-by-Word Meaning (English)
 - Provide WBW meaning in English or {state['language']} if requested.  
 
+#### 📜 Word-by-Word Meaning ({state['language']})
+- Provide WBW meaning {state['language']} if requested.  
+
 #### 📜 Detailed Notes / Purport
 - Summarize and translate explanatory notes/purports if present in context.  
 
 ---
 
 ⚠️ Rules:
+- For a follow-up question, if the user does not specify a context in the question, assume it is for the verse returned by the previous response.For e.g. "word by word meaning" implies that the user wants to know "the word by word meaning for the above pasuram".
 - Do not duplicate content across sections.  
 - Do not invent verses, meanings, or purports.  
 - If no context found → reply in {state['language']}:  
@@ -127,7 +135,7 @@ End with a short list of follow-up prompts:
                 )
             )
         )
-        state["initialized"] = True        
+        state["initialized"] = True
 
     state["tool_calls"] = 0
     state["seen_tool_calls"] = set()

nalayiram_helper.py

@@ -20,7 +20,7 @@ def get_standardized_prabandham_names() -> list[Pasuram]:
 
     return final_azhwars
 
-def get_standardized_azhwar_names() -> list[Pasuram]:
+def get_standardized_azhwar_names() -> list[str]:
     """
     Get a list of azhwar names along with the pasurams they have authored in divya_prabandham
     """
@@ -28,12 +28,12 @@ def get_standardized_azhwar_names() -> list[Pasuram]:
         azhwars = json.load(f)
         header = azhwars[0]
         rows = azhwars[1:]
-        final_azhwars = [Pasuram(**dict(zip(header, row))) for row in rows]
+        final_azhwars = [row[1] for row in rows] ## 2nd field is the azhwar name
 
-    return final_azhwars
+    return sorted(set(final_azhwars))
 
 
-def get_standardized_divya_desam_names() -> list[dict]:
+def get_standardized_divya_desam_names() -> list[str]:
     """
     Get a list of divya desam names in divya_prabandham
     """
@@ -52,7 +52,8 @@ def get_standardized_divya_desam_names() -> list[dict]:
         "sampradayam",
         "divya_desam",
     ]
-    return [{key : row[key] for key in selected_fields if key in row} for row in divya_desams["pageProps"]["hits"]]
+    data = [{key : row[key] for key in selected_fields if key in row} for row in divya_desams["pageProps"]["hits"]]
+    return sorted(set([row["title"] for row in data]))
 
 
 if __name__ == "__main__":

sanatan_assistant.py

@@ -104,96 +104,52 @@ Respond in **Markdown** format only. Ensure Sanskrit/Tamil verses are always cle
     return prompt
 
 
-def query(collection_name: allowedCollections, query: str, n_results=3):
-    """
-    Search a scripture collection.
-
-    Parameters:
-    - collection_name (str): The name of the scripture collection to search. ...
-    - query (str): The search query.
-    - n_results (int): Number of results to return. Default is 3.
-
-    Returns:
-    - A list of matching results.
-    """
-    logger.info("Semantic Search: Searching collection [%s] for [%s]", collection_name, query)
-    response = sanatanDatabase.search(
-        collection_name=collection_name, query=query, n_results=n_results
-    )
-
-    return "\n\n".join(
-        f"Document: {doc}\nMetadata: {meta}\nID: {id_}"
-        for doc, meta, id_ in zip(
-            response["documents"], response["metadatas"], response["ids"]
-        )
-    )
-
-def query_by_metadata_field(
+def query(
     collection_name: allowedCollections,
-    query: str,
-    metadata_where_clause : MetadataWhereClause,
+    query: str | None = None,
+    metadata_where_clause: MetadataWhereClause | None = None,
     n_results=3,
+    search_type: Literal["semantic", "literal", "random"] = "semantic",
 ):
     """
-    Search a scripture collection by metadata. Do NOT use this for semantic search. Only use when a specific metadata field is provided.
+    Search a scripture collection.
 
     Parameters:
-    - collection_name (str): The name of the scripture collection to search. ...
-    - query (str): The search query.
-    - metadata_where_clause: the filter which is an array of the following type
-        - metadata_field (str) : The name of the metadata field. e.g. azhwar_name
-        - metadata_search_operator (str) : The search operator e.g. $eq or $in. DO NOT use $regex.
-        - metadata_value : Value to search for can be any primitive datatype like str or int (or a list[str] if metadata_search_operator = '$in'). for e.g. Thirumangai Azhwar or '2233' or 2233
+    - collection_name (str): The name of the scripture collection to search (use the exact name from the metadata configuration. ...
+    - query (str): The search query - this is the semantic or literal query you want to search for. if you want to perform a random search or just want to search by metadata only, can be passed as None ..
+    - metadata_where_clause: MetadataWhereClause - Set to None if no metadata filters are requested. Always set when user mentions a specific prabandham, azhwar, or any other known field from the configuration. Example: {\"prabandham_name\": \"Thiruvaimozhi\"}. use the `conditional_operator` to filter based on $and or $or conditions. use `groups` to combine multiple queries into one.
     - n_results (int): Number of results to return. Default is 3.
+    - search_type: can be one of semantic, literal or random.
 
     Returns:
     - A list of matching results.
     """
-    logger.info("Searching collection [%s] for [%s]", collection_name, query)
-
+    logger.info(
+        "%s Search: collection [%s] for [%s] | metadata_where_clause=%s",
+        search_type,
+        collection_name,
+        query,
+        metadata_where_clause,
+    )
+    if search_type != "random" and metadata_where_clause is None and query is None:
+        raise Exception(
+            "Invalid input: when search type is not random, either metadata_where_clause or query should be provided"
+        )
     try:
-        sanatanConfig.is_metadata_field_allowed(collection_name=collection_name, metadata_where_clause=metadata_where_clause)
+        if metadata_where_clause is not None:
+            sanatanConfig.is_metadata_field_allowed(
+                collection_name=collection_name,
+                metadata_where_clause=metadata_where_clause,
+            )
     except:
         raise
 
-    response = sanatanDatabase.search_by_metadata(
+    response = sanatanDatabase.search(
         collection_name=collection_name,
         query=query,
         metadata_where_clause=metadata_where_clause,
         n_results=n_results,
-    )
-
-    return "\n\n".join(
-        f"Document: {doc}\nMetadata: {meta}\nID: {id_}"
-        for doc, meta, id_ in zip(
-            response["documents"], response["metadatas"], response["ids"]
-        )
-    )
-
-
-def query_by_literal_text(
-    collection_name: allowedCollections,
-    literal_to_search_for: str,
-    n_results=3,
-):
-    """
-    Search a scripture collection by a literal. Do NOT use this for semantic search. Only use when the user specifically asks for literal search.
-
-    Parameters:
-    - collection_name (str): The name of the scripture collection to search. ...
-    - literal_to_search_for (str): The search query.
-    - n_results (int): Number of results to return. Default is 3.
-
-    Returns:
-    - A list of matching results.
-    """
-    logger.info("Performing literal search in collection [%s] for [%s]", collection_name, literal_to_search_for)
-
-
-    response = sanatanDatabase.search_for_literal(
-        collection_name=collection_name,
-        literal_to_search_for=literal_to_search_for,
-        n_results=n_results,
+        search_type=search_type,
     )
 
     return "\n\n".join(

server.py

@@ -6,7 +6,7 @@ from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
 import pycountry
 from pydantic import BaseModel
-from app import chat
+from chat_utils import chat
 from config import SanatanConfig
 from db import SanatanDatabase
 

tests/test_config.py

@@ -0,0 +1,51 @@
+# Example test questions
+TEST_QUESTIONS = [
+    {
+        "q": "one pasuram on thirukudandai and another from srirangam both written by thirumangai azhwar",
+        "type": "composite",
+        "difficulty": "complex",
+        "expected_answer_summary": "Should return one pasuram from Thirukudanthai and another from Srirangam, both authored by Thirumangai Azhwar.",
+        "expected_sources": ["Thirukudanthai", "Srirangam"],
+        "expected_azhwar": ["Thirumangai Azhwar"],
+        "n_results": 2,
+    },
+    {
+        "q": "give me 2 pasurams, one written by thirumazhisai alwar and the other by thirumangai azhwar, both written on divya desam Srirangam",
+        "type": "composite",
+        "difficulty": "complex",
+        "expected_answer_summary": "Should return two pasurams on Srirangam: one by Thirumazhisai Azhwar and the other by Thirumangai Azhwar.",
+        "expected_sources": ["Srirangam"],
+        "expected_azhwar": ["Thirumazhisai Azhwar", "Thirumangai Azhwar"],
+        "n_results": 2,
+    },
+    {
+        "q": "a pasuram from nanmugan thiruvandhadhi that talks about Krishna playing flute",
+        "type": "semantic",
+        "difficulty": "medium",
+        "expected_answer_summary": "Should return 1 pasuram from Nanmukan Thiruvanthathi.",
+        "expected_sources": ["Nanmukan Thiruvanthathi"],
+        "expected_azhwar": ["Thirumazhisai Azhwar"],
+        "expected_topics": ["Krishna", "Flute"],
+        "n_results": 1,
+    },
+    {
+        "q": "varaha avatar in nanmugan thiruvandhadhi",
+        "type": "semantic",
+        "difficulty": "medium",
+        "expected_answer_summary": "Should return 1 pasuram from Nanmukan Thiruvanthathi.",
+        "expected_sources": ["Nanmukan Thiruvanthathi"],
+        "expected_azhwar": ["Thirumazhisai Azhwar"],
+        "expected_keywords": ["boar"],
+        "n_results": 1,
+    },
+    {
+        "q": "varaha avatar in nanmugan thiruvandadhi and perumal thirumozhi",
+        "type": "semantic+composite",
+        "difficulty": "medium",
+        "expected_answer_summary": "Should return 2 pasurams. One from Nanmukan Thiruvanthathi and another from perumal thirumozhi.",
+        "expected_sources": ["Nanmukan Thiruvanthathi", "perumal thirumozhi"],
+        "expected_azhwar": ["Thirumazhisai Azhwar", "Kulasekhara Azhwar"],
+        "expected_keywords": ["boar"],
+        "n_results": 2,
+    },
+]

tests/test_evaluator.py

@@ -0,0 +1,108 @@
+import os
+from datetime import datetime
+import openai
+import json
+
+from chat_utils import chat
+from tests.test_config import TEST_QUESTIONS
+
+def validate_with_ai(test_entry, bot_response):
+    """
+    Validator works with narrative bot responses. 
+    The bot does not need to output JSON.
+    The LLM analyzes the bot response and returns a JSON validation.
+    """
+    prompt = f"""
+You are a validator AI. The user provided the following bot response:
+
+Bot Response:
+\"\"\"{bot_response}\"\"\"
+
+Expected attributes:
+- Sources: {test_entry.get('expected_sources', [])}
+- Azhwar: {test_entry.get('expected_azhwar', [])}
+- Topics: {test_entry.get('expected_topics', [])}
+- Keywords: {test_entry.get('expected_keywords', [])}
+- Number of results: {test_entry.get('n_results', 1)}
+
+Check the bot response and answer **only** in JSON with two fields:
+{{
+  "valid": true/false,   // True if bot response matches the expected attributes
+  "feedback": "short explanation why it passed or failed"
+}}
+
+Do **not** ask the bot to output the JSON itself. You should parse the narrative internally and return JSON.
+"""
+    resp = openai.chat.completions.create(
+        model="gpt-5-nano",
+        messages=[{"role": "user", "content": prompt}],
+    )
+    try:
+        content = resp.choices[0].message.content
+        return json.loads(content)
+    except Exception as e:
+        return {"valid": False, "feedback": f"Validator parsing error: {e}"}
+
+def run_tests(debug_mode=False):
+    history = []
+    thread_id = "test_thread"
+
+    # Create log directory if it doesn't exist
+    log_dir = "outputs/tests"
+    os.makedirs(log_dir, exist_ok=True)
+
+    # Markdown log file with timestamp
+    run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
+    log_file_path = os.path.join(log_dir, f"{run_id}.md")
+
+    # Keep track of summary
+    total_tests = len(TEST_QUESTIONS)
+    passed_tests = 0
+    results_summary = []
+
+    with open(log_file_path, "w", encoding="utf-8") as f:
+        f.write(f"# Sanatan AI Test Run - {run_id}\n\n")
+        for idx, test in enumerate(TEST_QUESTIONS, start=1):
+            f.write(f"## Test {idx}: {test['q']}\n\n")
+            f.write(f"**Type:** {test['type']}  \n")
+            f.write(f"**Difficulty:** {test['difficulty']}  \n")
+            f.write(f"**Expected Summary:** {test.get('expected_answer_summary', '')}\n\n")
+
+            print(f"\n=== Testing Question ===\n{test['q']}")
+            bot_response = chat(debug_mode, test["q"], history, thread_id)
+            f.write(f"### Bot Response\n```\n{bot_response}\n```\n\n")
+
+            validation = validate_with_ai(test, bot_response)
+            f.write(f"### Validation\n- **Valid:** {validation['valid']}\n- **Feedback:** {validation['feedback']}\n\n")
+
+            print(f"Valid: {validation['valid']}\nFeedback: {validation['feedback']}")
+
+            # Track results for summary
+            results_summary.append({
+                "question": test['q'],
+                "valid": validation['valid']
+            })
+            if validation['valid']:
+                passed_tests += 1
+
+        # Write run summary
+        failed_tests = total_tests - passed_tests
+        pass_rate = (passed_tests / total_tests) * 100 if total_tests > 0 else 0
+        f.write(f"# Run Summary\n\n")
+        f.write(f"- **Total Tests:** {total_tests}\n")
+        f.write(f"- **Passed:** {passed_tests}\n")
+        f.write(f"- **Failed:** {failed_tests}\n")
+        f.write(f"- **Pass Rate:** {pass_rate:.2f}%\n\n")
+
+        # Optional: Table of all test results
+        f.write("## Test Results Table\n\n")
+        f.write("| Test | Question | Valid |\n")
+        f.write("|------|----------|-------|\n")
+        for i, res in enumerate(results_summary, start=1):
+            valid_str = "✅" if res['valid'] else "❌"
+            f.write(f"| {i} | {res['question']} | {valid_str} |\n")
+
+    print(f"\nTest run complete. Markdown log saved to {log_file_path}")
+
+if __name__ == "__main__":
+    run_tests(debug_mode=True)

tools.py

@@ -13,8 +13,6 @@ from serperdev_helper import search as search_web
 from sanatan_assistant import (
     format_scripture_answer,
     query,
-    query_by_metadata_field,
-    query_by_literal_text,
 )
 
 tool_push = Tool(
@@ -25,52 +23,37 @@ allowed_collections = [s["collection_name"] for s in SanatanConfig.scriptures]
 
 tool_search_db = StructuredTool.from_function(
     query,
-    name="tool_semantic_vector_search",
+    name="tool_search_db",
     description=(
-        "🚫 Never use this tool if the user asks for a verse number, pasuram number, or any explicit metadata field "
-        "(like azhwar name, prabandham, divya desam, decade, or chapter). "
-        "✅ Only use this tool when the query is vague or thematic, e.g. "
-        "'Which pasurams talk about Krishna's childhood?' or 'Show me verses about compassion'. "
+        "🚫 use this tool to fetch any data from the database."
+        "rules for metadata_where_clause:"
+        """
+        - ⚠️ Every time you include a metadata_where_clause argument, you must first call the appropriate standardization tool (tool_get_standardized_divya_desam_names,tool_get_standardized_prabandham_names,tool_get_standardized_azhwar_names). Never insert raw values directly. Even if the input already looks correct, you must still call the tool. If you fail to do this, the query will be invalid.
+            > Standardization Step 1: Call the standardization tool to get the canonical Divya Desam name.
+            |--Example:
+            |----standardized_divya_desams = tool_get_standardized_divya_desam_names()
+            |----standardized_divya_desam =  look for closest match to "Thirukkudandai" in standardized_divya_desams
+            > Standardization Step 2: Use the standardized name in your DB search argument for metadata_where_clause for the field divya_desams.
+        - When choosing collection_name argument for the tool_search_db, make sure you choose the exact collection_name from the metadata configuration above
+        - Always prefer a single tool call with composite filters rather than multiple calls. 
+        - For MetadataWhereClause.filters.$.metadata_search_operator do not use $regex as angument. use semantic search option by using query argument instead.
+        - If user posts a thematic question, do not ignore the theme when you pass `query` arguments. 
+        - Use `MetadataWhereClause` recursively with `filters` and `groups` to build nested conditions.
+        """        
+        "- Always set metadata filters  when user mentions a specific divya desam, prabandham, azhwar, or any other known field from the configuration. Example: {\"prabandham_name\": \"Thiruvaimozhi\"}."
+        "- Multiple metadata filters can be passed at the same time."
+        "- If passing '$in' as metadata_search_operator, the metadata_value should always be of type array. for instance {'metadata_field': 'divya_desams', 'metadata_search_operator': '$in', 'metadata_value': []'Srirangam']}"
+        "- Set metadata filters as None if no metadata filter is requested.\n"
+        "rules for search_type:"
+        "- use `random` if user does not provide a thematic/semantic search request. For e.g. 'any pasuram' or 'any pasuram from thiruvaimozhi'"
+        "- use `semantic` if user provides thematic/semantic search request"
+        "- use `literal` ONLY if user specifically requests for a literal search."
+        "\n"
         f"The collection_name must be one of: {', '.join(allowed_collections)}."
     ),
 )
 
 
-tool_search_db_for_literal = StructuredTool.from_function(
-    query_by_literal_text,
-    name="tool_search_db_by_literal_text",
-    description=(
-        "🚫 Never use this tool by default."
-        " ✅ Only use this tool if the user explicitly requests a 'literal match', 'exact phrase search', or uses words like 'match exactly', 'find the exact string', 'verbatim', or 'literal text'."
-        " If the user simply asks for a verse number (e.g., verse 34, pasuram 2.3.5, sahasranamam verse 20), you must NOT use this tool — instead you must use `tool_search_db_by_metadata`."
-        " Do not fall back to this tool if semantic or metadata search seems difficult or fails — it is reserved strictly for explicit literal match requests."
-        f" The collection_name must be one of: {', '.join(allowed_collections)}."
-    ),
-)
-
-
-
-tool_search_db_by_metadata = StructuredTool.from_function(
-    query_by_metadata_field,
-    name="tool_search_db_by_metadata",
-    description=(
-        "Use this tool **only when the user provides explicit metadata criteria**, such as: azhwar name, pasuram number, verse number, decade, prabandham name, or divya desam name."
-        " This is not meant for general queries."
-        f" The collection_name must be one of: {', '.join(allowed_collections)}."
-        "You *MUST* ALWAYS call one of the standardization tools available to get the correct entity name before using this tool."
-        "If the user asks for a specific azhwar, use `tool_get_standardized_azhwar_names` first."
-        "If the user asks for a specific prabandham, use `tool_get_standardized_prabandham_names` first."
-        "If the user mentions a divya desam, use `tool_get_standardized_divya_desam_names` first."
-        "If you set metadata_search_operator to $in, then metadata_value must always be a list — even if it contains only a single item."
-        """🔒 Important:
-        When using the tool_get_standardized_azhwar_names, tool_get_standardized_divya_desam_names, or any similar standardization tool, you must use the standardized name exactly as returned by the tool — without modifying, reformatting, translating, or simplifying it in any way.
-        For example, if the tool returns Thirumālirum Solai, you must pass that exact string to tool_search_db_by_metadata. Do not change it to Thirumalirum Solai, Tirumalirumsolai, or anything else.
-        🔍 This is critical for the search to return results correctly.
-        🚫 Any deviation will cause the search to fail or miss results."""
-    ),
-)
-
-
 tool_search_web = Tool(
     name="search_web", description="Search the web for information", func=search_web
 )