app.py

import os
import gradio as gr
import requests
import pandas as pd
import torch
import base64
from io import BytesIO
import numexpr  # Using a dedicated and safe math library

from llama_index.core.tools import FunctionTool
from llama_index.llms.huggingface import HuggingFaceLLM
from llama_index.core.agent import ReActAgent
from llama_index.tools.duckduckgo import DuckDuckGoSearchToolSpec
from youtube_transcript_api import YouTubeTranscriptApi
from PIL import Image

# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
IMAGE_ANALYSIS_API_URL = (
    "https://api-inference.huggingface.co/models/llava-hf/llava-1.5-7b-hf"
)

# --- Helper Functions for Tools ---

# HF_TOKEN must be set as a Space Secret in Hugging Face
HF_TOKEN = os.getenv("HF_TOKEN")


def get_video_transcript(youtube_url: str):
    """Fetches the transcript of a YouTube video given its URL."""
    try:
        if "v=" not in youtube_url:
            return "Error: Invalid YouTube URL, missing 'v='."
        video_id = youtube_url.split("v=")[1].split("&")[0]
        transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
        transcript = " ".join([d["text"] for d in transcript_list])
        return transcript
    except Exception as e:
        return f"Error fetching transcript: {e}"


def analyze_image_url(image_url: str, question: str):
    """Analyzes an image from a URL using the Hugging Face Inference API."""
    if not HF_TOKEN:
        return (
            "Error: Hugging Face token is not set. Cannot use the image analysis tool."
        )
    try:
        response = requests.get(image_url)
        response.raise_for_status()
        image_bytes = BytesIO(response.content).getvalue()
        headers = {"Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "image/png"}
        response = requests.post(
            IMAGE_ANALYSIS_API_URL, headers=headers, data=image_bytes
        )
        response.raise_for_status()
        result = response.json()
        generated_text = result[0].get("generated_text", "").strip()
        final_answer = generated_text.split("ASSISTANT:")[-1].strip()
        return f"The image description is: {final_answer}. Now, answer the original question based on this."
    except Exception as e:
        return f"Error analyzing image: {e}"


# NEW: A custom, reliable math tool using a safe evaluator
def evaluate_math_expression(expression: str):
    """Evaluates a mathematical expression safely."""
    try:
        # Using numexpr for safe evaluation of numerical expressions
        result = numexpr.evaluate(expression).item()
        return result
    except Exception as e:
        return f"Error evaluating expression: {e}"


# --- Tool Definitions ---
youtube_tool = FunctionTool.from_defaults(
    fn=get_video_transcript,
    name="youtube_transcript_tool",
    description="Use this tool to get the transcript of a YouTube video.",
)
image_analyzer_tool = FunctionTool.from_defaults(
    fn=analyze_image_url,
    name="image_analyzer_tool",
    description="Use this tool to analyze an image when you are given a URL. Provide both the image URL and the question about the image.",
)
math_tool = FunctionTool.from_defaults(
    fn=evaluate_math_expression,
    name="math_evaluator_tool",
    description="Use this tool to evaluate simple mathematical expressions (e.g., '3 * (4 + 2)').",
)


# --- LlamaIndex Agent Definition ---
class LlamaIndexAgent:
    def __init__(self):
        print("Initializing LlamaIndexAgent with Final Tools...")
        ddg_spec = DuckDuckGoSearchToolSpec()
        self.tools = [
            youtube_tool,
            image_analyzer_tool,
            math_tool,
        ] + ddg_spec.to_tool_list()
        system_prompt = """
        You are a helpful assistant tasked with answering questions.
        You have access to a set of tools to help you. These tools include:
        - A web search tool.
        - A YouTube video transcriber.
        - An image analyzer for URLs.
        - A safe calculator for mathematical expressions.
        Use a tool if it is helpful. When you have the final answer, you MUST use the following template: FINAL ANSWER: [YOUR FINAL ANSWER].
        YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list.
        """
        self.llm = HuggingFaceLLM(
            model_name="HuggingFaceH4/zephyr-7b-beta",
            tokenizer_name="HuggingFaceH4/zephyr-7b-beta",
            device_map="auto",
            model_kwargs={"torch_dtype": torch.float16, "load_in_8bit": True},
        )
        self.agent = ReActAgent.from_tools(
            tools=self.tools, llm=self.llm, verbose=True, system_prompt=system_prompt
        )
        print("LlamaIndexAgent initialized successfully.")

    def __call__(self, question: str) -> str:
        print(f"Agent received question: {question[:80]}...")
        response = self.agent.chat(question)
        answer = str(response).strip()
        if "FINAL ANSWER:" in answer:
            final_answer = answer.split("FINAL ANSWER:")[-1].strip()
        else:
            print(
                f"Warning: Agent did not use the 'FINAL ANSWER:' template. Raw output: {answer}"
            )
            final_answer = answer
        return final_answer


# --- Main Gradio App Logic ---
def run_and_submit_all(profile: gr.OAuthProfile | None):
    if not HF_TOKEN:
        return (
            "ERROR: The `HF_TOKEN` secret is not set in this Space. The image analysis tool will fail. Please set it in Settings > Secrets.",
            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"
    try:
        # We instantiate our new powerful agent instead of the BasicAgent
        agent = LlamaIndexAgent()
    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(questions_url, 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 = []
    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,
    }
    try:
        response = requests.post(submit_url, json=submission_data, timeout=180)
        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.')}"
        )
        return final_status, pd.DataFrame(results_log)
    except Exception as e:
        return f"An unexpected error occurred during submission: {e}", pd.DataFrame(
            results_log
        )


# --- Build Gradio Interface using Blocks ---
# UI HAS BEEN REVERTED TO THE INITIAL TEMPLATE AS REQUESTED
with gr.Blocks() as demo:
    gr.Markdown("# Basic Agent Evaluation Runner")
    gr.Markdown(
        """
        **Instructions:**

        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.

        ---
        **Disclaimers:**
        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
        """
    )
    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)
    if not HF_TOKEN:
        print(
            "⚠️ WARNING: The `HF_TOKEN` secret is not set. The image analysis tool will be unavailable."
        )
    else:
        print("✅ `HF_TOKEN` secret is set.")
    print("Launching Gradio Interface...")
    demo.launch(debug=True, share=False)