app.py

import os
import gradio as gr
import requests
import inspect
import time
import pandas as pd
from smolagents import DuckDuckGoSearchTool
import threading
from typing import Dict, List, Optional, Tuple, Union
import json
from huggingface_hub import InferenceClient
import base64
from PIL import Image
import io

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

# --- Global Cache for Answers ---
cached_answers = {}
cached_questions = []
processing_status = {"is_processing": False, "progress": 0, "total": 0}

# --- Image Processing Tool ---
class ImageAnalysisTool:
    def __init__(self, model_name: str = "microsoft/Florence-2-large"):
        self.client = InferenceClient(model=model_name)
        
    def analyze_image(self, image_path: str, prompt: str = "Describe this image in detail") -> str:
        """
        Analyze an image and return a description.
        """
        try:
            # Open and process the image
            with open(image_path, "rb") as f:
                image_bytes = f.read()
            
            # Use the vision model to analyze the image
            response = self.client.image_to_text(
                image=image_bytes,
                model="microsoft/Florence-2-large"
            )
            
            return response.get("generated_text", "Could not analyze image")
            
        except Exception as e:
            try:
                # Fallback: use a different vision model
                response = self.client.image_to_text(
                    image=image_bytes,
                    model="Salesforce/blip-image-captioning-large"
                )
                return response.get("generated_text", f"Image analysis error: {e}")
            except:
                return f"Image analysis failed: {e}"

    def extract_text_from_image(self, image_path: str) -> str:
        """
        Extract text from an image using OCR.
        """
        try:
            with open(image_path, "rb") as f:
                image_bytes = f.read()
            
            # Use an OCR model
            response = self.client.image_to_text(
                image=image_bytes,
                model="microsoft/trocr-base-printed"
            )
            
            return response.get("generated_text", "No text found in image")
            
        except Exception as e:
            return f"OCR failed: {e}"

# --- Audio Processing Tool ---
class AudioTranscriptionTool:
    def __init__(self, model_name: str = "openai/whisper-large-v3"):
        self.client = InferenceClient(model=model_name)
        
    def transcribe_audio(self, audio_path: str) -> str:
        """
        Transcribe audio file to text.
        """
        try:
            with open(audio_path, "rb") as f:
                audio_bytes = f.read()
            
            # Use Whisper for transcription
            response = self.client.automatic_speech_recognition(
                audio=audio_bytes
            )
            
            return response.get("text", "Could not transcribe audio")
            
        except Exception as e:
            try:
                # Fallback to a different ASR model
                response = self.client.automatic_speech_recognition(
                    audio=audio_bytes,
                    model="facebook/wav2vec2-large-960h-lv60-self"
                )
                return response.get("text", f"Audio transcription error: {e}")
            except:
                return f"Audio transcription failed: {e}"

# --- Enhanced Intelligent Agent with Media Processing ---
class IntelligentAgent:
    def __init__(self, debug: bool = True, model_name: str = "meta-llama/Llama-3.1-8B-Instruct"):
        self.search = DuckDuckGoSearchTool()
        self.client = InferenceClient(model=model_name, provider="sambanova")
        self.image_tool = ImageAnalysisTool()
        self.audio_tool = AudioTranscriptionTool()
        self.debug = debug
        if self.debug:
            print(f"IntelligentAgent initialized with model: {model_name}")

    def _chat_completion(self, prompt: str, max_tokens: int = 500, temperature: float = 0.3) -> str:
        """
        Use chat completion instead of text generation to avoid provider compatibility issues.
        """
        try:
            messages = [{"role": "user", "content": prompt}]
            
            # Try chat completion first
            try:
                response = self.client.chat_completion(
                    messages=messages,
                    max_tokens=max_tokens,
                    temperature=temperature
                )
                return response.choices[0].message.content.strip()
            except Exception as chat_error:
                if self.debug:
                    print(f"Chat completion failed: {chat_error}, trying text generation...")
                
                # Fallback to text generation
                response = self.client.conversational(
                    prompt,
                    max_new_tokens=max_tokens,
                    temperature=temperature,
                    do_sample=temperature > 0
                )
                return response.strip()
                
        except Exception as e:
            if self.debug:
                print(f"Both chat completion and text generation failed: {e}")
            raise e

    def _process_media_files(self, image_files: List[str] = None, audio_files: List[str] = None) -> str:
        """
        Process attached media files and return their content as text.
        """
        media_content = []
        
        # Process images
        if image_files:
            for image_file in image_files:
                if image_file and os.path.exists(image_file):
                    try:
                        # Analyze the image
                        image_description = self.image_tool.analyze_image(image_file)
                        media_content.append(f"Image Analysis: {image_description}")
                        
                        # Try to extract text from image
                        extracted_text = self.image_tool.extract_text_from_image(image_file)
                        if extracted_text and "No text found" not in extracted_text:
                            media_content.append(f"Text from Image: {extracted_text}")
                            
                    except Exception as e:
                        media_content.append(f"Error processing image {image_file}: {e}")
        
        # Process audio files
        if audio_files:
            for audio_file in audio_files:
                if audio_file and os.path.exists(audio_file):
                    try:
                        # Transcribe the audio
                        transcription = self.audio_tool.transcribe_audio(audio_file)
                        media_content.append(f"Audio Transcription: {transcription}")
                        
                    except Exception as e:
                        media_content.append(f"Error processing audio {audio_file}: {e}")
        
        return "\n\n".join(media_content) if media_content else ""

    def _should_search(self, question: str, media_context: str = "") -> bool:
        """
        Use LLM to determine if search is needed for the question, considering media context.
        Returns True if search is recommended, False otherwise.
        """
        decision_prompt = f"""Analyze this question and decide if it requires real-time information, recent data, or specific facts that might not be in your training data.

SEARCH IS NEEDED for:
- Current events, news, recent developments
- Real-time data (weather, stock prices, sports scores)
- Specific factual information that changes frequently
- Recent product releases, company information
- Current status of people, organizations, or projects
- Location-specific current information

SEARCH IS NOT NEEDED for:
- General knowledge questions
- Mathematical calculations
- Programming concepts and syntax
- Historical facts (older than 1 year)
- Definitions of well-established concepts
- How-to instructions for common tasks
- Creative writing or opinion-based responses
- Questions that can be answered from attached media content

Question: "{question}"

{f"Media Context Available: {media_context[:500]}..." if media_context else "No media context available."}

Respond with only "SEARCH" or "NO_SEARCH" followed by a brief reason (max 20 words).

Example responses:
- "SEARCH - Current weather data needed"
- "NO_SEARCH - Mathematical concept, general knowledge sufficient"
- "NO_SEARCH - Can be answered from attached image content"
"""

        try:
            response = self._chat_completion(decision_prompt, max_tokens=50, temperature=0.1)
            
            decision = response.strip().upper()
            should_search = decision.startswith("SEARCH")
            time.sleep(5)
            
            if self.debug:
                print(f"Decision for '{question}': {decision}")
                
            return should_search
            
        except Exception as e:
            if self.debug:
                print(f"Error in search decision: {e}, defaulting to search")
            # Default to search if decision fails
            return True

    def _answer_with_llm(self, question: str, media_context: str = "") -> str:
        """
        Generate answer using LLM without search, considering media context.
        """
        context_section = f"\n\nMedia Context:\n{media_context}" if media_context else ""
        
        answer_prompt = f"""You are a general AI assistant. I will ask you a question. YOUR ANSWER should be a number OR as few words as possible 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.

{context_section}

Question: {question}

Answer:"""

        try:
            response = self._chat_completion(answer_prompt, max_tokens=500, temperature=0.3)
            return response
            
        except Exception as e:
            return f"Sorry, I encountered an error generating the response: {e}"

    def _answer_with_search(self, question: str, media_context: str = "") -> str:
        """
        Generate answer using search results and LLM, considering media context.
        """
        try:
            # Perform search
            time.sleep(10)
            search_results = self.search(question)
            
            if self.debug:
                print(f"Search results type: {type(search_results)}")
            
            if not search_results:
                return "No search results found. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, media_context)

            # Format search results - handle different result formats
            if isinstance(search_results, str):
                search_context = search_results
            else:
                # Handle list of results
                formatted_results = []
                for i, result in enumerate(search_results[:3]):  # Use top 3 results
                    if isinstance(result, dict):
                        title = result.get("title", "No title")
                        snippet = result.get("snippet", "").strip()
                        link = result.get("link", "")
                        formatted_results.append(f"Title: {title}\nContent: {snippet}\nSource: {link}")
                    elif isinstance(result, str):
                        formatted_results.append(result)
                    else:
                        formatted_results.append(str(result))
                
                search_context = "\n\n".join(formatted_results)

            # Generate answer using search context and media context
            context_section = f"\n\nMedia Context:\n{media_context}" if media_context else ""
            
            answer_prompt = f"""You are a general AI assistant. I will ask you a question. Based on the search results below, provide an answer to the question. If the search results don't fully answer the question, you can supplement with your general knowledge.
            Your ANSWER should be a number OR as few words as possible 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.

Question: {question}

Search Results:
{search_context}

{context_section}

Answer:"""

            try:
                response = self._chat_completion(answer_prompt, max_tokens=600, temperature=0.3)
                return response
                
            except Exception as e:
                if self.debug:
                    print(f"LLM generation error: {e}")
                # Fallback to simple search result formatting
                if search_results:
                    if isinstance(search_results, str):
                        return search_results
                    elif isinstance(search_results, list) and len(search_results) > 0:
                        first_result = search_results[0]
                        if isinstance(first_result, dict):
                            title = first_result.get("title", "Search Result")
                            snippet = first_result.get("snippet", "").strip()
                            link = first_result.get("link", "")
                            return f"**{title}**\n\n{snippet}\n\n{f'Source: {link}' if link else ''}"
                        else:
                            return str(first_result)
                    else:
                        return str(search_results)
                else:
                    return "Search completed but no usable results found."

        except Exception as e:
            return f"Search failed: {e}. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, media_context)

    def __call__(self, question: str, image_files: List[str] = None, audio_files: List[str] = None) -> str:
        """
        Main entry point - process media files, decide whether to search, and generate appropriate response.
        """
        if self.debug:
            print(f"Agent received question: {question}")
            print(f"Image files: {image_files}")
            print(f"Audio files: {audio_files}")

        # Early validation
        if not question or not question.strip():
            return "Please provide a valid question."

        try:
            # Process media files first
            media_context = self._process_media_files(image_files, audio_files)
            
            if self.debug and media_context:
                print(f"Media context: {media_context[:200]}...")

            # Decide whether to search
            if self._should_search(question, media_context):
                if self.debug:
                    print("Using search-based approach")
                answer = self._answer_with_search(question, media_context)
            else:
                if self.debug:
                    print("Using LLM-only approach")
                answer = self._answer_with_llm(question, media_context)

        except Exception as e:
            answer = f"Sorry, I encountered an error: {e}"

        if self.debug:
            print(f"Agent returning answer: {answer[:100]}...")
        
        return answer

def fetch_questions() -> Tuple[str, Optional[pd.DataFrame]]:
    """
    Fetch questions from the API and cache them.
    """
    global cached_questions
    
    api_url = DEFAULT_API_URL
    questions_url = f"{api_url}/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:
            return "Fetched questions list is empty.", None
            
        cached_questions = questions_data
        
        # Create DataFrame for display
        display_data = []
        for item in questions_data:
            display_data.append({
                "Task ID": item.get("task_id", "Unknown"),
                "Question": item.get("question", "")
            })
        
        df = pd.DataFrame(display_data)
        status_msg = f"Successfully fetched {len(questions_data)} questions. Ready to generate answers."
        
        return status_msg, df
        
    except requests.exceptions.RequestException as e:
        return f"Error fetching questions: {e}", None
    except Exception as e:
        return f"An unexpected error occurred: {e}", None

def generate_answers_async(model_name: str = "meta-llama/Llama-3.1-8B-Instruct", progress_callback=None):
    """
    Generate answers for all cached questions asynchronously using the intelligent agent.
    """
    global cached_answers, processing_status
    
    if not cached_questions:
        return "No questions available. Please fetch questions first."
    
    processing_status["is_processing"] = True
    processing_status["progress"] = 0
    processing_status["total"] = len(cached_questions)
    
    try:
        agent = IntelligentAgent(debug=True, model_name=model_name)
        cached_answers = {}
        
        for i, item in enumerate(cached_questions):
            if not processing_status["is_processing"]:  # Check if cancelled
                break
                
            task_id = item.get("task_id")
            question_text = item.get("question")
            
            if not task_id or question_text is None:
                continue
                
            try:
                answer = agent(question_text)
                cached_answers[task_id] = {
                    "question": question_text,
                    "answer": answer
                }
            except Exception as e:
                cached_answers[task_id] = {
                    "question": question_text,
                    "answer": f"AGENT ERROR: {e}"
                }
            
            processing_status["progress"] = i + 1
            if progress_callback:
                progress_callback(i + 1, len(cached_questions))
                
    except Exception as e:
        print(f"Error in generate_answers_async: {e}")
    finally:
        processing_status["is_processing"] = False

def start_answer_generation(model_choice: str):
    """
    Start the answer generation process in a separate thread.
    """
    if processing_status["is_processing"]:
        return "Answer generation is already in progress."
    
    if not cached_questions:
        return "No questions available. Please fetch questions first."
    
    # Map model choice to actual model name
    model_map = {
        "Llama 3.1 8B": "meta-llama/Llama-3.1-8B-Instruct",
        "Mistral 7B": "mistralai/Mistral-7B-Instruct-v0.3"
    }
    
    selected_model = model_map.get(model_choice, "meta-llama/Llama-3.1-8B-Instruct")
    
    # Start generation in background thread
    thread = threading.Thread(target=generate_answers_async, args=(selected_model,))
    thread.daemon = True
    thread.start()
    
    return f"Answer generation started using {model_choice}. Check progress."

def get_generation_prog