app.py

import os
import gradio as gr
import requests
import pandas as pd
import json
import re
import base64
from typing import Optional, Dict, List, Any
import anthropic

# API URL для GAIA
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

class GAIAAgent:
    def __init__(self):
        print("Initializing GAIA Agent powered by Claude...")
        # Получение API-ключа Claude из переменных окружения
        self.claude_key = os.environ.get("ANTHROPIC_API_KEY")
        if not self.claude_key:
            raise ValueError("ANTHROPIC_API_KEY not found in environment variables")
        
        # Инициализация клиента Claude
        self.client = anthropic.Anthropic(api_key=self.claude_key)
        
        # API URL для GAIA
        self.api_url = DEFAULT_API_URL
        
        # Словарь для кеширования результатов поиска и ответов
        self.search_cache = {}
        self.file_cache = {}
        
        # Системный промпт для Claude
        self.system_prompt = """
        You are an AI assistant specially designed to answer questions from the GAIA benchmark with exceptional accuracy. 
        The GAIA benchmark evaluates AI's ability to perform real-world tasks that require reasoning, web browsing, and tool use.
        
        Your goal is to provide the EXACT answer in the format requested by each question. GAIA uses exact matching for evaluation.
        
        Guidelines for GAIA answers:
        1. Provide ONLY the final answer, with NO explanations, reasoning, or additional text
        2. Format is critical - follow the instructions in the question precisely
        3. For comma-separated lists, provide "item1, item2, item3" with no quotes or extra punctuation
        4. For numeric answers, provide just the number without units unless specifically requested
        5. Maintain exact capitalization and spacing as requested in the question
        6. If asked to order items, follow the requested ordering precisely
        
        Examples of correct formatting:
        - If asked for fruits in alphabetical order: "apples, bananas, oranges"
        - If asked for a single word: "photosynthesis"
        - If asked for a number: "42"
        - If asked for a date in MM/DD/YY format: "05/04/25"
        
        Remember, your score depends on exact matching against the reference answer.
        """

    def search_web(self, query: str) -> str:
        """Improved web search function with caching"""
        if query in self.search_cache:
            print(f"Using cached search results for: {query}")
            return self.search_cache[query]
        
        print(f"Performing web search for: {query}")
        try:
            # DuckDuckGo Instant Answer API
            response = requests.get(
                "https://api.duckduckgo.com/",
                params={"q": query, "format": "json"},
                timeout=10
            )
            data = response.json()
            
            # Собираем результаты из разных полей
            results = []
            if data.get("AbstractText"):
                results.append(f"Abstract: {data['AbstractText']}")
            if data.get("RelatedTopics"):
                topics = data.get("RelatedTopics", [])
                for i, topic in enumerate(topics[:5]):  # Ограничиваем 5 результатами
                    if isinstance(topic, dict) and topic.get("Text"):
                        results.append(f"Related Topic {i+1}: {topic['Text']}")
            
            result_text = "\n\n".join(results) if results else "No results found"
            
            # Вторичный поиск с использованием серпапи.com (если бы у нас был ключ API)
            # В реальном приложении здесь можно было бы использовать другой поисковый API
            
            # Кешируем и возвращаем результаты
            self.search_cache[query] = result_text
            return result_text
        except Exception as e:
            print(f"Web search error: {e}")
            return f"Web search failed: {str(e)}"

    def fetch_file(self, task_id: str) -> Optional[Dict[str, Any]]:
        """Fetches and processes a file associated with a task"""
        if task_id in self.file_cache:
            print(f"Using cached file for task: {task_id}")
            return self.file_cache[task_id]
            
        print(f"Fetching file for task: {task_id}")
        try:
            response = requests.get(f"{self.api_url}/files/{task_id}", timeout=15)
            
            if response.status_code == 200:
                file_content = response.content
                file_info = {
                    "content": file_content,
                    "content_type": response.headers.get("Content-Type", ""),
                    "size": len(file_content)
                }
                
                # Определяем тип файла и обрабатываем соответственно
                content_type = file_info["content_type"].lower()
                
                if "image" in content_type:
                    # Преобразуем изображение в base64 для Claude
                    file_info["base64"] = base64.b64encode(file_content).decode('utf-8')
                    file_info["type"] = "image"
                    print(f"Processed image file ({file_info['size']} bytes)")
                elif "pdf" in content_type:
                    # Для PDF мы можем только сказать, что это PDF
                    file_info["type"] = "pdf"
                    print(f"Detected PDF file ({file_info['size']} bytes)")
                elif "text" in content_type or "json" in content_type or "csv" in content_type:
                    # Для текстовых файлов пытаемся декодировать
                    try:
                        file_info["text"] = file_content.decode('utf-8')
                        file_info["type"] = "text"
                        print(f"Processed text file ({file_info['size']} bytes)")
                    except UnicodeDecodeError:
                        file_info["type"] = "binary"
                        print(f"Could not decode text file ({file_info['size']} bytes)")
                else:
                    file_info["type"] = "binary"
                    print(f"Detected binary file ({file_info['size']} bytes, {content_type})")
                
                # Кешируем файл
                self.file_cache[task_id] = file_info
                return file_info
            else:
                print(f"Failed to fetch file, status code: {response.status_code}")
                print(f"Response: {response.text[:1000]}")
                return None
        except Exception as e:
            print(f"Error fetching file: {e}")
            return None

    def extract_answer(self, response_text: str) -> str:
        """Extract just the final answer from Claude's response"""
        # Удаляем очевидные вводные фразы
        cleaned = re.sub(r'^(final answer|the answer is|answer|Here\'s the answer|response):?\s*', '', response_text, flags=re.IGNORECASE)
        
        # Удаляем объяснения в конце
        cleaned = re.sub(r'\n.*?explain.*?$', '', cleaned, flags=re.IGNORECASE | re.DOTALL)
        
        # Проверяем на многострочный ответ и берем только первую строку, если она содержит ответ
        lines = cleaned.strip().split('\n')
        if len(lines) > 1:
            first_line = lines[0].strip()
            # Если первая строка выглядит как полный ответ, возвращаем только её
            if len(first_line) > 5 and not first_line.startswith('I ') and not first_line.startswith('The '):
                return first_line
        
        # Вычищаем кавычки в начале и конце
        cleaned = cleaned.strip()
        if cleaned.startswith('"') and cleaned.endswith('"'):
            cleaned = cleaned[1:-1]
            
        return cleaned.strip()

    def process_question(self, question: str, task_id: str = None) -> Dict[str, Any]:
        """Processes a question to extract relevant information and prepare for Claude"""
        question_info = {
            "original": question,
            "task_id": task_id,
            "has_file": False,
            "file_info": None,
            "contains_math": bool(re.search(r'calculate|compute|sum|average|mean|median|formula|equation', question, re.IGNORECASE)),
            "requires_list": bool(re.search(r'list|order|sequence|rank|items|elements|values', question, re.IGNORECASE)),
            "format_requirements": None
        }
        
        # Извлекаем формат, если указан
        format_match = re.search(r'(format|in the format|formatted as|as a|in) ([^\.]+)', question, re.IGNORECASE)
        if format_match:
            question_info["format_requirements"] = format_match.group(2).strip()
            
        # Проверяем наличие файла
        if task_id and self.fetch_file(task_id):
            question_info["has_file"] = True
            question_info["file_info"] = self.fetch_file(task_id)
        
        return question_info

    def __call__(self, question: str, task_id: str = None) -> str:
        """Main method to process a question and return an answer"""
        if task_id is None:
            # Пытаемся извлечь task_id из вопроса, если он там есть
            match = re.search(r'task[\s_-]?id:?\s*(\w+)', question, re.IGNORECASE)
            if match:
                task_id = match.group(1)
                
        print(f"Processing question for task_id: {task_id}")
        print(f"Question: {question[:100]}...")
        
        # Обработка вопроса
        question_info = self.process_question(question, task_id)
        
        try:
            # Подготовка сообщения для Claude
            messages = []
            
            # Подготовка контента сообщения
            user_content = [{
                "type": "text",
                "text": f"""
Question from GAIA benchmark: {question}

Remember:
1. Provide ONLY the final answer
2. Format exactly as requested
3. No explanations or reasoning
"""
            }]
            
            # Добавляем результаты поиска, если нужно
            web_results = self.search_web(question)
            if web_results:
                user_content.append({
                    "type": "text",
                    "text": f"""
Web search results related to this question:

{web_results}
"""
                })
            
            # Добавляем файл, если он есть
            if question_info["has_file"] and question_info["file_info"]:
                file_info = question_info["file_info"]
                
                if file_info["type"] == "image":
                    # Добавляем изображение для Claude
                    user_content.append({
                        "type": "image",
                        "source": {
                            "type": "base64", 
                            "media_type": file_info["content_type"],
                            "data": file_info["base64"]
                        }
                    })
                    
                    user_content.append({
                        "type": "text",
                        "text": "The above image is part of the question. Please analyze it carefully."
                    })
                elif file_info["type"] == "text" and "text" in file_info:
                    # Для текстовых файлов добавляем содержимое
                    user_content.append({
                        "type": "text",
                        "text": f"""
The question includes a text file with the following content:

{file_info["text"][:4000]}  # ограничиваем, чтобы не превысить лимиты токенов
"""
                    })
            
            # Добавляем форматирование, если указано
            if question_info["format_requirements"]:
                user_content.append({
                    "type": "text",
                    "text": f"""
Important format requirement: {question_info["format_requirements"]}
Make sure your answer follows this format EXACTLY.
"""
                })
            
            messages.append({
                "role": "user",
                "content": user_content
            })
            
            # Запрос к Claude
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                system=self.system_prompt,
                messages=messages,
                temperature=0.1,  # Низкая температура для точных ответов
                max_tokens=4096
            )
            
            # Получаем ответ
            raw_answer = response.content[0].text.strip()
            
            # Вычищаем ответ от лишнего
            clean_answer = self.extract_answer(raw_answer)
            
            print(f"Raw answer: {raw_answer}")
            print(f"Clean answer: {clean_answer}")
            
            return clean_answer
        except Exception as e:
            print(f"Error in agent: {e}")
            import traceback
            traceback.print_exc()
            return f"Error processing question: {str(e)}"


# Используем наш агент как BasicAgent для совместимости с остальным кодом
class BasicAgent(GAIAAgent):
    pass


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"

    # 1. Instantiate Agent ( modify this part to create your agent)
    try:
        agent = BasicAgent()
    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

    # 3. Run your Agent
    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)

    # 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=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.")
        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}"
        print(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."
        print(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}"
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df
    except Exception as e:
        status_message = f"An unexpected error occurred during submission: {e}"
        print(status_message)
        results_df = pd.DataFrame(results_log)
        return status_message, results_df


# --- Build Gradio Interface using Blocks ---
with gr.Blocks() as demo:
    gr.Markdown("# GAIA Benchmark Agent Evaluation")
    gr.Markdown(
        """
        **Instructions:**
        1. Log in to your Hugging Face account using the button below. This uses your HF username for submission.
        2. Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
        
        This agent uses Claude 3.5 Sonnet to solve GAIA benchmark tasks.
        """
    )

    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)
    # 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 GAIA Agent Evaluation...")
    demo.launch(debug=True, share=False)