import re import requests import pandas as pd import torch from transformers import AutoTokenizer, AutoModelForSeq2SeqLM import json import logging import time import sys import os from functools import lru_cache # Настройка логирования logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("GAIA-Mastermind") # Конфигурация DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" MODEL_NAME = "google/flan-t5-small" # Используем меньшую модель для быстрой загрузки API_RETRIES = 3 API_TIMEOUT = 30 # Настройка кэширования моделей os.environ["TRANSFORMERS_CACHE"] = "/tmp/transformers_cache" os.environ["HF_HOME"] = "/tmp/hf_home" class GAIAExpert: _instance = None _is_initialized = False def __new__(cls): # Паттерн Singleton для предотвращения повторной загрузки модели if cls._instance is None: cls._instance = super(GAIAExpert, cls).__new__(cls) return cls._instance def __init__(self): # Инициализируем только один раз if not GAIAExpert._is_initialized: self.device = "cuda" if torch.cuda.is_available() else "cpu" logger.info(f"Инициализация модели на {self.device.upper()}") # Отложенная инициализация - токенизатор загружаем сразу, модель - по требованию self.tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) self.model = None GAIAExpert._is_initialized = True def _ensure_model_loaded(self): """Ленивая загрузка модели только при необходимости""" if self.model is None: try: logger.info("Загрузка модели...") # Оптимизированная загрузка модели self.model = AutoModelForSeq2SeqLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float16 if self.device == "cuda" else torch.float32, low_cpu_mem_usage=True, device_map="auto" # Автоматическое распределение на доступные устройства ).eval() logger.info("Модель успешно загружена") except Exception as e: logger.exception("Ошибка загрузки модели") raise RuntimeError(f"Ошибка инициализации: {str(e)}") @lru_cache(maxsize=100) # Кэширование ответов для повторяющихся вопросов def process_question(self, question: str) -> str: """Обработка вопроса с оптимизацией и кэшированием""" try: # Загружаем модель только при первом вызове self._ensure_model_loaded() # Оптимизированная обработка токенов inputs = self.tokenizer( f"Вопрос: {question}\nОтвет:", return_tensors="pt", max_length=256, truncation=True, padding="max_length" ) # Перемещаем тензоры на нужное устройство if self.device == "cuda": inputs = {k: v.to(self.device) for k, v in inputs.items()} # Оптимизированная генерация with torch.no_grad(): # Отключаем вычисление градиентов для экономии памяти outputs = self.model.generate( **inputs, max_new_tokens=50, num_beams=1, # Ускорение генерации early_stopping=True, do_sample=False # Детерминированная генерация для скорости ) answer = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return json.dumps({"final_answer": answer.strip()}) except Exception as e: return json.dumps({"final_answer": f"ERROR: {str(e)}"}) class GAIAEvaluator: def __init__(self, api_url: str = DEFAULT_API_URL): self.api_url = api_url self.questions_url = f"{api_url}/questions" self.submit_url = f"{api_url}/submit" self.session = requests.Session() self.session.headers.update({"Content-Type": "application/json"}) # Настройка повторных попыток и таймаутов self.session.mount('https://', requests.adapters.HTTPAdapter(max_retries=API_RETRIES)) def run_evaluation(self, username: str, agent_code: str): """Консольный процесс оценки без интерфейса""" # Создаем агента только при необходимости agent = GAIAExpert() # Получение вопросов с повторными попытками questions = self._fetch_questions_with_retry() if not isinstance(questions, list): logger.error(f"Ошибка получения вопросов: {questions}") return 0, 0 # Обработка вопросов answers = [] for i, q in enumerate(questions): task_id = q.get("task_id", f"task_{i}") logger.info(f"Обработка задачи {i+1}/{len(questions)}: {q['question'][:50]}...") try: json_response = agent.process_question(q["question"]) response_obj = json.loads(json_response) answer = response_obj.get("final_answer", "") answers.append({ "task_id": task_id, "answer": str(answer)[:300] }) except Exception as e: logger.error(f"Ошибка обработки: {str(e)}") answers.append({ "task_id": task_id, "answer": f"ERROR: {str(e)}" }) # Отправка ответов с повторными попытками return self._submit_answers_with_retry(username, agent_code, answers) def _fetch_questions_with_retry(self, max_retries=3): """Получение вопросов с API с повторными попытками""" for attempt in range(max_retries): try: response = self.session.get(self.questions_url, timeout=API_TIMEOUT) if response.status_code == 200: return response.json() logger.warning(f"HTTP error {response.status_code}, попытка {attempt+1}/{max_retries}") time.sleep(2 ** attempt) # Экспоненциальная задержка между попытками except Exception as e: logger.warning(f"Connection error: {str(e)}, попытка {attempt+1}/{max_retries}") time.sleep(2 ** attempt) return f"Failed after {max_retries} attempts" def _submit_answers_with_retry(self, username: str, agent_code: str, answers: list, max_retries=3): """Отправка ответов на сервер с повторными попытками""" for attempt in range(max_retries): try: payload = { "username": username.strip(), "agent_code": agent_code.strip(), "answers": answers } response = self.session.post( self.submit_url, json=payload, timeout=API_TIMEOUT * 2 ) if response.status_code == 200: result = response.json() score = result.get("score", 0) return score, len(answers) logger.warning(f"HTTP error {response.status_code}, попытка {attempt+1}/{max_retries}") time.sleep(2 ** attempt) except Exception as e: logger.error(f"Ошибка отправки: {str(e)}, попытка {attempt+1}/{max_retries}") time.sleep(2 ** attempt) return 0, len(answers) if __name__ == "__main__": # Параметры запуска USERNAME = "yoshizen" AGENT_CODE = "https://huggingface.co/spaces/yoshizen/FinalTest" logger.info(f"Запуск оценки для {USERNAME}") start_time = time.time() evaluator = GAIAEvaluator() score, total = evaluator.run_evaluation(USERNAME, AGENT_CODE) elapsed = time.time() - start_time logger.info(f"Оценка завершена за {elapsed:.1f} сек") logger.info(f"Результат: {score}/{total} правильных ответов") if total > 0: logger.info(f"Точность: {score/total*100:.1f}%") else: logger.error("Не удалось обработать ни одного вопроса")