Update agent.py

agent.py

@@ -491,48 +491,274 @@ def get_llm(provider: str, config: dict):
 # ----------------------------------------------------------------
 # Planning & Execution Logic
 # ----------------------------------------------------------------
-def planner(question: str) -> list:
-    if "calculate" in question or any(op in question for op in ["add", "subtract", "multiply", "divide", "modulus"]):
-        return ["math"]
-    elif "wiki" in question or "who is" in question.lower():
-        return ["wiki_search"]
-    else:
-        return ["default"]
+def planner(question: str, tools: list) -> list:
+    question = question.lower().strip()
+
+    # Define general keywords for various intents (without hardcoding the tool names)
+    intent_keywords = {
+        "math": ["calculate", "evaluate", "add", "subtract", "multiply", "divide", "modulus", "plus", "minus", "times"],
+        "wiki_search": ["who is", "what is", "define", "explain", "tell me about", "overview of"],
+        "web_search": ["search", "find", "look up", "google", "latest news", "current info"],
+        "arxiv": ["arxiv", "research paper", "scientific paper", "preprint"],
+        "youtube": ["youtube", "watch", "play video", "show me a video"],
+        "video_analysis": ["analyze video", "summarize video", "video content"],
+        "data_analysis": ["analyze", "plot", "graph", "data", "visualize"],
+        "wikidata_query": ["wikidata", "sparql", "run sparql", "query wikidata"],
+        "general_qa": ["why", "how", "difference between", "compare", "what happens", "reason for", "cause of", "effect of"]
+    }
+
+    matched_tools = []
+
+    # Loop over tools and match based on descriptions
+    for tool in tools:
+        # Get tool description
+        tool_description = getattr(tool, "description", "").lower()
+
+        # Check if any keywords match tool's description or if they fit general intent categories
+        for intent, keywords in intent_keywords.items():
+            if any(keyword in question for keyword in keywords) and intent in tool_description:
+                matched_tools.append(tool)
+                break  # No need to check other keywords for this tool once matched
+
+    # If no matched tool found, fallback to general-purpose tools or default
+    if not matched_tools:
+        matched_tools = [tool for tool in tools if "default" in getattr(tool, "name", "").lower() or "qa" in getattr(tool, "description", "").lower()]
+
+    return matched_tools if matched_tools else [tools[0]]  # Return the first tool as a last resort
+
 
 
 def task_classifier(question: str) -> str:
-    if any(op in question.lower() for op in ["add", "subtract", "multiply", "divide", "modulus"]):
+    question = question.lower().strip()
+
+    # Context-aware intent patterns
+    if any(phrase in question for phrase in [
+        "calculate", "how much is", "what is the result of", "evaluate", "solve"
+    ]) or any(op in question for op in ["add", "subtract", "multiply", "divide", "modulus", "plus", "minus", "times"]):
         return "math"
-    elif "who" in question.lower() or "what is" in question.lower():
+
+    elif any(phrase in question for phrase in [
+        "who is", "what is", "define", "explain", "tell me about", "give me an overview of"
+    ]):
         return "wiki_search"
-    else:
-        return "default"
+
+    elif any(phrase in question for phrase in [
+        "search", "find", "look up", "google", "get the latest", "current news", "trending"
+    ]):
+        return "web_search"
+
+    elif any(phrase in question for phrase in [
+        "arxiv", "latest research", "scientific paper", "research paper", "preprint"
+    ]):
+        return "arxiv"
+
+    elif any(phrase in question for phrase in [
+        "youtube", "watch", "play the video", "show me a video"
+    ]):
+        return "youtube"
+
+    elif any(phrase in question for phrase in [
+        "analyze video", "summarize video", "what happens in the video", "video content"
+    ]):
+        return "video_analysis"
+
+    elif any(phrase in question for phrase in [
+        "analyze", "visualize", "plot", "graph", "inspect data", "explore dataset"
+    ]):
+        return "data_analysis"
+
+    elif any(phrase in question for phrase in [
+        "sparql", "wikidata", "query wikidata", "run sparql", "wikidata query"
+    ]):
+        return "wikidata_query"
+
+    return "default"
+
 
 # Function to extract math operation from the question
 def extract_math_from_question(question: str):
-    """Extract numbers and operator from a math question."""
-    match = re.search(r'(\d+)\s*(\+|\-|\*|\/|\%)\s*(\d+)', question)
+    question = question.lower()
+    # Map word-based operations to symbols
+    ops = {
+        "add": "+", "plus": "+",
+        "subtract": "-", "minus": "-",
+        "multiply": "*", "times": "*",
+        "divide": "/", "divided by": "/",
+        "modulus": "%", "mod": "%"
+    }
+
+    for word, symbol in ops.items():
+        question = question.replace(word, symbol)
+
+    # Match expressions like "4 + 5"
+    match = re.search(r'(\d+)\s*([\+\-\*/%])\s*(\d+)', question)
     if match:
         num1 = int(match.group(1))
         operator = match.group(2)
         num2 = int(match.group(3))
         return num1, operator, num2
-    else:
-        return None
+    return None
+
+
+# Example tool set (adjust these to match your actual tool names)
+tools = {
+    "math": calc_tool,          # Example tool for math tasks
+    "wiki_search": wiki_tool,   # Example tool for wiki search tasks
+    "retriever": retriever_tool, # Example tool for retriever tasks
+    "default": default_tool     # Fallback tool
+}
+
+# The task order can also include the tools for each task
+priority_order = [
+    {"task": "math", "tool": "math"},        # Priority task and tool
+    {"task": "wiki_search", "tool": "wiki_search"},
+    {"task": "retriever", "tool": "retriever"},
+    {"task": "default", "tool": "default"}  # Fallback tool
+]
 
 def decide_task(state: dict) -> str:
-    return planner(state["question"])[0]
+    """Decides which task to perform based on the current state."""
+    
+    # Get the list of tasks from the planner
+    tasks = planner(state["question"])
+    print(f"Available tasks: {tasks}")  # Debugging: show all possible tasks
+    
+    # Check if the tasks list is empty or invalid
+    if not tasks:
+        print("❌ No valid tasks were returned from the planner.")
+        return "default"  # Return a default task if no tasks were generated
+
+    # If there are multiple tasks, we can prioritize based on certain conditions
+    task = tasks[0]  # Default to the first task in the list
+    if len(tasks) > 1:
+        print(f"⚠️ Multiple tasks found. Deciding based on priority.")
+        # Example logic to prioritize tasks, adjust based on your use case
+        task = prioritize_tasks(tasks)
+    
+    print(f"Decided on task: {task}")  # Debugging: show the final task
+    return task
+
 
+def prioritize_tasks(tasks: list) -> str:
+    """Prioritize tasks based on certain conditions or criteria, including tools."""
+    # Sort tasks based on priority_order mapping
+    for priority in priority_order:
+        # Check if any task matches the priority task type
+        for task in tasks:
+            if priority["task"] in task:
+                print(f"✅ Prioritizing task: {task} with tool: {priority['tool']}")  # Debugging: show the chosen task and tool
+                # Assign the correct tool based on the task
+                tool = tools.get(priority["tool"], tools["default"])  # Default to 'default_tool' if not found
+                return task, tool
+
+    # If no priority task is found, return the first task with its default tool
+    return tasks[0], tools["default"]
+
+
+def process_question(question: str):
+    """Process the question and route it to the appropriate tool."""
+    # Get the tasks from the planner
+    tasks = planner(question)
+    print(f"Tasks to perform: {tasks}")
+
+    task_type, tool = decide_task({"question": question})
+    print(f"Next task: {task_type} with tool: {tool}")
+
+    if node_skipper({"question": question}):
+        print(f"Skipping task: {task_type}")
+        return "Task skipped."
+
+    try:
+        # Execute the corresponding tool for the task type
+        if task_type == "wiki_search":
+            response = tool.run(question)  # Assuming tool is wiki_tool
+        elif task_type == "math":
+            response = tool.run(question)  # Assuming tool is calc_tool
+        elif task_type == "retriever":
+            response = tool.run(question)  # Assuming tool is retriever_tool
+        else:
+            response = tool.run(question)  # Default tool
+
+        return generate_final_answer({"question": question}, {task_type: response})
+
+    except Exception as e:
+        print(f"❌ Error: {e}")
+        return f"Sorry, I encountered an error: {str(e)}"
+
+
+
+
+# To store previously asked questions and timestamps (simulating state persistence)
+recent_questions = {}
 
 def node_skipper(state: dict) -> bool:
+    """
+    Determines whether to skip the task based on the state.
+    This could include:
+    1. Repeated or similar questions
+    2. Irrelevant or empty questions
+    3. Tasks that have already been processed recently
+    """
+    question = state.get("question", "").strip()
+    
+    if not question:
+        print("❌ Skipping: Empty or invalid question.")
+        return True  # Skip if no valid question
+
+    # 1. Skip if the question has already been asked recently (within a given time window)
+    # Here, we're using a simple example with a 5-minute window (300 seconds).
+    if question in recent_questions:
+        last_asked_time = recent_questions[question]
+        time_since_last_ask = time.time() - last_asked_time
+        if time_since_last_ask < 300:  # 5-minute threshold
+            print(f"❌ Skipping: The question has been asked recently. Time since last ask: {time_since_last_ask:.2f} seconds.")
+            return True  # Skip if the question was asked within the last 5 minutes
+    
+    # 2. Skip if the question is irrelevant or not meaningful enough
+    irrelevant_keywords = ["blah", "nothing", "invalid", "nonsense"]
+    if any(keyword in question.lower() for keyword in irrelevant_keywords):
+        print("❌ Skipping: Irrelevant or nonsense question.")
+        return True  # Skip if the question contains irrelevant keywords
+
+    # 3. Skip if the task has already been completed for this question (based on a unique task identifier)
+    if "last_response" in state and state["last_response"]:
+        print("❌ Skipping: Task has already been processed recently.")
+        return True  # Skip if a response has already been given
+
+    # 4. Skip based on a condition related to the task itself
+    # Example: Skip math-related tasks if the result is already known or trivial
+    if "math" in state.get("question", "").lower():
+        # If math is trivial (like "What is 2+2?")
+        trivial_math = ["2 + 2", "1 + 1", "3 + 3"]
+        if any(trivial_question in question for trivial_question in trivial_math):
+            print(f"❌ Skipping trivial math question: {question}")
+            return True  # Skip if the math question is trivial
+    
+    # 5. Skip based on external factors (e.g., current time, system load, etc.)
+    # Example: Avoid processing tasks at night if that's part of the business logic
+    current_hour = time.localtime().tm_hour
+    if current_hour >= 22 or current_hour < 6:
+        print("❌ Skipping: It's night time, not processing tasks.")
+        return True  # Skip tasks during night time (e.g., between 10 PM and 6 AM)
+
+    # If none of the conditions matched, don't skip the task
     return False
 
+# Update recent questions (for simulating repeated question check)
+def update_recent_questions(question: str):
+    """Update the recent questions dictionary with the current timestamp."""
+    recent_questions[question] = time.time()
+
+
 
 def generate_final_answer(state: dict, task_results: dict) -> str:
+    """Generate a final answer based on the results of the task."""
     if "wiki_search" in task_results:
         return f"📚 Wiki Summary:\n{task_results['wiki_search']}"
     elif "math" in task_results:
         return f"🧮 Math Result: {task_results['math']}"
+    elif "retriever" in task_results:
+        return f"🔍 Retrieved Info: {task_results['retriever']}"
     else:
         return "🤖 Unable to generate a specific answer."
 
@@ -541,28 +767,27 @@ def answer_question(question: str) -> str:
     """Process a single question and return the answer."""
     print(f"Processing question: {question[:50]}...")  # Debugging: show first 50 chars
 
-    # Wrap the question in a HumanMessage from langchain_core
+    # Wrap the question in a HumanMessage from langchain_core (assuming langchain is used)
     messages = [HumanMessage(content=question)]
-    messages = graph.invoke({"messages": messages})  # Assuming `graph` is defined elsewhere
+    response = graph.invoke({"messages": messages})  # Assuming `graph` is defined elsewhere
     
     # Extract the answer from the response
-    answer = messages['messages'][-1].content
+    answer = response['messages'][-1].content
     return answer[14:]  # Assuming 'answer[14:]' is correct based on your example
 
 
 def process_all_tasks(tasks: list):
     """Process a list of tasks."""
     results = {}
-    
+
     for task in tasks:
-        # Ensure task has a question and process it
         question = task.get("question", "").strip()
         if not question:
             print(f"Skipping task with missing or empty 'question': {task}")
             continue
-        
+
         print(f"\n🟢 Processing Task: {task['task_id']} - Question: {question}")
-        
+
         # Call the existing process_question logic
         response = process_question(question)
         
@@ -573,41 +798,9 @@ def process_all_tasks(tasks: list):
 
 
 
-def process_question(question: str):
-    tasks = planner(question)
-    print(f"Tasks to perform: {tasks}")
-
-    task_type = task_classifier(question)
-    print(f"Task type: {task_type}")
-
-    state = {"question": question, "last_response": "", "messages": [HumanMessage(content=question)]}
-    next_task = decide_task(state)
-    print(f"Next task: {next_task}")
-
-    if node_skipper(state):
-        print(f"Skipping task: {next_task}")
-        return "Task skipped."
-
-    try:
-        if task_type == "wiki_search":
-            response = wiki_tool.run(question)
-        elif task_type == "math":
-            # You should dynamically parse these inputs in real use
-            response = calc_tool.run(question)
-        elif task_type == "retriever":
-            retrieval_result = retriever(state)
-            response = retrieval_result["messages"][-1].content
-        else:
-            response = "Default fallback answer."
-
-        return generate_final_answer(state, {task_type: response})
-
-    except Exception as e:
-        print(f"❌ Error: {e}")
-        return "Sorry, I encountered an error processing your request."
-        
 
 
+## Langgraph
 
 # Build graph function
 provider = "huggingface"