import uuid 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"{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_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 = ["🚩", " "] 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"
πŸ€”{random.choice(thinking_verbs)} ...
" 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)} ...
" 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 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 def init_session(): return str(uuid.uuid4()) MAX_MESSAGES_IN_CONVERSATION = 5 async def limited_chat_wrapper( message, history, thread_id, debug, preferred_language, count ): # increment **after processing the message** count += 1 # warn before reset if count == MAX_MESSAGES_IN_CONVERSATION - 1: yield [ { "role": "system", "content": "⚠️ You are allowed to ask one more follow-up. The next question will be considered a new conversation. Please wait ... processing your request ...", } ], thread_id, count await asyncio.sleep(1) # reset if count > MAX_MESSAGES_IN_CONVERSATION: thread_id = init_session() history = [] count = 1 yield [ { "role": "system", "content": "πŸ”„ This is now considered a new question. Don't worry, your message shall still be processed! If I am giving irrelevant responses, you know why :-)", } ], thread_id, count await asyncio.sleep(1) # normal flow: stream from your original chat_wrapper final_chunk = [] async for chunk in chat_wrapper( message, history, thread_id, debug, preferred_language ): yield chunk, thread_id, count final_chunk = chunk # Simulating LLM Response # for i in range(5): # final_chunk += [{ # "role": "assistant", # "content": f"Simulated LLM output {i+1}", # }] # yield final_chunk, thread_id, count # await asyncio.sleep(0.25) yield final_chunk, thread_id, count