Update app.py

app.py

@@ -1,23 +1,195 @@
 import os
+import re
 import gradio as gr
 import requests
-import inspect
 import pandas as pd
+from bs4 import BeautifulSoup
+
+try:
+    from dotenv import load_dotenv
+    load_dotenv()
+except ImportError:
+    pass  # .env not loaded; use os.getenv (e.g. HF Secrets)
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+REACT_MAX_STEPS = 10
+# Use a model available on HF serverless inference (410 = model retired). Alternatives: google/gemma-2-9b-it, mistralai/Mistral-7B-Instruct-v0.2
+LLM_MODEL = "google/gemma-2-2b-it"
+
+
+# --- Tools (DuckDuckGo search, web page view, code agent) ---
+def tool_web_search(query: str, max_results: int = 5) -> str:
+    """Search the web using DuckDuckGo. Input: search query string."""
+    try:
+        from duckduckgo_search import DDGS
+        results = list(DDGS().text(query, max_results=max_results))
+        if not results:
+            return "No search results found."
+        out = []
+        for i, r in enumerate(results, 1):
+            out.append(f"{i}. {r.get('title', '')}\n   URL: {r.get('href', '')}\n   {r.get('body', '')}")
+        return "\n\n".join(out)
+    except Exception as e:
+        return f"Web search error: {e}"
+
+
+def tool_web_page_view(url: str) -> str:
+    """View the main text content of a web page. Input: full URL string."""
+    try:
+        headers = {"User-Agent": "Mozilla/5.0 (compatible; ReActAgent/1.0)"}
+        r = requests.get(url, timeout=15, headers=headers)
+        r.raise_for_status()
+        soup = BeautifulSoup(r.text, "html.parser")
+        for tag in soup(["script", "style", "nav", "footer", "header"]):
+            tag.decompose()
+        text = soup.get_text(separator="\n", strip=True)
+        return text[:8000] if len(text) > 8000 else text or "No text content found."
+    except Exception as e:
+        return f"Web page view error: {e}"
+
+
+def tool_code_agent(code: str) -> str:
+    """Run Python code to compute an answer. Input: a single Python expression or block (e.g. print(2+2)). No file or network access."""
+    import builtins
+    import io
+    import sys
+    safe_builtins = {
+        "abs": builtins.abs, "all": builtins.all, "any": builtins.any,
+        "bin": builtins.bin, "bool": builtins.bool, "chr": builtins.chr,
+        "dict": builtins.dict, "divmod": builtins.divmod, "enumerate": builtins.enumerate,
+        "filter": builtins.filter, "float": builtins.float, "format": builtins.format,
+        "hash": builtins.hash, "int": builtins.int, "len": builtins.len,
+        "list": builtins.list, "map": builtins.map, "max": builtins.max,
+        "min": builtins.min, "next": builtins.next, "pow": builtins.pow,
+        "print": builtins.print, "range": builtins.range, "repr": builtins.repr,
+        "reversed": builtins.reversed, "round": builtins.round, "set": builtins.set,
+        "sorted": builtins.sorted, "str": builtins.str, "sum": builtins.sum,
+        "tuple": builtins.tuple, "zip": builtins.zip,
+    }
+    try:
+        code = code.strip()
+        if not code.startswith("print(") and "print(" not in code:
+            code = f"print({code})"
+        buf = io.StringIO()
+        old_stdout = sys.stdout
+        sys.stdout = buf
+        try:
+            exec(code, {"__builtins__": safe_builtins, "print": builtins.print}, {})
+        finally:
+            sys.stdout = old_stdout
+        return buf.getvalue().strip() or "Code ran (no printed output)."
+    except Exception as e:
+        return f"Code error: {e}"
+
+
+TOOLS = {
+    "web_search": tool_web_search,
+    "web_page_view": tool_web_page_view,
+    "code_agent": tool_code_agent,
+}
+
+TOOL_DESCRIPTIONS = """Available tools:
+- web_search: search the web with DuckDuckGo. Input: search query (string).
+- web_page_view: get main text from a web page. Input: URL (string).
+- code_agent: run Python code (math, string ops). Input: code (string)."""
+
+
+# --- ReAct Agent: Plan -> Act -> Observe -> Reflect ---
+class ReActAgent:
+    def __init__(self, token: str | None = None, model: str = LLM_MODEL, max_steps: int = REACT_MAX_STEPS):
+        self.token = (token or HF_TOKEN or "").strip()
+        self.model = model
+        self.max_steps = max_steps
+        print("ReActAgent initialized (plan -> act -> observe -> reflect).")
+
+    def _llm(self, messages: list[dict]) -> str:
+        if not self.token:
+            return "Error: HF_TOKEN not set. Add your token in .env to use the LLM."
+        url = f"https://api-inference.huggingface.co/models/{self.model}"
+        headers = {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
+        payload = {"inputs": self._messages_to_prompt(messages), "parameters": {"max_new_tokens": 512, "return_full_text": False}}
+        try:
+            r = requests.post(url, json=payload, headers=headers, timeout=60)
+            r.raise_for_status()
+            data = r.json()
+            if isinstance(data, list) and len(data) > 0:
+                return (data[0].get("generated_text") or "").strip()
+            if isinstance(data, dict) and "generated_text" in data:
+                return (data["generated_text"] or "").strip()
+            return ""
+        except Exception as e:
+            return f"LLM error: {e}"
+
+    def _messages_to_prompt(self, messages: list[dict]) -> str:
+        out = []
+        for m in messages:
+            role = m.get("role", "user")
+            content = m.get("content", "")
+            if role == "system":
+                out.append(f"System: {content}")
+            elif role == "user":
+                out.append(f"User: {content}")
+            else:
+                out.append(f"Assistant: {content}")
+        out.append("Assistant:")
+        return "\n\n".join(out)
+
+    def _parse_action(self, text: str) -> tuple[str | None, str | None, str | None]:
+        """Returns (thought, action, action_input) or (None, None, final_answer)."""
+        text = text.strip()
+        final_match = re.search(r"Final Answer\s*:\s*(.+?)(?=\n\n|\Z)", text, re.DOTALL | re.IGNORECASE)
+        if final_match:
+            return None, None, final_match.group(1).strip()
+        action_match = re.search(r"Action\s*:\s*(\w+)", text, re.IGNORECASE)
+        input_match = re.search(r"Action Input\s*:\s*(.+?)(?=\n\n|\nThought:|\Z)", text, re.DOTALL | re.IGNORECASE)
+        thought = None
+        thought_match = re.search(r"Thought\s*:\s*(.+?)(?=\nAction:|\Z)", text, re.DOTALL | re.IGNORECASE)
+        if thought_match:
+            thought = thought_match.group(1).strip()
+        action = action_match.group(1).strip() if action_match else None
+        action_input = input_match.group(1).strip() if input_match else None
+        if action_input:
+            action_input = action_input.strip().strip('"\'')
+        return thought, action, action_input
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class BasicAgent:
-    def __init__(self):
-        print("BasicAgent initialized.")
     def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
+        print(f"ReAct agent received question (first 50 chars): {question[:50]}...")
+        if not self.token:
+            return "HF_TOKEN not set. Add your Hugging Face token in .env to run the ReAct agent."
+        system = (
+            "You are a ReAct agent. For each turn you must either:\n"
+            "1. Output: Thought: <reasoning> then Action: <tool_name> then Action Input: <input>\n"
+            "2. Or when you have the answer: Final Answer: <your answer>\n\n"
+            + TOOL_DESCRIPTIONS
+        )
+        messages = [
+            {"role": "system", "content": system},
+            {"role": "user", "content": f"Question: {question}\n\nFirst, plan which tool(s) to use, then take action, then observe, then reflect. Give your final answer when done."},
+        ]
+        for step in range(self.max_steps):
+            response = self._llm(messages)
+            thought, action, action_input = self._parse_action(response)
+            if thought is None and action is None and action_input is not None:
+                return action_input  # Final Answer
+            if not action or action not in TOOLS:
+                messages.append({"role": "assistant", "content": response})
+                messages.append({"role": "user", "content": "You must use one of the tools (Action: tool_name, Action Input: input) or give Final Answer: your answer. Try again."})
+                continue
+            try:
+                observation = TOOLS[action](action_input)
+            except Exception as e:
+                observation = f"Tool error: {e}"
+            observation = (observation[:3000] + "...") if len(observation) > 3000 else observation
+            messages.append({"role": "assistant", "content": response})
+            messages.append({"role": "user", "content": f"Observation: {observation}\n\nReflect: does this answer the question? If yes, reply with Final Answer: <answer>. If not, use another tool (Thought / Action / Action Input)."})
+        last_assistant = next((m["content"] for m in reversed(messages) if m.get("role") == "assistant"), "")
+        final = self._parse_action(last_assistant)
+        if final[2] and final[0] is None and final[1] is None:
+            return final[2]
+        return last_assistant[:500] if last_assistant else "ReAct agent reached max steps without a final answer."
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -40,7 +212,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = BasicAgent()
+        agent = ReActAgent(token=os.getenv("HF_TOKEN"), max_steps=REACT_MAX_STEPS)
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -142,19 +314,12 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
 # --- Build Gradio Interface using Blocks ---
 with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner")
+    gr.Markdown("# ReAct 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.
-
-        ---
-        **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.
+        **Multi-step ReAct agent:** Plan → Act (tools) → Observe → Reflect. The agent has access to:
+        **DuckDuckGo search**, **web page view**, and **code agent** (safe Python). Set `HF_TOKEN` in Secrets (or .env) to enable the LLM.
+        1. Log in with the button below. 2. Click 'Run Evaluation & Submit All Answers'. Submission can take a while while the agent runs on all questions.
         """
     )