Update app.py

app.py

@@ -5,518 +5,182 @@ import pandas as pd
 from smolagents import ToolCallingAgent, tool
 from duckduckgo_search import DDGS
 import math
-import openai
 import re
-import json
 from datetime import datetime, timedelta
 import time
 
-# --- Enhanced Tools ---
+# --- Enhanced Tools with Proper Error Handling ---
 @tool
 def duck_search(query: str) -> str:
-    """
-    Searches the web using DuckDuckGo and returns detailed information.
+    """Searches the web using DuckDuckGo.
     
     Args:
         query: The search query string.
     
     Returns:
-        A string with comprehensive search results including titles, snippets, and URLs.
+        A formatted string with search results.
     """
     try:
         with DDGS() as ddgs:
-            results = ddgs.text(query, max_results=5)  # Increased results
-            if not results:
-                return "No results found."
-            
-            formatted_results = []
-            for i, r in enumerate(results, 1):
-                formatted_results.append(
-                    f"Result {i}:\n"
-                    f"Title: {r['title']}\n"
-                    f"Content: {r['body']}\n"
-                    f"URL: {r['href']}\n"
-                    f"---"
-                )
-            return "\n".join(formatted_results)
-    except Exception as e:
-        return f"Search error: {e}"
-
-@tool
-def focused_search(query: str, topic: str = "") -> str:
-    """
-    Performs a more focused search with specific keywords for better results.
-    
-    Args:
-        query: The main search query
-        topic: Additional topic context to improve search accuracy
-    
-    Returns:
-        Focused search results
-    """
-    try:
-        # Enhance query with topic context
-        enhanced_query = f"{query} {topic}".strip()
-        
-        with DDGS() as ddgs:
-            results = ddgs.text(enhanced_query, max_results=3)
-            if not results:
-                # Try alternative search if no results
-                results = ddgs.text(query, max_results=3)
-            
-            if not results:
-                return "No results found for focused search."
-            
-            summaries = []
-            for r in results:
-                summaries.append(f"**{r['title']}**\n{r['body']}\nSource: {r['href']}")
-            
-            return "\n\n".join(summaries)
+            results = ddgs.text(query, max_results=3)
+            return "\n\n".join(
+                f"Title: {r['title']}\nContent: {r['body']}\nURL: {r['href']}"
+                for r in results
+            ) if results else "No results found."
     except Exception as e:
-        return f"Focused search error: {e}"
+        return f"Search error: {str(e)}"
 
 @tool
 def advanced_calculator(expression: str) -> str:
-    """
-    Enhanced calculator with support for complex mathematical operations.
+    """Evaluates mathematical expressions.
     
     Args:
-        expression: A mathematical expression or calculation
+        expression: The math expression to evaluate.
     
     Returns:
-        The calculated result with detailed steps when possible
+        The result as a string.
     """
     try:
-        # Clean the expression
-        expression = expression.strip()
-        
-        # Handle common mathematical functions and constants
-        safe_dict = {
-            "__builtins__": {},
-            **math.__dict__,
-            "abs": abs,
-            "round": round,
-            "min": min,
-            "max": max,
-            "sum": sum,
-            "pow": pow,
-        }
-        
-        # Try to evaluate the expression
-        result = eval(expression, safe_dict)
-        
-        # Format the result nicely
-        if isinstance(result, float):
-            if result.is_integer():
-                return str(int(result))
-            else:
-                return f"{result:.10g}"  # Remove trailing zeros
-        
+        # Safe evaluation environment
+        safe_dict = {k: v for k, v in math.__dict__.items() if not k.startswith("__")}
+        safe_dict.update({
+            '__builtins__': None,
+            'abs': abs,
+            'round': round,
+            'min': min,
+            'max': max
+        })
+        
+        # Handle percentage expressions
+        if '%' in expression:
+            expression = expression.replace('%', '/100')
+            
+        result = eval(expression, {'__builtins__': None}, safe_dict)
         return str(result)
-        
-    except Exception as e:
-        # Try to handle percentage calculations
-        if "%" in expression:
-            try:
-                # Convert percentage expressions
-                expr_mod = expression.replace("%", "/100")
-                result = eval(expr_mod, safe_dict)
-                return str(result)
-            except:
-                pass
-        
-        return f"Calculation error: {e}. Please check the mathematical expression."
-
-@tool
-def date_calculator(date_expression: str) -> str:
-    """
-    Calculates dates, time differences, and handles date-related queries.
-    
-    Args:
-        date_expression: A date calculation or query
-    
-    Returns:
-        The calculated date or time difference
-    """
-    try:
-        current_date = datetime.now()
-        
-        # Handle relative date expressions
-        if "days ago" in date_expression.lower():
-            days_match = re.search(r'(\d+)\s*days?\s*ago', date_expression.lower())
-            if days_match:
-                days = int(days_match.group(1))
-                target_date = current_date - timedelta(days=days)
-                return target_date.strftime("%Y-%m-%d (%A)")
-        
-        elif "days from now" in date_expression.lower():
-            days_match = re.search(r'(\d+)\s*days?\s*from\s*now', date_expression.lower())
-            if days_match:
-                days = int(days_match.group(1))
-                target_date = current_date + timedelta(days=days)
-                return target_date.strftime("%Y-%m-%d (%A)")
-        
-        elif "weeks ago" in date_expression.lower():
-            weeks_match = re.search(r'(\d+)\s*weeks?\s*ago', date_expression.lower())
-            if weeks_match:
-                weeks = int(weeks_match.group(1))
-                target_date = current_date - timedelta(weeks=weeks)
-                return target_date.strftime("%Y-%m-%d (%A)")
-        
-        # Current date info
-        elif "today" in date_expression.lower() or "current date" in date_expression.lower():
-            return current_date.strftime("%Y-%m-%d (%A)")
-        
-        return f"Current date: {current_date.strftime('%Y-%m-%d (%A)')}"
-        
-    except Exception as e:
-        return f"Date calculation error: {e}"
-
-@tool
-def text_analyzer(text: str) -> str:
-    """
-    Analyzes text for patterns, extracts information, and provides insights.
-    
-    Args:
-        text: The text to analyze
-    
-    Returns:
-        Analysis results including word count, patterns, and extracted information
-    """
-    try:
-        if not text:
-            return "No text provided for analysis."
-        
-        # Basic statistics
-        word_count = len(text.split())
-        char_count = len(text)
-        sentence_count = len([s for s in text.split('.') if s.strip()])
-        
-        # Extract numbers
-        numbers = re.findall(r'-?\d+(?:\.\d+)?', text)
-        
-        # Extract dates
-        date_patterns = re.findall(r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b|\b\d{4}[/-]\d{1,2}[/-]\d{1,2}\b', text)
-        
-        # Extract emails
-        emails = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', text)
-        
-        analysis = f"Text Analysis:\n"
-        analysis += f"- Words: {word_count}\n"
-        analysis += f"- Characters: {char_count}\n"
-        analysis += f"- Sentences: {sentence_count}\n"
-        
-        if numbers:
-            analysis += f"- Numbers found: {', '.join(numbers[:10])}{'...' if len(numbers) > 10 else ''}\n"
-        
-        if date_patterns:
-            analysis += f"- Dates found: {', '.join(date_patterns)}\n"
-        
-        if emails:
-            analysis += f"- Emails found: {', '.join(emails)}\n"
-        
-        return analysis
-        
     except Exception as e:
-        return f"Text analysis error: {e}"
+        return f"Calculation error: {str(e)}"
 
-# --- Enhanced Agent ---
-class ImprovedWebSearchAgent:
+# --- Simplified Agent Class ---
+class GAIAAgent:
     def __init__(self):
-        """Initialize the enhanced agent with better reasoning capabilities."""
-        
-        # Use more powerful model if available
-        model_name = "gpt-4o-mini"  # Fallback to gpt-3.5-turbo if needed
-        
-        # Enhanced system prompt for better reasoning
-        system_prompt = """You are an advanced AI assistant designed to solve complex problems by breaking them down systematically. 
-
-Key capabilities:
-1. **Multi-step Reasoning**: Break complex problems into smaller, manageable steps
-2. **Information Synthesis**: Combine information from multiple sources
-3. **Verification**: Double-check calculations and facts
-4. **Context Awareness**: Understand the broader context of questions
-
-Problem-solving approach:
-1. Analyze the question carefully to understand what's being asked
-2. Identify what information you need to find
-3. Use available tools strategically (search, calculate, analyze)
-4. Verify your findings and reasoning
-5. Provide a clear, accurate answer
-
-When using tools:
-- Use focused_search for specific factual information
-- Use duck_search for broader context
-- Use advanced_calculator for any mathematical operations
-- Use date_calculator for time-related queries
-- Use text_analyzer when you need to extract information from text
-
-Always think step-by-step and explain your reasoning process."""
-
         try:
             self.agent = ToolCallingAgent(
-                name="ImprovedGAIAAgent",
-                description=system_prompt,
-                tools=[duck_search, focused_search, advanced_calculator, date_calculator, text_analyzer],
-                model=model_name,
-                planning_interval=3,  # More frequent planning
+                name="GAIA_Agent",
+                description="Agent for GAIA benchmark tasks",
+                tools=[duck_search, advanced_calculator],
+                model="gpt-3.5-turbo",  # Use this as default
+                max_iterations=5
             )
-            print(f"✅ Enhanced agent initialized with {model_name}")
+            print("✅ Agent initialized successfully")
         except Exception as e:
-            print(f"⚠️ Error initializing with {model_name}, trying fallback...")
-            try:
-                self.agent = ToolCallingAgent(
-                    name="ImprovedGAIAAgent",
-                    description=system_prompt,
-                    tools=[duck_search, focused_search, advanced_calculator, date_calculator, text_analyzer],
-                    model="gpt-3.5-turbo",
-                    planning_interval=3,
-                )
-                print("✅ Enhanced agent initialized with gpt-3.5-turbo")
-            except Exception as e2:
-                print(f"❌ Agent initialization failed: {e2}")
-                raise e2
+            print(f"❌ Agent initialization failed: {str(e)}")
+            raise
 
     def __call__(self, question: str) -> str:
-        """
-        Process a question with enhanced reasoning and error handling.
-        
-        Args:
-            question: The question to answer
-            
-        Returns:
-            A comprehensive answer
-        """
-        print(f"🔍 Processing question: {question}")
-        
+        """Process a question with proper error handling."""
         try:
-            # Add some preprocessing to understand question type
-            question_lower = question.lower()
-            
-            # Enhance the question with context clues
-            enhanced_question = self._enhance_question(question)
-            
-            # Run the agent with timeout protection
-            start_time = time.time()
-            max_time = 120  # 2 minutes max per question
+            # Simple preprocessing
+            question = question.strip()
             
-            result = self.agent.run(enhanced_question)
+            # Run the agent
+            response = self.agent.run(question)
             
-            elapsed_time = time.time() - start_time
-            print(f"⏱️ Question processed in {elapsed_time:.1f} seconds")
-            
-            # Post-process the result
-            final_answer = self._post_process_answer(result, question)
-            
-            return final_answer
+            # Basic post-processing
+            if not response:
+                return "Could not generate an answer."
+                
+            return str(response)[:1000]  # Limit response length
             
         except Exception as e:
-            print(f"❌ Agent error: {e}")
-            # Try a simpler approach as fallback
-            return self._fallback_answer(question, str(e))
-
-    def _enhance_question(self, question: str) -> str:
-        """Add context and instructions to improve question processing."""
-        
-        enhanced = f"""Please solve this step by step:
-
-Question: {question}
-
-Instructions:
-1. Read the question carefully and identify what type of answer is needed
-2. Break down complex problems into steps
-3. Use the available tools to gather information or perform calculations
-4. Verify your answer makes sense
-5. Provide a clear, concise final answer
-
-If this is a factual question, search for current information.
-If this involves calculations, show your work.
-If this requires multiple steps, explain each step clearly."""
-
-        return enhanced
-
-    def _post_process_answer(self, result: str, original_question: str) -> str:
-        """Clean and improve the agent's response."""
-        
-        if not result or len(result.strip()) < 10:
-            return f"I need more information to properly answer: {original_question}"
-        
-        # Clean up the response
-        result = result.strip()
-        
-        # Ensure we have a clear answer
-        if "final answer" not in result.lower() and "answer:" not in result.lower():
-            # Try to extract the most relevant part
-            lines = result.split('\n')
-            if lines:
-                # Look for the most substantive line as the answer
-                best_line = max(lines, key=len, default=result)
-                if len(best_line) > 20:
-                    result = f"{result}\n\nFinal Answer: {best_line}"
-        
-        return result
-
-    def _fallback_answer(self, question: str, error: str) -> str:
-        """Provide a fallback response when the main agent fails."""
-        
-        question_lower = question.lower()
-        
-        # Try simple keyword-based responses for common question types
-        if any(word in question_lower for word in ['calculate', 'math', '+', '-', '*', '/', 'equals']):
-            return f"This appears to be a mathematical question. Error occurred: {error}. Please verify the calculation manually."
-        
-        elif any(word in question_lower for word in ['when', 'date', 'year', 'time']):
-            return f"This appears to be a date/time related question. Error occurred: {error}. Please search for current information."
-        
-        elif any(word in question_lower for word in ['who', 'what', 'where', 'how']):
-            return f"This appears to be a factual question. Error occurred: {error}. Please search for current information."
-        
-        else:
-            return f"I encountered an error while processing your question: {error}. Please try rephrasing your question."
-
-# --- Constants ---
-DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+            print(f"⚠️ Processing error: {str(e)}")
+            return f"Error processing question: {str(e)}"
 
 # --- Evaluation & Submission ---
 def run_and_submit_all(profile: gr.OAuthProfile | None):
-    space_id = os.getenv("SPACE_ID")
-    if profile:
-        username = profile.username
-        print(f"👤 User: {username}")
-    else:
+    if not profile:
         return "Please login to Hugging Face.", None
 
-    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
-    questions_url = f"{DEFAULT_API_URL}/questions"
-    submit_url = f"{DEFAULT_API_URL}/submit"
-
     try:
-        agent = ImprovedWebSearchAgent()
+        agent = GAIAAgent()
     except Exception as e:
-        return f"Agent initialization error: {e}", None
+        return f"Agent initialization failed: {str(e)}", None
 
     try:
-        response = requests.get(questions_url, timeout=15)
-        response.raise_for_status()
+        response = requests.get(
+            "https://agents-course-unit4-scoring.hf.space/questions",
+            timeout=30
+        )
         questions = response.json()
         if not questions:
             return "No questions received.", None
-        
-        print(f"📝 Received {len(questions)} questions")
-        
     except Exception as e:
-        return f"Failed to fetch questions: {e}", None
+        return f"Failed to fetch questions: {str(e)}", None
 
-    results_log = []
-    answers_payload = []
-
-    for i, item in enumerate(questions, 1):
+    results = []
+    answers = []
+    
+    for item in questions[:20]:  # Process first 20 questions for testing
         task_id = item.get("task_id")
         question = item.get("question")
         
         if not task_id or not question:
             continue
             
-        print(f"\n📋 Processing question {i}/{len(questions)}: {task_id}")
-        
         try:
             answer = agent(question)
-            
-            # Ensure answer is not empty
-            if not answer or len(answer.strip()) < 2:
-                answer = "Unable to determine answer from available information."
-            
-            results_log.append({
-                "Task ID": task_id, 
-                "Question": question[:100] + "..." if len(question) > 100 else question,
-                "Submitted Answer": answer[:200] + "..." if len(answer) > 200 else answer
-            })
-            
-            answers_payload.append({
-                "task_id": task_id, 
+            answers.append({
+                "task_id": task_id,
                 "submitted_answer": answer
             })
-            
-            print(f"✅ Answer generated for {task_id}")
-            
+            results.append({
+                "Task ID": task_id,
+                "Question": question[:100],
+                "Answer": answer[:200]
+            })
         except Exception as e:
-            error_msg = f"Agent error: {str(e)[:100]}"
-            print(f"❌ Error for {task_id}: {error_msg}")
-            
-            results_log.append({
-                "Task ID": task_id, 
-                "Question": question[:100] + "..." if len(question) > 100 else question,
-                "Submitted Answer": error_msg
+            answers.append({
+                "task_id": task_id,
+                "submitted_answer": f"Error: {str(e)}"
             })
-            
-            answers_payload.append({
-                "task_id": task_id, 
-                "submitted_answer": "Error processing question"
+            results.append({
+                "Task ID": task_id,
+                "Question": question[:100],
+                "Answer": f"Error: {str(e)}"
             })
 
-    if not answers_payload:
-        return "No answers were generated.", pd.DataFrame(results_log)
-
-    print(f"\n🚀 Submitting {len(answers_payload)} answers...")
-
     try:
-        response = requests.post(submit_url, json={
-            "username": username.strip(),
-            "agent_code": agent_code,
-            "answers": answers_payload
-        }, timeout=120)  # Increased timeout
-        
-        response.raise_for_status()
-        result = response.json()
-        
-        score = result.get('score', 0)
-        correct_count = result.get('correct_count', 0)
-        total_attempted = result.get('total_attempted', len(answers_payload))
-        
-        status = (
-            f"✅ Submission Successful!\n"
-            f"User: {result.get('username')}\n"
-            f"Score: {score}% ({correct_count}/{total_attempted} correct)\n"
-            f"Message: {result.get('message', 'No message')}\n"
-            f"Total questions processed: {len(questions)}"
+        response = requests.post(
+            "https://agents-course-unit4-scoring.hf.space/submit",
+            json={
+                "username": profile.username,
+                "agent_code": f"https://huggingface.co/spaces/{os.getenv('SPACE_ID')}",
+                "answers": answers
+            },
+            timeout=60
+        )
+        data = response.json()
+        return (
+            f"Submitted {len(answers)} answers\n"
+            f"Score: {data.get('score', 'N/A')}%\n"
+            f"Correct: {data.get('correct_count', 0)}/{data.get('total_attempted', 0)}\n"
+            f"Message: {data.get('message', '')}",
+            pd.DataFrame(results)
         )
-        
-        print(f"🎯 Final Score: {score}%")
-        
-        return status, pd.DataFrame(results_log)
-        
     except Exception as e:
-        error_msg = f"❌ Submission failed: {e}"
-        print(error_msg)
-        return error_msg, pd.DataFrame(results_log)
+        return f"Submission failed: {str(e)}", pd.DataFrame(results)
 
-# --- UI ---
-with gr.Blocks(title="Enhanced GAIA Agent") as demo:
-    gr.Markdown("# 🤖 Enhanced GAIA Agent with Advanced Reasoning")
-    gr.Markdown("""
-    **Improvements in this version:**
-    - 🧠 Enhanced multi-step reasoning capabilities
-    - 🔍 Multiple specialized search tools
-    - 🧮 Advanced calculator with better math support
-    - 📅 Date and time calculation tools
-    - 📝 Text analysis capabilities
-    - ⚡ Better error handling and fallback mechanisms
-    - 🎯 Optimized for GAIA benchmark performance
-    """)
-    
+# --- Gradio Interface ---
+with gr.Blocks() as demo:
+    gr.Markdown("# GAIA Agent")
     gr.LoginButton()
+    submit_btn = gr.Button("Run & Submit", variant="primary")
+    output = gr.Textbox(label="Results")
+    table = gr.DataFrame(label="Details")
     
-    with gr.Row():
-        run_btn = gr.Button("🚀 Run Enhanced Evaluation & Submit", variant="primary", scale=2)
-    
-    status_box = gr.Textbox(label="📊 Status & Results", lines=8, interactive=False)
-    result_table = gr.DataFrame(label="📋 Agent Answers Log", interactive=False)
-
-    run_btn.click(
-        fn=run_and_submit_all, 
-        outputs=[status_box, result_table],
-        show_progress=True
+    submit_btn.click(
+        fn=run_and_submit_all,
+        outputs=[output, table]
     )
 
 if __name__ == "__main__":
-    demo.launch(debug=True, share=False)
+    demo.launch()