Update app.py

app.py

@@ -69,8 +69,8 @@ class BasicAgent:
         self.tools = [
             WikipediaSearchTool(),
                         # DuckDuckGoSearchTool(),
-            PythonInterpreterTool(),
-                      VisitWebpageTool()
+            # PythonInterpreterTool(),
+                      # VisitWebpageTool()
                      ]
         
         # Create CodeAgent
@@ -117,435 +117,6 @@ class BasicAgent:
 
 
 
-            
-#  # from langgraph.gr import StateGraph, END
-# from langgraph.graph import StateGraph, END
-# from langchain_core.messages import HumanMessage, AIMessage
-# from langchain_openai import AzureChatOpenAI
-# from langchain_community.tools import WikipediaQueryRun, DuckDuckGoSearchRun
-# from langchain_community.utilities import WikipediaAPIWrapper, DuckDuckGoSearchAPIWrapper
-# from langchain_experimental.tools import PythonREPLTool
-# from langchain_core.tools import tool
-# import os
-# import math
-# import numpy as np
-# import re
-# from typing import Optional, Dict, Any, List
-# from langchain_core.agents import AgentAction, AgentFinish
-
-# class BasicAgent:
-#     def __init__(self, model_id: Optional[str] = None, api_key: Optional[str] = None):
-#         """
-#         Initialize BasicAgent optimized for GAIA benchmark success.
-#         """        
-        
-#         # Initialize model
-#         self.model = AzureChatOpenAI(
-#             deployment_name="o3-mini",
-#             azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
-#             api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
-#             api_version="2024-12-01-preview"
-#         )
-        
-#         # Initialize tools
-#         self.tools = self._initialize_tools()
-        
-#         # Create LangGraph workflow
-#         self.workflow = self._create_workflow()
-#         self.app = self.workflow.compile()
-        
-#     def _initialize_tools(self):
-#         """Initialize tools with GAIA-specific optimizations."""
-        
-#         @tool
-#         def web_search(query: str) -> str:
-#             """
-#             Search for current information on the web. Use specific, targeted queries.
-#             Best for: recent events, current data, specific facts, news.
-#             """
-#             try:
-#                 ddg = DuckDuckGoSearchAPIWrapper(max_results=5)
-#                 results = ddg.run(query)
-#                 return results[:1500]
-#             except Exception as e:
-#                 return f"Search failed: {str(e)}"
-        
-#         @tool  
-#         def wikipedia_search(query: str) -> str:
-#             """
-#             Search Wikipedia for established facts, definitions, historical data.
-#             Best for: biographical info, historical events, scientific concepts, definitions.
-#             """
-#             try:
-#                 wiki = WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=1000)
-#                 result = wiki.run(query)
-#                 return result
-#             except Exception as e:
-#                 return f"Wikipedia search failed: {str(e)}"
-        
-#         @tool
-#         def python_calculator(code: str) -> str:
-#             """
-#             Execute Python code for calculations, data processing, file operations.
-#             Best for: complex math, data analysis, file processing, calculations.
-#             Always include print() statements to see results.
-#             """
-#             try:
-#                 # Enhanced Python environment
-#                 exec_globals = {
-#                     '__builtins__': __builtins__,
-#                     'math': math,
-#                     'np': np,
-#                     'numpy': np,
-#                     'pd': None,  # Will try to import if needed
-#                     'os': os,
-#                     're': re
-#                 }
-                
-#                 # Try to import common libraries
-#                 try:
-#                     import pandas as pd
-#                     exec_globals['pd'] = pd
-#                     exec_globals['pandas'] = pd
-#                 except:
-#                     pass
-                
-#                 # Capture output
-#                 import io
-#                 import sys
-#                 old_stdout = sys.stdout
-#                 sys.stdout = captured_output = io.StringIO()
-                
-#                 # Execute code
-#                 exec(code, exec_globals)
-                
-#                 # Get output
-#                 sys.stdout = old_stdout
-#                 output = captured_output.getvalue()
-                
-#                 return output if output.strip() else "Code executed successfully (no output)"
-                
-#             except Exception as e:
-#                 return f"Python execution error: {str(e)}"
-        
-#         @tool
-#         def simple_math(expression: str) -> str:
-#             """
-#             Evaluate simple mathematical expressions quickly.
-#             Best for: basic arithmetic, simple calculations.
-#             Examples: "2+3*4", "sqrt(16)", "sin(pi/4)"
-#             """
-#             try:
-#                 # Safe evaluation environment
-#                 allowed_names = {
-#                     k: v for k, v in math.__dict__.items() if not k.startswith("__")
-#                 }
-#                 allowed_names.update({
-#                     "abs": abs, "round": round, "min": min, "max": max,
-#                     "sum": sum, "pow": pow, "divmod": divmod
-#                 })
-                
-#                 result = eval(expression, {"__builtins__": {}}, allowed_names)
-#                 return str(result)
-#             except Exception as e:
-#                 return f"Math error: {str(e)}"
-        
-#         @tool
-#         def file_analyzer(task: str) -> str:
-#             """
-#             Analyze files in the current directory.
-#             Best for: examining uploaded files, extracting data from files.
-#             """
-#             try:
-#                 # List available files
-#                 files = [f for f in os.listdir('.') if os.path.isfile(f)]
-                
-#                 result = f"Available files: {files}\n"
-#                 result += f"Task: {task}\n"
-#                 result += "Use python_calculator for detailed file processing."
-                
-#                 return result
-#             except Exception as e:
-#                 return f"File analysis error: {str(e)}"
-        
-#         return [
-#             # web_search, 
-#             wikipedia_search, python_calculator
-#             # , simple_math
-#             # , file_analyzer
-#         ]
-    
-#     def _create_workflow(self):
-#         """Create optimized LangGraph workflow."""
-#         workflow = StateGraph(dict)
-        
-#         workflow.add_node("planner", self._planner_node)
-#         workflow.add_node("executor", self._executor_node)
-#         workflow.add_node("validator", self._validator_node)
-        
-#         workflow.set_entry_point("planner")
-        
-#         workflow.add_conditional_edges(
-#             "planner",
-#             self._plan_decision,
-#             {
-#                 "execute": "executor",
-#                 "final": "validator"
-#             }
-#         )
-        
-#         workflow.add_conditional_edges(
-#             "executor", 
-#             self._execution_decision,
-#             {
-#                 "continue": "planner",
-#                 "validate": "validator"
-#             }
-#         )
-        
-#         workflow.add_edge("validator", END)
-        
-#         return workflow
-    
-#     def _planner_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
-#         """Enhanced planning node focused on GAIA success patterns."""
-#         messages = state.get("messages", [])
-#         step_count = state.get("step_count", 0)
-#         max_steps = state.get("max_steps", 4)
-#         plan_history = state.get("plan_history", [])
-        
-#         if step_count >= max_steps:
-#             return {
-#                 **state,
-#                 "final_answer": "Maximum steps reached. Providing best available answer.",
-#                 "action_type": "final"
-#             }
-        
-#         planning_prompt = f"""
-#                                 You are a general AI assistant. I will ask you a question.
-#                                 Report your thoughts, and finish your answer with the following template:
-#                                 FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings.
-#                                 If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. 
-#                                 If you are asked for a string, don't use articles,
-#                                 neither abbreviations (e.g. for cities),
-#                                 and write the digits in plain text unless specified otherwise.
-#                                 If you are asked for a comma separated list, 
-#                                 apply the above rules depending of whether the element to be put in the list is a number or a string.
-
-#                                 QUESTION: {messages[0]['content'] if messages else 'No question provided'}
-                                
-#                                 EXECUTION HISTORY: {plan_history}\n
-
-#                                 RÈGLES ABSOLUES:
-#                                 1. Lis la question 3 fois avant de commencer
-#                                 2. Décompose TOUJOURS la question en sous-problèmes identifiables
-#                                 3. Vérifie CHAQUE information avec au moins 2 sources différentes
-#                                 4. Pour les calculs: utilise OBLIGATOIREMENT Python pour tous les calculs numériques
-#                                 5. Pour les dates: vérifie l'année actuelle (nous sommes en 2025)
-#                                 6. JAMAIS de réponse approximative - sois précis au maximum
-                                
-#                                 PROCESSUS OBLIGATOIRE:
-#                                 1. ANALYSE: Que demande exactement la question? Quel type de réponse?
-#                                 2. RECHERCHE: Quelles informations spécifiques me manquent?
-#                                 3. VÉRIFICATION: Les sources sont-elles cohérentes entre elles?
-#                                 4. CALCUL: Si nécessaire, utilise Python pour calculs précis
-#                                 5. SYNTHÈSE: Donne une réponse finale précise et concise
-                                
-#                                 FORMAT DE RÉPONSE FINAL:
-#                                 - Si c'est un nombre: donne UNIQUEMENT le nombre (ex: "42")
-#                                 - Si c'est un nom: donne UNIQUEMENT le nom (ex: "Paris")
-#                                 - Si c'est une date: format précis demandé
-#                                 - Pas d'explication supplémentaire dans la réponse finale
-#                             """
-
-#         response = self.model.invoke([{"role": "system", "content": planning_prompt}])
-#         content = response.content.strip()
-        
-#         if content.startswith("FINAL:"):
-#             answer = content.replace("FINAL:", "").strip()
-#             return {
-#                 **state,
-#                 "final_answer": answer,
-#                 "action_type": "final",
-#                 "step_count": step_count
-#             }
-#         elif content.startswith("EXECUTE:"):
-#             # Parse execution command
-#             try:
-#                 parts = content.replace("EXECUTE:", "").split("|")
-#                 tool_name = parts[0].split()[0].strip()
-#                 input_part = [p for p in parts if p.strip().startswith("INPUT:")][0]
-#                 tool_input = input_part.replace("INPUT:", "").strip()
-#                 goal_part = [p for p in parts if p.strip().startswith("GOAL:")][0] if len(parts) > 2 else ""
-#                 goal = goal_part.replace("GOAL:", "").strip() if goal_part else ""
-                
-#                 return {
-#                     **state,
-#                     "current_tool": tool_name,
-#                     "current_input": tool_input,
-#                     "current_goal": goal,
-#                     "action_type": "execute",
-#                     "step_count": step_count + 1
-#                 }
-#             except Exception as e:
-#                 return {
-#                     **state,
-#                     "final_answer": f"Planning error: {str(e)}",
-#                     "action_type": "final"
-#                 }
-#         else:
-#             return {
-#                 **state,
-#                 "final_answer": content,
-#                 "action_type": "final"
-#             }
-    
-#     def _executor_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
-#         """Execute the planned action."""
-#         tool_name = state.get("current_tool", "")
-#         tool_input = state.get("current_input", "")
-#         goal = state.get("current_goal", "")
-#         plan_history = state.get("plan_history", [])
-        
-#         # Find and execute tool
-#         tool_map = {tool.name: tool for tool in self.tools}
-        
-#         # Add flexible matching
-#         tool_matches = {
-#             # "web_search": ["web", "search", "google", "internet"],
-#             "wikipedia_search": ["wiki", "wikipedia"],  
-#             "python_calculator": ["python", "code", "calc", "calculate"],
-#             # "simple_math": ["math", "arithmetic"],
-#             # "file_analyzer": ["file", "analyze"]
-#         }
-        
-#         matched_tool = None
-#         for tool_real_name, aliases in tool_matches.items():
-#             if tool_name.lower() in aliases or tool_name.lower() == tool_real_name.lower():
-#                 matched_tool = tool_map.get(tool_real_name)
-#                 break
-        
-#         if not matched_tool:
-#             matched_tool = tool_map.get(tool_name)
-        
-#         if matched_tool:
-#             try:
-#                 result = matched_tool.run(tool_input)
-#                 execution_record = f"STEP: Used {tool_name} with '{tool_input}' -> {result[:200]}..."
-#                 plan_history.append(execution_record)
-                
-#                 return {
-#                     **state,
-#                     "last_result": result,
-#                     "plan_history": plan_history,
-#                     "action_type": "continue"
-#                 }
-#             except Exception as e:
-#                 error_msg = f"Tool {tool_name} failed: {str(e)}"
-#                 plan_history.append(f"ERROR: {error_msg}")
-#                 return {
-#                     **state,
-#                     "last_result": error_msg,
-#                     "plan_history": plan_history,
-#                     "action_type": "validate"
-#                 }
-#         else:
-#             available = list(tool_map.keys())
-#             error_msg = f"Tool '{tool_name}' not found. Available: {available}"
-#             plan_history.append(f"ERROR: {error_msg}")
-#             return {
-#                 **state,
-#                 "last_result": error_msg,
-#                 "plan_history": plan_history,
-#                 "action_type": "validate"
-#             }
-    
-#     def _validator_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
-#         """Validate and finalize the answer."""
-#         final_answer = state.get("final_answer", "")
-#         plan_history = state.get("plan_history", [])
-#         last_result = state.get("last_result", "")
-        
-#         if not final_answer and last_result:
-#             # Extract answer from last result
-#             validation_prompt = f"""Extract the EXACT answer from this result for the GAIA question.
-
-# QUESTION: {state.get('messages', [{}])[0].get('content', '')}
-# TOOL RESULT: {last_result}
-
-# Provide ONLY the precise answer - no explanations, no context, just the exact answer required.
-# Examples:
-# - If asked for a number: "42"
-# - If asked for a name: "John Smith"  
-# - If asked for a date: "1969"
-# - If asked for a yes/no: "Yes"
-
-# EXACT ANSWER:"""
-            
-#             response = self.model.invoke([{"role": "user", "content": validation_prompt}])
-#             final_answer = response.content.strip()
-        
-#         # Clean up the answer
-#         final_answer = self._clean_answer(final_answer)
-        
-#         return {
-#             **state,
-#             "final_answer": final_answer,
-#             "completed": True
-#         }
-    
-#     def _clean_answer(self, answer: str) -> str:
-#         """Clean and format the final answer for GAIA."""
-#         if not answer:
-#             return "No answer found"
-            
-#         # Remove common prefixes
-#         prefixes = [
-#             "the answer is", "answer:", "final answer:", "result:", 
-#             "exact answer:", "solution:", "response:", "output:"
-#         ]
-        
-#         cleaned = answer.strip()
-#         for prefix in prefixes:
-#             if cleaned.lower().startswith(prefix):
-#                 cleaned = cleaned[len(prefix):].strip()
-        
-#         # Remove quotes if they wrap the entire answer
-#         if cleaned.startswith('"') and cleaned.endswith('"'):
-#             cleaned = cleaned[1:-1]
-#         if cleaned.startswith("'") and cleaned.endswith("'"):
-#             cleaned = cleaned[1:-1]
-            
-#         return cleaned
-    
-#     def _plan_decision(self, state: Dict[str, Any]) -> str:
-#         """Decide whether to execute or finalize."""
-#         return state.get("action_type", "execute")
-    
-#     def _execution_decision(self, state: Dict[str, Any]) -> str:
-#         """Decide next step after execution."""
-#         return state.get("action_type", "continue")
-    
-#     def run(self, question: str, max_steps: int = 4) -> str:
-#         """
-#         Run the agent with GAIA-optimized settings.
-#         """
-#         initial_state = {
-#             "messages": [{"role": "user", "content": question}],
-#             "step_count": 0,
-#             "max_steps": max_steps,
-#             "plan_history": [],
-#             "completed": False
-#         }
-        
-#         try:
-#             result = self.app.invoke(initial_state)
-#             return result.get("final_answer", "No answer generated")
-            
-#         except Exception as e:
-#             return f"Error: {str(e)}"
-
-