Upload 2 files

chat_langraph.py

@@ -0,0 +1,187 @@
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_ollama import ChatOllama
+from langchain_core.messages import BaseMessage,ChatMessage,ToolMessage,AIMessage,SystemMessage,HumanMessage,messages_from_dict
+from langgraph.graph import StateGraph
+from langgraph.graph import add_messages , START , END
+from langgraph.checkpoint.base import CheckpointTuple
+from typing import TypedDict,Annotated , List
+from langchain_core.tools import tool
+from langgraph.prebuilt.tool_node import ToolNode
+from langgraph.checkpoint.sqlite import SqliteSaver
+from langchain_core.prompts import ChatPromptTemplate
+import sqlite3
+import subprocess
+import requests
+from chat_image_analyser import image_analyser
+from erpfunctionalities import *
+from htmltomarkdown import *
+from datetime import datetime , date
+class chatstate(TypedDict):
+    messages : Annotated[List[BaseMessage], add_messages] 
+
+user_id = "23IT441"
+password = "09122005"
+api = "AIzaSyA5zvErF4vUmAoslVzkOBUfvSCSoW0vjEA"
+LANGSEARCH_API_KEY = "sk-f1a8f996f9e44b43adf9943e43e8582b"
+
+# llm = ChatOllama(model = "qwen3:8b" , temperature = 0.5)
+llm = ChatGoogleGenerativeAI(model = "gemini-2.5-flash" ,  temperature = 0.2 , api_key = api)
+system = SystemMessage(
+    content=
+    f"""
+    --> Todays date is : {datetime.today()} + " \n" + Dayno : {datetime.today().date().weekday()}
+    You are an assistant that is practical, tool-aware, and outcome-oriented. Aim for correctness and clarity; avoid hallucinations.  
+Rules (short & general):
+
+1. Prefer text answers and code when the user only asks for examples or explanations.  
+2. If the user explicitly asks you to *create, save, run, or modify* a file or perform a system action (e.g., "create sales.xlsx", "save as report.py", "run this script now"), **call the appropriate tool(s) generally go with python tool to achieve the tasks which are not available directly.** to make that happen. Do not return code only in those cases.  
+3. If you produce code via a tool, and the user has asked for the file/action to be created or executed, write the file and execute it (write_file + run_cmd_command or the relevant tools) unless the user explicitly says "only give code".  
+4. Do not run or write obviously destructive commands (delete, format, rm -rf, wipe, format C:, etc.) without explicit, separate confirmation from the user. For any command that could be destructive, always ask one short confirmation (yes/no).  
+5. Keep tool inputs minimal and focused. When calling a tool, include only the parameters needed. After the tool returns, always validate the result and summarize it for the user (one-line result + any errors).  
+6. If a requested action fails (permission, path, syntax error), explain the failure, propose a fix, and offer to retry when the user approves.  
+7. Do not make assumptions — ask the user if in doubt. If you need clarification, ask a single short question; otherwise make a best-effort decision to proceed.  
+8. If you are unsure whether to call a tool, prefer asking a *single* clarifying question; otherwise make a best-effort decision to proceed.  
+9. Do not make assumptions about the system of the user , u have command prompt access it to gather the information about the system of the user..
+10. Respond in the same language as the user.
+
+Tone: concise, helpful, and decisive. 
+
+""" 
+)
+
+conn = sqlite3.connect("chatbot.db" , check_same_thread=False)
+checkpointer = SqliteSaver(conn=conn)
+@tool
+def add(a : int , b:int):
+    """Adds two numbers """
+    return a+b
+@tool 
+def reverse(string : str):
+    """Reverses a string..."""
+    return string[::-1]
+@tool 
+def evaluate(string : str):
+    """Evaluates a simple mathematical expression but it should be passed as a string..."""
+    print(eval(string))
+    return eval(string)
+@tool 
+def write_file(name : str , extension : str , content : str):
+    """Writes a file given name extension and content of the file , 
+    it only supports txt files and coding files currently , Use the python script to create other file types."""
+    with open(f"{name}.{extension}" , "w" , encoding='utf-8') as f:
+        f.write(content)
+    return f"Content :  {content}  is writtened and saved to {name}.{extension} ."
+
+
+@tool
+def run_python_file(filename:str):
+    """Runs the python file when given the filename of it."""
+    if(filename[-1:-3:-1] != "py"):
+        filename += ".py"
+    msg = run_cmd_command(f"python {filename}")
+    return msg
+@tool
+def run_cmd_command(command: str) -> str:
+
+    """Any windows command is allowed"""
+    try:
+        # Execute the command and capture the output
+        result = subprocess.run(command, shell=True, check=True, text=True, capture_output=True)
+        print(result)
+        return result
+    except subprocess.CalledProcessError as e:
+        return f"An error occurred while executing the command: {e} "
+@tool 
+def search_tool(query:str):
+    """Searches the query on web and gives brief  description (in json) and url names...."""
+    response = requests.post("https://api.langsearch.com/v1/web-search",
+    headers={
+        "Authorization": f"Bearer {LANGSEARCH_API_KEY}",
+        "Content-Type":"application/json"
+    },
+    json={
+        "query":query,
+        "num_results":2
+    })
+
+    return response.json()
+
+@tool
+def image_analysis(query : str)->str:
+    """Analyzes an image given a query and generates a description based on conversational context.
+    It only analyses the file present in the current directory named as image.png , so if user asks for specific file rename the file and save to curr dir , otherwise just use this tool directly if not said anything...
+    Args: query : Any query regarding the image...
+    You dont have to worry about image name and anything it will be handled automatically. Finally just explain the returned text nicely .
+    Returns: str: A textual description of the image. Please give the final ans in detail and in your own words..."""
+    return image_analyser(query)
+
+
+@tool
+def get_fees_tool() -> str:
+    """Gives the fees detail of the student who is chatting..."""
+    return  get_fees_details(user_id, password)
+@tool
+def get_attendance_tool() -> str:
+    """Gives the attendance details of the student who is chatting."""
+    return  attendance(user_id, password)
+@tool
+def get_timetable_tool(target: str) -> str:
+    """Gives the timetable details of the student  who is chatting when given a date in dd-mm-yy format 
+    Timetable contains the holidays information as well."""
+    return  timetable(user_id, password, target)
+@tool
+def get_student_details_tool() -> str:
+    """Gives the information about the student who is chatting."""
+    return  student_details(user_id, password)
+@tool
+def get_result_tool(semester: int) -> str:
+    """Gets the result of the student who is chatting."""
+    return  get_result_details(user_id, password, semester)
+def shouldcontinue(state:chatstate):
+    if(state['messages'][-1].content == 'end'):
+        return 'end'
+    else:
+        return 'llmresponse' 
+def input_node(state:chatstate):
+    return {"messages":state['messages']}
+def llmresponse(state:chatstate):
+    response = llm.invoke(state['messages'])
+    print(response.content)
+    return {'messages':[response]}
+def checktool(state: chatstate):
+    last_msg = state['messages'][-1]
+    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
+        fc = last_msg.additional_kwargs.get('function_call')
+        if fc and 'name' in fc:
+            print("Calling the tool", fc['name'])
+        elif last_msg.tool_calls[0]['name'] :
+            print("Calling the tool " ,last_msg.tool_calls[0]['name'] )
+        return "tool_node"
+    return "end"
+
+tools = [add,reverse,evaluate,run_cmd_command,search_tool,write_file,image_analysis]
+tool_node = ToolNode(tools=tools)
+llm = llm.bind_tools(tools)
+graph = StateGraph(chatstate)
+graph.add_node('input_node' , input_node)
+graph.add_node('llmresponse',llmresponse)
+graph.add_node('tool_node',tool_node)
+graph.add_edge(START , "input_node")
+graph.add_edge("input_node" , "llmresponse")
+graph.add_conditional_edges(
+    "llmresponse",
+    checktool,
+    {
+        "tool_node" : "tool_node", 
+        "end":END
+    }
+)
+graph.add_edge("tool_node", "llmresponse")
+
+workflow = graph.compile(checkpointer=checkpointer)
+res = checkpointer.list(None)
+def get_all_chat_ids():
+    s = set()
+    for chkpoint in res:
+        s.add(chkpoint.config.get('configurable').get('thread_id'))
+    return list(s)

frontend_streamlit.py

@@ -0,0 +1,129 @@
+import streamlit as st
+from chat_langraph import system, workflow, HumanMessage, AIMessage, get_all_chat_ids, ToolMessage
+import uuid
+from chat_title_giver import give_me_the_title
+import os
+import base64  
+import re
+
+st.title("College chatbot")
+st.set_page_config(layout='wide')
+
+def set_title(messages):
+    if messages:
+        title = give_me_the_title(messages)
+        st.session_state.chat_dict[st.session_state.current_chat_id] = title
+
+def set_config():
+    return {'configurable': {'thread_id': st.session_state.current_chat_id}}
+
+
+def load_session_state():
+    if "chats" not in st.session_state:
+        st.session_state.chats = get_all_chat_ids()
+    if "current_chat_id" not in st.session_state:
+        if len(st.session_state.chats) > 0:
+            st.session_state.current_chat_id = st.session_state.chats[-1]
+        else:
+           
+            new_id = str(uuid.uuid4())
+            st.session_state.chats.append(new_id)
+            st.session_state.current_chat_id = new_id
+    if "chat_dict" not in st.session_state:
+        st.session_state.chat_dict = {}
+    if "uploaded_image" not in st.session_state:
+        st.session_state.uploaded_image = None
+
+def render_sidebar():
+    with st.sidebar:
+        st.title("Chats")
+        if st.button("➕ New Chat"):
+            new_id = str(uuid.uuid4())
+            st.session_state.chats.append(new_id)
+            st.session_state.current_chat_id = new_id
+            config = {'configurable': {'thread_id': new_id}}
+            workflow.update_state(config, {"messages": [system]})
+            st.session_state.chat_dict[new_id] = "New Chat"
+            st.rerun()  
+        for chat_id in st.session_state.chats:
+            if st.button(st.session_state.chat_dict.get(chat_id, "New Chat"), key=chat_id):
+                st.session_state.current_chat_id = chat_id
+
+def loadchats():
+    if "current_chat_id" not in st.session_state:
+        return []
+    config = {'configurable': {'thread_id': st.session_state.current_chat_id}}
+    state = workflow.get_state(config)
+    messages = state.values.get("messages", [])
+    for message in messages:
+        if isinstance(message, HumanMessage):
+            if message.content and message.content.strip():
+                with st.chat_message("human"):
+                    st.write(message.content)
+        elif isinstance(message, AIMessage):
+            content = st.expander("Thinking...", expanded=False)
+            if message.content and message.content.strip():
+                text = message.content
+                m = re.search(r'</\s*think\s*>', text, re.IGNORECASE)
+                if m:
+                    thinking_text = text[:m.start()].strip()
+                    rest = text[m.end():].strip()
+                    if thinking_text:
+                        with content:
+                            st.write(thinking_text)
+                    if rest:
+                        with st.chat_message("assistant"):
+                            st.write(rest)
+                else:
+                    with st.chat_message("assistant"):
+                        st.write(text)
+        elif isinstance(message, ToolMessage):
+            with st.chat_message("assistant"):
+                st.info(f"Used a tool")
+    return messages
+
+load_session_state()
+render_sidebar()
+if("current_chat_id" in st.session_state):
+    loadchats()
+    user_input = st.chat_input("Your message: ")
+    if(user_input is not None):
+        response_placeholder = st.empty() 
+        with st.chat_message("human"):
+            st.write(user_input)
+        
+        full_response = ""
+        answer_response = ""
+        is_thinking = True
+        with st.chat_message("assistant"):
+            content = st.expander("Thinking...", expanded=True)
+            resp = st.empty() 
+            resp2 = None
+            with content:
+                resp2 = st.empty()
+            for messages,metadata in workflow.stream({"messages":[system,HumanMessage(user_input)]} , config={'configurable': {'thread_id': st.session_state.current_chat_id}} , stream_mode="messages"):
+                if(isinstance(messages , AIMessage) and messages.content.strip()):
+                    if(messages.content.startswith("<think>")):
+                        is_thinking = True
+                        resp2.markdown(full_response + "|")
+                    if is_thinking:
+                        full_response += messages.content
+                        with content:
+                            resp2.markdown(full_response + "|")
+                        if "</think>" in full_response:
+                            is_thinking = False
+                            full_response = full_response.replace("</think>", "")
+                            resp2.markdown(full_response)
+                    else:
+                        answer_response += messages.content
+                        resp.markdown(answer_response + "|")
+                if(isinstance(messages ,ToolMessage) and messages.content.strip()):
+                    st.info(f"Using appropriate tool")
+
+        st.rerun()    
+
+        
+
+
+
+