import json
import os
import random
import asyncio
import logging
import time
import traceback
import uuid
from html import escape
import gradio as gr
from dotenv import load_dotenv
from langchain_core.messages.ai import AIMessageChunk, AIMessage
from langchain_core.messages.system import SystemMessage
from langchain_core.messages.tool import ToolMessage
from config import SanatanConfig
from db import SanatanDatabase
from drive_downloader import ZipDownloader
from graph_helper import generate_graph
# Logging
logging.basicConfig()
logger = logging.getLogger()
logger.setLevel(logging.INFO)
graph = generate_graph()
def init():
load_dotenv(override=True)
try:
SanatanDatabase().test_sanity()
except Exception as e:
logger.warning("Sanity Test Failed - %s", e)
logger.info("Downloading database ...")
downloader = ZipDownloader(
service_account_json=os.getenv("GOOGLE_SERVICE_ACCOUNT_JSON")
)
zip_path = downloader.download_zip_from_drive(
file_id=os.getenv("CHROMADB_FILE_ID"),
output_path=SanatanConfig.dbStorePath,
)
downloader.unzip(zip_path, extract_to="./")
def init_session():
return str(uuid.uuid4())
def render_message_with_tooltip(content: str, max_chars=200):
short = escape(content[:max_chars]) + ("β¦" if len(content) > max_chars else "")
return f"
{short}
"
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):
if debug:
async for chunk in chat_streaming(debug, message, history, thread_id):
yield chunk
else:
response = chat(debug, message, history, thread_id)
yield response
def chat(debug_mode, message, history, thread_id):
config = {"configurable": {"thread_id": thread_id}}
response = graph.invoke(
{"debug_mode": debug_mode, "messages": [{"role": "user", "content": message}]},
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("'", "'")
node_with_tooltip = (
f"{node_label}"
)
else:
node_with_tooltip = node_label
node_tree[-1] = node_with_tooltip
node_tree.append(" ")
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):
state = {
"debug_mode": debug_mode,
"messages": (history or []) + [{"role": "user", "content": message}],
}
config = {"configurable": {"thread_id": thread_id}}
start_time = time.time()
streamed_response = ""
final_response = ""
final_node = "validator"
MAX_CONTENT = 500
try:
node_tree = ["π©", " "]
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:{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"π€{random.choice(thinking_verbs)} ...
"
# f""
# f"{node}:{name or ''}:"
# f"Looking for : [{message}] {truncated or '...'}"
# f"
"
)
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"π€ {msg.name} tool: {random.choice(thinking_verbs)} ...
"
# f""
# f"Looking for : [{message}]
"
# f"Tool Args: {tooltip or '(no args)'}
"
# f"{truncated or '...'}"
# f"
"
)
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{escape(truncate_middle(streamed_response))}
"
else:
# Stream intermediate messages with transparent style
if node != final_node:
streamed_response += msg.content
yield f"### { ' β '.join(node_tree) }\n{escape(truncate_middle(streamed_response))}
"
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"π€ {random.choice(thinking_verbs)} ...
"
f""
f"Telling myself: {truncated or '...'}"
f"
"
)
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
if final_response:
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)
supported_scriptures = "\n - ".join(
[
f"π **{scripture['title']}** [source]({scripture['source']})"
for scripture in SanatanConfig.scriptures
]
)
init()
message_textbox = gr.Textbox(
placeholder="Search the scriptures ...", submit_btn=True, stop_btn=True
)
with gr.Blocks(
theme=gr.themes.Citrus(),
title="Sanatan-Ai",
css="""
table {
border-collapse: collapse;
width: 90%;
}
table, th, td {
border: 1px solid #ddd;
padding: 6px;
font-size: small;
}
td {
word-wrap: break-word;
white-space: pre-wrap; /* preserves line breaks but wraps long lines */
max-width: 300px; /* control width */
vertical-align: top;
}
.spinner {
display: inline-block;
width: 1em;
height: 1em;
border: 2px solid transparent;
border-top: 2px solid #333;
border-radius: 50%;
animation: spin 0.8s 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;
animation: pulse 1.5s infinite;
margin-bottom: 5px;
}
@keyframes pulse {
0% { opacity: 0.3; }
50% { opacity: 1; }
100% { opacity: 0.3; }
}
.node-label {
cursor: help;
border-bottom: 1px dotted #aaa;
}
.intermediate-output {
opacity: 0.4;
font-style: italic;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
""",
) as app:
show_sidebar = gr.State(True)
# with gr.Column(scale=1, visible=show_sidebar.value) as sidebar_container:
with gr.Sidebar(open=show_sidebar.value) as sidebar:
# session_id = gr.Textbox(value=f"Thread: {thread_id}")
# gr.Markdown(value=f"{'\n'.join([msg['content'] for msg in intro_messages])}")
gr.Markdown(
value="Namaskaram π I am Sanatan-Bot and I can help you explore the following scriptures:\n\n"
)
async def populate_chat_input(text: str):
buffer = ""
for c in text:
buffer += c
yield buffer
await asyncio.sleep(0.05)
return
def close_side_bar():
print("close_side_bar invoked")
yield gr.update(open=False)
for scripture in sorted(SanatanConfig.scriptures, key=lambda d: d.get("title")):
with gr.Accordion(label=f"{scripture['title']}", open=False):
gr.Markdown(f"* Source: [π click here]({scripture['source']})")
gr.Markdown(f"* Language: {scripture['language']}")
gr.Markdown(f"* Examples :")
with gr.Row():
for example_label, example_text in zip(
scripture["example_labels"], scripture["examples"]
):
btn = gr.Button(value=f"{example_label}", size="sm")
btn.click(close_side_bar,outputs=[sidebar]).then(
populate_chat_input,
inputs=[gr.State(example_text)],
outputs=[message_textbox],
)
gr.Markdown(value="------")
debug_checkbox = gr.Checkbox(label="Debug (Streaming)", value=True)
chatbot = gr.Chatbot(
elem_id="chatbot",
avatar_images=("assets/avatar_user.png", "assets/adiyen_bot.png"),
# value=intro_messages,
label="Sanatan-AI-Bot",
show_copy_button=True,
show_copy_all_button=True,
type="messages",
height=700,
render_markdown=True,
)
chatInterface = gr.ChatInterface(
title="Sanatan-AI",
fn=chat_wrapper,
additional_inputs=[thread_id, debug_checkbox],
chatbot=chatbot,
textbox=message_textbox,
)
app.launch()