""" Агент для взаимодействия с официальным 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()