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
import json
from huggingface_hub import InferenceClient

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

# --- Intelligent Agent with Conditional Search ---
class IntelligentAgent:
    def __init__(self, debug: bool = False, model_name: str = "meta-llama/Llama-3.1-8B-Instruct"):
        self.search = DuckDuckGoSearchTool()
        self.client = InferenceClient(model=model_name)
        self.debug = debug
        if self.debug:
            print(f"IntelligentAgent initialized with model: {model_name}")

    def _should_search(self, question: str) -> bool:
        """
        Use LLM to determine if search is needed for the question.
        Returns True if search is recommended, False otherwise.
        """
        decision_prompt = f"""You are an AI assistant that decides whether a web search is needed to answer questions accurately.

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

Question: "{question}"

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"
"""

        try:
            response = self.client.text_generation(
                decision_prompt,
                max_new_tokens=50,
                temperature=0.1,
                do_sample=False
            )
            
            decision = response.strip().upper()
            should_search = decision.startswith("SEARCH")
            
            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) -> str:
        """
        Generate answer using LLM without search.
        """
        answer_prompt = f"""You are a helpful AI assistant. Answer the following question based on your knowledge. Be accurate, concise, and helpful. If you're not certain about something, acknowledge the uncertainty.

Question: {question}

Answer:"""

        try:
            response = self.client.text_generation(
                answer_prompt,
                max_new_tokens=500,
                temperature=0.3,
                do_sample=True
            )
            return response.strip()
            
        except Exception as e:
            return f"Sorry, I encountered an error generating the response: {e}"

    def _answer_with_search(self, question: str) -> str:
        """
        Generate answer using search results and LLM.
        """
        try:
            # Perform search
            search_results = self.search(question)
            
            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)

            # Format search results
            formatted_results = []
            for i, result in enumerate(search_results[:3]):  # Use top 3 results
                title = result.get("title", "No title")
                snippet = result.get("snippet", "").strip()
                link = result.get("link", "")
                
                formatted_results.append(f"Result {i+1}:\nTitle: {title}\nContent: {snippet}\nSource: {link}")

            search_context = "\n\n".join(formatted_results)

            # Generate answer using search context
            answer_prompt = f"""You are a helpful AI assistant. Use the provided search results to answer the question accurately. Synthesize information from multiple sources when relevant, and cite sources when appropriate.

Question: {question}

Search Results:
{search_context}

Based on the search results above, provide a comprehensive answer to the question. If the search results don't fully answer the question, you can supplement with your general knowledge but clearly indicate what comes from the search results vs. your knowledge.

Answer:"""

            try:
                response = self.client.text_generation(
                    answer_prompt,
                    max_new_tokens=600,
                    temperature=0.3,
                    do_sample=True
                )
                return response.strip()
                
            except Exception as e:
                # Fallback to simple search result formatting
                top_result = search_results[0]
                title = top_result.get("title", "No title")
                snippet = top_result.get("snippet", "").strip()
                link = top_result.get("link", "")
                
                return f"**{title}**\n\n{snippet}\n\nSource: {link}"

        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)

    def __call__(self, question: str) -> str:
        """
        Main entry point - decide whether to search and generate appropriate response.
        """
        if self.debug:
            print(f"Agent received question: {question}")

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

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

        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.", None
    
    if not cached_questions:
        return "No questions available. Please fetch questions first.", None
    
    # Map model choice to actual model name
    model_map = {
        "Llama 3.1 8B": "meta-llama/Llama-3.1-8B-Instruct",
        "Llama 3.1 70B": "meta-llama/Llama-3.1-70B-Instruct",
        "Mistral 7B": "mistralai/Mistral-7B-Instruct-v0.3",
        "CodeLlama 7B": "codellama/CodeLlama-7b-Instruct-hf"
    }
    
    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 below.", None

def get_generation_progre