app.py

#!/usr/bin/env python3

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

from smolagents import CodeAgent, TransformersModel, 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: A string containing a math expression like "2 + 2 * 3".

    Returns:
        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: A valid timezone string (e.g., "Europe/Paris").

    Returns:
        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:
    import yaml
    with open("prompts.yaml", "r") as fh:
        prompt_templates = yaml.safe_load(fh)
except Exception:
    prompt_templates = None


# -------------------------
# TransformersModel + CodeAgent minimal
# -------------------------
model = TransformersModel(model_id="HuggingFaceTB/SmolLM-135M-Instruct")

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 True only for explicit calculation requests, not 'how many' type questions."""
        if not q:
            return False
        ql = q.lower()
        calc_triggers = ["calculate", "compute", "evaluate", "what is", "what's"]
        if any(tr in ql for tr in calc_triggers) and re.search(r"\d", ql):
            return True
        if re.search(r"\d\s*[\+\-\*\/\%\^]\s*\d", q):
            return True
        return False

    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 _call_llm(self, q: str) -> str:
        """Wrapper to call LLM and return result or proper error if token/API missing."""
        try:
            resp = self.code_agent.run(q)
        except Exception as e:
            msg = str(e)
            if "api_key" in msg.lower() or "auth" in msg.lower():
                return json.dumps({"error": "LLM error: missing HF API token. Set HF_API_TOKEN secret or login with HF."})
            return json.dumps({"error": f"LLM runtime error: {msg}"})
        if resp is None:
            return json.dumps({"error": "LLM returned no output"})
        if isinstance(resp, dict):
            for key in ("final_answer", "answer", "result", "output"):
                if key in resp:
                    return str(resp[key])
            return json.dumps(resp)
        if isinstance(resp, (int, float)):
            return str(resp)
        s = str(resp).strip()
        if s == "":
            return json.dumps({"error": "LLM returned empty string"})
        return s

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

            # 1) Calculator
            if self._is_calc(q):
                m = re.search(r'([0-9\.\s\+\-\*\/\^\%\(\)]+)', q)
                expr = m.group(1).strip() if m else ""
                if not expr or not re.search(r'[\+\-\*\/\%\^]', expr):
                    return self._call_llm(q)
                expr = expr.replace('^', '**').replace('\n', ' ').strip()
                if not re.fullmatch(r"[0-9\.\s\+\-\*\/\%\(\)\*]+", expr):
                    return json.dumps({"error": "Expression contains invalid characters or is not a simple math expression", "original": expr})
                return calculator(expr)

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

            # 3) fallback LLM
            return self._call_llm(q)

        except Exception as e:
            return json.dumps({"error": f"Agent internal error: {str(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)