fixing

app.py

@@ -9,25 +9,26 @@ from pdfminer.high_level import extract_text
 from bs4 import BeautifulSoup
 from typing import List, Dict, Optional, Tuple
 from dotenv import load_dotenv
-from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig
+from transformers import AutoModelForCausalLM, AutoTokenizer
 import torch
 import time
 import gc
 
-# --- Load Environment Variables ---
+# --- Configuration ---
 load_dotenv()
 SERPER_API_KEY = os.getenv("SERPER_API_KEY")
+MODEL_NAME = "microsoft/Phi-3-mini-4k-instruct"
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Constants ---
-DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 MAX_STEPS = 6
 MAX_TOKENS = 256
-MODEL_NAME = "microsoft/Phi-3-mini-4k-instruct"
 TIMEOUT_PER_QUESTION = 45
 MAX_RESULT_LENGTH = 500
+MAX_ATTEMPTS = 2
 
-# --- Fixed Model Loading ---
-print("Loading model with fixed configuration...")
+# --- Model Initialization ---
+print("Initializing model with fixed cache configuration...")
 start_time = time.time()
 
 model = AutoModelForCausalLM.from_pretrained(
@@ -49,30 +50,35 @@ if tokenizer.pad_token is None:
 
 print(f"Model loaded in {time.time() - start_time:.2f} seconds")
 
-# --- Tools Implementation ---
+# --- Tool Implementations ---
 def web_search(query: str) -> str:
-    """Enhanced web search with better error handling"""
     try:
-        if SERPER_API_KEY:
-            params = {'q': query, 'num': 3}
-            headers = {'X-API-KEY': SERPER_API_KEY}
-            response = requests.post(
-                'https://google.serper.dev/search',
-                headers=headers,
-                json=params,
-                timeout=10
-            )
-            results = response.json()
-            if 'organic' in results:
-                return "\n".join([f"{r['title']}: {r['snippet']}" for r in results['organic'][:3]])[:MAX_RESULT_LENGTH]
-            return "No search results found"
-        else:
+        if not SERPER_API_KEY:
             return "Search API key not configured"
+            
+        params = {'q': query, 'num': 3}
+        headers = {'X-API-KEY': SERPER_API_KEY}
+        response = requests.post(
+            'https://google.serper.dev/search',
+            headers=headers,
+            json=params,
+            timeout=10
+        )
+        response.raise_for_status()
+        results = response.json()
+        
+        if 'organic' not in results or not results['organic']:
+            return "No relevant results found"
+            
+        output = []
+        for r in results['organic'][:3]:
+            if 'title' in r and 'snippet' in r:
+                output.append(f"Title: {r['title']}\nSnippet: {r['snippet']}")
+        return "\n\n".join(output)[:MAX_RESULT_LENGTH]
     except Exception as e:
         return f"Search error: {str(e)}"
 
 def calculator(expression: str) -> str:
-    """Safe mathematical evaluation"""
     try:
         expression = re.sub(r'[^\d+\-*/().^%,\s]', '', expression)
         if not expression:
@@ -82,17 +88,22 @@ def calculator(expression: str) -> str:
         return f"Calculation error: {str(e)}"
 
 def read_webpage(url: str) -> str:
-    """Robust webpage content extraction"""
     try:
+        if not re.match(r'^https?://', url):
+            return "Invalid URL format"
+            
         headers = {'User-Agent': 'Mozilla/5.0'}
-        response = requests.get(url, timeout=10, headers=headers)
-        soup = BeautifulSoup(response.text, 'html.parser')
+        response = requests.get(url, timeout=15, headers=headers)
+        response.raise_for_status()
         
-        for element in soup(['script', 'style', 'nav', 'footer']):
+        soup = BeautifulSoup(response.text, 'html.parser')
+        for element in soup(['script', 'style', 'nav', 'footer', 'aside']):
             element.decompose()
             
-        text = soup.get_text(separator='\n', strip=True)
-        return re.sub(r'\n{3,}', '\n\n', text)[:MAX_RESULT_LENGTH]
+        main_content = soup.find('main') or soup.find('article') or soup
+        text = main_content.get_text(separator='\n', strip=True)
+        text = re.sub(r'\n{3,}', '\n\n', text)
+        return text[:MAX_RESULT_LENGTH]
     except Exception as e:
         return f"Webpage error: {str(e)}"
 
@@ -102,25 +113,26 @@ TOOLS = {
     "read_webpage": read_webpage
 }
 
-# --- Fixed GAIA Agent ---
+# --- GAIA Agent Class ---
 class GAIA_Agent:
     def __init__(self):
         self.tools = TOOLS
-        self.system_prompt = """You are an advanced GAIA problem solver. Follow these steps:
+        self.system_prompt = """You are an advanced problem solver. Follow these steps:
 1. Analyze the question
-2. Choose the best tool
-3. Process results
-4. Provide final answer
+2. Select the best tool
+3. Execute with proper arguments
+4. Interpret results
+5. Provide final answer
 
 Tools:
-- web_search: For general knowledge
-- calculator: For math
-- read_webpage: For web content
+- web_search(query): For general knowledge
+- calculator(expression): For math
+- read_webpage(url): For web content
 
 Tool format: ```json
-{"tool": "tool_name", "args": {"arg1": value}}```
+{"tool": "tool_name", "args": {"arg": value}}```
 
-Always end with: Final Answer: [answer]"""
+Always conclude with: Final Answer: [answer]"""
 
     def __call__(self, question: str) -> str:
         start_time = time.time()
@@ -150,61 +162,75 @@ Always end with: Final Answer: [answer]"""
             
             return "Maximum steps reached"
         except Exception as e:
-            return f"Error: {str(e)}"
+            return f"Agent error: {str(e)}"
 
     def _build_prompt(self, history: List[str]) -> str:
         return f"<|system|>\n{self.system_prompt}<|end|>\n<|user|>\n" + "\n".join(history) + "<|end|>\n<|assistant|>"
 
     def _call_model(self, prompt: str) -> str:
-        inputs = tokenizer(
-            prompt,
-            return_tensors="pt",
-            truncation=True,
-            max_length=3072,
-            padding=False
-        )
-        
-        # Fixed generation config without problematic parameters
-        outputs = model.generate(
-            inputs.input_ids,
-            max_new_tokens=MAX_TOKENS,
-            temperature=0.3,
-            top_p=0.9,
-            do_sample=True,
-            pad_token_id=tokenizer.pad_token_id,
-            attention_mask=inputs.attention_mask
-        )
-        
-        return tokenizer.decode(outputs[0], skip_special_tokens=True).split("<|assistant|>")[-1].strip()
+        for attempt in range(MAX_ATTEMPTS):
+            try:
+                inputs = tokenizer(
+                    prompt,
+                    return_tensors="pt",
+                    truncation=True,
+                    max_length=3072,
+                    padding=False
+                )
+                
+                outputs = model.generate(
+                    inputs.input_ids,
+                    max_new_tokens=MAX_TOKENS,
+                    temperature=0.3,
+                    top_p=0.9,
+                    do_sample=True,
+                    pad_token_id=tokenizer.pad_token_id,
+                    attention_mask=inputs.attention_mask
+                )
+                
+                return tokenizer.decode(outputs[0], skip_special_tokens=True).split("<|assistant|>")[-1].strip()
+            except Exception as e:
+                if attempt < MAX_ATTEMPTS - 1:
+                    time.sleep(0.5)
+                    continue
+                return f"Model error: {str(e)}"
 
     def _parse_tool_call(self, text: str) -> Optional[Tuple[str, Dict]]:
         try:
             json_match = re.search(r'```json\s*({.+?})\s*```', text, re.DOTALL)
-            if json_match:
-                tool_call = json.loads(json_match.group(1))
-                if "tool" in tool_call and "args" in tool_call:
-                    return tool_call["tool"], tool_call["args"]
+            if not json_match:
+                return None
+                
+            tool_call = json.loads(json_match.group(1))
+            if not isinstance(tool_call, dict):
+                return None
+            if "tool" not in tool_call or "args" not in tool_call:
+                return None
+            if not isinstance(tool_call["args"], dict):
+                return None
+                
+            return tool_call["tool"], tool_call["args"]
         except:
             return None
-        return None
 
     def _use_tool(self, tool_name: str, args: Dict) -> str:
         if tool_name not in self.tools:
             return f"Unknown tool: {tool_name}"
         
         try:
-            # Handle URL extraction for webpage reading
             if tool_name == "read_webpage" and "url" not in args:
-                if "http" in str(args):
-                    url = re.search(r'https?://[^\s]+', str(args)).group()
-                    args = {"url": url}
+                url_match = re.search(r'https?://[^\s]+', str(args))
+                if url_match:
+                    args = {"url": url_match.group()}
+                else:
+                    return "Missing URL argument"
             
             return str(self.tools[tool_name](**args))[:MAX_RESULT_LENGTH]
         except Exception as e:
             return f"Tool error: {str(e)}"
 
-# --- Evaluation Runner ---
-def run_and_submit_all(profile: gr.OAuthProfile | None):
+# --- Evaluation Function ---
+def run_evaluation(profile: gr.OAuthProfile | None):
     if not profile:
         return "Please login first", None
     
@@ -213,8 +239,11 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
     submit_url = f"{DEFAULT_API_URL}/submit"
     
     try:
-        response = requests.get(questions_url, timeout=15)
+        response = requests.get(questions_url, timeout=20)
+        response.raise_for_status()
         questions_data = response.json()
+        if not questions_data:
+            return "No questions available", None
     except Exception as e:
         return f"Failed to get questions: {str(e)}", None
     
@@ -245,28 +274,34 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
     }
     
     try:
-        response = requests.post(submit_url, json=submission, timeout=30)
+        response = requests.post(submit_url, json=submission, timeout=60)
+        response.raise_for_status()
         result = response.json()
-        return f"Submitted! Score: {result.get('score', 'N/A')}", pd.DataFrame(results)
+        status = (f"✅ Submission Successful!\n"
+                 f"Score: {result.get('score', 'N/A')}%\n"
+                 f"Correct: {result.get('correct_count', '?')}/{result.get('total_attempted', '?')}")
+        return status, pd.DataFrame(results)
     except Exception as e:
-        return f"Submission failed: {str(e)}", pd.DataFrame(results)
+        return f"❌ Submission failed: {str(e)}", pd.DataFrame(results)
 
 # --- Gradio Interface ---
-with gr.Blocks(title="Fixed GAIA Agent") as demo:
-    gr.Markdown("## 🛠️ Fixed GAIA Agent")
-    gr.Markdown("Resolved the 'DynamicCache' error with improved configuration")
+with gr.Blocks(title="Fixed GAIA Agent", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🚀 GAIA Agent Evaluation")
     
     with gr.Row():
         gr.LoginButton()
         run_btn = gr.Button("Run Evaluation", variant="primary")
     
-    output_status = gr.Textbox(label="Status")
+    status_output = gr.Textbox(label="Status")
     results_table = gr.DataFrame(label="Results")
     
     run_btn.click(
-        run_and_submit_all,
-        outputs=[output_status, results_table]
+        run_evaluation,
+        outputs=[status_output, results_table]
     )
 
 if __name__ == "__main__":
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860
+    )