Update chat_langraph.py

chat_langraph.py

@@ -1,187 +1,126 @@
-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)
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_core.messages import BaseMessage, ToolMessage, AIMessage, SystemMessage, HumanMessage
+from langgraph.graph import StateGraph, add_messages, START, END
+from langgraph.checkpoint.sqlite import SqliteSaver
+from typing import TypedDict, Annotated, List
+from langchain_core.tools import tool
+from langgraph.prebuilt.tool_node import ToolNode
+import sqlite3
+import subprocess
+import requests
+from datetime import datetime
+
+class chatstate(TypedDict):
+    messages: Annotated[List[BaseMessage], add_messages]
+
+
+api = "AIzaSyA5zvErF4vUmAoslVzkOBUfvSCSoW0vjEA"
+LANGSEARCH_API_KEY = "sk-f1a8f996f9e44b43adf9943e43e8582b"
+
+llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0.2, api_key=api)
+
+system = SystemMessage(
+    content=f"""
+    --> Today's date: {datetime.today()}  
+    Day number: {datetime.today().date().weekday()}
+    You are a practical, tool-aware assistant. Aim for correctness and clarity. Avoid hallucinations.
+    Do not provide internal information of the system . 
+Rules:
+1. Prefer text answers and code when examples/explanations are asked.
+2. Explicit requests to create/run files → call appropriate tool.
+3. Avoid destructive commands without confirmation.
+4. Keep tool inputs minimal.
+Tone: concise, helpful, decisive.
+"""
+)
+
+conn = sqlite3.connect("chatbot.db", check_same_thread=False)
+checkpointer = SqliteSaver(conn=conn)
+
+
+@tool
+def add(a: int, b: int):
+    return a + b
+
+
+@tool
+def reverse(string: str):
+    return string[::-1]
+
+
+@tool
+def evaluate(string: str):
+    return eval(string)
+
+
+@tool
+def write_file(name: str, extension: str, content: str):
+    with open(f"{name}.{extension}", "w", encoding="utf-8") as f:
+        f.write(content)
+    return f"Content saved to {name}.{extension}"
+
+
+@tool
+def run_cmd_command(command: str) -> str:
+    """Run a safe shell command on linux """
+    try:
+        result = subprocess.run(command, shell=True, check=True, text=True, capture_output=True)
+        return result.stdout
+    except subprocess.CalledProcessError as e:
+        return f"Error: {e}"
+
+
+@tool
+def search_tool(query: str):
+    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()
+
+
+def shouldcontinue(state: chatstate):
+    return "end" if state["messages"][-1].content == "end" else "llmresponse"
+
+
+def input_node(state: chatstate):
+    return {"messages": state["messages"]}
+
+
+def llmresponse(state: chatstate):
+    response = llm.invoke(state["messages"])
+    return {"messages": [response]}
+
+
+def checktool(state: chatstate):
+    last_msg = state["messages"][-1]
+    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
+        return "tool_node"
+    return "end"
+
+
+tools = [add, reverse, evaluate, run_cmd_command, search_tool, write_file]
+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)
+
+
+def get_all_chat_ids():
+    s = set()
+    for chkpoint in checkpointer.list(None):
+        s.add(chkpoint.config.get("configurable").get("thread_id"))
+    return list(s)