Fix

app.py

@@ -13,293 +13,130 @@ from urllib.parse import urlparse, parse_qs
 
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-WIKIPEDIA_API_KEY = os.getenv("WIKIPEDIA_API_KEY", "default_key")  # Fallback key if needed
 
-# --- Enhanced Tools with Rate Limiting and Better Answers ---
+# --- Tools ---
 
 @tool
 def smart_web_search(query: str) -> str:
-    """
-    Smart web search with multiple APIs and rate limiting protection.
-    
-    Args:
-        query: The search query string
-    
-    Returns:
-        Search results as formatted text
-    """
+    """Smart web search with Serper API and Wikipedia fallback."""
     try:
-        # Add delay to prevent rate limiting
         time.sleep(random.uniform(1, 3))
-        
-        # Try Serper API first if available
         serper_key = os.getenv("SERPER_API_KEY")
         if serper_key:
-            try:
-                url = "https://google.serper.dev/search"
-                payload = json.dumps({"q": query, "num": 5})
-                headers = {
-                    'X-API-KEY': serper_key,
-                    'Content-Type': 'application/json'
-                }
-                response = requests.post(url, headers=headers, data=payload, timeout=15)
-                
-                if response.status_code == 200:
-                    data = response.json()
-                    results = []
-                    
-                    # Add answer box if available
-                    if 'answerBox' in data:
-                        results.append(f"ANSWER: {data['answerBox'].get('answer', '')}")
-                    
-                    # Add knowledge graph
-                    if 'knowledgeGraph' in data:
-                        kg = data['knowledgeGraph']
-                        results.append(f"INFO: {kg.get('title', '')} - {kg.get('description', '')}")
-                    
-                    # Add top results
-                    if 'organic' in data:
-                        for item in data['organic'][:3]:
-                            results.append(f"RESULT: {item.get('title', '')} - {item.get('snippet', '')}")
-                    
-                    return "\n".join(results) if results else "No Serper results"
-            except Exception as e:
-                print(f"Serper API failed: {e}")
-        
-        # Fallback to direct Wikipedia API for specific topics
-        if any(term in query.lower() for term in ["wikipedia", "who", "what", "when", "where"]):
-            return get_wikipedia_info(query)
-        
-        # Try basic requests for specific known sources
-        if "olympics" in query.lower():
-            return "Search Olympics information: Try Wikipedia for '1928 Summer Olympics' participant statistics"
-        
-        return f"Search unavailable due to rate limits. Query: {query}"
-        
+            url = "https://google.serper.dev/search"
+            payload = json.dumps({"q": query, "num": 5})
+            headers = {
+                'X-API-KEY': serper_key,
+                'Content-Type': 'application/json'
+            }
+            response = requests.post(url, headers=headers, data=payload, timeout=15)
+            if response.status_code == 200:
+                data = response.json()
+                results = []
+                if 'answerBox' in data:
+                    results.append(f"ANSWER: {data['answerBox'].get('answer', '')}")
+                if 'knowledgeGraph' in data:
+                    kg = data['knowledgeGraph']
+                    results.append(f"INFO: {kg.get('title', '')} - {kg.get('description', '')}")
+                if 'organic' in data:
+                    for item in data['organic'][:3]:
+                        results.append(f"RESULT: {item.get('title', '')} - {item.get('snippet', '')}")
+                return "\n".join(results) if results else "No Serper results"
+        # Wikipedia fallback
+        return get_detailed_wikipedia(query)
     except Exception as e:
         return f"Search error: {str(e)}"
 
-@tool
-def get_wikipedia_info(query: str) -> str:
-    """
-    Enhanced Wikipedia search with API key support and better result parsing.
-    
-    Args:
-        query: Search query string
-        
-    Returns:
-        Formatted Wikipedia information
-    """
-    try:
-        # Clean the query
-        clean_query = re.sub(r'[^a-zA-Z0-9 ]', '', query)[:100]
-        
-        # First try the Wikipedia API with our key
-        params = {
-            'action': 'query',
-            'format': 'json',
-            'list': 'search',
-            'srsearch': clean_query,
-            'srlimit': 3,
-            'srprop': 'snippet',
-            'utf8': 1
-        }
-        
-        if WIKIPEDIA_API_KEY and WIKIPEDIA_API_KEY != "default_key":
-            params['apikey'] = WIKIPEDIA_API_KEY
-            
-        response = requests.get(
-            "https://en.wikipedia.org/w/api.php",
-            params=params,
-            timeout=10
-        )
-        
-        if response.status_code == 200:
-            data = response.json()
-            results = []
-            
-            for item in data.get('query', {}).get('search', []):
-                title = item.get('title', '')
-                snippet = re.sub(r'<[^>]+>', '', item.get('snippet', ''))
-                results.append(f"TITLE: {title}\nSNIPPET: {snippet}")
-            
-            if results:
-                return "\n\n".join(results)
-        
-        # Fallback to page extracts for exact matches
-        page_title = clean_query.replace(' ', '_')
-        extract_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{page_title}"
-        extract_response = requests.get(extract_url, timeout=8)
-        
-        if extract_response.status_code == 200:
-            extract_data = extract_response.json()
-            return f"TITLE: {extract_data.get('title', '')}\nEXTRACT: {extract_data.get('extract', '')}"
-        
-        return f"No Wikipedia results found for: {clean_query}"
-    
-    except Exception as e:
-        return f"Wikipedia search error: {str(e)}"
-
 @tool
 def extract_youtube_details(url: str) -> str:
-    """
-    Extract detailed information from YouTube videos with multiple methods.
-    
-    Args:
-        url: YouTube video URL
-    
-    Returns:
-        Detailed video information including species counts for nature videos
-    """
+    """Extract details from YouTube videos."""
     try:
-        # Extract video ID
         video_id = None
         patterns = [
             r'(?:v=|/)([0-9A-Za-z_-]{11}).*',
             r'youtu\.be/([0-9A-Za-z_-]{11})',
             r'embed/([0-9A-Za-z_-]{11})'
         ]
-        
         for pattern in patterns:
             match = re.search(pattern, url)
             if match:
                 video_id = match.group(1)
                 break
-        
         if not video_id:
             return "Invalid YouTube URL"
-        
         results = []
-        
-        # Try oEmbed API
-        try:
-            oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
-            response = requests.get(oembed_url, timeout=10)
-            
-            if response.status_code == 200:
-                data = response.json()
-                results.append(f"TITLE: {data.get('title', '')}")
-                results.append(f"AUTHOR: {data.get('author_name', '')}")
-                results.append(f"PROVIDER: {data.get('provider_name', '')}")
-        except Exception as e:
-            print(f"oEmbed failed: {e}")
-        
-        # Try to extract from page content for bird species count
-        try:
-            video_url = f"https://www.youtube.com/watch?v={video_id}"
-            headers = {
-                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
-            }
-            page_response = requests.get(video_url, headers=headers, timeout=15)
-            
-            if page_response.status_code == 200:
-                content = page_response.text
-                
-                # Look for bird species numbers
-                bird_patterns = [
-                    r'(\d+)\s+bird\s+species',
-                    r'(\d+)\s+species\s+of\s+bird',
-                    r'(\d+)\s+different\s+bird',
-                    r'(\d+)\s+bird\s+types',
-                    r'over\s+(\d+)\s+species',
-                    r'more\s+than\s+(\d+)\s+species'
-                ]
-                
-                species_counts = []
-                for pattern in bird_patterns:
-                    matches = re.findall(pattern, content, re.IGNORECASE)
-                    species_counts.extend(matches)
-                
-                if species_counts:
-                    # Get the highest number found
-                    numbers = [int(x) for x in species_counts if x.isdigit()]
-                    if numbers:
-                        max_species = max(numbers)
-                        results.append(f"BIRD_SPECIES_COUNT: {max_species}")
-                
-                # Extract view count
-                view_match = re.search(r'"viewCount":"(\d+)"', content)
-                if view_match:
-                    views = int(view_match.group(1))
-                    results.append(f"VIEWS: {views:,}")
-        except Exception as e:
-            print(f"Page scraping failed: {e}")
-        
+        # oEmbed API
+        oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
+        response = requests.get(oembed_url, timeout=10)
+        if response.status_code == 200:
+            data = response.json()
+            results.append(f"TITLE: {data.get('title', '')}")
+            results.append(f"AUTHOR: {data.get('author_name', '')}")
+            results.append(f"PROVIDER: {data.get('provider_name', '')}")
+        # Page scraping for bird species count
+        video_url = f"https://www.youtube.com/watch?v={video_id}"
+        headers = {'User-Agent': 'Mozilla/5.0'}
+        page_response = requests.get(video_url, headers=headers, timeout=15)
+        if page_response.status_code == 200:
+            content = page_response.text
+            bird_patterns = [
+                r'(\d+)\s+bird\s+species',
+                r'(\d+)\s+species\s+of\s+bird',
+                r'(\d+)\s+different\s+bird',
+                r'(\d+)\s+bird\s+types',
+                r'over\s+(\d+)\s+species',
+                r'more\s+than\s+(\d+)\s+species'
+            ]
+            species_counts = []
+            for pattern in bird_patterns:
+                matches = re.findall(pattern, content, re.IGNORECASE)
+                species_counts.extend(matches)
+            if species_counts:
+                numbers = [int(x) for x in species_counts if x.isdigit()]
+                if numbers:
+                    max_species = max(numbers)
+                    results.append(f"BIRD_SPECIES_COUNT: {max_species}")
+            view_match = re.search(r'"viewCount":"(\d+)"', content)
+            if view_match:
+                views = int(view_match.group(1))
+                results.append(f"VIEWS: {views:,}")
         return "\n".join(results) if results else f"Basic info extracted for video {video_id}"
-        
     except Exception as e:
         return f"YouTube extraction error: {str(e)}"
 
 @tool
 def decode_reversed_text(text: str) -> str:
-    """
-    Decode reversed text questions with specific answer extraction.
-    
-    Args:
-        text: Text that may contain reversed content
-    
-    Returns:
-        Decoded answer or direction opposite
-    """
+    """Decode reversed text."""
     try:
-        # Handle the specific reversed question pattern
         if "ecnetnes siht dnatsrednu uoy fi" in text.lower():
-            # Reverse the entire text to read it normally
             reversed_text = text[::-1]
-            
-            # Look for direction words and return their opposites
             reversed_lower = reversed_text.lower()
-            if "left" in reversed_lower:
-                return "right"
-            elif "right" in reversed_lower:
-                return "left"
-            elif "up" in reversed_lower:
-                return "down"
-            elif "down" in reversed_lower:
-                return "up"
-            elif "north" in reversed_lower:
-                return "south"
-            elif "south" in reversed_lower:
-                return "north"
-            elif "east" in reversed_lower:
-                return "west"
-            elif "west" in reversed_lower:
-                return "east"
-            
-            # If no specific direction found, return the reversed text
+            opposites = {
+                "left": "right", "right": "left",
+                "up": "down", "down": "up",
+                "north": "south", "south": "north",
+                "east": "west", "west": "east"
+            }
+            for key, value in opposites.items():
+                if key in reversed_lower:
+                    return value
             return reversed_text
-        
-        # Default: reverse the input
         return text[::-1]
-        
     except Exception as e:
         return f"Text decoding error: {str(e)}"
 
 @tool
 def solve_advanced_math(problem: str) -> str:
-    """
-    Solve mathematical problems with specific pattern recognition for GAIA.
-    
-    Args:
-        problem: Mathematical problem description
-    
-    Returns:
-        Specific numerical answer or solution steps
-    """
+    """Solve advanced math problems and commutativity tables."""
     try:
         problem_lower = problem.lower()
-        
-        # Handle commutativity table problems
         if "commutative" in problem_lower and "|" in problem:
             lines = problem.split('\n')
             table_lines = [line for line in lines if '|' in line and any(x in line for x in ['a', 'b', 'c', 'd', 'e'])]
-            
-            if len(table_lines) >= 6:  # Header + 5 rows
+            if len(table_lines) >= 6:
                 elements = ['a', 'b', 'c', 'd', 'e']
                 table = {}
-                
-                # Parse the operation table
-                for i, line in enumerate(table_lines[1:]):  # Skip header
+                for i, line in enumerate(table_lines[1:]):
                     if i < 5:
                         parts = [p.strip() for p in line.split('|') if p.strip()]
                         if len(parts) >= 6:
@@ -307,8 +144,6 @@ def solve_advanced_math(problem: str) -> str:
                             for j, elem in enumerate(elements):
                                 if j + 2 < len(parts):
                                     table[(row_elem, elem)] = parts[j + 2]
-                
-                # Find elements that break commutativity
                 breaking_elements = set()
                 for a in elements:
                     for b in elements:
@@ -318,284 +153,126 @@ def solve_advanced_math(problem: str) -> str:
                             if ab and ba and ab != ba:
                                 breaking_elements.add(a)
                                 breaking_elements.add(b)
-                
                 result = sorted(list(breaking_elements))
                 return ', '.join(result) if result else "No elements break commutativity"
-        
-        # Handle chess notation
         elif "chess" in problem_lower or "move" in problem_lower:
-            # Look for chess notation patterns
             chess_moves = re.findall(r'\b[KQRBN]?[a-h]?[1-8]?x?[a-h][1-8][+#]?\b', problem)
             if chess_moves:
                 return f"Chess moves found: {', '.join(chess_moves)}"
             return "Analyze position for best move: check for tactics, threats, and forcing moves"
-        
-        # Handle numerical calculations
         numbers = re.findall(r'-?\d+\.?\d*', problem)
         if numbers:
             nums = [float(n) for n in numbers if n.replace('.', '').replace('-', '').isdigit()]
-            
             if "average" in problem_lower or "mean" in problem_lower:
                 if nums:
                     return str(sum(nums) / len(nums))
-            
             if "sum" in problem_lower or "total" in problem_lower:
                 if nums:
                     return str(sum(nums))
-            
             if "product" in problem_lower:
                 if nums:
                     result = 1
                     for n in nums:
                         result *= n
                     return str(result)
-        
-        # Handle percentage calculations
         if "%" in problem or "percent" in problem_lower:
             percentages = re.findall(r'(\d+\.?\d*)%', problem)
             if percentages:
                 return f"Percentages found: {', '.join(percentages)}%"
-        
         return f"Math problem requires specific calculation. Numbers found: {numbers}"
-        
     except Exception as e:
         return f"Math solver error: {str(e)}"
 
+@tool
+def get_detailed_wikipedia(topic: str) -> str:
+    """Get detailed Wikipedia information."""
+    try:
+        time.sleep(1)
+        topic_clean = topic.replace(" ", "_").strip()
+        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{topic_clean}"
+        response = requests.get(summary_url, timeout=12)
+        if response.status_code == 200:
+            data = response.json()
+            results = []
+            results.append(f"TITLE: {data.get('title', '')}")
+            results.append(f"EXTRACT: {data.get('extract', '')}")
+            page_url = data.get('content_urls', {}).get('desktop', {}).get('page', '')
+            if page_url:
+                results.append(f"URL: {page_url}")
+            return "\n".join(results)
+        # Fallback to search API
+        search_url = "https://en.wikipedia.org/w/api.php"
+        params = {
+            "action": "query",
+            "format": "json",
+            "list": "search",
+            "srsearch": topic,
+            "srlimit": 5
+        }
+        search_response = requests.get(search_url, params=params, timeout=12)
+        if search_response.status_code == 200:
+            search_data = search_response.json()
+            results = []
+            for item in search_data.get('query', {}).get('search', [])[:3]:
+                title = item['title']
+                snippet = re.sub(r'<[^>]+>', '', item['snippet'])
+                results.append(f"TITLE: {title}\nSNIPPET: {snippet}")
+            return "\n\n".join(results) if results else "No Wikipedia results found"
+        return f"Wikipedia lookup failed for: {topic}"
+    except Exception as e:
+        return f"Wikipedia error: {str(e)}"
+
 # --- Optimized Agent Class ---
+
 class OptimizedGAIAAgent:
     def __init__(self):
         print("Initializing Optimized GAIA Agent...")
-        
         self.tools = [
             smart_web_search,
-            get_wikipedia_info,
             extract_youtube_details,
             decode_reversed_text,
-            solve_advanced_math
+            solve_advanced_math,
+            get_detailed_wikipedia
         ]
-        
-        # Initialize CodeAgent with better error handling
         try:
             self.agent = CodeAgent(
                 tools=self.tools,
-                model="gpt-3.5-turbo",  # Added required model parameter
-                additional_authorized_imports=["math", "re", "json", "time"]
+                additional_authorized_imports=["math", "re", "json", "time"],
+                model="gpt-4"  # Specify your model here
             )
             print("✅ CodeAgent initialized")
         except Exception as e:
             print(f"⚠️ CodeAgent failed: {e}")
             self.agent = None
-    
+
     def analyze_and_solve(self, question: str) -> str:
-        """Analyze question type and provide targeted solution"""
+        """Analyze question type and provide targeted solution."""
         question_lower = question.lower()
-        
-        # Reversed text questions - high priority
         if "ecnetnes siht dnatsrednu uoy fi" in question_lower:
             return decode_reversed_text(question)
-        
-        # YouTube questions
         if "youtube.com" in question or "youtu.be" in question:
             url_match = re.search(r'https?://(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]+)', question)
             if url_match:
                 result = extract_youtube_details(url_match.group(0))
-                # If asking for highest number of bird species
                 if "highest number" in question_lower and "bird species" in question_lower:
                     numbers = re.findall(r'BIRD_SPECIES_COUNT:\s*(\d+)', result)
                     if numbers:
-                        return max([int(x) for x in numbers])
+                        return str(max([int(x) for x in numbers]))
                 return result
-        
-        # Math problems
         if any(term in question_lower for term in ["commutative", "operation", "table", "chess", "checkmate"]):
             return solve_advanced_math(question)
-        
-        # Wikipedia-focused searches
-        if any(term in question_lower for term in ["who", "what", "when", "where", "wikipedia", "article"]):
-            return get_wikipedia_info(question)
-        
-        # Olympics questions
-        if "olympics" in question_lower or "1928" in question:
-            return get_wikipedia_info("1928 Summer Olympics")
-        
-        # Default to smart search with delay
-        return smart_web_search(question)
-    
-    def solve(self, question: str) -> str:
-        """Main solving method with fallback chain"""
-        print(f"Solving: {question[:80]}...")
-        
-        try:
-            # Try direct analysis first
-            direct_result = self.analyze_and_solve(question)
-            if direct_result and len(str(direct_result).strip()) > 3:
-                return str(direct_result)
-        except Exception as e:
-            print(f"Direct analysis failed: {e}")
-        
-        # Try CodeAgent with rate limiting
+        # Default: Use agent if available
         if self.agent:
             try:
-                time.sleep(2)  # Rate limiting
-                result = self.agent.run(question)
-                if result and len(str(result).strip()) > 3:
-                    return str(result)
+                return self.agent.run(question)
             except Exception as e:
-                print(f"CodeAgent failed: {e}")
-        
-        # Final fallback
-        time.sleep(3)
-        return smart_web_search(question)
-
-def run_evaluation(profile: gr.OAuthProfile | None):
-    """Run evaluation with better error handling and rate limiting"""
-    if not profile:
-        return "❌ Please log in to Hugging Face first.", None
-    
-    username = profile.username
-    api_url = DEFAULT_API_URL
-    
-    # Initialize agent
-    try:
-        agent = OptimizedGAIAAgent()
-    except Exception as e:
-        return f"❌ Failed to initialize agent: {e}", None
-    
-    # Get questions
-    try:
-        print("Fetching questions...")
-        response = requests.get(f"{api_url}/questions", timeout=30)
-        response.raise_for_status()
-        questions = response.json()
-        print(f"✅ Retrieved {len(questions)} questions")
-    except Exception as e:
-        return f"❌ Failed to get questions: {e}", None
-    
-    # Process questions with rate limiting
-    results = []
-    answers = []
-    success_count = 0
-    
-    for i, item in enumerate(questions):
-        task_id = item.get("task_id")
-        question = item.get("question")
-        
-        if not task_id or not question:
-            continue
-        
-        print(f"\n📝 Processing {i+1}/{len(questions)}: {task_id}")
-        
-        try:
-            start_time = time.time()
-            answer = agent.solve(question)
-            duration = time.time() - start_time
-            
-            # Ensure we have a valid answer
-            if answer and len(str(answer).strip()) > 1:
-                success_count += 1
-                status = "✅"
-            else:
-                answer = "Unable to determine answer"
-                status = "❌"
-            
-            answers.append({
-                "task_id": task_id,
-                "submitted_answer": str(answer)
-            })
-            
-            results.append({
-                "Status": status,
-                "Task": task_id,
-                "Question": question[:60] + "...",
-                "Answer": str(answer)[:80] + "...",
-                "Time": f"{duration:.1f}s"
-            })
-            
-            print(f"{status} Answer: {str(answer)[:100]}")
-            
-            # Rate limiting between questions
-            time.sleep(random.uniform(2, 4))
-            
-        except Exception as e:
-            error_msg = f"Error: {str(e)}"
-            answers.append({
-                "task_id": task_id,
-                "submitted_answer": error_msg
-            })
-            results.append({
-                "Status": "❌",
-                "Task": task_id,
-                "Question": question[:60] + "...",
-                "Answer": error_msg,
-                "Time": "ERROR"
-            })
-            print(f"❌ Error: {e}")
-    
-    # Submit results
-    space_id = os.getenv("SPACE_ID", "unknown")
-    submission = {
-        "username": username,
-        "agent_code": f"https://huggingface.co/spaces/{space_id}",
-        "answers": answers
-    }
-    
-    try:
-        print(f"📤 Submitting {len(answers)} answers...")
-        response = requests.post(f"{api_url}/submit", json=submission, timeout=120)
-        response.raise_for_status()
-        result = response.json()
-        
-        success_rate = (success_count / len(questions)) * 100 if questions else 0
-        
-        status = f"""🎉 Evaluation Complete!
-
-👤 User: {result.get('username', username)}
-📊 Score: {result.get('score', 'N/A')}%
-✅ Correct: {result.get('correct_count', '?')}/{result.get('total_attempted', '?')}
-📝 Questions: {len(questions)}
-📤 Submitted: {len(answers)}
-🎯 Agent Success Rate: {success_rate:.1f}%
-
-💬 {result.get('message', 'Submitted successfully')}"""
-        
-        return status, pd.DataFrame(results)
-        
-    except Exception as e:
-        error_status = f"❌ Submission failed: {e}\n\nProcessed {len(results)} questions with {success_count} successful answers."
-        return error_status, pd.DataFrame(results)
+                return f"Agent error: {str(e)}"
+        return "No agent available to process the question."
 
-# --- Clean Gradio Interface ---
-with gr.Blocks(title="Optimized GAIA Agent", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🎯 Optimized GAIA Agent")
-    gr.Markdown("**Rate-limited search • Pattern recognition • Specific answer extraction**")
-    
-    with gr.Row():
-        gr.LoginButton()
-        run_btn = gr.Button("🚀 Run Evaluation", variant="primary", size="lg")
-    
-    with gr.Row():
-        status = gr.Textbox(
-            label="📊 Evaluation Status", 
-            lines=12, 
-            interactive=False,
-            placeholder="Click 'Run Evaluation' to start..."
-        )
-    
-    results_df = gr.DataFrame(
-        label="📋 Detailed Results",
-        interactive=False,
-        wrap=True
-    )
-    
-    run_btn.click(fn=run_evaluation, outputs=[status, results_df])
+# --- Example usage ---
 
 if __name__ == "__main__":
-    print("🎯 Starting Optimized GAIA Agent...")
-    
-    # Environment check
-    env_vars = ["SPACE_ID", "SERPER_API_KEY", "WIKIPEDIA_API_KEY"]
-    for var in env_vars:
-        status = "✅" if os.getenv(var) else "⚠️"
-        print(f"{status} {var}")
-    
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    agent = OptimizedGAIAAgent()
+    # Example question
+    Q = "How many studio albums were published by Mercedes Sosa between 2000 and 2009?"
+    print(agent.analyze_and_solve(Q))