upload v1.2

agent/agent_graph/Graph_Nodes.py

@@ -1,152 +1,146 @@
-import os
-from typing import TypedDict, List, Dict, Any, Optional
-from langgraph.graph import StateGraph, START, END
-from agent.agent_graph.StateTasks import *
-from agent.llm.prompts import *
-from typing import get_type_hints
-import json
-from langchain_core.messages import HumanMessage,SystemMessage
-from agent.agent_graph.Graph_Utils import get_egp_to_usd
-from agent.rag.rag import *
-
-def answer_question(state:ProblemState):
-  question = state["question"]
-
-
-  if state["question_type"] == Available_Tasks.LAPTOP_CHOOSE.value:
-      guide_prompt = Tasks_prompts.LAPTOP_THINK.value
-  elif state["question_type"] == Available_Tasks.ROADMAP.value:
-      guide_prompt = Tasks_prompts.ROADMAP.value
-  elif ("node_output_article" in state.keys()):
-      guide_prompt = Tasks_prompts.ROADMAP.value  +"<source>" + state["node_output_article"] +"</source>"
-
-  else:
-      guide_prompt = Tasks_prompts.RAG.value
-
-  
-
-
-  state["answer"] = get_llm_answer(model_llm=state["llm"],messages=state["memory"] + [HumanMessage(content=(guide_prompt + "طلب المستخدم:\n" + question + "\nاهم معلومات المستخدم لاستخدامها ف الدلالة (بالنسبة للسعر هو نفس السعر لكن بالدولار فدائما ركز على السعر بالدولار\n)"+ str(state) + Route_prompts.FINALIZER_PROMPT.value))])
-
-
-  return state
-
-
-def update_context(state:ProblemState):
-    # Control what keys can be modified to preven hullicination
-    keys = get_type_hints(ProblemState).keys()
-    keys_modifiable =  [] # keep it inside the function to make sure append to it not affected by other calls
-
-    _is_rag=False
-    _rag_text_if_exist = ""
-    
-    for i in list(keys):
-
-        # 2️⃣ Prevent modifying question_type after chat starts
-        if i in ["question", "answer", "node_output_article", "memory"]:
-            continue
- 
-        # 2️⃣ Prevent modifying question_type after chat starts
-        # - question_type is only allowed at the very start of the chat
-        # - If it already exists in state → chat has started → modification not allowed
-        # if question exists in rag data skip it
-        if i == "question_type":
-          if "question_type" in state.keys():
-              continue
-          
-          rag_text = state["rag_model"].get_relevant_question(state["question"])
-
-          if bool(rag_text):
-              _is_rag = True
-              _rag_text_if_exist = rag_text # store text to prevent calling twice
-              continue
-
-        # 3️⃣ Any key that reaches here is allowed to be returned by the LLM
-        keys_modifiable.append(i)
-
-    # Make the prompt
-    prompt_llm_new_info = Route_prompts.Context_UPDATOR.value + "\n <KEYS> \n" +str(keys_modifiable) +"\n </KEYS> <Text>"+state["question"]+"</Text>"
-
-
-   
-    llm_new_info = get_llm_answer(model_llm=state["llm"],messages = [HumanMessage(prompt_llm_new_info)])
-
-
-    # Save and Process the returned json to prevent hallucination
-    try:
-        llm_new_info = json.loads(llm_new_info)
-        for key in llm_new_info.keys():
-            if key in keys_modifiable:
-                state[key] = llm_new_info[key]
-
-        # check if rag
-        if _is_rag:
-          state["question_type"] = Available_Tasks.QUESTION.value
-          state["node_output_article"] = _rag_text_if_exist 
-
-        # Check if all_ok can be answered now
-        last_question = state.get("answer", "")
-        check_finalized_prompt = (
-            "عندك ده اخر سوال واجابة"+
-            last_question +
-            state["question"] +
-            "هل معنى ذلك ان المستخدم اكد على الفهم الصحيح؟" +
-            "رجع فقط BOOL (True/False)"
-        )
-
-        check_finalized = get_llm_answer(model_llm=state["llm"], messages = [HumanMessage(
-            check_finalized_prompt
-        )])
-        state['all_ok'] =  check_finalized.strip().lower() == "true" # If wrong parsed it's false
-
-
-
-    except Exception as e:
-       print("Context was not updated due to error : ",e)
-
-
-    return state
-
-
-
-def convertPriceToDollar(state:ProblemState):
-    if "price" in state.keys():
-      state["price"] = get_egp_to_usd(state["price"])
-
-    return state
-
-def step(state:ProblemState):
-    next_topic = None
-
-    if "question_type" not in state.keys():
-        next_topic = "question_type"
-
-
-    else:
-        for i in task_steps[state.get("question_type")]:
-            if i not in state.keys():
-                next_topic = i
-                break
-    # Only after finishing the to do list of the question type we can ask for all_ok to confirm
-    if (not next_topic) and ("all_ok" not in state.keys() or (state["all_ok"]==False))  and "question_type" in state.keys():
-
-            next_topic = "all_ok"
-
-
-
-    step_prompt = (System_prompts.STATE_DESCRIBE.value + f"<order>{next_topic} </order> <state>{state}</state>" + Route_prompts.FINALIZER_PROMPT_STEP.value)
-    state['answer'] =  get_llm_answer(model_llm=state["llm"],messages = [HumanMessage(step_prompt)])
-
-    return state
-
-def search_knowledgebase(state:ProblemState):
-    """
-    Search the vector database for relevant contexts.
-    """
-    # fetch top 3 relevant docs
-    state["node_output_article"] = state["rag_model"].get_relevant_question(state["question"])
-
-    return state
-
-def get_llm_answer(model_llm=None,messages=[HumanMessage(content="hi")]):
-    return model_llm.invoke(messages).content
+import os
+from typing import TypedDict, List, Dict, Any, Optional
+from langgraph.graph import StateGraph, START, END
+from agent.agent_graph.StateTasks import *
+from agent.llm.prompts import *
+from typing import get_type_hints
+import json
+from langchain_core.messages import HumanMessage,SystemMessage,AIMessage
+from agent.rag.rag import *
+from typing import Any
+
+class GraphNode:
+    def __init__(self,llm_report,llm_intro_with_tools,llm_exe_with_tools):
+        self.llm_report = llm_report
+        self.llm_intro_with_tools = llm_intro_with_tools
+        self.llm_exe_with_tools = llm_exe_with_tools
+        
+    def make_report(self,state) -> dict[str,Any]:
+        """
+        Generate a final report of roadmap or Laptop file.
+
+        Returns:
+            {"final_ans_report": string of response}
+        """
+
+        print(state)
+
+        prompt: str | None = None  # Initialize prompt to None with a type hint
+
+        intent = state.get("intent")
+
+        if intent == Available_Tasks.ROADMAP.value:
+            prompt = prompt_roadmap_maker
+        elif intent == Available_Tasks.LAPTOP_CHOOSE.value:
+            prompt = prompt_laptop_maker
+
+        if prompt is None:
+            final = state.get("final_ans_report", "")
+
+            if hasattr(final, "content"):
+                final = final.content
+
+            return {"final_ans_report": final}
+
+        # Ensure final_ans_report is a string by accessing its content attribute if it's an AIMessage object
+        final_ans_report_content = state["final_ans_report"]
+        if hasattr(final_ans_report_content, 'content'):
+            final_ans_report_content = final_ans_report_content.content
+
+        messages = prompt + "<case>" + final_ans_report_content + "</case>"
+
+
+        response = self.llm_report.invoke(messages)
+
+
+        return {"final_ans_report": response.content}
+
+
+
+    def Conv_Manager_Agent_Node(self,state: AgentState) -> dict:
+        """The 'Thought' step of the ReAct loop.
+
+        Reads the full message history, calls the LLM (with tools bound), and returns
+        its response. LangGraph's `add_messages` reducer appends this response to the
+        state -- it does not replace the history.
+        """
+        messages = [SystemMessage(content=Conv_Manager_Prompt)] + state["messages"]
+        response = self.llm_intro_with_tools.invoke(messages)
+
+        updates = {
+            "messages": [response]
+        }
+
+        try:
+            data = json.loads(response.content)
+            if "intent" in data:
+                updates["intent"] = data["intent"]
+        except:
+            pass
+
+        return updates
+
+
+    def Planner_Agent_Node(self,state: AgentState) -> dict:
+        """The 'Thought' step of the ReAct loop.
+
+        Reads the full message history, calls the LLM (with tools bound), and returns
+        its response. LangGraph's `add_messages` reducer appends this response to the
+        state -- it does not replace the history.
+        """
+        plan = ""
+        if state["intent"] == Available_Tasks.LAPTOP_CHOOSE.value:
+            plan = Laptop_plan
+        elif state["intent"] == Available_Tasks.QUESTION.value:
+            plan = question_plan
+        elif state["intent"] == Available_Tasks.ROADMAP.value:
+            plan = Roadmap_plan
+
+        updates = {"plan": plan}
+
+        # Explicitly carry intent forward
+        if "intent" in state:
+            updates["intent"] = state["intent"]
+        return updates
+
+
+
+    def ReACT_Agent_Node(self,state: AgentState) -> dict:
+        """The 'Thought' step of the ReAct loop.
+
+        Reads the full message history, calls the LLM (with tools bound), and returns
+        its response. LangGraph's `add_messages` reducer appends this response to the
+        state -- it does not replace the history.
+        """
+        messages = [SystemMessage(content=state['plan'])] + state["messages"]
+        response = self.llm_exe_with_tools.invoke(messages)
+
+        updates = {
+            "messages": [response]
+        }
+
+        # Only attempt to extract final_ans_report if it's an AIMessage and has content
+        if isinstance(response, AIMessage) and response.content:
+            # If the LLM returned tool calls, it's an intermediate step, not the final report.
+            if not getattr(response, "tool_calls", None):
+                try:
+                    data = json.loads(response.content)
+                    # Check for the specific final_ans_report key and if it's a string
+                    if "final_ans_report" in data and isinstance(data["final_ans_report"], str):
+                        updates["final_ans_report"] = data["final_ans_report"]
+                    else: # Generic JSON not matching final_ans_report structure or a question
+                        pass # Don't set final_ans_report
+                except json.JSONDecodeError:
+                    # If it's not a JSON, and there are no tool calls, it might be a plain text answer.
+                    if state["intent"] == Available_Tasks.QUESTION.value:
+                        # For QUESTION intent, plain text is the final answer.
+                        updates["final_ans_report"] = response.content
+                    pass # Not JSON, not a QUESTION final answer, and not a tool call, so no final_ans_report set.
+
+        # Explicitly carry intent and plan forward
+        if "intent" in state:
+            updates["intent"] = state["intent"]
+        if "plan" in state:
+            updates["plan"] = state["plan"]
+
+        return updates

agent/agent_graph/Graph_Routes.py

@@ -1,19 +1,29 @@
-from agent.agent_graph.StateTasks import *
-
-def is_question_clear(state:ProblemState): # not check type , only check to do list
-  is_clear = True # as used with and later
-
-  is_clear = is_clear and ("question" in state.keys())
-
-  is_clear = is_clear and ("question_type" in state.keys()) and (state.get("question_type") in task_steps.keys())
-
-  if "question_type" in state.keys():
-      for step in task_steps[state.get("question_type")]:
-
-          is_clear = is_clear and (step in state.keys())
-
-  is_clear = is_clear and (state["all_ok"]==True)
-
-  return is_clear
-
-
+from agent.agent_graph.StateTasks import AgentState
+from langgraph.graph import StateGraph, MessagesState, START, END,state
+
+def conv_state_check(state:AgentState):
+    print("CONV RUN ", state)
+    last = state["messages"][-1]
+
+    if state.get("intent"):
+        print("intent =", state.get("intent"))
+        return "Planner_Agent"
+
+    if getattr(last, "tool_calls", None):
+        return "tools"
+
+    return END
+
+def react_check(state:AgentState):
+    print("REACT RUN ", state)
+
+    last = state["messages"][-1]
+
+    if state.get("final_ans_report"):
+        print("report =")
+        return "Report_Agent"
+
+    if getattr(last, "tool_calls", None):
+        return "tools"
+
+    return END

agent/agent_graph/Graph_Tools.py

@@ -0,0 +1,126 @@
+from langchain_core.tools import tool, StructuredTool
+from langgraph.types import interrupt, Command
+import requests
+import warnings
+from agent.rag.rag import init_rag, get_relevant_question
+
+warnings.filterwarnings("ignore", category=UserWarning)
+
+text_encoder_model = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+price_url = "https://open.er-api.com/v6/latest/EGP"
+
+class GraphTools:
+
+    def __init__(self,path_file:str,text_encoder_model:str=text_encoder_model,price_url:str=price_url):
+        print("Initializing GraphTools...")
+        self.text_encoder_model = text_encoder_model
+        self.path_file = path_file
+        self.price_url = price_url
+        self.model, self.corpus_embeddings, self.corpus, self.answers = self.init_rag()
+
+    ##############################
+    ## ASK USER Interrupt Tool
+    ##############################
+    def ask_user(self, question: str) -> str:
+        """Ask the user in Arabic one clear clarifying question when more information is needed.
+        Ask only one question per call and wait for the user's answer before asking
+        another if necessary.
+        """
+
+        answer = interrupt({"question": question})
+        return answer
+
+
+
+    ##############################
+    ## EGP to usd conv. tool
+    ##############################
+
+    def get_egp_to_usd(self, egp_amount:float) -> float :
+        """Convert an amount from Egyptian Pounds (EGP) to US Dollars (USD)
+        using the latest available exchange rate.
+
+        Use this tool whenever you found prices.
+
+        Args:
+            egp_amount: The amount in Egyptian Pounds (EGP) to convert.
+
+        Returns:
+            The equivalent amount in US Dollars (USD).
+
+            Returns:
+            - Converted USD amount on success.
+            - None if the exchange-rate service returns an unexpected response.
+            - -1 if the request fails or another error occurs.
+        """
+
+        
+
+        try:
+            response = requests.get(self.price_url, timeout=8)
+            response.raise_for_status()  # raise error for bad status
+            data = response.json()
+
+            if data.get("result") != "success":
+                print("API error:", data)
+                return None
+
+            rate = data["rates"]["USD"]
+            usd_amount = egp_amount * rate
+
+            return usd_amount
+
+        except Exception as e:
+            print("Error:", e)
+            return -1
+
+    ####################
+    ## RAG
+    ####################
+    def init_rag(self):
+       return init_rag(self.path_file, self.text_encoder_model)
+    
+
+    def get_relevant_question(self,query:str) -> str:
+        """
+        Retrieve the most relevant question-answer pair for a user's query
+        using semantic similarity.
+
+        Use this tool when the user asks a question that may already have an
+        existing answer in the knowledge base. The tool searches semantically
+        rather than by exact keyword matching.
+
+        Args:
+            query: The user's question or search query.
+
+        Returns:
+            A formatted string containing the most relevant question, its answer,
+            and the similarity score if a sufficiently similar match is found.
+
+            Returns None if no match meets the similarity threshold.
+        """
+        
+        return get_relevant_question(self.model, self.corpus_embeddings, self.corpus, self.answers, query)
+
+
+
+    def get_tools(self):
+        ask_user_tool = StructuredTool.from_function(
+                func=self.ask_user,
+                name="ask_user",
+                description=self.ask_user.__doc__,
+            )
+
+        dollar_tool =  StructuredTool.from_function(
+                        func=self.get_egp_to_usd,
+                        name="get_egp_to_usd",
+                        description=self.get_egp_to_usd.__doc__,
+                    )
+
+        rag_tool  = StructuredTool.from_function(
+                        func=self.get_relevant_question,
+                        name="get_relevant_question",
+                        description=self.get_relevant_question.__doc__,
+                    )
+    
+        return [ask_user_tool,dollar_tool,rag_tool]

agent/agent_graph/Graph_Utils.py

@@ -1,25 +1,16 @@
-import requests
-USD_url = "https://open.er-api.com/v6/latest/EGP"
-
-def get_egp_to_usd(egp_amount):
-
-    try:
-        response = requests.get(USD_url, timeout=8)
-        response.raise_for_status()  # raise error for bad status
-        data = response.json()
-
-        if data.get("result") != "success":
-            print("API error:", data)
-            return None
-
-        rate = data["rates"]["USD"]
-        usd_amount = egp_amount * rate
-
-        return usd_amount
-
-    except Exception as e:
-        print("Error:", e)
-        return None
-
-
-get_egp_to_usd(1)
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_huggingface import HuggingFaceEndpoint,ChatHuggingFace
+
+def get_llm_obj():
+    llm = HuggingFaceEndpoint(
+        repo_id="openai/gpt-oss-20b",#"deepseek-ai/DeepSeek-V3.2-Exp",#"openai/gpt-oss-20b",
+        task='conversational',
+        provider="auto",
+            max_new_tokens=2048
+    )
+    llm = ChatHuggingFace(llm=llm)
+
+
+    #llm = ChatGoogleGenerativeAI(model="gemini-3-flash-preview", google_api_key=GEMINI_TOKEN)
+
+    return llm

agent/agent_graph/StateTasks.py

@@ -1,49 +1,26 @@
-import os
-from typing import TypedDict, List, Dict, Any, Optional
-from enum import Enum
-
-# 1) Core Tasks
-class Available_Tasks(Enum):
-    LAPTOP_CHOOSE = "LAPTOP_CHOOSE"
-    QUESTION = "QUESTION" # EAG of well known
-    ROADMAP = "ROADMAP"
-    PROGRAMMING = "PROGRAMMING"
-    GENERAL = "GENERAL" # last option with notice
-    HELLO = "HELLO" # this is to generally detect hello , but in use update_context prompt control to not classify it as core type
-
-# 2) Task Steps
-task_steps = {
-    Available_Tasks.GENERAL.value : [],
-    Available_Tasks.LAPTOP_CHOOSE.value : ["price","usage","other_concern"],
-    Available_Tasks.QUESTION.value : [],
-    Available_Tasks.PROGRAMMING.value : [],
-    Available_Tasks.ROADMAP.value : ["career","level","experience","skills"],
-    Available_Tasks.HELLO.value : []
-}
-
-# 3) Tasks State
-class ProblemState(TypedDict):
-    # main
-    question: str
-    answer: Optional[str]
-    node_output_article : Optional[str]
-    memory: List[Dict[str, Any]] # filled by list of Human/Sytem messages ... as in gui
-
-    # to do lists
-    # same as defined in task_Steps
-    question_type : Optional[str] # --> must be of Available_Tasks values
-    price : Optional[str]
-    usage : Optional[str]
-    other_concern : Optional[str]
-    career : Optional[str]
-    level : Optional[str]
-    experience : Optional[List[str]]
-    skills : Optional[List[str]]
-    all_ok : Optional[bool]
-
-    # Models
-    llm: Optional[Any]
-    rag_model: Optional[Any]
-
-
-
+import os
+from typing import TypedDict, List, Dict, Any, Optional
+from enum import Enum
+from typing import Annotated
+from typing_extensions import TypedDict
+from langgraph.graph import add_messages
+
+
+class Available_Tasks(Enum):
+    LAPTOP_CHOOSE = "LAPTOP_CHOOSE"
+    QUESTION = "QUESTION"
+    ROADMAP = "ROADMAP"
+
+
+
+av_tasks = [i.value for i in Available_Tasks]
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list, add_messages]
+
+    intent: str
+
+    plan: str
+
+    final_ans_report :str

agent/agent_graph/graph.py

@@ -1,32 +1,91 @@
-from IPython.display import Image, display
-from langgraph.graph import StateGraph, START, END
-from agent.agent_graph.Graph_Nodes import *
-from agent.agent_graph.Graph_Routes import *
- 
-problem_graph = StateGraph(ProblemState)
-
-# Add nodes
-problem_graph.add_node("answer_question",answer_question)
-problem_graph.add_node("update_context",update_context)
-problem_graph.add_node("convertPriceToDollar",convertPriceToDollar)
-problem_graph.add_node("step",step)
-
-
-# Routes
-problem_graph.add_conditional_edges(
-    "update_context",
-    is_question_clear,
-    {
-        True: "convertPriceToDollar",
-        False: "step"
-    }
-)
-
-
-# Edges
-problem_graph.add_edge(START,"update_context")
-problem_graph.add_edge("convertPriceToDollar","answer_question")
-
-# Finalize
-compiled_graph = problem_graph.compile()
-#display(Image(compiled_graph.get_graph().draw_mermaid_png()))
+from IPython.display import Image, display
+from langgraph.graph import StateGraph, START, END
+from agent.agent_graph.Graph_Nodes import *
+from agent.agent_graph.Graph_Routes import *
+from agent.agent_graph.Graph_Utils import *
+from agent.agent_graph.Graph_Tools import GraphTools
+from agent.agent_graph.Graph_Nodes import GraphNode
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.prebuilt import ToolNode, tools_condition
+
+
+
+class AgentGraph:
+    def __init__(self,checkpointer,path_rag_file):
+        self.path_rag_file = path_rag_file
+        self.checkpointer = checkpointer
+        
+        #self.checkpointer.setup()   # only first time
+        self.graph = self._build_graph()
+
+    def _build_graph(self):
+
+        
+        llm_intro = get_llm_obj()
+        llm_exe = get_llm_obj()
+        llm_report = get_llm_obj()
+
+        
+        GraphToolsOBJ = GraphTools(path_file=self.path_rag_file)
+        ask_user_tool,dollar_tool,rag_tool = GraphToolsOBJ.get_tools()
+
+        tools_conv = [ask_user_tool]
+        tools_exe = [ask_user_tool, dollar_tool, rag_tool]
+        
+        llm_intro_with_tools = llm_intro.bind_tools(tools_conv)
+        llm_exe_with_tools = llm_exe.bind_tools(tools_exe)
+
+
+        GraphNodeOBJ = GraphNode(llm_report,llm_intro_with_tools,llm_exe_with_tools)
+
+        builder = StateGraph(AgentState)
+        builder.add_node("Conv_Manager_Agent",GraphNodeOBJ.Conv_Manager_Agent_Node)
+        builder.add_node("Planner_Agent",GraphNodeOBJ.Planner_Agent_Node)
+        builder.add_node("ReACT_Agent",GraphNodeOBJ.ReACT_Agent_Node)
+        builder.add_node("tools_conv", ToolNode(tools_conv))
+        builder.add_node("tools_react", ToolNode(tools_exe))
+
+        builder.add_node("make_report", GraphNodeOBJ.make_report)
+
+
+        builder.add_edge(START, "Conv_Manager_Agent")
+        builder.add_conditional_edges(
+            "Conv_Manager_Agent",
+            conv_state_check,       # inspects the last AIMessage for tool_calls
+            {
+                "tools": "tools_conv",
+                "Planner_Agent":"Planner_Agent",
+                END:END
+            },
+        )
+        builder.add_edge("tools_conv", "Conv_Manager_Agent")
+
+
+
+
+        builder.add_edge("Planner_Agent", "ReACT_Agent")
+
+        builder.add_conditional_edges(
+            "ReACT_Agent",
+            react_check,       # inspects the last AIMessage for tool_calls
+            {
+                "tools": "tools_react",
+                "Report_Agent":"make_report",
+                END:END
+            },
+        )
+
+        builder.add_edge("tools_react", "ReACT_Agent")
+
+        builder.add_edge("make_report",END)
+
+        
+       
+        graph = builder.compile(checkpointer=self.checkpointer)
+        #print("Graph compiled successfully ✅")
+        #display(Image(graph.get_graph().draw_mermaid_png()))
+        return graph
+
+    def get_graph(self):
+        return self.graph
+