Blame for app.py

import os
import re
import subprocess
import time

import gradio as gr
import pandas as pd
import requests
from smolagents import (
    CodeAgent,
    DuckDuckGoSearchTool,
    LiteLLMModel,
    VisitWebpageTool,
    tool,
)

DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
MODEL_ID = os.getenv("MODEL_ID", "gemini/gemini-2.0-flash")  # modele du LLM (changeable via les variables du Space)
PAUSE_BETWEEN_QUESTIONS = 8  # secondes, pour eviter les limites de requetes gratuites
FILES_DIR = "/tmp/task_files"
os.makedirs(FILES_DIR, exist_ok=True)


# ----------------------------------------------------------------------------
# Outils personnalises
# ----------------------------------------------------------------------------
@tool
def read_excel_file(path: str) -> str:
    """Reads an Excel file (.xlsx) and returns all its sheets as text.

    Args:
        path: Local path of the Excel file.
    """
    sheets = pd.read_excel(path, sheet_name=None)
    out = []
    for name, df in sheets.items():
        out.append(f"--- Sheet: {name} ---\n{df.to_string()}")
    return "\n\n".join(out)


@tool
def read_text_file(path: str) -> str:
    """Reads a text or code file (.py, .txt, .csv...) and returns its content.

    Args:
        path: Local path of the file.
    """
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        return f.read()


@tool
def run_python_file(path: str) -> str:
    """Runs a Python file and returns what it prints (stdout and stderr).

    Args:
        path: Local path of the Python file to run.
    """
    try:
        result = subprocess.run(
            ["python", path], capture_output=True, text=True, timeout=60
        )
        return (result.stdout + result.stderr).strip()
    except Exception as e:
        return f"Execution error: {e}"


@tool
def transcribe_audio(path: str) -> str:
    """Transcribes an audio file (mp3, wav...) to text using Whisper on Groq.

    Args:
        path: Local path of the audio file.
    """
    api_key = os.getenv("GROQ_API_KEY")
    if not api_key:
        return "Error: GROQ_API_KEY is not set in the Space secrets."
    with open(path, "rb") as f:
        response = requests.post(
            "https://api.groq.com/openai/v1/audio/transcriptions",
            headers={"Authorization": f"Bearer {api_key}"},
            files={"file": (os.path.basename(path), f)},
            data={"model": "whisper-large-v3"},
            timeout=120,
        )
    if response.status_code != 200:
        return f"Transcription error {response.status_code}: {response.text[:300]}"
    return response.json().get("text", "")


# ----------------------------------------------------------------------------
# Agent
# ----------------------------------------------------------------------------
SYSTEM_RULES = """You are answering questions from the GAIA benchmark. Be precise.
Rules for your final answer:
- Reply with ONLY the final answer. No explanation, no sentence, no 'FINAL ANSWER'.
- Numbers: write digits only, without units or thousands separators, unless the question asks otherwise.
- Lists: comma-separated, in the order requested (alphabetical or ascending if asked).
- Strings: no articles, no abbreviations unless the question asks for them.
- Use web search and visit_webpage to verify facts instead of guessing.
- If a file path is given, use the provided file tools to read it.
"""


class GaiaAgent:
    def __init__(self):
        model = LiteLLMModel(
            model_id=MODEL_ID,
            api_key=os.getenv("GEMINI_API_KEY"),
        )
        self.agent = CodeAgent(
            tools=[
                DuckDuckGoSearchTool(),
                VisitWebpageTool(),
                read_excel_file,
                read_text_file,
                run_python_file,
                transcribe_audio,
            ],
            model=model,
            max_steps=12,
            additional_authorized_imports=[
                "pandas", "numpy", "re", "json", "math", "datetime",
                "itertools", "collections", "statistics", "unicodedata",
                "requests", "bs4",
            ],
            verbosity_level=0,
        )

    def __call__(self, question: str, file_path: str | None = None) -> str:
        prompt = f"{SYSTEM_RULES}\nQuestion: {question}"
        if file_path:
            prompt += f"\n\nAn attached file is available locally at: {file_path}"
        answer = str(self.agent.run(prompt)).strip()
        answer = re.sub(r"^\s*final answer\s*:\s*", "", answer, flags=re.IGNORECASE)
        return answer.strip().strip('"').strip()


def download_task_file(api_url: str, task_id: str, file_name: str) -> str | None:
    """Telecharge le fichier associe a une question (si elle en a un)."""
    if not file_name:
        return None
    try:
        r = requests.get(f"{api_url}/files/{task_id}", timeout=60)
        r.raise_for_status()
        local_path = os.path.join(FILES_DIR, file_name)
        with open(local_path, "wb") as f:
            f.write(r.content)
        return local_path
    except Exception as e:
        print(f"Could not download file for {task_id}: {e}")
        return None


# ----------------------------------------------------------------------------
# Evaluation + soumission
# ----------------------------------------------------------------------------
def run_and_submit_all(profile: gr.OAuthProfile | None):
    space_id = os.getenv("SPACE_ID")

    if not profile:
        return "Please log in to Hugging Face with the button.", None
    username = profile.username

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

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

    try:
        response = requests.get(f"{api_url}/questions", timeout=30)
        response.raise_for_status()
        questions = response.json()
    except Exception as e:
        return f"Error fetching questions: {e}", None

    results_log = []
    answers_payload = []
    for i, item in enumerate(questions):
        task_id = item.get("task_id")
        question_text = item.get("question")
        file_name = item.get("file_name", "")
        if not task_id or question_text is None:
            continue
        file_path = download_task_file(api_url, task_id, file_name)
        try:
            answer = agent(question_text, file_path)
        except Exception as e:
            print(f"Error on task {task_id}: {e}")
            answer = "AGENT ERROR"
        answers_payload.append({"task_id": task_id, "submitted_answer": answer})
        results_log.append(
            {
                "Task ID": task_id,
                "Question": question_text[:150],
                "Submitted Answer": answer,
            }
        )
        print(f"[{i + 1}/{len(questions)}] {answer}")
        time.sleep(PAUSE_BETWEEN_QUESTIONS)

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

    submission = {
        "username": username.strip(),
        "agent_code": agent_code,
        "answers": answers_payload,
    }
    try:
        response = requests.post(f"{api_url}/submit", json=submission, timeout=120)
        response.raise_for_status()
        data = response.json()
        status = (
            "Submission successful!\n"
            f"User: {data.get('username')}\n"
            f"Score: {data.get('score')}% "
            f"({data.get('correct_count')}/{data.get('total_attempted')} correct)\n"
            f"Message: {data.get('message')}"
        )
        return status, pd.DataFrame(results_log)
    except Exception as e:
        return f"Submission failed: {e}", pd.DataFrame(results_log)


# ----------------------------------------------------------------------------
# Interface Gradio
# ----------------------------------------------------------------------------
with gr.Blocks() as demo:
    gr.Markdown("# Agent Evaluation Runner")
    gr.Markdown(
        "1. Log in with your Hugging Face account.\n"
        "2. Click the button: the agent answers the 20 questions and submits them.\n"
        "3. This can take 10 to 20 minutes. Keep the page open."
    )
    gr.LoginButton()
    run_button = gr.Button("Run Evaluation & Submit All Answers")
    status_output = gr.Textbox(label="Run Status / 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__":
    demo.launch(debug=True, share=False)