app.py

import os
import logging
import mimetypes
import subprocess

from typing import Any, List

import gradio as gr
import requests
import pandas as pd
import io
import torchaudio
import torchaudio.transforms as T
import whisper

from llama_index.core.agent.workflow import AgentWorkflow, ToolCallResult, ToolCall, AgentOutput
from llama_index.core.base.llms.types import ChatMessage, TextBlock, ImageBlock, AudioBlock
from llama_index.llms.openai import OpenAI

from agents.video_analyzer_agent import initialize_video_analyzer_agent

os.environ["TOKENIZERS_PARALLELISM"] = "false"

# Assuming agent initializers are in the same directory or a known path
# Adjust import paths if necessary based on deployment structure
try:
    # Existing agents
    from agents.image_analyzer_agent import initialize_image_analyzer_agent
    from agents.reasoning_agent import initialize_reasoning_agent
    from agents.text_analyzer_agent import initialize_text_analyzer_agent
    from agents.code_agent import initialize_code_agent
    from agents.math_agent import initialize_math_agent
    from agents.planner_agent import initialize_planner_agent
    from agents.research_agent import initialize_research_agent
    from agents.role_agent import initialize_role_agent
    # New agents
    from agents.advanced_validation_agent import initialize_advanced_validation_agent
    from agents.long_context_management_agent import initialize_long_context_management_agent
    from agents.synthesis_agent import initialize_synthesis_agent
    AGENT_IMPORT_PATH = "local"
except ImportError as e:
    print(f"Import Error: Could not find agent modules. Tried local and final_project paths. Error: {e}")
    # Set initializers to None or raise error to prevent app start
    initialize_image_analyzer_agent = None
    # ... set all others to None ...
    raise RuntimeError(f"Failed to import agent modules: {e}")

# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# --- Constants ---
DEFAULT_API_URL = os.getenv("GAIA_API_URL", "https://agents-course-unit4-scoring.hf.space")

# --- Helper Functions ---
_whisper_model = whisper.load_model("small")

def transcribe_audio(audio_bytes: bytes) -> str:
    logger.info(f"Attempting to transcribe audio file")

    file_like = io.BytesIO(audio_bytes)

    waveform, sample_rate = torchaudio.load(file_like)

    waveform = waveform.mean(dim=0, keepdim=True)  # [1, samples]

    if sample_rate != 16000:
        resampler = T.Resample(orig_freq=sample_rate, new_freq=16000)
        waveform = resampler(waveform)

    waveform = waveform.squeeze(0)

    print(f"Tensor shape : {waveform.shape}, Frequency : {sample_rate} Hz")

    # Load the Whisper model (lazy loading)
    model: whisper.Whisper = _whisper_model  # Uses default size "base" or WHISPER_MODEL_SIZE env var
    if model is None:
        return "Error: Failed to load Whisper model."

    try:
        # Perform transcription
        # The transcribe function handles various audio formats via ffmpeg
        result = whisper.transcribe(model=model, audio=waveform)
        transcribed_text = result["text"]
        detected_language = result.get("language", "unknown")  # Get detected language if available
        logger.info(
            f"Audio transcription successful. Detected language: {detected_language}. Text length: {len(transcribed_text)}")
        return transcribed_text

    except Exception as e:
        # Check if it might be an ffmpeg issue
        if "ffmpeg" in str(e).lower():
            logger.error(f"Error during transcription, possibly ffmpeg issue: {e}", exc_info=True)
            # Check if ffmpeg is installed using shell command
            try:
                subprocess.run(["ffmpeg", "-version"], check=True, capture_output=True)
                # If ffmpeg is installed, the error is likely something else
                return f"Error during transcription (ffmpeg seems installed): {e}"
            except (FileNotFoundError, subprocess.CalledProcessError):
                logger.error("ffmpeg command not found or failed. Please ensure ffmpeg is installed and in PATH.")
                return "Error: ffmpeg not found or not working. Please install ffmpeg."
        else:
            logger.error(f"Unexpected error during transcription: {e}", exc_info=True)
            return f"Error during transcription: {e}"

# --- Agent Initialization (Singleton Pattern) ---
# Initialize the agent workflow once
AGENT_WORKFLOW = None
try:
    logger.info(f"Initializing GAIA Multi-Agent Workflow (import path: {AGENT_IMPORT_PATH})...")
    # Existing agents
    # role_agent = initialize_role_agent()
    code_agent = initialize_code_agent()
    math_agent = initialize_math_agent()
    planner_agent = initialize_planner_agent()
    research_agent = initialize_research_agent()
    text_analyzer_agent = initialize_text_analyzer_agent()
    image_analyzer_agent = initialize_image_analyzer_agent()
    reasoning_agent = initialize_reasoning_agent()
    # New agents
    advanced_validation_agent = initialize_advanced_validation_agent()
    long_context_management_agent = initialize_long_context_management_agent()
    video_analyzer_agent = initialize_video_analyzer_agent()
    synthesis_agent = initialize_synthesis_agent()

    # Check if all agents initialized successfully
    all_agents = [
        code_agent, math_agent, planner_agent, research_agent,
        text_analyzer_agent, image_analyzer_agent, reasoning_agent,
        advanced_validation_agent, long_context_management_agent,
        video_analyzer_agent, synthesis_agent
    ]
    if not all(all_agents):
        raise RuntimeError("One or more agents failed to initialize.")

    AGENT_WORKFLOW = AgentWorkflow(
        agents=all_agents,
        root_agent="reasoning_agent", # Keep planner as root as per plan
        initial_state={
            "research_content": []
        }
    )
    logger.info("GAIA Multi-Agent Workflow initialized successfully.")
except Exception as e:
    logger.error(f"FATAL: Error initializing agent workflow: {e}", exc_info=True)
    # AGENT_WORKFLOW remains None, BasicAgent init will fail

# --- Basic Agent Definition (Wrapper for Workflow) ---
class BasicAgent:
    def __init__(self, workflow: AgentWorkflow):
        if workflow is None:
            logger.error("AgentWorkflow is None, initialization likely failed.")
            raise RuntimeError("AgentWorkflow failed to initialize. Check logs for details.")
        self.agent_workflow = workflow
        logger.info("BasicAgent wrapper initialized.")

    async def __call__(self, question: str | ChatMessage) -> Any:
        if isinstance(question, ChatMessage):
            log_question = str(question.blocks[0].text)[:100] if question.blocks and hasattr(question.blocks[0], "text") else str(question)[:100]
            logger.info(f"Agent received question (first 100 chars): {log_question}...")
        else:
            logger.info(f"Agent received question (first 100 chars): {question[:100]}...")

        handler = self.agent_workflow.run(user_msg=question)

        current_agent = None
        async for event in handler.stream_events():
            if (
                    hasattr(event, "current_agent_name")
                    and event.current_agent_name != current_agent
            ):
                current_agent = event.current_agent_name
                logger.info(f"{'=' * 50}")
                logger.info(f"🤖 Agent: {current_agent}")
                logger.info(f"{'=' * 50}\n")

            # Optional detailed logging (uncomment if needed)
            # from llama_index.core.agent.runner.base import AgentStream, AgentInput
            # if isinstance(event, AgentStream):
            #     if event.delta:
            #         logger.debug(f"STREAM: {event.delta}") # Use debug level
            # elif isinstance(event, AgentInput):
            #     logger.debug(f"📥 Input: {event.input}") # Use debug level
            elif isinstance(event, AgentOutput):
                if event.response and hasattr(event.response, 'content') and event.response.content:
                    logger.info(f"📤 Output: {event.response.content}")
                if event.tool_calls:
                    logger.info(
                        f"🛠️  Planning to use tools: {[call.tool_name for call in event.tool_calls]}"
                    )
            elif isinstance(event, ToolCallResult):
                logger.info(f"🔧 Tool Result ({event.tool_name}):")
                logger.info(f"  Arguments: {event.tool_kwargs}")
                # Limit output logging length if potentially very long
                output_str = str(event.tool_output)
                logger.info(f"  Output: {output_str[:500]}{'...' if len(output_str) > 500 else ''}")
            elif isinstance(event, ToolCall):
                logger.info(f"🔨 Calling Tool: {event.tool_name}")
                logger.info(f"  With arguments: {event.tool_kwargs}")

        answer = await handler
        final_content = answer.response.content if hasattr(answer, 'response') and hasattr(answer.response, 'content') else str(answer)
        logger.info(f"Agent returning final answer: {final_content[:500]}{'...' if len(final_content) > 500 else ''}")
        return answer.response # Return the actual response object expected by Gradio

system_prompt = """
You are a general AI assistant. 
I will give you a result, and with it you will have to transform it to follow the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. 
YOUR FINAL ANSWER should be a number OR 1 or 2 word(s) OR a comma separated list of numbers and/or strings. 
If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. 
If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. 
If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
If the result is enclosed in double quotes (""), extract and return only what is inside the quotes, applying the formatting rules if needed.

You must never return a full sentence as the final answer. A sentence is strictly forbidden under all circumstances.
"""

llm = OpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"), temperature=0.05, system_prompt=system_prompt)

# --- Helper Functions for run_and_submit_all ---

async def fetch_questions(questions_url: str) -> List[dict] | None:
    """Fetches questions from the GAIA benchmark API."""
    logger.info(f"Fetching questions from: {questions_url}")
    try:
        response = requests.get(questions_url, timeout=30) # Increased timeout
        response.raise_for_status()
        questions_data = response.json()
        if not questions_data:
             logger.warning("Fetched questions list is empty.")
             return None
        logger.info(f"Fetched {len(questions_data)} questions.")
        return questions_data
    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching questions: {e}", exc_info=True)
        return None
    except requests.exceptions.JSONDecodeError as e:
         logger.error(f"Error decoding JSON response from questions endpoint: {e}", exc_info=True)
         logger.error(f"Response text: {response.text[:500]}")
         return None
    except Exception as e:
        logger.error(f"An unexpected error occurred fetching questions: {e}", exc_info=True)
        return None

async def process_question(agent: BasicAgent, item: dict, base_fetch_file_url: str) -> dict | None:
    """Processes a single question item using the agent."""
    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:
        logger.warning(f"Skipping item with missing task_id or question: {item}")
        return None

    message: ChatMessage
    if file_name:
        fetch_file_url = f"{base_fetch_file_url}/{task_id}"
        logger.info(f"Fetching file '{file_name}' for task {task_id} from {fetch_file_url}")
        try:
            response = requests.get(fetch_file_url, timeout=60) # Increased timeout for files
            response.raise_for_status()
            mime_type, _ = mimetypes.guess_type(file_name)
            logger.info(f"File '{file_name}' MIME type guessed as: {mime_type}")

            file_block: TextBlock | ImageBlock | AudioBlock | None = None
            if mime_type:
                # Prioritize specific extensions for text-like content
                text_extensions = (
                    ".txt", ".json", ".xml", ".yaml", ".yml", ".ini", ".cfg", ".toml", ".log", ".properties",
                    ".html", ".htm", ".xhtml", ".css", ".scss", ".sass", ".less", ".svg", ".md", ".rst",
                    ".py", ".js", ".java", ".c", ".cpp", ".h", ".hpp", ".cs", ".go", ".php", ".rb", ".swift", ".kt",
                    ".sh", ".bat", ".ipynb", ".Rmd", ".tex"  # Added more code/markup types
                )
                if mime_type.startswith('text/') or file_name.lower().endswith(text_extensions):
                    try:
                        file_content = response.content.decode('utf-8') # Try UTF-8 first
                    except UnicodeDecodeError:
                        try:
                            file_content = response.content.decode('latin-1') # Fallback
                            logger.warning(f"Decoded file {file_name} using latin-1 fallback.")
                        except Exception as decode_err:
                            logger.error(f"Could not decode file {file_name}: {decode_err}")
                            file_content = f"[Error: Could not decode file content for {file_name}]"
                    file_block = TextBlock(block_type="text", text=f"[File: {file_name}]\n[Content]:\n{file_content}")
                elif mime_type.startswith('image/'):
                    # Pass image content directly for multi-modal models
                    file_block = ImageBlock(url=fetch_file_url, image=response.content)
                elif mime_type.startswith('audio/'):
                    # Pass audio content directly
                    audio_text = transcribe_audio(response.content)
                    file_block = TextBlock(text=f"[Transcribed Audio: {audio_text}]")
                elif mime_type == 'application/pdf':
                    # PDF: Pass a text block indicating the URL for agents to handle
                    logger.info(f"PDF file detected: {file_name}. Passing reference URL.")
                    file_block = TextBlock(text=f"[Reference PDF file available at: {fetch_file_url}]")
                elif file_name.lower().endswith((".xlsx", ".xls", ".csv")):
                    logger.info(f"Data file detected: {file_name}. Passing reference URL.")
                    file_block = TextBlock(text=f"[Reference Data file available at: {fetch_file_url}]")

                # Add handling for other types like video if needed
                # elif mime_type.startswith('video/'):
                #     logger.info(f"Video file detected: {file_name}. Passing reference URL.")
                #     file_block = TextBlock(text=f"[Reference Video file available at: {fetch_file_url}]")

            if file_block:
                 blocks = [TextBlock(text=question_text), file_block]
                 message = ChatMessage(role="user", blocks=blocks)
            else:
                 logger.warning(f"File type for '{file_name}' (MIME: {mime_type}) not directly supported for block creation or no block created (e.g., unsupported). Passing text question only.")
                 message = ChatMessage(role="user", blocks=[TextBlock(text=question_text)])

        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching file for task {task_id}: {e}", exc_info=True)
            return {"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: Failed to fetch file {file_name} - {e}"}
        except Exception as e:
             logger.error(f"Error processing file for task {task_id}: {e}", exc_info=True)
             return {"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: Failed to process file {file_name} - {e}"}
    else:
        # No file associated with the question
        message = ChatMessage(role="user", blocks=[TextBlock(text=question_text)])

    # Run the agent on the prepared message
    try:
        logger.info(f"Running agent on task {task_id}...")
        submitted_answer_response = await agent(message)
        # Extract content safely
        submitted_answer = submitted_answer_response.content if hasattr(submitted_answer_response, 'content') else str(submitted_answer_response)

        prompt = f"""
        You are a general AI assistant. 
        I will give you a result, and with it you will have to transform it to follow the following template: [YOUR FINAL ANSWER]. 
        YOUR FINAL ANSWER should be a number OR 1 or 2 word(s) OR a comma separated list of numbers and/or strings. 
        If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. 
        If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. 
        If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
        If the result is enclosed in double quotes (""), extract and return only what is inside the quotes, applying the formatting rules if needed.
        
        You must never return a full sentence as the final answer. A sentence is strictly forbidden under all circumstances.
        QUESTION: {question_text}
        ANSWER: {submitted_answer}
        INSTRUCTIONS: Based on the provided question and answer, generate a final answer that is clear, concise, and directly addresses the question.
        [YOUR FINAL ANSWER]
        """

        final_answer = llm.complete(prompt)

        logger.info(f"👍 Agent submitted answer for task {task_id}: {final_answer.text[:200]}{'...' if len(final_answer.text) > 200 else ''}")
        return {"Task ID": task_id, "Question": question_text, "Submitted Answer": final_answer.text}
    except Exception as e:
         logger.error(f"Error running agent on task {task_id}: {e}", exc_info=True)
         return {"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}

async def run_and_submit_all( profile: gr.OAuthProfile | None):
    """
    Fetches all questions, runs the BasicAgent on them, submits all answers,
    and displays the results.
    """
    # --- 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

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

    results_log = []
    answers_payload = []

    try:
        agent = BasicAgent(AGENT_WORKFLOW)
    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)

    questions_data = await fetch_questions(questions_url)
    if not questions_data:
        return "Failed to fetch questions.", None

    # 3. Process Questions
    # questions_data = [questions_data[3]]
    for item in questions_data:
        answers = await process_question(agent, item, fetch_file_url)
        results_log.append(answers)
        answers_payload.append({"task_id": answers["Task ID"], "submitted_answer": answers["Submitted Answer"]})

    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. Submitting {len(answers_payload)} answers for user '{username}'..."
    print(status_update)

    # 5. Submit
    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")

    try:
        response = requests.post(submit_url, json=submission_data, timeout=120) # Increased timeout
        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.')}"
        )
        logger.info("Submission successful.")
        results_df = pd.DataFrame(results_log)
        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}"
        logger.error(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except requests.exceptions.Timeout:
        status_message = "Submission Failed: The request timed out."
        logger.error(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except requests.exceptions.RequestException as e:
        status_message = f"Submission Failed: Network error - {e}"
        logger.error(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except Exception as e:
        status_message = f"Submission Failed: An unexpected error occurred during submission - {e}"
        logger.error(status_message, exc_info=True)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df

# --- Gradio Interface ---
def create_gradio_interface():
    """Creates and returns the Gradio interface."""
    # --- 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.  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)
        # Removed max_rows=10 from DataFrame constructor
        results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)

        run_button.click(
            fn=run_and_submit_all,
            outputs=[status_output, results_table]
        )

    return demo

# --- Main Execution ---
if __name__ == "__main__":
    if not AGENT_WORKFLOW:
        print("ERROR: Agent Workflow failed to initialize. Cannot start Gradio app.")
        print("Please check logs for initialization errors (e.g., missing API keys, import issues).")
    else:
        gradio_app = create_gradio_interface()
        # Launch Gradio app
        # Share=True creates a public link (use with caution)
        # Set server_name="0.0.0.0" to allow access from network
        gradio_app.launch(server_name="0.0.0.0", server_port=7860)