Update app.py

app.py

@@ -26,18 +26,47 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 @tool
 def web_search(query: str) -> str:
-    """Search the web using DuckDuckGo. Use for current events, facts, and general knowledge."""
+    """Search the web using DuckDuckGo for current facts, news, specific data, recent information, and verification. Returns top search results."""
     try:
         return DuckDuckGoSearchRun().run(query)
     except Exception as e:
         return f"Search error: {e}"
  
  
+def _wikipedia_api_query(title: str) -> str:
+    """Get plain text extract from English Wikipedia for a page title."""
+    import urllib.parse
+    url = (
+        "https://en.wikipedia.org/w/api.php"
+        "?action=query&format=json&prop=extracts&explaintext=1&titles="
+        + urllib.parse.quote(title)
+    )
+    headers = {"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0; +https://example.com/bot)"}
+    r = requests.get(url, timeout=15, headers=headers)
+    try:
+        r.raise_for_status()
+    except requests.exceptions.HTTPError as ee:
+        print(f"Warning: Wikipedia API HTTP error {ee}")
+        return ""
+    data = r.json()
+    pages = data.get("query", {}).get("pages", {})
+    if not pages:
+        return ""
+    text = next(iter(pages.values())).get("extract", "")
+    return text or ""
+
+
+@tool
+def wikipedia_api_query(title: str) -> str:
+    """Get plain text extract from English Wikipedia for a page title."""
+    return _wikipedia_api_query(title)
+
+
 @tool
 def wikipedia_search(query: str) -> str:
-    """Search Wikipedia for encyclopedic knowledge, historical facts, biographies, science."""
+    """Search Wikipedia for encyclopedic knowledge: historical facts, biographies, dates, definitions, figures, scientific information. Provides structured text summaries."""
     try:
-        wiki = WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=3000)
+        wiki = WikipediaAPIWrapper(top_k_results=3, doc_content_chars_max=3000)
         return wiki.run(query)
     except Exception as e:
         return f"Wikipedia error: {e}"
@@ -46,9 +75,9 @@ def wikipedia_search(query: str) -> str:
 @tool
 def python_repl(code: str) -> str:
     """
-    Execute Python code for math calculations, data processing, logic.
-    Always use print() to show the result.
-    Example: print(2 + 2)
+    Execute Python code for mathematical calculations, data processing, logic operations, and transformations.
+    Always use print() to output results. 
+    Examples: print(2**10), print([1,2,3].count(2)), data=[1,2,3]; print(sum(data)/len(data))
     """
     import io, sys
     old_stdout = sys.stdout
@@ -67,8 +96,8 @@ def python_repl(code: str) -> str:
 @tool
 def calculator(expression: str) -> str:
     """
-    Evaluate a simple math expression.
-    Examples: '2 + 2', '100 * 1.07 ** 5', 'math.sqrt(144)'
+    Evaluate a mathematical expression quickly. Use for simple arithmetic and compound calculations.
+    Examples: '2 + 2', '100 * 1.07 ** 5', 'math.sqrt(144)', '(50 + 30) / 2'
     """
     try:
         return str(eval(expression, {"math": math, "__builtins__": {}}))
@@ -79,8 +108,9 @@ def calculator(expression: str) -> str:
 @tool
 def get_task_file(task_id: str) -> str:
     """
-    Fetch the file attached to a GAIA task by its task_id.
-    Use this when the question mentions an attached file or document.
+    Fetch the file or document attached to a GAIA task by its task_id.
+    Use this when the question mentions an attached file, document, PDF, or any attachment.
+    Returns text content for text/JSON files, or indicates binary file type.
     """
     try:
         import requests as req
@@ -100,23 +130,44 @@ def get_task_file(task_id: str) -> str:
 class AgentState(TypedDict):
     messages: Annotated[list[AnyMessage], add_messages]
 
-SYSTEM_PROMPT = """You are a precise expert AI solving GAIA benchmark questions.
- 
-## Answer Format (CRITICAL)
-- Give ONLY the bare answer: a number, word, name, date, or short phrase.
-- NO explanations, NO punctuation at the end, NO "The answer is...".
-- Correct examples: `42`, `Marie Curie`, `Paris`, `1969`, `blue`
-- For lists: `item1, item2, item3`
- 
-## Strategy
-1. Read carefully — identify exactly what is asked.
-2. Use tools to find and verify the answer.
-3. Double-check calculations with calculator or python_repl.
-4. If the question mentions a file or attachment, use get_task_file.
- 
-## Final Answer
-Always end with:
-FINAL ANSWER: <your answer here>
+SYSTEM_PROMPT = """You are a highly capable GAIA benchmark solver. Your goal is to answer questions accurately and precisely.
+
+## How to Solve Questions - Step by Step
+
+1. **Understand the Question**: Read carefully and identify:
+   - What type of answer is expected (number, text, list, date, etc.)
+   - Key constraints or special formats mentioned
+   - Whether a file or document is attached
+
+2. **Choose Your Approach**:
+   - For arithmetic/math: Use `calculator` or `python_repl`
+   - For current facts/events: Use `web_search`
+   - For historical/encyclopedic knowledge: Use `wikipedia_search`
+   - For attached files: Use `get_task_file`
+   - For complex logic/data processing: Use `python_repl`
+
+3. **Use Tools Effectively**:
+   - Search for key facts and verify information from multiple sources
+   - Extract relevant data from search results
+   - Perform calculations or transformations
+   - Cross-check results when possible
+
+4. **Format Your Final Answer**:
+   - For numbers: just the number (e.g., "42", "3.14", "-5")
+   - For text: exact text without extra punctuation (e.g., "Paris", "Monday")
+   - For lists: comma-separated values (e.g., "item1, item2, item3")
+   - For dates: use the format specified in the question
+   - If completely unsure: respond with just "Unknown"
+
+5. **End Response**:
+   After your reasoning, output a clean final answer on a new line:
+   FINAL ANSWER: <your answer>
+
+## Important Rules
+- Never make up facts - always search or calculate
+- Verify key numbers and spelling with web search
+- If a calculation is involved, always show the work
+- Be concise in your reasoning but thorough in verification
 """
 
 def _tool_to_openai_schema(t) -> dict:
@@ -146,11 +197,16 @@ class BasicAgent:
  
         # Mappa nome → funzione tool per esecuzione
         self.tools_by_name = {t.name: t for t in self.tools_list}
- 
-        # InferenceClient diretto — usa la Serverless Inference API HF
-        self.client = InferenceClient(
-            api_key=os.getenv("HF_TOKEN"),
-        )
+
+        hf_token = os.getenv("HF_TOKEN")
+        if not hf_token:
+            print("WARNING: HF_TOKEN non impostata. L'agente userà fallback locale e risposte molto limitate.")
+            self.client = None
+        else:
+            # InferenceClient diretto — usa la Serverless Inference API HF
+            self.client = InferenceClient(
+                api_key=hf_token,
+            )
          
         # Schema OpenAI dei tool per passarli al client
         self.tools_schema = [_tool_to_openai_schema(t) for t in self.tools_list]
@@ -204,12 +260,12 @@ class BasicAgent:
         hf_messages = self._messages_to_hf_format([sys_msg] + state["messages"])
  
         response = self.client.chat_completion(
-            model="Qwen/Qwen2.5-7B-Instruct",            
+            model="Qwen/Qwen2.5-72B-Instruct",            
             messages=hf_messages,
             tools=self.tools_schema,
             tool_choice="auto",
-            max_tokens=100,
-            temperature=0,
+            max_tokens=1000,
+            temperature=0.1,
         )
  
         choice = response.choices[0].message
@@ -230,25 +286,134 @@ class BasicAgent:
             tool_calls=tool_calls,
         )
         return {"messages": [ai_message]}
- 
+
+    def _local_fallback_answer(self, question: str) -> str:
+        """
+        Minimal fallback when inference client is unavailable.
+        Attempts basic arithmetic only, otherwise returns Unknown.
+        """
+        q = question.lower().strip()
+        
+        # Try simple arithmetic if it looks like a math problem
+        if re.search(r"(?:how\s+many|calculate|compute|what\s+is).*\d+", q):
+            try:
+                # Try to extract and evaluate a simple expression
+                numbers = re.findall(r"\d+\.?\d*", question)
+                if len(numbers) >= 2:
+                    # Don't try to hardcode logic - just return Unknown
+                    pass
+            except Exception:
+                pass
+
+        return "Unknown"
+
     def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        print(f"Agent received question (first 100 chars): {question[:100]}...")
+
+        if self.client is None:
+            print("No HF InferenceClient configured; using local fallback logic")
+            return self._local_fallback_answer(question)
+
+        q_lower = question.lower().strip()
+
+        # Numeric math shortcut: explicit arithmetic detection reduces hallucination.
+        arithmetic = self._extract_arithmetic_expression(question)
+        if arithmetic:
+            calc = calculator(arithmetic)
+            if not calc.startswith("Calculation error"):
+                normalized = self._normalize_answer(calc)
+                print(f"Arithmetic shortcut using calculator: {arithmetic} -> {normalized}")
+                return normalized
+
+        # The main RL loop
+        answer = self._run_agent(question)
+
+        # Retry with explicit “Unknown” handling and chain-of-thought guidance
+        if answer == "Unknown":
+            print("Got Unknown on first pass; retrying with more explicit reasoning request")
+            replay_question = question + "\n\nPlease reason step by step with tool calls and provide FINAL ANSWER only."  # gentle prompt nudge
+            answer = self._run_agent(replay_question)
+
+        print(f"Agent returning answer: '{answer}'")
+        return answer
+
+    def _run_agent(self, question: str) -> str:
         try:
-            result = self.graph.invoke({
-                "messages": [HumanMessage(content=question)]
-            })
-            last_message = result["messages"][-1].content
-            print(f"Agent raw output: {last_message[:200]}...")
- 
-            # Estrai FINAL ANSWER se presente, altrimenti ultima riga
-            match = re.search(r"FINAL ANSWER:\s*(.+?)(?:\n|$)", last_message, re.IGNORECASE)
-            answer = match.group(1).strip() if match else last_message.strip().split("\n")[-1]
- 
-            print(f"Agent returning answer: {answer}")
-            return answer
+            result = self.graph.invoke({"messages": [HumanMessage(content=question)]})
+            last_message = result["messages"][-1]
+            response_text = last_message.content if isinstance(last_message, AIMessage) else str(last_message)
+            print(f"Agent raw output (first 300 chars): {response_text[:300]}...")
+            return self._extract_answer(response_text)
         except Exception as e:
-            print(f"Agent error: {e}")
-            return f"AGENT ERROR: {e}"
+            error_message = str(e)
+            print(f"Agent error during run: {error_message}")
+            if "402" in error_message or "Payment Required" in error_message:
+                fallback = self._local_fallback_answer(question)
+                print(f"Payment required detected; using local fallback answer: {fallback}")
+                return fallback
+            return "Unknown"
+
+    def _extract_answer(self, text: str) -> str:
+        """
+        Extract the final answer from agent output using multiple strategies.
+        """
+        # Strategy 1: Look for explicit "FINAL ANSWER:" marker
+        match = re.search(r"FINAL ANSWER:\s*(.+?)(?:\n|$)", text, re.IGNORECASE)
+        if match:
+            answer = self._normalize_answer(match.group(1).strip())
+            if answer and answer != "Unknown":
+                return answer
+
+        # Strategy 2: Look at the last few lines
+        lines = [line.strip() for line in text.split('\n') if line.strip()]
+        if lines:
+            for candidate in reversed(lines[-4:]):
+                if candidate and not any(phrase in candidate.lower() for phrase in ["i'm not sure", "error", "failed", "final answer"]):
+                    normalized = self._normalize_answer(candidate)
+                    if normalized and normalized != "Unknown":
+                        return normalized
+
+        # Fallback
+        return "Unknown"
+
+    def _normalize_answer(self, answer: str) -> str:
+        """Normalize answer text (strip punctuation, normalize choices etc.)."""
+        answer_clean = answer.strip().strip('"\'').rstrip('.?,;')
+
+        if not answer_clean:
+            return "Unknown"
+
+        # Multiple-choice token: take plain option text if present
+        mc = re.match(r"^([A-D])\s*[:\)]\s*(.+)$", answer_clean, re.IGNORECASE)
+        if mc:
+            return mc.group(2).strip()
+
+        # Numeric decision: enforce numeric format for numeric questions
+        if re.match(r"^-?\d+(\.\d+)?$", answer_clean):
+            return answer_clean
+
+        return answer_clean
+
+    def _extract_arithmetic_expression(self, question: str) -> Optional[str]:
+        """Extract simple arithmetic expression candidate from a question for calculator use."""
+        m = re.search(r"([-+]?\d+(?:\.\d+)?(?:\s*[-+*/]\s*\d+(?:\.\d+)?)+)", question)
+        if not m:
+            return None
+        expr = m.group(1).replace("^", "**")
+        if re.search(r"[a-zA-Z]", expr):
+            return None
+        return expr
+
+    def _detect_question_type(self, question: str) -> str:
+        q = question.lower().strip()
+        if any(tok in q for tok in ["how many", "calculate", "compute", "sum", "difference", "times", "per cent", "%"]):
+            return "numeric"
+        if any(tok in q for tok in ["when", "year", "date", "born", "died"]):
+            return "date"
+        if any(tok in q for tok in ["which of the following", "option", "choose", "select"]):
+            return "multiple_choice"
+        return "factual"
+
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -371,42 +536,44 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
         return status_message, results_df
 
 
-# --- Build Gradio Interface using Blocks ---
-with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner")
-    gr.Markdown(
-        """
-        **Instructions:**
+def build_gradio_ui():
+    with gr.Blocks() as demo:
+        gr.Markdown("# Basic Agent Evaluation Runner")
+        gr.Markdown(
+            """
+            **Instructions:**
 
-        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
-        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
-        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
+            1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
+            2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
+            3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
 
-        ---
-        **Disclaimers:**
-        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
-        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
-        """
-    )
+            ---
+            **Disclaimers:**
+            Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
+            This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
+            """
+        )
 
-    gr.LoginButton()
+        gr.LoginButton()
 
-    run_button = gr.Button("Run Evaluation & Submit All Answers")
+        run_button = gr.Button("Run Evaluation & Submit All Answers")
 
-    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
-    # Removed max_rows=10 from DataFrame constructor
-    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
+        status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
+        results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
 
-    run_button.click(
-        fn=run_and_submit_all,
-        outputs=[status_output, results_table]
-    )
+        run_button.click(
+            fn=run_and_submit_all,
+            outputs=[status_output, results_table]
+        )
+
+    return demo
 
 if __name__ == "__main__":
     print("\n" + "-"*30 + " App Starting " + "-"*30)
+
     # Check for SPACE_HOST and SPACE_ID at startup for information
     space_host_startup = os.getenv("SPACE_HOST")
-    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
+    space_id_startup = os.getenv("SPACE_ID")
 
     if space_host_startup:
         print(f"✅ SPACE_HOST found: {space_host_startup}")
@@ -414,7 +581,7 @@ if __name__ == "__main__":
     else:
         print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
 
-    if space_id_startup: # Print repo URLs if SPACE_ID is found
+    if space_id_startup:
         print(f"✅ SPACE_ID found: {space_id_startup}")
         print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
         print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
@@ -424,4 +591,5 @@ if __name__ == "__main__":
     print("-"*(60 + len(" App Starting ")) + "\n")
 
     print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+    demo = build_gradio_ui()
+    demo.launch(debug=True, share=False)