app.py

"""Basic Agent Evaluation Runner"""

import inspect
import os
from typing import Any

import gradio as gr
import pandas as pd
import requests

from gagent.agents import registry
from gagent.config import settings

# (Keep Constants as is)
# --- Constants ---
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

# --- Basic Agent Definition ---
# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------


class BasicAgent:
    """A langgraph agent."""

    def __init__(self, agent_type: str, **kwargs):
        print(f"BasicAgent initialized with type: {agent_type}")
        self.agent = registry.get_agent(agent_type=agent_type, **kwargs)

    def __call__(self, question: str, question_number: int | None, total_questions: int | None) -> str:
        print(
            f"\n{':' * 20}Agent received question ({question_number}/{total_questions}){':' * 20}\n{question}\n{'-' * 100}"
        )
        answer = self.agent.run(question, question_number=question_number, total_questions=total_questions)
        return answer


def get_agent_parameters(agent_type: str) -> dict[str, Any]:
    """Get the parameters for a specific agent type."""
    if agent_type not in registry._agent_classes:
        return {}

    agent_class = registry._agent_classes[agent_type]
    init_signature = inspect.signature(agent_class.__init__)

    parameters = {}
    for name, param in init_signature.parameters.items():
        if name == "self":
            continue

        # Get default value if available
        default = param.default if param.default != inspect.Parameter.empty else None

        # Get parameter type
        param_type = param.annotation if param.annotation != inspect.Parameter.empty else str

        # Get parameter description from docstring if available
        description = ""
        if agent_class.__doc__:
            doc_lines = agent_class.__doc__.split("\n")
            for line in doc_lines:
                if f"{name}:" in line:
                    description = line.split(":")[1].strip()
                    break

        parameters[name] = {
            "type": param_type,
            "default": default,
            "description": description,
        }

    return parameters


def get_settings_value(param_name: str) -> str:
    """Get the value of a parameter from settings if available."""
    return getattr(settings, param_name.upper(), "")


def run_and_submit_all(request: gr.Request, profile: gr.OAuthProfile | None, *args):
    """
    Fetches all questions, runs the BasicAgent on them, submits all answers,
    and displays the results. Optionally skips submission.
    """
    # --- Determine HF Space Runtime URL and Repo URL ---
    space_id = os.getenv("SPACE_ID")  # Get the SPACE_ID for sending link to the code

    if profile:
        username = f"{profile.username}"
        print(f"User logged in: {username}")
    else:
        print("User not logged in.")
        return "Please Login to Hugging Face with the button.", None

    # Get available agents from registry
    available_agents = registry.list_available_agents()
    if not available_agents:
        return "No agents available in registry.", None

    agent_type = agent_type_dropdown.value

    # Validate agent type
    if not agent_type or agent_type not in available_agents:
        print(f"Invalid agent type: {agent_type}, using first available agent")
        agent_type = available_agents[0]

    print(f"Running agent with type: {agent_type}")  # Debug log

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

    # Get parameters from args
    parameters = {}
    agent_params = get_agent_parameters(agent_type)
    print(f"Agent {agent_type} parameters: {agent_params}")  # Debug log

    # Map input values to their corresponding parameters
    for i, (param_name, param_info) in enumerate(agent_params.items()):
        if i < len(parameter_inputs):
            parameters[param_name] = parameter_inputs[param_name].value
            print(f"Setting parameter {param_name} = {parameter_inputs[param_name].value}")  # Debug log

    print(f"Agent parameters: {parameters}")  # Debug log

    # 1. Instantiate Agent
    try:
        print(f"Initializing agent with type: {agent_type}")
        agent = BasicAgent(agent_type=agent_type, **parameters)
    except Exception as e:
        print(f"Error instantiating agent: {e}")
        return f"Error initializing agent: {e}", None
    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
    print(agent_code)

    # 2. Fetch Questions
    print(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=15)
        response.raise_for_status()
        questions_data = response.json()
        if not questions_data:
            print("Fetched questions list is empty.")
            return "Fetched questions list is empty or invalid format.", None
        print(f"Fetched {len(questions_data)} questions.")
    except requests.exceptions.RequestException as e:
        print(f"Error fetching questions: {e}")
        return f"Error fetching questions: {e}", None
    except requests.exceptions.JSONDecodeError as e:
        print(f"Error decoding JSON response from questions endpoint: {e}")
        print(f"Response text: {response.text[:500]}")
        return f"Error decoding server response for questions: {e}", None
    except Exception as e:
        print(f"An unexpected error occurred fetching questions: {e}")
        return f"An unexpected error occurred fetching questions: {e}", None

    # # TODO: Remove this
    # questions_data = questions_data[:3]

    # 3. Run your Agent
    results_log = []
    answers_payload = []
    total_questions = len(questions_data)
    print(f"Running agent on {total_questions} questions...")

    # Create a progress bar
    progress = gr.Progress()

    for i, item in enumerate(questions_data, 1):
        task_id = item.get("task_id")
        question_text = item.get("question")
        if not task_id or question_text is None:
            print(f"Skipping item with missing task_id or question: {item}")
            continue
        try:
            # Update progress
            progress((i - 1) / total_questions)

            # Run agent with progress info
            submitted_answer = agent(question_text, question_number=i, total_questions=total_questions)
            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:
            print(f"Error running agent on task {task_id}: {e}")
            results_log.append(
                {
                    "Task ID": task_id,
                    "Question": question_text,
                    "Submitted Answer": f"AGENT ERROR: {e}",
                }
            )

    # Complete progress bar
    progress(1.0)

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

    # 4. Prepare Submission
    submission_data = {
        "username": username.strip(),
        "agent_code": agent_code,
        "answers": answers_payload,
    }
    status_update = f"Agent finished. Preparing {len(answers_payload)} answers for user '{username}'..."
    print(status_update)

    # 5. Submit (or Skip)
    results_df = pd.DataFrame(results_log)
    if skip_submission:
        final_status = "Submission skipped as requested."
        print(final_status)
        return final_status, results_df

    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
    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.')}"
        )
        print("Submission successful.")
        return final_status, results_df
    except requests.exceptions.HTTPError as e:
        error_detail = f"Server responded with status {e.response.status_code}."
        try:
            error_json = e.response.json()
            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
        except requests.exceptions.JSONDecodeError:
            error_detail += f" Response: {e.response.text[:500]}"
        status_message = f"Submission Failed: {error_detail}"
        print(status_message)
        return status_message, results_df
    except requests.exceptions.Timeout:
        status_message = "Submission Failed: The request timed out."
        print(status_message)
        return status_message, results_df
    except requests.exceptions.RequestException as e:
        status_message = f"Submission Failed: Network error - {e}"
        print(status_message)
        return status_message, results_df
    except Exception as e:
        status_message = f"An unexpected error occurred during submission: {e}"
        print(status_message)
        return status_message, results_df


# Dictionary to store parameter inputs for each agent type
all_parameter_inputs = {}

# Initialize parameter inputs dictionary
parameter_inputs = {}

skip_submission = True

# --- Build Gradio Interface using Blocks ---
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.  Select your agent type and configure its parameters.
        4.  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()

    with gr.Row():
        with gr.Column():
            # Get available agents from registry
            available_agents = registry.list_available_agents()
            if not available_agents:
                raise ValueError("No agents found in registry. Please check your agent implementations.")

            # Get default agent from settings
            default_agent = settings.DEFAULT_AGENT
            if default_agent not in available_agents:
                default_agent = available_agents[0]  # Fallback to first available agent
                print(f"Default agent '{settings.DEFAULT_AGENT}' not available, using '{default_agent}' instead")

            # Create agent type dropdown with change handler
            def on_agent_type_change(agent_type: str):
                """Handle agent type change."""
                print(f"Agent type changed to: {agent_type}")
                if not agent_type:
                    return gr.Column(visible=False)

                param_col = create_parameter_inputs(agent_type)
                return param_col

            agent_type_dropdown = gr.Dropdown(
                choices=available_agents,
                label="Agent Type",
                value=default_agent,  # Use default agent from settings
            )

            # Create a container for parameter inputs
            parameter_container = gr.Column()

            def create_parameter_inputs(agent_type: str):
                """Create parameter inputs for the selected agent type."""
                global parameter_inputs

                if not agent_type:
                    return gr.Column(visible=False)

                print(f"Creating parameter inputs for agent type: {agent_type}")

                parameters = get_agent_parameters(agent_type)

                # Check if we already have inputs for this agent type
                if agent_type in all_parameter_inputs:
                    parameter_inputs = all_parameter_inputs[agent_type]
                else:
                    # Create new parameter inputs
                    parameter_inputs = {}

                    # Create a new column for parameters
                    with gr.Column(visible=True) as param_col:
                        for param_name, param_info in parameters.items():
                            # Determine input type based on parameter type
                            if param_info["type"] == bool:
                                input_component = gr.Checkbox(
                                    label=param_name,
                                    value=param_info["default"] or False,
                                    info=param_info["description"],
                                )
                            elif param_info["type"] == int:
                                input_component = gr.Number(
                                    label=param_name,
                                    value=param_info["default"] or 0,
                                    info=param_info["description"],
                                )
                            elif param_info["type"] == float:
                                input_component = gr.Number(
                                    label=param_name,
                                    value=param_info["default"] or 0.0,
                                    info=param_info["description"],
                                )
                            else:  # Default to text input
                                # Check if this is likely an API key
                                is_api_key = any(key in param_name.lower() for key in ["api", "key", "token"])
                                input_component = gr.Textbox(
                                    label=param_name,
                                    value=get_settings_value(param_name) or param_info["default"] or "",
                                    type="password" if is_api_key else "text",
                                    info=param_info["description"],
                                )

                            input_component.placeholder = "Leave blank for default from environment variable"
                            parameter_inputs[param_name] = input_component

                    # Store in our dictionary
                    all_parameter_inputs[agent_type] = parameter_inputs

                return param_col

            # Create initial parameter inputs for default agent
            initial_params = create_parameter_inputs(default_agent)
            parameter_container = initial_params

            # Update parameter inputs when agent type changes
            def update_parameter_inputs(agent_type):
                global parameter_inputs
                # Update the parameter_inputs reference
                if agent_type in all_parameter_inputs:
                    parameter_inputs = all_parameter_inputs[agent_type]
                return on_agent_type_change(agent_type)

            agent_type_dropdown.change(
                fn=update_parameter_inputs,
                inputs=[agent_type_dropdown],
                outputs=[parameter_container],
            )

    run_button = gr.Button("Run Evaluation & Submit All Answers")
    skip_submission_checkbox = gr.Checkbox(
        label="Skip Submission",
        value=skip_submission,
        info="Check this box to run the agent without submitting answers to the scoring API.",
    )

    def update_skip_submission(val: bool):
        global skip_submission
        skip_submission = val

    skip_submission_checkbox.change(
        fn=update_skip_submission,
        inputs=[skip_submission_checkbox],
        outputs=[],
    )

    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,
        inputs=[gr.State(), gr.State()],
        outputs=[status_output, results_table],
    )

if __name__ == "__main__":
    print("\n" + "-" * 30 + " App Starting " + "-" * 30)
    # Check for SPACE_HOST and SPACE_ID at startup for information
    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID")  # Get SPACE_ID at startup

    if space_host_startup:
        print(f"✅ SPACE_HOST found: {space_host_startup}")
        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
    else:
        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")

    if space_id_startup:  # Print repo URLs if SPACE_ID is found
        print(f"✅ SPACE_ID found: {space_id_startup}")
        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
    else:
        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")

    print("-" * (60 + len(" App Starting ")) + "\n")

    print("Launching Gradio Interface for Basic Agent Evaluation...")
    demo.launch(debug=True, share=False)