app.py

import json
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
from langchain_groq import ChatGroq
from typing import Annotated, Sequence, TypedDict, Literal, Dict
from contextlib import redirect_stdout, redirect_stderr
from dotenv import load_dotenv
import os, base64, mimetypes, re
import gradio as gr
import requests
import pandas as pd
import json
import traceback
import io
from tavily import TavilyClient
from langchain_core.tools import tool
import time

# --- Constants ---
load_dotenv()
client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
groq_token = os.getenv("GROQ_API_KEY")
tavily_token = os.getenv("TAVILY_API_KEY")

MAX_BYTES = 20 * 1024 * 1024  # 20MB cap for inline base64 payloads
_YT_ID = re.compile(r"(?:v=|youtu\.be/|shorts/)([A-Za-z0-9_-]{11})")

# --- System messystem_prompt = 
system_prompt = """
You are Supervisor.

Rules:

- Use web_research_agent only when external information is required.
- Use python_code_runner_agent only for code execution.
- Answer simple reasoning questions directly.

If the answer is a number:
return only the number.

If the answer is text:
return only the text.

Never explain.
Never say "The answer is".
"""
prompt_search = """
You are a web research agent.

You have one tool:

web_search(query)

Always use the tool when factual information is required.

After receiving search results:
- Extract the answer.
- Return only the final answer.
- If multiple sources disagree, choose the most reliable source.
- If the answer is numeric, return only the number.
"""

prompt_python_execute = """You are a python code execution agent.
                           INSTRUCTIONS:
                           Assist ONLY with tasks related to running python code, DO NOT do any math
                           After you're done with your tasks, respond to the supervisor directly
                           Respond ONLY with the results of your work, do NOT include ANY other text."""

prompt_chess = """You are a chess position reviewing agent.
                   INSTRUCTIONS:
                   Assist ONLY with tasks related to chess position reviewing
                   After you're done with your tasks, respond to the supervisor directly
                   Respond ONLY with the results of your work, do NOT include ANY other text."""


# --- Tools ---


CHESSVISION_TO_FEN_URL = "http://app.chessvision.ai/predict"
CHESS_MOVE_API = "https://chess-api.com/v1"


def _extract_yt_id(url: str):
    m = _YT_ID.search(url)
    return m.group(1) if m else None


def download_file_as_base64(task_id: str) -> str:
    url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
    response = requests.get(url)
    if response.status_code == 200:
        encoded_bytes = base64.b64encode(response.content)
        return encoded_bytes.decode('utf-8')
    else:
        raise Exception(f"Failed to download the file. Status code: {response.status_code}")


def download_file_as_string(task_id: str) -> str:
    url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
    response = requests.get(url)
    if response.status_code == 200:
        return response.content.decode('utf-8')
    else:
        raise Exception(f"Failed to download the file. Status code: {response.status_code}")

@tool
def web_search(query: str) -> str:
    """
    Search the web and return relevant results.
    """
    try:
        result = client.search(
    query=query,
    max_results=3,
    search_depth="basic"
)
        snippets = []

        for item in result.get("results", []):
            snippets.append(
                f"Title: {item.get('title','')}\n"
                f"Content: {item.get('content','')}\n"
                f"URL: {item.get('url','')}\n"
            )

        return "\n\n".join(snippets)

    except Exception as e:
        return f"ERROR: {str(e)}"
from langchain_core.tools import tool


@tool("python_code_runner_tool")
def python_code_runner_tool(task_id: str) -> str:
    """
    Download and run python code, capturing the output.
    Args:
        task_id: Task ID necessary to retrieve the python code to be run.
    Returns:
        String with the output of the python code.
    """
    print(f"python_code_runner_tool called with task_id: {task_id}")
    python_code = download_file_as_string(task_id)
    print(f"python_code: {python_code}")

    saida = io.StringIO()
    erros = io.StringIO()

    try:
        with redirect_stdout(saida), redirect_stderr(erros):
            exec(python_code, {'__name__': '__main__'})

        saida_valor = saida.getvalue()
        erro_valor = erros.getvalue()

        if erro_valor:
            return f"[ERRO DE EXECUÇÃO]:\n{erro_valor}"

        return saida_valor if saida_valor.strip() else "[SEM SAÍDA]"

    except Exception:
        return f"[EXCEÇÃO DURANTE EXECUÇÃO]:\n{traceback.format_exc()}"


def chess_image_to_fen_tool(
    task_id: str,
    current_player: Literal["black", "white"]
) -> Dict[str, str]:
    """
    Convert a chess board image into FEN notation.

    Args:
        task_id: Task identifier containing the chess image.
        current_player: white or black.

    Returns:
        Dictionary containing the FEN position.
    """

    print(f"chess_image_to_fen_tool called: task_id={task_id}, current_player={current_player}")
def chess_fen_get_best_next_move_tool(fen: str, current_player: Literal["black", "white"]) -> str:
    """
    Return the best move in algebraic notation.
    Args:
        fen: FEN notation string.
        current_player: Whose turn it is ('black' or 'white').
    Returns:
        Best move in algebraic notation.
    """
    if not fen:
        raise ValueError("fen must be provided.")
    if current_player not in ["black", "white"]:
        raise ValueError("current_player must be 'black' or 'white'")

    payload = {"fen": fen}
    print(f"Fetching best move from {CHESS_MOVE_API} with payload: {payload}")

    response = requests.post(CHESS_MOVE_API, json=payload)
    if response.status_code == 200:
        dados = response.json()
        move_algebraic = dados.get("san")
        move = dados.get("text")
        print(f"Best move from chess-api.com: {move}")
        return move_algebraic
    else:
        raise Exception(f"Request error: {response.status_code}")


# --- LLM Definition (Groq only) ---
groq_llm = ChatGroq(
    model="openai/gpt-oss-20b",
    temperature=0,
    max_retries=2,
    api_key=groq_token,
)
print("Registered tool:", web_search.name)
# --- Agents ---
web_research_agent = create_react_agent(
    model=groq_llm,
    tools=[web_search],
    prompt=prompt_search,
    name="web_research_agent"
)

python_code_runner_agent = create_react_agent(
    model=groq_llm,
    tools=[python_code_runner_tool],
    prompt=prompt_python_execute,
    name="python_code_runner_agent"
)



SupAgent = create_supervisor(
    model=groq_llm,
    agents=[
        web_research_agent,
        python_code_runner_agent
    ],
    prompt=system_prompt,
    add_handoff_back_messages=False,
    output_mode="last_message",
).compile()


class BasicAgent:
    def __init__(self):
        print('agent initialized')

    def __call__(self, question: str, task_id: str) -> str:
        question = "Question: " + question + " Task ID: " + task_id

        messages = {
            "messages": [
                {
                    "role": "user",
                    "content": question
                }
            ]
        }
        print(f"Input messages: {messages}.")
        events = SupAgent.stream(
            messages,
            stream_mode="values",
        )
        listMessages = []

        for event in events:
            listMessages.extend(event["messages"])
        print(f"messages: {listMessages}\n")

        answer = ""
        for msg in reversed(listMessages):
            if hasattr(msg, "content"):
                content = str(msg.content).strip()
                if content and \
                "Successfully transferred" not in content and \
                "Transferring back" not in content:
                 answer = content
                 break
        print(f"Answer: {answer}\n")
        return answer


def run_and_submit_all(profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the BasicAgent on them, submits all answers,
    and displays the results.
    """
    space_id = os.getenv("SPACE_ID")

    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

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

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

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

    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: {e}")
        return f"Error decoding server response for questions: {e}", None
    except Exception as e:
        print(f"Unexpected error fetching questions: {e}")
        return f"An unexpected error occurred 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:
            print(f"Skipping item with missing task_id or question: {item}")
            continue
        try:
            submitted_answer = agent(question_text, task_id)
            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}"})
        

    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)

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

    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, pd.DataFrame(results_log)
    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, pd.DataFrame(results_log)
    except requests.exceptions.Timeout:
        status_message = "Submission Failed: The request timed out."
        print(status_message)
        return status_message, pd.DataFrame(results_log)
    except requests.exceptions.RequestException as e:
        status_message = f"Submission Failed: Network error - {e}"
        print(status_message)
        return status_message, pd.DataFrame(results_log)
    except Exception as e:
        status_message = f"An unexpected error occurred during submission: {e}"
        print(status_message)
        return status_message, pd.DataFrame(results_log)


# --- Gradio Interface ---
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.
        """
    )

    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)
    space_host_startup = os.getenv("SPACE_HOST")
    space_id_startup = os.getenv("SPACE_ID")
    print(space_host_startup, space_id_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(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?).")

    print("-" * (60 + len(" App Starting ")) + "\n")
    print("Launching Gradio Interface for Basic Agent Evaluation...")
    demo.launch(debug=True, share=False)