gaia_certification_agent.py

"""
Агент для взаимодействия с официальным API сертификации курса Hugging Face Agents.
Этот код правильно получает вопросы и отправляет ответы в систему сертификации.
"""

import os
import json
import logging
import requests
import time
import gradio as gr
from typing import Dict, List, Any, Optional, Tuple
from agent_implementation import AgentController

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("gaia_certification.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("gaia_certification")

# URL API сертификации Hugging Face
API_BASE_URL = "https://huggingface.co/api/courses/agents"

class GAIACertificationClient:
    """Клиент для взаимодействия с API сертификации курса Hugging Face Agents."""
    
    def __init__(self, username: str, hf_token: Optional[str] = None):
        """
        Инициализация клиента.
        
        Args:
            username: Имя пользователя Hugging Face
            hf_token: Токен Hugging Face для аутентификации
        """
        self.username = username
        self.token = hf_token or os.environ.get("HF_TOKEN")
        
        if not self.token:
            logger.error("HF_TOKEN не найден ни в параметрах, ни в переменных окружения. Авторизация невозможна.")
            print("ОШИБКА: HF_TOKEN не найден. Пожалуйста, укажите токен в параметрах или установите переменную окружения HF_TOKEN.")
        else:
            logger.info(f"Токен HF_TOKEN успешно получен для пользователя {username}")
        
        # Корректный формат заголовка Authorization
        self.headers = {
            "Authorization": f"Bearer {self.token}" if self.token else "",
            "Content-Type": "application/json"
        }
        
        logger.info(f"Инициализирован клиент сертификации для пользователя {username}")
    
    def get_questions(self) -> List[Dict[str, Any]]:
        """
        Получение списка вопросов для сертификации.
        
        Returns:
            Список вопросов
        """
        logger.info("Получение списка вопросов")
        
        if not self.token:
            logger.error("Невозможно получить вопросы: отсутствует токен авторизации")
            return []
        
        try:
            max_retries = 3
            retry_delay = 2
            
            for attempt in range(max_retries):
                try:
                    response = requests.get(
                        f"{API_BASE_URL}/questions",
                        headers=self.headers,
                        timeout=30
                    )
                    
                    logger.info(f"Статус ответа при получении вопросов: {response.status_code}")
                    
                    if response.status_code == 401:
                        logger.error("Ошибка авторизации при получении вопросов. Проверьте валидность токена.")
                        return []
                    
                    if response.status_code == 404:
                        logger.error("API не найден. Проверьте корректность URL.")
                        return []
                    
                    response.raise_for_status()
                    
                    questions = response.json()
                    logger.info(f"Получено {len(questions)} вопросов")
                    return questions
                    
                except requests.exceptions.RequestException as e:
                    if attempt < max_retries - 1:
                        logger.warning(f"Попытка {attempt+1} получения вопросов не удалась: {e}. Повторная попытка через {retry_delay} сек...")
                        time.sleep(retry_delay)
                        retry_delay *= 2  # Экспоненциальная задержка
                    else:
                        logger.error(f"Все попытки получения вопросов не удались: {e}")
                        return []
                
        except Exception as e:
            logger.error(f"Ошибка при получении вопросов: {e}")
            return []
    
    def get_random_question(self) -> Dict[str, Any]:
        """
        Получение случайного вопроса.
        
        Returns:
            Случайный вопрос
        """
        logger.info("Получение случайного вопроса")
        
        if not self.token:
            logger.error("Невозможно получить случайный вопрос: отсутствует токен авторизации")
            return {}
        
        try:
            max_retries = 3
            retry_delay = 2
            
            for attempt in range(max_retries):
                try:
                    response = requests.get(
                        f"{API_BASE_URL}/random-question",
                        headers=self.headers,
                        timeout=30
                    )
                    
                    logger.info(f"Статус ответа при получении случайного вопроса: {response.status_code}")
                    
                    if response.status_code == 401:
                        logger.error("Ошибка авторизации при получении случайного вопроса. Проверьте валидность токена.")
                        return {}
                    
                    if response.status_code == 404:
                        logger.error("API не найден. Проверьте корректность URL.")
                        return {}
                    
                    response.raise_for_status()
                    
                    question = response.json()
                    logger.info(f"Получен вопрос: {question.get('task_id', 'unknown')}")
                    return question
                    
                except requests.exceptions.RequestException as e:
                    if attempt < max_retries - 1:
                        logger.warning(f"Попытка {attempt+1} получения случайного вопроса не удалась: {e}. Повторная попытка через {retry_delay} сек...")
                        time.sleep(retry_delay)
                        retry_delay *= 2
                    else:
                        logger.error(f"Все попытки получения случайного вопроса не удались: {e}")
                        return {}
                
        except Exception as e:
            logger.error(f"Ошибка при получении случайного вопроса: {e}")
            return {}
    
    def get_file(self, task_id: str) -> bytes:
        """
        Получение файла для задания.
        
        Args:
            task_id: Идентификатор задания
            
        Returns:
            Содержимое файла
        """
        logger.info(f"Получение файла для задания {task_id}")
        
        if not self.token:
            logger.error(f"Невозможно получить файл для задания {task_id}: отсутствует токен авторизации")
            return b""
        
        try:
            max_retries = 3
            retry_delay = 2
            
            for attempt in range(max_retries):
                try:
                    response = requests.get(
                        f"{API_BASE_URL}/files/{task_id}",
                        headers=self.headers,
                        timeout=30
                    )
                    
                    logger.info(f"Статус ответа при получении файла: {response.status_code}")
                    
                    if response.status_code == 401:
                        logger.error("Ошибка авторизации при получении файла. Проверьте валидность токена.")
                        return b""
                    
                    if response.status_code == 404:
                        logger.error(f"Файл для задания {task_id} не найден. Проверьте корректность task_id.")
                        return b""
                    
                    response.raise_for_status()
                    
                    logger.info(f"Файл для задания {task_id} успешно получен")
                    return response.content
                    
                except requests.exceptions.RequestException as e:
                    if attempt < max_retries - 1:
                        logger.warning(f"Попытка {attempt+1} получения файла не удалась: {e}. Повторная попытка через {retry_delay} сек...")
                        time.sleep(retry_delay)
                        retry_delay *= 2
                    else:
                        logger.error(f"Все попытки получения файла не удались: {e}")
                        return b""
                
        except Exception as e:
            logger.error(f"Ошибка при получении файла для задания {task_id}: {e}")
            return b""
    
    def submit_answers(self, answers: List[Dict[str, str]], agent_code_url: str) -> Dict[str, Any]:
        """
        Отправка ответов для оценки.
        
        Args:
            answers: Список ответов в формате [{"task_id": "...", "submitted_answer": "..."}]
            agent_code_url: URL кода агента в публичном Space
            
        Returns:
            Результат отправки
        """
        logger.info(f"Отправка {len(answers)} ответов")
        
        if not self.token:
            error_msg = "Невозможно отправить ответы: отсутствует токен авторизации"
            logger.error(error_msg)
            return {"error": error_msg}
        
        # Проверка URL кода агента
        if not agent_code_url:
            error_msg = "URL кода агента не указан"
            logger.error(error_msg)
            return {"error": error_msg}
            
        # Обязательная проверка и коррекция URL для соответствия формату /tree/main
        if "/tree/main" not in agent_code_url:
            # Автоматическое исправление URL
            agent_code_url = agent_code_url.rstrip("/") + "/tree/main"
            logger.info(f"URL кода агента скорректирован: {agent_code_url}")
        
        # Дополнительная проверка формата URL
        if not agent_code_url.startswith("https://huggingface.co/spaces/"):
            logger.warning(f"URL кода агента имеет нестандартный формат: {agent_code_url}")
            # Не исправляем автоматически, но предупреждаем
        
        # Валидация ответов
        for answer in answers:
            if "task_id" not in answer or "submitted_answer" not in answer:
                logger.error(f"Некорректный формат ответа: {answer}")
                return {"error": "Некорректный формат ответов"}
            
            # Дополнительная проверка на пустые ответы
            if not answer.get("submitted_answer"):
                logger.warning(f"Пустой ответ для задания {answer.get('task_id')}")
        
        try:
            # Структура данных строго в соответствии с требованиями API
            payload = {
                "username": self.username,
                "agent_code": agent_code_url,
                "answers": answers
            }
            
            logger.debug(f"Отправляемые данные: {json.dumps(payload, ensure_ascii=False)[:1000]}...")
            
            # Добавление повторных попыток при временных сбоях
            max_retries = 3
            retry_delay = 2
            
            for attempt in range(max_retries):
                try:
                    response = requests.post(
                        f"{API_BASE_URL}/submit",
                        headers=self.headers,
                        json=payload,
                        timeout=30  # Добавление таймаута
                    )
                    
                    # Подробное логирование ответа
                    logger.info(f"Статус ответа: {response.status_code}")
                    
                    if response.status_code == 401:
                        logger.error("Ошибка авторизации. Проверьте валидность токена.")
                        return {"error": "Ошибка авторизации. Проверьте валидность токена."}
                    
                    if response.status_code == 404:
                        logger.error("API не найден. Проверьте корректность URL.")
                        return {"error": "API не найден. Проверьте корректность URL."}
                    
                    response.raise_for_status()
                    
                    result = response.json()
                    logger.info(f"Ответы успешно отправлены. Результат: {result}")
                    return result
                    
                except requests.exceptions.RequestException as e:
                    if attempt < max_retries - 1:
                        logger.warning(f"Попытка {attempt+1} не удалась: {e}. Повторная попытка через {retry_delay} сек...")
                        time.sleep(retry_delay)
                        retry_delay *= 2  # Экспоненциальная задержка
                    else:
                        logger.error(f"Все попытки отправки ответов не удались: {e}")
                        return {"error": f"Ошибка при отправке ответов: {str(e)}"}
            
        except Exception as e:
            logger.error(f"Неожиданная ошибка при отправке ответов: {e}")
            return {"error": str(e)}


class GAIACertificationAgent:
    """Агент для прохождения сертификации GAIA."""
    
    def __init__(self, agent_controller: AgentController, username: str, hf_token: Optional[str] = None):
        """
        Инициализация агента.
        
        Args:
            agent_controller: Контроллер агента для обработки вопросов
            username: Имя пользователя Hugging Face
            hf_token: Токен Hugging Face для аутентификации
        """
        self.agent = agent_controller
        self.client = GAIACertificationClient(username, hf_token)
        self.username = username
        
        logger.info(f"Инициализирован агент сертификации для пользователя {username}")
    
    def process_question(self, question: Dict[str, Any]) -> str:
        """
        Обработка вопроса агентом.
        
        Args:
            question: Вопрос для обработки
            
        Returns:
            Ответ агента
        """
        task_id = question.get("task_id", "unknown")
        question_text = question.get("question", "")
        
        logger.info(f"Обработка вопроса {task_id}")
        
        try:
            # Здесь должна быть реализация обработки вопроса агентом
            # В данном примере просто возвращаем заглушку
            answer = self.agent.process_question(question_text)
            
            # ВАЖНО: Ответ должен быть в точном формате, без "FINAL ANSWER" или других префиксов
            # Система проверяет ответы методом EXACT MATCH
            
            # Очистка ответа от префиксов и суффиксов
            prefixes = [
                "FINAL ANSWER:", "ANSWER:", "The answer is:", 
                "My answer is:", "I believe the answer is:",
                "The correct answer is:", "Answer:", "Final answer:"
            ]
            
            result = answer.strip()
            
            # Поиск и удаление префиксов (регистронезависимый)
            for prefix in prefixes:
                if result.upper().startswith(prefix.upper()):
                    result = result[len(prefix):].strip()
                    logger.info(f"Удален префикс '{prefix}' из ответа")
                    break
            
            # Удаление кавычек в начале и конце
            if (result.startswith('"') and result.endswith('"')) or \
               (result.startswith("'") and result.endswith("'")):
                result = result[1:-1].strip()
                logger.info("Удалены кавычки из ответа")
            
            # Удаление точки в конце, если она есть
            if result.endswith("."):
                result = result[:-1].strip()
                logger.info("Удалена точка в конце ответа")
            
            logger.info(f"Вопрос {task_id} обработан")
            return result
            
        except Exception as e:
            logger.error(f"Ошибка при обработке вопроса {task_id}: {e}")
            return ""
    
    def run_certification(self, agent_code_url: str) -> Dict[str, Any]:
        """
        Запуск процесса сертификации.
        
        Args:
            agent_code_url: URL кода агента в публичном Space
            
        Returns:
            Результат сертификации
        """
        logger.info("Запуск процесса сертификации")
        
        # Проверка URL кода агента
        if not agent_code_url:
            error_msg = "URL кода агента не указан"
            logger.error(error_msg)
            return {"error": error_msg}
            
        # Обязательная проверка и коррекция URL для соответствия формату /tree/main
        if "/tree/main" not in agent_code_url:
            # Автоматическое исправление URL
            agent_code_url = agent_code_url.rstrip("/") + "/tree/main"
            logger.info(f"URL кода агента скорректирован: {agent_code_url}")
        
        try:
            # Получение вопросов
            questions = self.client.get_questions()
            
            if not questions:
                error_msg = "Не удалось получить вопросы от API"
                logger.error(error_msg)
                return {"error": error_msg}
            
            logger.info(f"Получено {len(questions)} вопросов")
            
            # Обработка вопросов
            answers = []
            for question in questions:
                task_id = question.get("task_id", "unknown")
                answer = self.process_question(question)
                
                answers.append({
                    "task_id": task_id,
                    "submitted_answer": answer
                })
            
            logger.info(f"Обработано {len(answers)} вопросов")
            
            # Отправка ответов
            result = self.client.submit_answers(answers, agent_code_url)
            
            logger.info(f"Результат сертификации: {result}")
            return result
            
        except Exception as e:
            error_msg = f"Ошибка при запуске сертификации: {e}"
            logger.error(error_msg)
            return {"error": error_msg}


def create_certification_gradio_interface(agent_controller: Optional[AgentController] = None, hf_token: Optional[str] = None):
    """
    Создание Gradio интерфейса для сертификации GAIA.
    
    Args:
        agent_controller: Контроллер агента для обработки вопросов
        hf_token: Токен Hugging Face для аутентификации
        
    Returns:
        Gradio интерфейс
    """
    # Функция для запуска сертификации
    def run_certification(username: str, agent_code_url: str, token: str) -> str:
        """
        Запуск сертификации через Gradio интерфейс.
        
        Args:
            username: Имя пользователя Hugging Face
            agent_code_url: URL кода агента в публичном Space
            token: Токен Hugging Face для аутентификации
            
        Returns:
            Текст результата
        """
        if not username:
            return "Ошибка: Введите имя пользователя Hugging Face"
        
        if not agent_code_url:
            return "Ошибка: Введите URL кода агента в публичном Space"
        
        # Приоритет токена: параметр -> переменная окружения
        effective_token = token or hf_token or os.environ.get("HF_TOKEN")
        if not effective_token:
            return "Ошибка: Токен Hugging Face не указан. Пожалуйста, введите токен или установите переменную окружения HF_TOKEN."
        
        try:
            # Создание агента
            if agent_controller is None:
                agent = AgentController(model_name="gpt-3.5-turbo", username=username)
            else:
                agent = agent_controller
                
            certification_agent = GAIACertificationAgent(agent, username, effective_token)
            
            # Запуск сертификации
            result = certification_agent.run_certification(agent_code_url)
            
            if "error" in result:
                return f"Ошибка: {result['error']}"
            
            # Форматирование результата
            result_text = "### Результаты сертификации GAIA\n\n"
            
            if "score" in result:
                result_text += f"**Общая точность:** {result['score']:.2f}%\n"
            
            if "correct_answers" in result and "total_questions" in result:
                result_text += f"**Правильных ответов:** {result['correct_answers']}/{result['total_questions']}\n\n"
            
            if "passed" in result:
                if result["passed"]:
                    result_text += "**✅ Поздравляем! Вы прошли сертификацию.**\n"
                    result_text += "Теперь вы можете получить свой сертификат на странице: https://huggingface.co/spaces/agents-course/Unit4-Final-Certificate\n"
                else:
                    result_text += "**❌ К сожалению, вы не прошли сертификацию.**\n"
                    result_text += "Для получения сертификата необходимо набрать не менее 30% точности.\n"
            
            return result_text
            
        except Exception as e:
            return f"Произошла ошибка при запуске сертификации: {str(e)}"
    
    # Создание интерфейса Gradio
    with gr.Blocks() as demo:
        gr.Markdown("# Сертификация GAIA для курса Hugging Face Agents")
        
        with gr.Row():
            with gr.Column():
                username_input = gr.Textbox(label="Имя пользователя Hugging Face", placeholder="Введите ваше имя пользователя")
                agent_code_url_input = gr.Textbox(
                    label="URL кода агента", 
                    placeholder="Введите URL вашего публичного Space (например, https://huggingface.co/spaces/username/space-name/tree/main)"
                )
                token_input = gr.Textbox(
                    label="Токен Hugging Face (опционально)", 
                    placeholder="Введите ваш токен Hugging Face или оставьте пустым, если он установлен в переменной окружения HF_TOKEN"
                )
                run_button = gr.Button("Запустить сертификацию")
            
            with gr.Column():
                results_output = gr.Markdown(label="Результаты")
        
        run_button.click(fn=run_certification, inputs=[username_input, agent_code_url_input, token_input], outputs=results_output)
        
        gr.Markdown("""
        ## Инструкции по использованию
        
        1. **Введите ваше имя пользователя Hugging Face** - это имя будет использоваться для идентификации ваших результатов.
        
        2. **Введите URL вашего публичного Space** - это должна быть ссылка на код вашего агента в публичном Space на Hugging Face.
           Формат: `https://huggingface.co/spaces/username/space-name/tree/main`
        
        3. **Введите ваш токен Hugging Face** (опционально) - если у вас установлена переменная окружения HF_TOKEN, можно оставить пустым.
        
        4. **Нажмите "Запустить сертификацию"** - система получит вопросы, обработает их вашим агентом и отправит ответы в систему сертификации.
        
        5. **Проверьте результаты** - если вы набрали не менее 30% точности, вы можете получить сертификат на официальной странице.
        
        ## Важные замечания
        
        - Ваш Space должен быть **публичным**, иначе система сертификации не сможет проверить ваш код.
        - Система проверяет ответы методом **EXACT MATCH**, поэтому формат ответов критически важен.
        - Для получения сертификата необходимо набрать **не менее 30% точности**.
        """)
    
    return demo


# Пример использования
if __name__ == "__main__":
    # Создание агента
    agent = AgentController(
        model_name="gpt-3.5-turbo",
        username="your_username"
    )
    
    # Создание интерфейса
    demo = create_certification_gradio_interface(agent)
    
    # Запуск интерфейса
    demo.launch()