app.py

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

from smolagents import CodeAgent, LiteLLMModel, DuckDuckGoSearchTool, VisitWebpageTool, tool

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


# ---------------------------------------------------------------------------
# Custom tools for file-based question types (audio, images, python files)
# ---------------------------------------------------------------------------

@tool
def transcribe_audio(file_path: str) -> str:
    """
    Transcribes an audio file (mp3/wav) to text using OpenAI Whisper.

    Args:
        file_path: Local path to the audio file to transcribe.

    Returns:
        The transcribed text.
    """
    from openai import OpenAI
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
    with open(file_path, "rb") as f:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=f
        )
    return transcript.text


@tool
def analyze_image(file_path: str, question: str) -> str:
    """
    Analyzes an image (e.g. a chess position) using a vision-capable LLM
    and answers a question about it.

    Args:
        file_path: Local path to the image file.
        question: The question to answer about the image.

    Returns:
        The model's answer about the image.
    """
    import base64
    from openai import OpenAI
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

    with open(file_path, "rb") as f:
        b64_image = base64.b64encode(f.read()).decode("utf-8")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64_image}"}}
            ]
        }]
    )
    return response.choices[0].message.content


@tool
def run_python_file(file_path: str) -> str:
    """
    Reads and returns the contents of a Python (.py) file so the agent
    can analyze or trace through the code to determine its output.

    Args:
        file_path: Local path to the python file.

    Returns:
        The raw source code as text.
    """
    with open(file_path, "r") as f:
        return f.read()


@tool
def read_excel_file(file_path: str) -> str:
    """
    Reads an Excel (.xlsx) file and returns its contents as a string table.

    Args:
        file_path: Local path to the Excel file.

    Returns:
        A string representation of the spreadsheet data.
    """
    df = pd.read_excel(file_path)
    return df.to_string()


# ---------------------------------------------------------------------------
# The Agent
# ---------------------------------------------------------------------------

class BasicAgent:
    def __init__(self):
        print("BasicAgent initialized.")

        # LiteLLM lets you swap providers by just changing model_id, e.g.:
        # "gpt-4o-mini", "claude-3-5-sonnet-20241022", "huggingface/Qwen/Qwen2.5-72B-Instruct"
        self.model = LiteLLMModel(
            model_id="gpt-4o-mini",
            api_key=os.environ.get("OPENAI_API_KEY"),
        )

        self.agent = CodeAgent(
            model=self.model,
            tools=[
                DuckDuckGoSearchTool(),
                VisitWebpageTool(),
                transcribe_audio,
                analyze_image,
                run_python_file,
                read_excel_file,
            ],
            max_steps=8,
        )

    def __call__(self, question: str, file_path: str = None) -> str:
        print(f"Agent received question (first 80 chars): {question[:80]}...")

        prompt = question
        if file_path:
            prompt += f"\n\nA file has been downloaded for this question at local path: {file_path}. Use the appropriate tool to read it before answering."

        prompt += "\n\nIMPORTANT: Respond with ONLY the final answer. No explanation, no 'FINAL ANSWER:' prefix, just the answer itself, formatted exactly as requested in the question."

        try:
            answer = self.agent.run(prompt)
        except Exception as e:
            print(f"Agent error: {e}")
            answer = "ERROR"

        answer = str(answer).strip()
        print(f"Agent returning answer: {answer}")
        return answer


# ---------------------------------------------------------------------------
# Evaluation + submission logic
# ---------------------------------------------------------------------------

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

    if profile:
        username = profile.username
        print(f"User logged in: {username}")
    else:
        return "Please log in to Hugging Face first.", None

    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/questions"
    files_url = f"{api_url}/files"
    submit_url = f"{api_url}/submit"

    agent = BasicAgent()
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"

    # 1. Fetch questions
    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 = []

    for item in questions_data:
        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 = None
        if file_name:
            try:
                file_resp = requests.get(f"{files_url}/{task_id}", timeout=30)
                file_resp.raise_for_status()
                file_path = f"/tmp/{file_name}"
                with open(file_path, "wb") as f:
                    f.write(file_resp.content)
            except Exception as e:
                print(f"Could not download file for {task_id}: {e}")

        try:
            submitted_answer = agent(question_text, file_path)
        except Exception as e:
            submitted_answer = f"AGENT ERROR: {e}"

        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})

    if not answers_payload:
        return "No answers were generated.", pd.DataFrame(results_log)

    # 2. Submit
    submission_data = {
        "username": username.strip(),
        "agent_code": agent_code,
        "answers": answers_payload
    }

    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', '')}"
        )
        return final_status, pd.DataFrame(results_log)
    except Exception as e:
        return f"Submission failed: {e}", pd.DataFrame(results_log)


# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------

with gr.Blocks() as demo:
    gr.Markdown("# Basic Agent Evaluation Runner")
    gr.Markdown(
        """
        **Instructions:**
        1. This Space defines your agent's logic, tools, and required packages.
        2. Log in to your Hugging Face account using the button below.
        3. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
        """
    )

    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)