""" Минимальная версия приложения для отправки ответов на сервер GAIA Без зависимостей от внешних модулей и файлов """ import os import json import requests import time from typing import Dict, Any, List # Импортируем агента from enhanced_gaia_agent_v3 import EnhancedGAIAAgent # Константы API_URL = "https://gaia-challenge.huggingface.co/api/submit" CACHE_FILE = "submission_cache.json" class GAIASubmitter: """ Класс для отправки ответов на сервер GAIA """ def __init__(self, username: str, agent_code: str, use_cache: bool = True): """ Инициализация отправителя Args: username: Имя пользователя для отправки agent_code: Код агента для отправки use_cache: Использовать ли кэширование ответов """ self.username = username self.agent_code = agent_code self.use_cache = use_cache self.cache = self._load_cache() if use_cache else {} # Инициализируем агента print(f"Initializing agent for submission...") self.agent = EnhancedGAIAAgent(use_cache=True) def _load_cache(self) -> Dict[str, Any]: """ Загружает кэш ответов из файла Returns: Dict[str, Any]: Словарь с кэшированными ответами """ if os.path.exists(CACHE_FILE): try: with open(CACHE_FILE, 'r', encoding='utf-8') as f: print(f"Loading submission cache from {CACHE_FILE}") return json.load(f) except Exception as e: print(f"Error loading submission cache: {e}") return {} else: print(f"Submission cache file {CACHE_FILE} not found, creating new cache") return {} def _save_cache(self) -> None: """ Сохраняет кэш ответов в файл """ try: with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(self.cache, f, ensure_ascii=False, indent=2) print(f"Submission cache saved to {CACHE_FILE}") except Exception as e: print(f"Error saving submission cache: {e}") def submit_answers(self, questions: List[Dict[str, str]]) -> Dict[str, Any]: """ Отправляет ответы на сервер GAIA Args: questions: Список вопросов для ответа Returns: Dict[str, Any]: Ответ сервера """ # Подготавливаем ответы answers = {} print(f"Processing {len(questions)} questions...") for i, question in enumerate(questions): task_id = question["task_id"] question_text = question["question"] print(f"Question {i+1}/{len(questions)}: {task_id} - {question_text[:50]}...") # Получаем ответ от агента try: json_response = self.agent(question_text, task_id) response_obj = json.loads(json_response) answer = response_obj.get("final_answer", "") print(f"Answer: {answer}") answers[task_id] = answer except Exception as e: print(f"Error processing question {task_id}: {e}") answers[task_id] = f"ERROR: {e}" # Подготавливаем данные для отправки submission_data = { "username": self.username, "agent_code": self.agent_code, "answers": answers } # Сохраняем данные для отладки with open("submission_data.json", 'w', encoding='utf-8') as f: json.dump(submission_data, f, ensure_ascii=False, indent=2) print("Submission data saved to submission_data.json") # Проверяем наличие ответа в кэше cache_key = json.dumps(submission_data) if self.use_cache and cache_key in self.cache: print("Using cached submission response") return self.cache[cache_key] # Отправляем запрос на сервер с повторными попытками max_retries = 3 for attempt in range(max_retries): try: print(f"Submitting answers to {API_URL} (attempt {attempt+1}/{max_retries})...") response = requests.post(API_URL, json=submission_data, timeout=10) # Сохраняем ответ для отладки with open("response_content.txt", 'w', encoding='utf-8') as f: f.write(response.text) print("Response content saved to response_content.txt") # Проверяем статус ответа if response.status_code == 200: print("Submission successful!") result = response.json() # Сохраняем результат для отладки with open("results_response.txt", 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) print("Results saved to results_response.txt") # Сохраняем в кэш if self.use_cache: self.cache[cache_key] = result self._save_cache() return result else: error_msg = f"HTTP error {response.status_code}: {response.text}" print(error_msg) except requests.exceptions.RequestException as e: error_msg = f"Network error: {str(e)}" print(error_msg) # Экспоненциальная задержка перед следующей попыткой delay = 2 ** attempt print(f"Retrying in {delay} seconds...") time.sleep(delay) return {"error": f"All {max_retries} attempts failed"} def main(): """ Основная функция для отправки ответов """ # Получаем имя пользователя из аргументов командной строки import sys if len(sys.argv) > 1: username = sys.argv[1] else: # Используем значение по умолчанию вместо интерактивного запроса username = "yoshizen" # Замените на ваше имя пользователя print(f"Using default username: {username}") # Код агента для отправки agent_code = "enhanced_gaia_agent_v3" # Создаем отправителя submitter = GAIASubmitter(username, agent_code, use_cache=True) # Создаем минимальный набор тестовых вопросов # Это нужно только для локального тестирования # В реальной системе GAIA вопросы будут предоставлены системой questions = [ { "task_id": "test_1", "question": "What is the capital of France?", "type": "location" }, { "task_id": "test_2", "question": "Who wrote 'War and Peace'?", "type": "name" } ] # Отправляем ответы result = submitter.submit_answers(questions) # Выводим результат print("\n=== Submission Result ===") if "error" in result: print(f"Error: {result['error']}") else: print(f"Score: {result.get('score', 'N/A')}") print(f"Message: {result.get('message', 'No message')}") # Выводим детальные результаты, если есть if "details" in result: print("\nDetails:") for task_id, detail in result["details"].items(): print(f"- {task_id}: {detail}") if __name__ == "__main__": main()