Upload folder using huggingface_hub

app.py

@@ -107,6 +107,11 @@ def chat(message, history, thread_id):
     return response["messages"][-1].content
 
 
+def add_node_to_tree(node_tree : list[str], node : str) -> list[str]:
+    node_tree[-1] = node
+    node_tree.append("<span class='spinner'>⏳</span>")
+    return node_tree
+
 async def chat_streaming(message, history, thread_id):
     state = {"messages": (history or []) + [{"role": "user", "content": message}]}
     config = {"configurable": {"thread_id": thread_id}}
@@ -117,48 +122,58 @@ async def chat_streaming(message, history, thread_id):
 
     try:
         tool_calls = []
+        node_tree = ["__start__","<span class='spinner'>⏳</span>"]
         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:{node}"
+            tool_label =f"{name or ''}"
+            if(tool_label):
+                node_label = node_label + f":{tool_label}"
+            label = f"{node_icon} {node_label}"
+            # 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)
             full: str = escape(msg.content)
             truncated = (full[:MAX_CONTENT] + "…") if len(full) > MAX_CONTENT else full
 
-            processing_message = (
+            def generate_processing_message():
+                return (
                 f"<div class='thinking-bubble'><em>🤔{random.choice(thinking_verbs)} ...</em></div>"
                 f"<div style='opacity: 0.1' title='{full}'>"
                 f"<span>{node}:{name or ''}:</span>"
                 f"<strong>Looking for : [{message}]</strong> {truncated or '...'}"
                 f"</div>"
             )
+
             if (
                 not isinstance(msg, ToolMessage)
                 and not isinstance(msg, SystemMessage)
                 and not isinstance(msg, AIMessageChunk)
             ):
                 logger.info("msg = %s", msg)
-            # yield processing_message
             if isinstance(msg, ToolMessage):
                 logger.debug("tool message = %s", msg)
                 html = (
-                    f"<div class='thinking-bubble'><em>🤔{name} : {random.choice(thinking_verbs)} ...</em></div>"
+                    f"<div class='thinking-bubble'><em>🤔{name} tool: {random.choice(thinking_verbs)} ...</em></div>"
                     f"<div style='opacity: 0.5'>"
                     f"<strong>Looking for : [{message}]</strong> {truncated or '...'}"
                     f"</div>"
                 )
-                yield html
-                # yield f"""
-                # <div class='thinking-bubble'>🤔 {random.choice(thinking_verbs)}...</div>
-                # <div style='opacity: 0.5'><strong>[{node} - {name}]</strong>: {escape(msg.content)}</div>"""
-
+                yield f"### { " → ".join(node_tree)}\n{html}"
             elif isinstance(msg, AIMessageChunk):
                 if not msg.content:
                     # logger.warning("*** No Message Chunk!")
-                    yield processing_message
+                    yield f"### { " → ".join(node_tree)}\n{generate_processing_message()}"
                 else:
                     streamed_response += msg.content
-                    yield streamed_response
+                    yield f"### { " → ".join(node_tree)}\n{streamed_response}"
                 if(msg.tool_calls):
                     tool_calls.append(msg.tool_calls)
             elif isinstance(msg, AIMessage):                    
@@ -176,9 +191,10 @@ async def chat_streaming(message, history, thread_id):
                     f"<strong>Telling myself:</strong> {truncated or '...'}"
                     f"</div>"
                 )
-                yield html
+                yield f"### { " → ".join(node_tree)}\n{html}"
 
-        yield streamed_response
+        node_tree[-1] = "✅"
+        yield f"### { " → ".join(node_tree)}\n{streamed_response}"
     except Exception as e:
         yield f"Error processing request {str(e)}"
 
@@ -236,6 +252,24 @@ chatInterface = gr.ChatInterface(
     chatbot=chatbot,
     css="""
 <style>
+.spinner {
+  display: inline-block;
+  width: 1em;
+  height: 1em;
+  border: 2px solid #ccc;
+  border-top: 2px solid #333;
+  border-radius: 50%;
+  animation: spin 1s linear infinite;
+  vertical-align: middle;
+  margin-left: 0.5em;
+}
+
+@keyframes spin {
+  0% { transform: rotate(0deg); }
+  100% { transform: rotate(360deg); }
+}
+
+
 .thinking-bubble {
   opacity: 0.5;
   font-style: italic;

db.py

@@ -1,6 +1,6 @@
 from typing import Literal
 import chromadb
-
+import re, unicodedata
 from config import SanatanConfig
 from embeddings import get_embedding
 import logging
@@ -38,6 +38,60 @@ class SanatanDatabase:
         )
         return response
 
+    def search_for_literal(
+        self, collection_name: str, literal_to_search_for: str, n_results=2
+    ):
+        logger.info(
+            "Searching literally for [%s] in [%s]",
+            literal_to_search_for,
+            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_texts=[""],
+            where_document={"$contains": literal_to_search_for},
+            n_results=n_results,
+        )
+
+        if response["documents"] and any(response["documents"]):
+            return response
+
+        # 2. Regex fallback (normalized)
+        regex = re.compile(re.escape(normalize(literal_to_search_for)))
+
+        all_docs = collection.get()
+        matched_docs = []
+
+        for doc, metadata, ids in zip(
+            all_docs["documents"], all_docs["metadatas"], all_docs["ids"]
+        ):
+            for i, d in enumerate(doc):
+                if regex.search(normalize(d)):
+                    matched_docs.append(
+                        {
+                            "id": ids[i],
+                            "document": d,
+                            "metadata": (
+                                metadata[i] if isinstance(metadata, list) else metadata
+                            ),
+                        }
+                    )
+                    if len(matched_docs) >= n_results:
+                        break
+            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,
@@ -81,10 +135,11 @@ if __name__ == "__main__":
         query = input("Search for: ")
         if query.strip() == "":
             break
-        response = database.search(
-            collection_name=collection_name, query=query, n_results=1
+        response = database.search_for_literal(
+            collection_name=collection_name, literal_to_search_for=query, n_results=1
         )
-        print("Document: ")
-        print(response["documents"][0][0])
-        print("Metadata: ")
-        print(response["metadatas"][0][0])
+        print("Matches" , response)
+        # print("Document: ")
+        # print(response["documents"][0][0])
+        # print("Metadata: ")
+        # print(response["metadatas"][0][0])

graph_helper.py

@@ -13,6 +13,7 @@ from tools import (
     tool_get_standardized_azhwar_names,
     tool_search_db_by_metadata,
     tool_get_standardized_divya_desam_names,
+    tool_search_db_for_literal
 )
 from langgraph.prebuilt import ToolNode, tools_condition
 from langchain_core.messages import SystemMessage, ToolMessage, HumanMessage
@@ -37,6 +38,7 @@ def generate_graph() -> CompiledStateGraph:
         tool_get_standardized_prabandham_names,
         tool_get_standardized_divya_desam_names,
         tool_search_db_by_metadata,
+        tool_search_db_for_literal
     ]
     llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2).bind_tools(tools)
 

sanatan_assistant.py

@@ -206,3 +206,36 @@ def query_by_metadata_field(
             response["documents"], response["metadatas"], response["ids"]
         )
     )
+
+
+def query_by_literal_text(
+    collection_name: allowedCollections,
+    literal_to_search_for: str,
+    n_results=5,
+):
+    """
+    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 5.
+
+    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,
+    )
+
+    return "\n\n".join(
+        f"Document: {doc}\nMetadata: {meta}\nID: {id_}"
+        for doc, meta, id_ in zip(
+            response["documents"], response["metadatas"], response["ids"]
+        )
+    )

tools.py

@@ -6,7 +6,7 @@ from config import SanatanConfig
 from nalayiram_helper import get_standardized_azhwar_names, get_standardized_divya_desam_names
 from push_notifications_helper import push
 from serperdev_helper import search as search_web
-from sanatan_assistant import format_scripture_answer, query, query_by_metadata_field
+from sanatan_assistant import format_scripture_answer, query, query_by_metadata_field, query_by_literal_text
 
 tool_push = Tool(
     name="push", description="Send a push notification to the user", func=push
@@ -17,7 +17,16 @@ allowed_collections = [s["collection_name"] for s in SanatanConfig.scriptures]
 tool_search_db = StructuredTool.from_function(
     query,
     description=(
-        "Do a vector Search within a specific scripture collection. "
+        "Do a semantic vector search within a specific scripture collection. "
+        f"The collection_name must be one of: {', '.join(allowed_collections)}."
+        "Use this to narrow down relevant scripture verses or explanations based on the given query."
+    ),
+)
+
+tool_search_db_for_literal = StructuredTool.from_function(
+    query_by_literal_text,
+    description=(
+        "Do a literal search within a specific scripture collection (only if user specifically asks for a literal search or if semantic search does not yield relevant results)."
         f"The collection_name must be one of: {', '.join(allowed_collections)}."
         "Use this to find relevant scripture verses or explanations based on the given query."
         # "If the query doesn't yield any relevant results, then call `tool_search_db_by_metadata` tool to search specifically by a given metadata field (only if specific field from metadata has been mentioned)."