app.py

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

from smolagents import (
    CodeAgent,
    DuckDuckGoSearchTool,
    InferenceClientModel,
    tool,
    LiteLLMModel,
)

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

@tool
def download_file_from_task(task_id: str) -> str:
    """
    Downloads a file associated with a GAIA task and returns its content as text.
    Use this when a question references an attached file.
    Args:
        task_id: The task ID string of the GAIA question.
    """
    url = f"{DEFAULT_API_URL}/files/{task_id}"
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        try:
            text = response.content.decode("utf-8")
            if len(text) > 8000:
                text = text[:8000] + "\n[... truncated ...]"
            return text
        except UnicodeDecodeError:
            return f"Binary file. Size: {len(response.content)} bytes."
    except Exception as e:
        return f"Error downloading file: {e}"

@tool
def python_calculator(code: str) -> str:
    """
    Executes a Python expression and returns the result.
    Use for arithmetic, unit conversions, date calculations.
    Args:
        code: A Python expression to evaluate (e.g. '2 ** 10')
    """
    import math, datetime
    allowed = {"__builtins__": {}, "math": math, "datetime": datetime,
               "abs": abs, "round": round, "int": int, "float": float,
               "str": str, "len": len, "sum": sum, "min": min, "max": max,
               "sorted": sorted, "range": range, "list": list, "dict": dict,
               "set": set, "zip": zip, "enumerate": enumerate}
    try:
        return str(eval(code, allowed))
    except Exception as e:
        return f"Error: {e}"

SYSTEM_PROMPT = """You are an expert research assistant answering GAIA benchmark questions.
CRITICAL RULES:
1. Your final answer must be EXACT and CONCISE. No explanations, no sentences.
2. If the answer is a number, return ONLY the number.
3. If the answer is a name, return ONLY the name.
4. If the answer is a list, return items separated by commas.
5. If a question references a file, use download_file_from_task with the task_id.
6. Always search the web for factual questions.
7. Never include FINAL ANSWER in your response.
8. Match the exact format requested in the question.
9. If you cannot find the answer, return your best single-word or single-number guess. Never return a long explanation.
"""

def build_agent():
    model = LiteLLMModel(          # ← changed from InferenceClientModel
        model_id="groq/llama-3.1-8b-instant",
        api_key=os.getenv("HF"),   # ← changed from token= to api_key=
        max_tokens=1024,
        temperature=0.1,
    )
    agent = CodeAgent(
        model=model,
        tools=[DuckDuckGoSearchTool(), download_file_from_task, python_calculator],
        max_steps=8,
        verbosity_level=1,
        additional_authorized_imports=["math", "datetime", "re", "json", "csv", "io"],
    )
    agent.prompt_templates["system_prompt"] = SYSTEM_PROMPT + "\n\n" + agent.prompt_templates["system_prompt"]
    return agent

def run_and_submit_all(profile: gr.OAuthProfile | None):
    space_id = os.getenv("SPACE_ID")
    if not profile:
        return "Please login to Hugging Face first.", None
    username = profile.username
    print(f"User logged in: {username}")
    try:
        agent = build_agent()
    except Exception as e:
        return f"Error initializing agent: {e}", None
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
    try:
        response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15)
        response.raise_for_status()
        questions_data = response.json()
    except Exception as e:
        return f"Error fetching questions: {e}", None

    results_log = []
    answers_payload = []

    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:
            raw = str(agent.run(f"[task_id: {task_id}]\n\n{question_text}")).strip()
            lines = [l.strip() for l in raw.split('\n') if l.strip()]
            submitted_answer = lines[0] if lines else raw
            for prefix in ["The answer is", "Answer:", "Result:", "Final answer:", "FINAL ANSWER:"]:
                if submitted_answer.lower().startswith(prefix.lower()):
                    submitted_answer = submitted_answer[len(prefix):].strip().strip(":")
        except Exception as e:
            submitted_answer = f"AGENT ERROR: {e}"
            print(f"Error on task {task_id}: {e}")

        answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
        results_log.append({"Task ID": task_id, "Question": question_text[:120], "Submitted Answer": submitted_answer})
        print(f"Task {task_id}: {submitted_answer[:80]}")

    if not answers_payload:
        return "Agent produced no answers.", pd.DataFrame(results_log)

    try:
        response = requests.post(f"{DEFAULT_API_URL}/submit",
            json={"username": username.strip(), "agent_code": agent_code, "answers": answers_payload},
            timeout=120)
        response.raise_for_status()
        r = response.json()
        return (f"Submission Successful!\nUser: {r.get('username')}\n"
                f"Score: {r.get('score')}% ({r.get('correct_count')}/{r.get('total_attempted')} correct)\n"
                f"Message: {r.get('message')}"), pd.DataFrame(results_log)
    except Exception as e:
        return f"Submission Failed: {e}", pd.DataFrame(results_log)

with gr.Blocks() as demo:
    gr.Markdown("# GAIA Agent - Unit 4 Final Assignment")
    gr.Markdown("""
    1. Log in with your Hugging Face account.
    2. Click Run Evaluation to start the agent on all 20 GAIA questions.
    3. Target: >= 30% to earn the certificate.
    """)
    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers", variant="primary")
    status_output = gr.Textbox(label="Submission Result", lines=6, 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)
    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID")
    if space_host_startup:
        print(f"SPACE_HOST: {space_host_startup}")
    if space_id_startup:
        print(f"SPACE_ID: {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")
    print("-"*74 + "\n")
    print("Launching Gradio Interface...")
    demo.launch(debug=True, share=False)