app.py

#!/usr/bin/env python3

import os
import re
import ast
import operator
import datetime
import pytz
import json
import requests
import yaml
import pandas as pd
import gradio as gr

from smolagents import CodeAgent, HfApiModel, tool

# -------------------------
# Minimal tools
# -------------------------
_allowed_ops = {
    ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
    ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg,
    ast.Mod: operator.mod,
}

def _eval_node(node):
    if isinstance(node, ast.Constant):
        return node.value
    if isinstance(node, ast.Num):
        return node.n
    if isinstance(node, ast.UnaryOp) and type(node.op) in _allowed_ops:
        return _allowed_ops[type(node.op)](_eval_node(node.operand))
    if isinstance(node, ast.BinOp) and type(node.op) in _allowed_ops:
        return _allowed_ops[type(node.op)](_eval_node(node.left), _eval_node(node.right))
    raise ValueError("Unsupported expression")

def safe_calc(expr: str):
    tree = ast.parse(expr, mode='eval')
    return _eval_node(tree.body)

@tool
def calculator(expr: str) -> str:
    """
    Safely evaluate a mathematical expression.

    Args:
        expr (str): A string containing a math expression like "2 + 2 * 3".

    Returns:
        str: JSON string with {"expression": expr, "result": value} or {"error": "..."} on failure.
    """
    try:
        val = safe_calc(expr)
        return json.dumps({"expression": expr, "result": float(val)})
    except Exception as e:
        return json.dumps({"error": f"Calc error: {e}"})


@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """
    Get the current local time in a specified timezone.

    Args:
        timezone (str): A valid timezone string (e.g., "Europe/Paris").

    Returns:
        str: JSON string with {"timezone": timezone, "local_time": "..."} or {"error": "..."} on failure.
    """
    try:
        tz = pytz.timezone(timezone)
        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
        return json.dumps({"timezone": timezone, "local_time": local_time})
    except Exception as e:
        return json.dumps({"error": f"Timezone error: {e}"})


# -------------------------
# Load prompts.yaml if exists
# -------------------------
prompt_templates = None
try:
    with open("prompts.yaml", "r") as fh:
        prompt_templates = yaml.safe_load(fh)
except Exception:
    prompt_templates = None

# -------------------------
# HfApiModel + CodeAgent minimal
# -------------------------
model = HfApiModel(
    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
    max_tokens=1024,
    temperature=0.0
)

code_agent = CodeAgent(
    model=model,
    tools=[calculator, get_current_time_in_timezone],
    max_steps=6,
    verbosity_level=0,
    prompt_templates=prompt_templates
)

# -------------------------
# GAIA Agent wrapper
# -------------------------
class GaiaAgentMinimal:
    def __init__(self, code_agent):
        self.code_agent = code_agent

    def _is_calc(self, q: str) -> bool:
        return bool(re.search(r'[\d]', q)) and any(op in q for op in ['+', '-', '*', '/', '%', '^'])

    def _is_time(self, q: str) -> bool:
        ql = q.lower()
        return "time" in ql or "heure" in ql or "quelle heure" in ql or "what time" in ql

    def run(self, question: str) -> str:
        try:
            q = question.strip()

            # Calculator queries
            if self._is_calc(q):
                m = re.search(r'([0-9\.\s\+\-\*\/\^\%\(\)]+)', q)
                expr = m.group(1) if m else q
                return calculator(expr)

            # Time queries
            if self._is_time(q):
                if "paris" in q.lower() or "france" in q.lower():
                    tz = "Europe/Paris"
                else:
                    tz = "UTC"
                return get_current_time_in_timezone(tz)

            # fallback LLM
            resp = self.code_agent.run(q)
            if isinstance(resp, dict):
                for key in ("final_answer", "answer", "result", "output"):
                    if key in resp:
                        return str(resp[key])
                return json.dumps(resp)
            return str(resp)
        except Exception as e:
            return json.dumps({"error": f"Agent internal error: {e}"})

# instantiate GAIA agent
gaia_agent = GaiaAgentMinimal(code_agent)

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

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

    if profile:
        username = f"{profile.username}"
    else:
        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"

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

    # Fetch questions
    try:
        response = requests.get(questions_url, timeout=15)
        response.raise_for_status()
        questions_data = response.json()
        if not questions_data:
            return "Fetched questions list is empty or invalid format.", None
    except Exception as e:
        return f"Error fetching questions: {e}", None

    # Run agent
    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:
            submitted_answer = gaia_agent.run(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}

    # Submit
    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:
        results_df = pd.DataFrame(results_log)
        return f"Submission failed: {e}", results_df

# -------------------------
# Gradio UI
# -------------------------
with gr.Blocks() as demo:
    gr.Markdown("# Minimal GAIA Agent Runner")
    gr.Markdown(
        "Log in to Hugging Face, click 'Run Evaluation & Submit All Answers' to fetch questions, run the agent, and submit answers."
    )
    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__":
    demo.launch(debug=True, share=False)