app.py

import os
import gradio as gr
import requests
import pandas as pd
import time

# --- Built-in Tool: Wikipedia Search using standard 'requests' ---
def search_wikipedia(query: str) -> str:
    """A simple search tool using the Wikipedia API without extra libraries."""
    print(f"  -> Tool Executing Search for: {query}")
    url = "https://en.wikipedia.org/w/api.php"
    params = {
        "action": "query",
        "format": "json",
        "list": "search",
        "srsearch": query,
        "utf8": 1,
        "srlimit": 3  # Return top 3 snippets
    }
    try:
        response = requests.get(url, params=params, timeout=5)
        data = response.json()
        snippets = [item['snippet'].replace('<span class="searchmatch">', '').replace('</span>', '') for item in data['query']['search']]
        if not snippets:
            return "Observation: No results found."
        return "Observation: " + " | ".join(snippets)
    except Exception as e:
        return f"Observation: Search error - {e}"

# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

# --- Vanilla Gemini ReAct Agent Definition ---
class GeminiReActAgent:
    def __init__(self):
        self.api_key = os.getenv("GEMINI_API_KEY")
        if not self.api_key:
            raise ValueError("GEMINI_API_KEY not set")
        
        # Direct REST API endpoint for Gemini 2.5 Flash
        self.url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent?key={self.api_key}"
        print("Vanilla ReAct Gemini Agent initialized with rate-limit handling.")

    def call_gemini(self, history) -> str:
        """Helper to make direct HTTP requests to the Gemini API with strict time pacing."""
        payload = {
            "contents": history,
            "generationConfig": {
                "temperature": 0.0,
                # Tell Gemini to stop generating when it's time for an observation
                "stopSequences": ["Observation:"] 
            }
        }
        
        max_retries = 3
        for attempt in range(max_retries):
            try:
                # ENFORCED TIME PACING: Wait 15 seconds to stay under the 5 Requests Per Minute limit.
                print("  -> Pacing API: Sleeping for 15 seconds to respect 5 RPM limit...")
                time.sleep(5) 
                
                response = requests.post(self.url, json=payload, timeout=20)
                
                # Check if we hit a 429
                if response.status_code == 429:
                    print(f"  -> WARNING: Rate limit hit (429). Entering 60-second cooldown... (Attempt {attempt + 1}/{max_retries})")
                    time.sleep(60)
                    continue 
                
                response.raise_for_status()
                data = response.json()
                return data["candidates"][0]["content"]["parts"][0]["text"].strip()
                
            except requests.exceptions.RequestException as e:
                print(f"Gemini API Network Error: {e}")
                if hasattr(e, 'response') and e.response is not None:
                    print(e.response.text)
                time.sleep(10)
            except Exception as e:
                print(f"Unexpected Error parsing Gemini response: {e}")
                return "Error"
                
        return "Error"

    def __call__(self, question: str) -> str:
        print(f"\nAgent processing question: {question[:50]}...")
        
        system_instruction = """You are an expert assistant for the GAIA benchmark. 
You must provide a short, factual, direct answer. No explanations.
You have access to a Wikipedia search tool, but you also have vast internal knowledge.

CRITICAL RULES:
1. If the question asks you to categorize, sort, or use logic, use your internal knowledge immediately to output the Final Answer. Do not use tools.
2. If the question mentions an attached image, video, audio, or Excel/CSV file, do your best to answer based purely on the text provided or by searching the internet. Do NOT say "I cannot analyze this."
3. You must output EXACTLY one of the two formats below. 

FORMAT 1 (To use the search tool):
Thought: <what you need to search>
Action: Search
Action Input: <short, specific search query>

FORMAT 2 (To give the final answer):
Thought: <your reasoning>
Final Answer: <the short, direct answer>"""

        history = [
            {"role": "user", "parts": [{"text": system_instruction + "\n\nQuestion: " + question}]}
        ]

        for iteration in range(5):
            reply = self.call_gemini(history)
            
            if reply == "Error":
                return "0"

            history.append({"role": "model", "parts": [{"text": reply}]})

            if "Final Answer:" in reply:
                answer = reply.split("Final Answer:")[-1].strip()
                return answer if answer else "0"
            
            elif "Action: Search" in reply and "Action Input:" in reply:
                query_lines = [line for line in reply.split('\n') if "Action Input:" in line]
                if query_lines:
                    query = query_lines[0].split("Action Input:")[-1].strip()
                    observation = search_wikipedia(query)
                    history.append({"role": "user", "parts": [{"text": observation}]})
                    continue
            
            else:
                history.append({
                    "role": "user", 
                    "parts": [{"text": "Format error. Please use 'Action: Search' or 'Final Answer:'"}]
                })
        
        return "0"

def run_and_submit_all(profile: gr.OAuthProfile | None):
    space_id = os.getenv("SPACE_ID") 

    if profile:
        username = f"{profile.username}"
        print(f"User logged in: {username}")
    else:
        print("User not logged in.")
        return "Please Login to Hugging Face with the button.", None

    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    submit_url = f"{api_url}/submit"

    try:
        agent = GeminiReActAgent()
    except Exception as e:
        print(f"Error instantiating agent: {e}")
        return f"Error initializing agent: {e}", None

    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"

    print(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=60)
        response.raise_for_status()
        questions_data = response.json()
        if not questions_data:
            return "Fetched questions list is empty or invalid format.", None
        print(f"Fetched {len(questions_data)} questions.")
    except Exception as e:
        return f"An unexpected error occurred fetching questions: {e}", None

    results_log = []
    answers_payload = []
    print(f"Running agent on {len(questions_data)} questions...")
    
    for item in questions_data:
        task_id = item.get("task_id")
        question_text = item.get("question")
        if not task_id or question_text is None:
            continue
        try:
            submitted_answer = agent(question_text)
            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:
            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})

    if not answers_payload:
        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)

    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
    print(f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'...")

    try:
        response = requests.post(submit_url, json=submission_data, timeout=60)
        response.raise_for_status()
        result_data = response.json()
        final_status = (
            f"Submission Successful!\n"
            f"User: {result_data.get('username')}\n"
            f"Overall Score: {result_data.get('score', 'N/A')}% "
            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
            f"Message: {result_data.get('message', 'No message received.')}"
        )
        results_df = pd.DataFrame(results_log)
        return final_status, results_df
    except Exception as e:
        status_message = f"Submission Failed: {e}"
        results_df = pd.DataFrame(results_log)
        return status_message, results_df

# --- Build Gradio Interface using Blocks ---
with gr.Blocks() as demo:
    gr.Markdown("# Gemini Rate-Limited Agent Evaluation")
    gr.Markdown(
        """
        **Disclaimers:**
        Due to Google's strict Free Tier rate limit (5 requests per minute), the agent forces a 15-second delay before every API call. 
        **This process will take roughly 15 to 25 minutes to complete 20 questions.** Please click 'Submit' and do not refresh the page.
        """
    )
    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers")
    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]
    )

if __name__ == "__main__":
    print("\n" + "-"*30 + " App Starting " + "-"*30)
    demo.launch(debug=True, share=False)