web-search-improved

agent.py

@@ -1,45 +1,52 @@
 from typing import TypedDict, Annotated
 import os
-from dotenv import load_dotenv
 from langgraph.graph.message import add_messages
-
-# Load environment variables from .env file
-load_dotenv()
 from langchain_core.messages import AnyMessage, HumanMessage, AIMessage, SystemMessage
 from langgraph.prebuilt import ToolNode
 from langgraph.graph import START, StateGraph
 from langgraph.checkpoint.memory import MemorySaver
 from langgraph.prebuilt import tools_condition
-from langchain_openai import ChatOpenAI
+from langchain_huggingface import HuggingFaceEndpoint, ChatHuggingFace
 from tools import agent_tools
-from utils import format_gaia_answer, log_agent_step
+from utils import format_gaia_answer, analyze_question_type, create_execution_plan, log_agent_step
 
-# Initialize OpenAI LLM with GPT-4o (most capable model)
-chat = ChatOpenAI(
-    model="gpt-4o",
+# Initialize LLM (same as unit3)
+llm = HuggingFaceEndpoint(
+    repo_id="Qwen/Qwen2.5-Coder-32B-Instruct",
+    huggingfacehub_api_token=os.environ.get("HUGGINGFACE_API_TOKEN"),
     temperature=0.1,
-    max_tokens=1024,
-    api_key=os.environ.get("OPENAI_API_KEY")
+    max_new_tokens=1024,
 )
 
+chat = ChatHuggingFace(llm=llm, verbose=True)
 chat_with_tools = chat.bind_tools(agent_tools)
 
-# System prompt for GAIA evaluation (exact format required by HF)
-SYSTEM_PROMPT = """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.
+# System prompt for intelligent question answering
+SYSTEM_PROMPT = """You are a highly capable AI assistant designed to answer questions accurately and helpfully.
+
+Your approach should include:
+- Multi-step reasoning and planning for complex questions
+- Intelligent tool usage when needed for web search, file processing, calculations, and analysis
+- Precise, factual answers based on reliable information
+- Breaking down complex questions into manageable steps
 
-You have access to tools that can help you:
-- Search the web for current information
-- Download and process files associated with task IDs
-- Analyze images
-- Perform calculations
-- Process text
+IMPORTANT GUIDELINES:
+1. Think step-by-step and use available tools when they can help provide better answers
+2. For current information: Search the web for up-to-date facts
+3. For files: Process associated files when task_id is provided
+4. For visual content: Analyze images carefully when present
+5. For calculations: Use computational tools for accuracy
+6. Provide concise, direct answers without unnecessary prefixes
+7. Focus on accuracy and helpfulness
+8. Be factual and avoid speculation
 
-IMPORTANT: You must provide a specific answer in the FINAL ANSWER format. Do not say you cannot find information or provide general approaches. Use web search to find the information you need, but limit yourself to 2-3 search attempts maximum. If you cannot find perfect information, make your best determination based on what you found and provide a concrete FINAL ANSWER. Always end with a specific FINAL ANSWER, never with explanations about not finding information."""
+Your goal is to be as helpful and accurate as possible while using the right tools for each task."""
 
 # Generate the AgentState
 class AgentState(TypedDict):
     messages: Annotated[list[AnyMessage], add_messages]
     task_id: str
+    question_analysis: dict
 
 def assistant(state: AgentState):
     """Main assistant function that processes messages and calls tools."""
@@ -73,7 +80,9 @@ def create_smart_agent():
     )
     builder.add_edge("tools", "assistant")
     
-    agent = builder.compile()
+    # Add memory
+    memory = MemorySaver()
+    agent = builder.compile(checkpointer=memory)
     
     return agent
 
@@ -82,41 +91,44 @@ class SmartAgent:
     
     def __init__(self):
         self.agent = create_smart_agent()
-        print("🤖 Smart Agent initialized with OpenAI GPT-4o and tools")
+        print("🤖 Smart Agent initialized with LangGraph and tools")
     
-    def __call__(self, question: str, task_id: str = None) -> tuple:
-        """Process a question and return the formatted answer and reasoning trace."""
+    def __call__(self, question: str, task_id: str = None) -> str:
+        """Process a question and return the formatted answer."""
         try:
             print(f"\n🎯 Processing question: {question[:100]}...")
             
+            # Analyze the question
+            analysis = analyze_question_type(question)
+            print(f"📊 Question analysis: {analysis}")
+            
+            # Create execution plan
+            plan = create_execution_plan(question, task_id)
+            print(f"📋 Execution plan: {plan}")
+            
+            # Prepare the question with task_id context if available
             enhanced_question = question
             if task_id:
-                enhanced_question = f"Task ID: {task_id}\n\nQuestion: {question}"
+                enhanced_question = f"Task ID: {task_id}\n\nQuestion: {question}\n\nNote: If this question involves files, use the file_download tool with task_id '{task_id}' to access associated files."
             
-            config = {
-                "recursion_limit": 15
-            }
+            # Invoke the agent
+            thread_id = f"task-{task_id}" if task_id else "general"
+            config = {"configurable": {"thread_id": thread_id}}
             
             initial_state = {
                 "messages": [HumanMessage(content=enhanced_question)],
-                "task_id": task_id or ""
+                "task_id": task_id or "",
+                "question_analysis": analysis
             }
             
             result = self.agent.invoke(initial_state, config=config)
             
+            # Extract the final answer
             if result and 'messages' in result and result['messages']:
                 final_message = result['messages'][-1]
                 raw_answer = final_message.content
-                
-                reasoning_trace = []
-                for msg in result['messages']:
-                    if hasattr(msg, 'content') and msg.content:
-                        reasoning_trace.append(msg.content)
-                
-                reasoning_text = "\n---\n".join(reasoning_trace)
             else:
                 raw_answer = "No response generated"
-                reasoning_text = "No reasoning trace available"
             
             # Format the answer for submission
             formatted_answer = format_gaia_answer(raw_answer)
@@ -124,16 +136,11 @@ class SmartAgent:
             print(f"✅ Raw answer: {raw_answer}")
             print(f"🎯 Formatted answer: {formatted_answer}")
             
-            # Validate the formatted answer
-            if not formatted_answer or formatted_answer.strip() == "":
-                print("⚠️ WARNING: Empty formatted answer!")
-                formatted_answer = "ERROR: No valid answer extracted"
-            
-            return formatted_answer, reasoning_text
+            return formatted_answer
             
         except Exception as e:
             error_msg = f"Error processing question: {str(e)}"
             print(f"❌ {error_msg}")
-            return error_msg, f"Error occurred: {str(e)}"
+            return error_msg
 
 smart_agent = SmartAgent()

app.py

@@ -71,22 +71,12 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
             print(f"Skipping item with missing task_id or question: {item}")
             continue
         try:
-            submitted_answer, reasoning_trace = agent(question_text, task_id)
-            answers_payload.append({
-                "task_id": task_id, 
-                "submitted_answer": submitted_answer,
-                "reasoning_trace": reasoning_trace
-            })
+            submitted_answer = agent(question_text, task_id)
+            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
         except Exception as e:
              print(f"Error running agent on task {task_id}: {e}")
-             error_answer = f"AGENT ERROR: {e}"
-             answers_payload.append({
-                "task_id": task_id, 
-                "submitted_answer": error_answer,
-                "reasoning_trace": f"Error occurred: {str(e)}"
-            })
-             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": error_answer})
+             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
 
     if not answers_payload:
         print("Agent did not produce any answers to submit.")

requirements.txt

@@ -1,7 +1,7 @@
 # Core dependencies from unit3
 langchain
 langchain-community
-langchain-openai
+langchain-huggingface
 langgraph
 huggingface_hub
 
@@ -10,11 +10,8 @@ gradio
 requests
 pillow
 PyPDF2
-ddgs
+duckduckgo-search
 python-dotenv
-beautifulsoup4
-faiss-cpu
-langchain-text-splitters
 
 # For image processing and multimodal capabilities
 transformers

test_local.py

@@ -1,238 +1,137 @@
 #!/usr/bin/env python3
 """
-Test script for validating agent performance on a random GAIA question.
-Fetches one random question and tests the complete pipeline without submitting.
+Local testing script for the GAIA agent.
+Run this to test the agent before deploying to HF Spaces.
 """
 
-import time
-from utils import fetch_random_question, format_gaia_answer
+import os
+import sys
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Add current directory to path for imports
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+from utils import fetch_random_question, analyze_question_type
 from agent import smart_agent
 
-def test_predefined_gaia_question():
-    """Test the agent with a predefined GAIA question to verify web search and answer format."""
-    
-    print("🧪 Testing predefined GAIA question (1928 Olympics)")
-    print("="*60)
+def test_question_analysis():
+    """Test the question analysis functionality."""
+    print("🧪 Testing question analysis...")
+    
+    test_questions = [
+        "What is the current population of Tokyo?",
+        "Calculate 15 * 23 + 45",
+        "Analyze the image shown in the document",
+        "Extract all dates from the provided text file"
+    ]
+    
+    for question in test_questions:
+        analysis = analyze_question_type(question)
+        print(f"Question: {question}")
+        print(f"Analysis: {analysis}")
+        print()
+
+def test_tools():
+    """Test individual tools."""
+    print("🔧 Testing individual tools...")
     
-    # Predefined question that requires web search
-    question = "What country had the least number of athletes at the 1928 Summer Olympics? If there's a tie for a number of athletes, return the first in alphabetical order. Give the IOC country code as your answer."
-    task_id = "predefined_test"
+    # Test calculator
+    from tools import calculator_tool
+    calc_result = calculator_tool.func("15 + 27")
+    print(f"Calculator test: {calc_result}")
     
-    print(f"❓ Question: {question}")
-    print()
-    
-    # Run the agent
-    print("🤖 Running smart agent on the predefined question...")
+    # Test web search (if available)
     try:
-        start_time = time.time()
-        answer, reasoning_trace = smart_agent(question, task_id)
-        end_time = time.time()
-        
-        processing_time = end_time - start_time
-        print(f"✅ Agent completed in {processing_time:.2f} seconds")
-        print()
-        
+        from tools import web_search_tool
+        search_result = web_search_tool.func("Python programming language")
+        print(f"Web search test: {search_result[:100]}...")
     except Exception as e:
-        print(f"❌ Error running agent: {e}")
-        return False
+        print(f"Web search test failed: {e}")
     
-    # Display results
-    print("📊 AGENT RESULTS")
-    print("-" * 40)
-    print(f"🎯 Formatted Answer: '{answer}'")
-    print(f"📝 Reasoning Length: {len(reasoning_trace)} characters")
-    print(f"⏱️  Processing Time: {processing_time:.2f}s")
     print()
+
+def test_agent_simple():
+    """Test the agent with a simple question."""
+    print("🤖 Testing Smart agent with simple question...")
     
-    # Show reasoning trace preview
-    print("🧠 REASONING TRACE PREVIEW")
-    print("-" * 40)
-    reasoning_preview = reasoning_trace[:400] + "..." if len(reasoning_trace) > 400 else reasoning_trace
-    print(reasoning_preview)
-    print()
-    
-    # Validate answer format for GAIA
-    print("✅ GAIA FORMAT VALIDATION")
-    print("-" * 40)
-    
-    # Check if answer is not empty
-    if answer and answer.strip():
-        print("✅ Answer is not empty")
-    else:
-        print("❌ Answer is empty or None")
-        return False
-    
-    # Check if answer looks like IOC country code (2-3 uppercase letters)
-    import re
-    if re.match(r'^[A-Z]{2,3}$', answer.strip()):
-        print(f"✅ Answer '{answer}' matches IOC country code format")
-    else:
-        print(f"⚠️  Answer '{answer}' may not be in correct IOC format (should be 2-3 uppercase letters)")
-    
-    # Check if web search was used (look for web_search in reasoning)
-    if "web_search" in reasoning_trace.lower() or "search" in reasoning_trace.lower():
-        print("✅ Agent appears to have used web search")
-    else:
-        print("⚠️  No clear evidence of web search usage")
-    
-    # Check answer length (should be short for country code)
-    if len(answer.strip()) <= 5:
-        print("✅ Answer length is appropriate for country code")
-    else:
-        print("⚠️  Answer seems too long for a country code")
+    test_question = "What is 25 + 17?"
+    try:
+        result = smart_agent(test_question)
+        print(f"Question: {test_question}")
+        print(f"Answer: {result}")
+        print("✅ Simple test passed!")
+    except Exception as e:
+        print(f"❌ Simple test failed: {e}")
     
     print()
-    
-    # Final validation
-    print("🏁 FINAL VALIDATION")
-    print("-" * 40)
-    
-    if answer and answer.strip() and len(answer.strip()) <= 5:
-        print("✅ PREDEFINED TEST PASSED - Answer format suitable for GAIA")
-        print(f"🎯 Agent produced: '{answer}' for 1928 Olympics question")
-        return True
-    else:
-        print("❌ PREDEFINED TEST FAILED - Answer format needs improvement")
-        return False
 
-def test_random_gaia_question():
-    """Test the agent with a random GAIA question and validate the complete pipeline."""
-    
-    print("🔧 GAIA Random Question Test")
-    print("="*60)
+def test_agent_with_api():
+    """Test the agent with a real GAIA question from the API."""
+    print("🌐 Testing with real GAIA question from API...")
     
-    # Step 1: Fetch a random question
-    print("📡 Fetching random question from GAIA API...")
     try:
         question_data = fetch_random_question()
         if not question_data:
-            print("❌ Failed to fetch random question")
-            return False
+            print("❌ Failed to fetch question from API")
+            return
         
-        task_id = question_data.get("task_id", "unknown")
-        question_text = question_data.get("question", "")
+        task_id = question_data.get("task_id")
+        question = question_data.get("question")
         
-        if not question_text:
-            print("❌ No question text in response")
-            return False
-            
-        print(f"✅ Successfully fetched question")
-        print(f"📋 Task ID: {task_id}")
-        print(f"❓ Question: {question_text}")
-        print()
+        print(f"Task ID: {task_id}")
+        print(f"Question: {question}")
         
-    except Exception as e:
-        print(f"❌ Error fetching question: {e}")
-        return False
-    
-    # Step 2: Run the agent
-    print("🤖 Running smart agent on the question...")
-    try:
-        start_time = time.time()
-        answer, reasoning_trace = smart_agent(question_text, task_id)
-        end_time = time.time()
-        
-        processing_time = end_time - start_time
-        print(f"✅ Agent completed in {processing_time:.2f} seconds")
-        print()
+        # Run the agent
+        answer = smart_agent(question, task_id)
+        print(f"Agent Answer: {answer}")
+        print("✅ API test completed!")
         
     except Exception as e:
-        print(f"❌ Error running agent: {e}")
-        return False
+        print(f"❌ API test failed: {e}")
     
-    # Step 3: Display results
-    print("📊 AGENT RESULTS")
-    print("-" * 40)
-    print(f"🎯 Formatted Answer: '{answer}'")
-    print(f"📝 Reasoning Length: {len(reasoning_trace)} characters")
-    print(f"⏱️  Processing Time: {processing_time:.2f}s")
     print()
+
+def check_environment():
+    """Check if all required environment variables are set."""
+    print("🔍 Checking environment...")
     
-    # Step 4: Show reasoning trace preview
-    print("🧠 REASONING TRACE PREVIEW")
-    print("-" * 40)
-    reasoning_preview = reasoning_trace[:300] + "..." if len(reasoning_trace) > 300 else reasoning_trace
-    print(reasoning_preview)
-    print()
+    required_vars = ["HUGGINGFACE_API_TOKEN"]
+    missing_vars = []
     
-    # Step 5: Validate answer format
-    print("✅ ANSWER VALIDATION")
-    print("-" * 40)
+    for var in required_vars:
+        if not os.getenv(var):
+            missing_vars.append(var)
+        else:
+            print(f"✅ {var} is set")
     
-    # Check if answer is not empty
-    if answer and answer.strip():
-        print("✅ Answer is not empty")
-    else:
-        print("❌ Answer is empty or None")
+    if missing_vars:
+        print(f"❌ Missing environment variables: {missing_vars}")
+        print("Please set these in your .env file or environment")
         return False
     
-    # Check if answer contains error messages
-    if "ERROR" in answer.upper() or "FAILED" in answer.upper():
-        print("⚠️  Answer contains error message")
-    else:
-        print("✅ Answer appears to be valid (no error messages)")
-    
-    # Check answer length (reasonable bounds)
-    if len(answer) > 1000:
-        print("⚠️  Answer is very long (>1000 chars) - might need review")
-    else:
-        print("✅ Answer length is reasonable")
-    
-    print()
-    
-    # Step 6: Show submission format
-    print("📡 SUBMISSION FORMAT PREVIEW")
-    print("-" * 40)
-    
-    submission_entry = {
-        "task_id": task_id,
-        "model_answer": answer,
-        "reasoning_trace": reasoning_trace
-    }
-    
-    # Validate required fields
-    required_fields = ["task_id", "model_answer"]
-    all_valid = True
-    
-    for field in required_fields:
-        if field in submission_entry and submission_entry[field]:
-            print(f"✅ {field}: '{submission_entry[field][:50]}{'...' if len(str(submission_entry[field])) > 50 else ''}'")
-        else:
-            print(f"❌ Missing or empty {field}")
-            all_valid = False
+    print("✅ All required environment variables are set")
+    return True
+
+def main():
+    """Run all tests."""
+    print("🚀 Starting GAIA Agent Local Tests")
+    print("=" * 50)
     
-    # Check optional fields
-    if "reasoning_trace" in submission_entry and submission_entry["reasoning_trace"]:
-        print(f"✅ reasoning_trace: Present ({len(submission_entry['reasoning_trace'])} chars)")
-    else:
-        print("ℹ️  reasoning_trace: Not present (optional)")
+    # Check environment first
+    if not check_environment():
+        print("❌ Environment check failed. Please fix and try again.")
+        return
     
     print()
     
-    # Step 7: Final validation
-    print("🏁 FINAL VALIDATION")
-    print("-" * 40)
-    
-    if all_valid and answer and answer.strip():
-        print("✅ ALL CHECKS PASSED - Agent is ready for submission!")
-        print("🚀 You can now run the full evaluation with confidence.")
-        return True
-    else:
-        print("❌ SOME CHECKS FAILED - Please review the issues above.")
-        return False
+    # Run tests
+    # test_question_analysis()
+    # test_tools()
+    # test_agent_simple()
+    test_agent_with_api()
 
 if __name__ == "__main__":
-    print("🧪 Testing agent with predefined GAIA question...")
-    print("This test validates web search functionality and answer formatting.")
-    print()
-    
-    # Test the predefined 1928 Olympics question
-    success = test_predefined_gaia_question()
-    
-    print("\n" + "="*60)
-    if success:
-        print("🎉 Predefined test completed successfully! Agent produces well-defined answers.")
-        print("💡 You can also run test_random_gaia_question() for additional testing.")
-    else:
-        print("⚠️  Predefined test revealed issues that need to be addressed.")
-    print("="*60)
+    main()

tools.py

@@ -4,21 +4,28 @@ import os
 from PIL import Image
 import io
 import base64
-from ddgs import DDGS
+from langchain_community.tools import DuckDuckGoSearchRun
 from typing import Optional
 import json
 import PyPDF2
 import tempfile
-import requests
-from bs4 import BeautifulSoup
-from langchain_community.vectorstores import FAISS
-from langchain_openai import OpenAIEmbeddings
-from langchain_text_splitters import RecursiveCharacterTextSplitter
-from langchain.schema import Document
-from dotenv import load_dotenv
 
-# Load environment variables
-load_dotenv()
+# Initialize web search tool
+search_tool = DuckDuckGoSearchRun()
+
+def web_search_tool_func(query: str) -> str:
+    """Searches the web for information using DuckDuckGo."""
+    try:
+        results = search_tool.run(query)
+        return results
+    except Exception as e:
+        return f"Web search failed: {str(e)}"
+
+web_search_tool = Tool(
+    name="web_search",
+    func=web_search_tool_func,
+    description="Searches the web for current information. Use this for factual questions, recent events, or when you need to find information not in your training data."
+)
 
 def file_download_tool_func(task_id: str) -> str:
     """Downloads a file associated with a GAIA task ID."""
@@ -101,6 +108,26 @@ image_analysis_tool = Tool(
     description="Analyzes images to extract information. Use this for questions involving visual content."
 )
 
+def calculator_tool_func(expression: str) -> str:
+    """Performs mathematical calculations safely."""
+    try:
+        # Basic safety check - only allow certain characters
+        allowed_chars = set('0123456789+-*/().= ')
+        if not all(c in allowed_chars for c in expression):
+            return f"Invalid characters in expression: {expression}"
+        
+        # Use eval safely for basic math
+        result = eval(expression)
+        return f"Calculation result: {expression} = {result}"
+    except Exception as e:
+        return f"Calculation failed for '{expression}': {str(e)}"
+
+calculator_tool = Tool(
+    name="calculator",
+    func=calculator_tool_func,
+    description="Performs mathematical calculations. Use this for numerical computations and math problems."
+)
+
 def text_processor_tool_func(text: str, operation: str = "summarize") -> str:
     """Processes text for various operations like summarization, extraction, etc."""
     try:
@@ -143,267 +170,11 @@ text_processor_tool = Tool(
     description="Processes text for various operations like summarization, number extraction, date extraction. Specify operation as second parameter."
 )
 
-def enhanced_web_retrieval_tool_func(query: str, backend: str = "bing") -> str:
-    """Enhanced web search with cascading fallback: Wikipedia first, then general web search."""
-    try:
-        print(f"🔍 Enhanced web retrieval for: {query}")
-        
-        # Step 1: Try Wikipedia search first
-        print("📚 Searching Wikipedia...")
-        wikipedia_results = get_wikipedia_search_urls(query, backend)
-        
-        if has_sufficient_results(wikipedia_results):
-            print(f"✅ Found {len(wikipedia_results)} Wikipedia results")
-            documents = fetch_and_process_results(wikipedia_results, "Wikipedia")
-            if documents:
-                return search_documents_with_vector_store(documents, query, "Wikipedia")
-        
-        # Step 2: Fallback to general web search
-        print("🌐 Wikipedia results insufficient, searching general web...")
-        web_results = get_general_web_search_urls(query, backend)
-        
-        if web_results:
-            print(f"✅ Found {len(web_results)} general web results")
-            documents = fetch_and_process_results(web_results, "General Web")
-            if documents:
-                return search_documents_with_vector_store(documents, query, "General Web")
-        
-        return "No sufficient results found in Wikipedia or general web search."
-        
-    except Exception as e:
-        return f"Enhanced web retrieval failed: {str(e)}"
-
-def get_wikipedia_search_urls(query: str, backend: str = "auto") -> list:
-    """Get search results from English Wikipedia using DDGS."""
-    try:
-        with DDGS() as ddgs:
-            # Create Wikipedia-specific search queries
-            wikipedia_queries = [
-                f"{query} site:en.wikipedia.org"
-            ]
-            
-            search_results = []
-            seen_urls = set()
-            
-            for wiki_query in wikipedia_queries:
-                try:
-                    results = list(ddgs.text(
-                        wiki_query, 
-                        max_results=8, 
-                        region="us-en", 
-                        backend=backend, 
-                        safesearch="moderate"
-                    ))
-                    
-                    for result in results:
-                        url = result.get('href', '')
-                        
-                        # Only include Wikipedia URLs and avoid duplicates
-                        if 'en.wikipedia.org' in url and url not in seen_urls:
-                            search_results.append({
-                                'url': url,
-                                'title': result.get('title', 'No title'),
-                                'snippet': result.get('body', 'No content')
-                            })
-                            seen_urls.add(url)
-                            
-                            # Limit to 6 unique Wikipedia pages
-                            if len(search_results) >= 6:
-                                break
-                    
-                    if len(search_results) >= 6:
-                        break
-                        
-                except Exception as e:
-                    print(f"Wikipedia search attempt failed: {e}")
-                    continue
-            
-            return search_results
-            
-    except Exception as e:
-        print(f"Wikipedia search URL retrieval failed: {e}")
-        return []
-
-def get_general_web_search_urls(query: str, backend: str = "auto") -> list:
-    """Get search results from general web using DDGS."""
-    try:
-        with DDGS() as ddgs:
-            search_results = []
-            seen_urls = set()
-            
-            try:
-                # General web search without site restriction
-                results = list(ddgs.text(
-                    query, 
-                    max_results=8, 
-                    region="us-en", 
-                    backend=backend, 
-                    safesearch="moderate"
-                ))
-                
-                for result in results:
-                    url = result.get('href', '')
-                    
-                    # Avoid duplicates and filter out low-quality sources
-                    if url not in seen_urls and is_quality_source(url):
-                        search_results.append({
-                            'url': url,
-                            'title': result.get('title', 'No title'),
-                            'snippet': result.get('body', 'No content')
-                        })
-                        seen_urls.add(url)
-                        
-                        # Limit to 6 unique web pages
-                        if len(search_results) >= 6:
-                            break
-                            
-            except Exception as e:
-                print(f"General web search attempt failed: {e}")
-            
-            return search_results
-            
-    except Exception as e:
-        print(f"General web search URL retrieval failed: {e}")
-        return []
-
-def is_quality_source(url: str) -> bool:
-    """Filter out low-quality or problematic sources."""
-    low_quality_domains = [
-        'pinterest.com', 'instagram.com', 'facebook.com', 'twitter.com',
-        'tiktok.com', 'youtube.com', 'reddit.com'
-    ]
-    
-    for domain in low_quality_domains:
-        if domain in url.lower():
-            return False
-    
-    return True
-
-def has_sufficient_results(results: list) -> bool:
-    """Check if search results are sufficient to proceed."""
-    if not results:
-        return False
-    
-    # Check for minimum number of results
-    if len(results) < 2:
-        return False
-    
-    # Check if results have meaningful content
-    meaningful_results = 0
-    for result in results:
-        snippet = result.get('snippet', '')
-        title = result.get('title', '')
-        
-        # Consider result meaningful if it has substantial content
-        if len(snippet) > 50 or len(title) > 10:
-            meaningful_results += 1
-    
-    return meaningful_results >= 2
-
-def fetch_and_process_results(results: list, source_type: str) -> list:
-    """Fetch and process webpage content from search results."""
-    documents = []
-    
-    for result in results[:4]:  # Process top 4 results
-        url = result.get('url', '')
-        title = result.get('title', 'No title')
-        
-        print(f"📄 Fetching content from: {title}")
-        content = fetch_webpage_content(url)
-        
-        if content and len(content.strip()) > 100:  # Ensure meaningful content
-            doc = Document(
-                page_content=content,
-                metadata={
-                    "source": url, 
-                    "title": title,
-                    "source_type": source_type
-                }
-            )
-            documents.append(doc)
-    
-    return documents
-
-def fetch_webpage_content(url: str) -> str:
-    """Fetch and extract clean text content from a webpage."""
-    try:
-        headers = {
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
-        }
-        
-        response = requests.get(url, headers=headers, timeout=10)
-        response.raise_for_status()
-        
-        # Parse HTML and extract text
-        soup = BeautifulSoup(response.content, 'html.parser')
-        
-        # Remove script and style elements
-        for script in soup(["script", "style"]):
-            script.decompose()
-        
-        # Get text content
-        text = soup.get_text()
-        
-        # Clean up text
-        lines = (line.strip() for line in text.splitlines())
-        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
-        text = ' '.join(chunk for chunk in chunks if chunk)
-        
-        return text[:30000]
-        
-    except Exception as e:
-        print(f"Failed to fetch content from {url}: {e}")
-        return ""
-
-def search_documents_with_vector_store(documents: list, query: str, source_type: str = "Web") -> str:
-    """Create vector store and search for relevant information."""
-    try:
-        # Split documents into chunks
-        text_splitter = RecursiveCharacterTextSplitter(
-            chunk_size=1000,
-            chunk_overlap=200,
-            length_function=len,
-        )
-        
-        splits = text_splitter.split_documents(documents)
-        
-        if not splits:
-            return "No content to process after splitting."
-        
-        # Create embeddings and vector store
-        embeddings = OpenAIEmbeddings()
-        vectorstore = FAISS.from_documents(splits, embeddings)
-        
-        # Search for relevant chunks with the original query
-        relevant_docs = vectorstore.similarity_search(query, k=5)
-        
-        # Format results with source type indication
-        results = []
-        results.append(f"🔍 Search Results from {source_type}:\n")
-        
-        for i, doc in enumerate(relevant_docs, 1):
-            source = doc.metadata.get('source', 'Unknown source')
-            title = doc.metadata.get('title', 'No title')
-            source_type_meta = doc.metadata.get('source_type', source_type)
-            content = doc.page_content[:2000]  # Increased content length
-            
-            results.append(f"Result {i} ({source_type_meta}) - {title}:\n{content}\nSource: {source}\n")
-        
-        return "\n---\n".join(results)
-        
-    except Exception as e:
-        return f"Vector search failed: {str(e)}"
-
-web_search_tool = Tool(
-    name="enhanced_web_retrieval",
-    func=enhanced_web_retrieval_tool_func,
-    description="Enhanced cascading web search with vector retrieval. First searches Wikipedia for reliable factual information, then falls back to general web search if insufficient results are found. Supports multiple search backends (auto, html, lite, bing) and uses semantic search to find relevant information. Ideal for comprehensive research on any topic."
-)
-
 # List of all tools for easy import
 agent_tools = [
     web_search_tool,
     file_download_tool,
     image_analysis_tool,
+    calculator_tool,
     text_processor_tool
 ]

utils.py

@@ -41,48 +41,81 @@ def submit_answers(username: str, agent_code: str, answers: List[Dict[str, str]]
 
 def format_gaia_answer(raw_answer: str) -> str:
     """Format the agent's raw answer for GAIA submission (exact match)."""
-    import re
+    # Remove common prefixes that might interfere with exact matching
+    prefixes_to_remove = [
+        "FINAL ANSWER:",
+        "Final Answer:",
+        "Answer:",
+        "The answer is:",
+        "The final answer is:",
+    ]
     
-    # Look for FINAL ANSWER: pattern (case insensitive)
-    final_answer_pattern = r'FINAL ANSWER:\s*(.+?)(?:\n|$)'
-    match = re.search(final_answer_pattern, raw_answer, re.IGNORECASE | re.DOTALL)
+    answer = raw_answer.strip()
     
-    if match:
-        answer = match.group(1).strip()
-    else:
-        # Fallback: try to extract from common patterns
-        fallback_patterns = [
-            r'(?:The\s+)?(?:final\s+)?answer\s+is:?\s*(.+?)(?:\n|$)',
-            r'(?:Answer|Result):\s*(.+?)(?:\n|$)',
-        ]
-        
-        answer = raw_answer.strip()
-        for pattern in fallback_patterns:
-            match = re.search(pattern, answer, re.IGNORECASE)
-            if match:
-                answer = match.group(1).strip()
-                break
-    
-    # Apply GAIA formatting rules
-    answer = answer.strip()
+    for prefix in prefixes_to_remove:
+        if answer.startswith(prefix):
+            answer = answer[len(prefix):].strip()
     
     # Remove trailing punctuation that might not be in ground truth
     while answer and answer[-1] in '.!?':
         answer = answer[:-1].strip()
     
-    # Remove quotes if they wrap the entire answer
-    if len(answer) >= 2 and answer[0] == answer[-1] and answer[0] in '"\'':
-        answer = answer[1:-1].strip()
+    return answer
+
+def analyze_question_type(question: str) -> Dict[str, bool]:
+    """Analyze what capabilities a question might need."""
+    question_lower = question.lower()
     
-    # Additional cleanup for common issues
-    # Remove "approximately" or similar qualifiers
-    answer = re.sub(r'^(?:approximately|about|roughly|around)\s+', '', answer, flags=re.IGNORECASE)
+    analysis = {
+        "needs_web_search": any(keyword in question_lower for keyword in [
+            "current", "recent", "latest", "today", "now", "2024", "2023"
+        ]),
+        "needs_file_processing": "file" in question_lower or "document" in question_lower,
+        "needs_calculation": any(keyword in question_lower for keyword in [
+            "calculate", "compute", "sum", "total", "average", "percentage", "multiply", "divide"
+        ]),
+        "needs_image_analysis": any(keyword in question_lower for keyword in [
+            "image", "picture", "photo", "visual", "shown", "displayed"
+        ]),
+        "needs_text_processing": any(keyword in question_lower for keyword in [
+            "extract", "find in", "search for", "list", "count"
+        ])
+    }
     
-    # For numbers, ensure no commas (as per GAIA rules)
-    if re.match(r'^[\d,]+(?:\.\d+)?$', answer):
-        answer = answer.replace(',', '')
+    return analysis
+
+def create_execution_plan(question: str, task_id: str = None) -> List[str]:
+    """Create a step-by-step execution plan for a GAIA question."""
+    analysis = analyze_question_type(question)
+    plan = []
     
-    return answer
+    # Always start with understanding the question
+    plan.append("Analyze the question to understand what information is needed")
+    
+    # Add file processing if needed
+    if task_id and analysis["needs_file_processing"]:
+        plan.append(f"Download and process any files associated with task {task_id}")
+    
+    # Add web search if needed
+    if analysis["needs_web_search"]:
+        plan.append("Search the web for current/recent information")
+    
+    # Add image analysis if needed
+    if analysis["needs_image_analysis"]:
+        plan.append("Analyze any images for visual information")
+    
+    # Add calculation if needed
+    if analysis["needs_calculation"]:
+        plan.append("Perform necessary calculations")
+    
+    # Add text processing if needed
+    if analysis["needs_text_processing"]:
+        plan.append("Process and extract specific information from text")
+    
+    # Always end with synthesis
+    plan.append("Synthesize all information to provide the final answer")
+    
+    return plan
 
 def log_agent_step(step: str, result: str, step_number: int = None):
     """Log agent execution steps for debugging."""