Upload fast parser and ml evaluator

fast_parser.py

@@ -0,0 +1,253 @@
+# 📄 src/core/intent_parser/fast_parser.py
+import re
+import logging
+from typing import Dict, Tuple, Optional
+from dataclasses import dataclass
+
+@dataclass
+class ParsedIntent:
+    """Универсальный контейнер для распознанного намерения"""
+    intent: str
+    confidence: float
+    original_text: str
+    normalized_text: str
+    parameters: Dict[str, any]
+    source: str = "fast_parser"
+
+class FastIntentParser:
+    """
+    Быстрый парсер намерений на основе ключевых слов и правил.
+    Обрабатывает 80-90% типичных запросов без использования ML.
+    """
+
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+        self._setup_domains()
+        self._setup_synonyms()
+        self._setup_patterns()
+
+    def _setup_domains(self):
+        """Настройка доменов и ключевых слов"""
+        self.domains = {
+            'greeting': {
+                'keywords': ['привет', 'здравствуй', 'добрый', 'хай', 'салют', 'здаров'],
+                'priority': 1,
+                'response_templates': [
+                    "Привет! Готов к работе.",
+                    "Здравствуйте! Чем могу помочь?",
+                    "Приветствую! Ariel на связи."
+                ]
+            },
+            'system': {
+                'keywords': ['будильник', 'таймер', 'открой', 'запусти', 'выключи', 'громкость'],
+                'priority': 2,
+                'subdomains': {
+                    'alarm': ['будильник', 'разбуди', 'напомни'],
+                    'app_launch': ['открой', 'запусти', 'включи'],
+                    'system_control': ['выключи', 'перезагрузи', 'громкость']
+                }
+            },
+            'visualization': {
+                'keywords': ['график', 'диаграмм', 'схем', 'визуализир', 'построй', 'нарисуй'],
+                'priority': 3,
+                'subdomains': {
+                    'plot': ['график', 'построй'],
+                    'chart': ['диаграмм', 'гистограмм'],
+                    'scheme': ['схем', 'блок-схем']
+                }
+            },
+            'knowledge': {
+                'keywords': ['что такое', 'как работает', 'объясни', 'найди информацию', 'база данных'],
+                'priority': 4
+            },
+            'creative': {
+                'keywords': ['расскажи', 'пошути', 'придумай', 'рекомендуй', 'советуй'],
+                'priority': 5
+            },
+            'help': {
+                'keywords': ['помощь', 'команды', 'что ты умеешь', 'справка'],
+                'priority': 6
+            }
+        }
+
+    def _setup_synonyms(self):
+        """Настройка синонимов для нормализации текста"""
+        self.synonyms = {
+            'привет': ['салют', 'здаров', 'хей', 'хай', 'здорово', 'добрый день'],
+            'будильник': ['будильничек', 'напоминание', 'оповещение', 'звонок'],
+            'поставь': ['заведи', 'установи', 'создай', 'активируй'],
+            'открой': ['запусти', 'включи', 'открой', 'запусти'],
+            'график': ['графики', 'графичек', 'плотик'],
+            'помощь': ['справка', 'хелп', 'помоги', 'подскажи']
+        }
+
+    def _setup_patterns(self):
+        """Настройка regex-паттернов для сложных случаев"""
+        self.patterns = {
+            'system': [
+                # Будильники и таймеры
+                r'(поставь|заведи|установи).*будильник.*(\d{1,2}:\d{2})',
+                r'будильник.*(\d{1,2}).*(утра|вечера|часов|час)',
+                r'разбуди.*(\d{1,2}).*(утра|вечера)',
+                # Запуск приложений
+                r'(открой|запусти).*(браузер|chrome|хром|firefox|файрфокс)',
+                r'(открой|запусти).*(терминал|cmd|командную строку)',
+                # Управление системой
+                r'(выключи|перезагрузи).*(компьютер|систему)',
+                r'(сделай|поставь).*(громче|тише)'
+            ],
+            'visualization': [
+                r'построй.*график.*(\w+).*от.*(\d+).*до.*(\d+)',
+                r'график.*(sin|синус|cos|косинус|tan|тангенс)',
+                r'диаграмм.*(кругова|столбчата|гистограмм)',
+                r'нарисуй.*схем.*работы'
+            ],
+            'knowledge': [
+                r'что такое (\w+)',
+                r'как работает (\w+)',
+                r'объясни.*(\w+)',
+                r'найди.*информацию.*о (\w+)'
+            ]
+        }
+
+    def normalize_text(self, text: str) -> str:
+        """Нормализация текста: приведение к нижнему регистру и замена синонимов"""
+        if not text:
+            return ""
+
+        text_lower = text.lower().strip()
+
+        # Замена синонимов на основные формы
+        for main_word, synonyms in self.synonyms.items():
+            for synonym in synonyms:
+                if synonym in text_lower:
+                    text_lower = text_lower.replace(synonym, main_word)
+                    self.logger.debug(f"Заменен синоним '{synonym}' -> '{main_word}'")
+
+        return text_lower
+
+    def extract_parameters(self, domain: str, text: str) -> Dict[str, any]:
+        """Извлечение параметров из текста команды"""
+        normalized_text = self.normalize_text(text)
+        parameters = {}
+
+        if domain == 'system':
+            # Извлечение времени для будильников
+            time_match = re.search(r'(\d{1,2})(?::(\d{2}))?\s*(утра|вечера|часов|час)?', normalized_text)
+            if time_match:
+                hour = int(time_match.group(1))
+                minute = int(time_match.group(2) or "0")
+                period = time_match.group(3) or ""
+
+                # Конвертация в 24-часовой формат
+                if period == 'вечера' and hour < 12:
+                    hour += 12
+
+                parameters['time'] = f"{hour:02d}:{minute:02d}"
+                parameters['period'] = period
+
+            # Извлечение названия приложения
+            app_matches = re.findall(r'(браузер|хром|chrome|терминал|cmd)', normalized_text)
+            if app_matches:
+                parameters['app'] = app_matches[0]
+
+        elif domain == 'visualization':
+            # Извлечение математической функции
+            func_match = re.search(r'(sin|синус|cos|косинус|tan|тангенс|x\^2)', normalized_text)
+            if func_match:
+                func_map = {'синус': 'sin', 'косинус': 'cos', 'тангенс': 'tan'}
+                parameters['function'] = func_map.get(func_match.group(1), func_match.group(1))
+
+            # Извлечение диапазона
+            range_match = re.search(r'от\s*(\d+)\s*до\s*(\d+)', normalized_text)
+            if range_match:
+                parameters['x_range'] = [float(range_match.group(1)), float(range_match.group(2))]
+
+        elif domain == 'knowledge':
+            # Извлечение темы для поиска
+            topic_match = re.search(r'что такое\s+(\w+)', normalized_text)
+            if not topic_match:
+                topic_match = re.search(r'как работает\s+(\w+)', normalized_text)
+            if not topic_match:
+                topic_match = re.search(r'объясни\s+(\w+)', normalized_text)
+
+            if topic_match:
+                parameters['topic'] = topic_match.group(1)
+
+        return parameters
+
+    def parse(self, text: str) -> Optional[ParsedIntent]:
+        """
+        Основной метод парсинга намерения из текста.
+        Возвращает ParsedIntent или None если намерение не распознано.
+        """
+        if not text or not text.strip():
+            return None
+
+        normalized_text = self.normalize_text(text)
+        self.logger.debug(f"Парсинг текста: '{text}' -> '{normalized_text}'")
+
+        # Сначала проверяем regex-паттерны (более точные)
+        domain_from_patterns = self._check_patterns(normalized_text)
+        if domain_from_patterns:
+            domain, subdomain, confidence = domain_from_patterns
+            parameters = self.extract_parameters(domain, normalized_text)
+
+            return ParsedIntent(
+                intent=domain,
+                confidence=confidence,
+                original_text=text,
+                normalized_text=normalized_text,
+                parameters=parameters
+            )
+
+        # Затем проверяем ключевые слова
+        domain_from_keywords = self._check_keywords(normalized_text)
+        if domain_from_keywords:
+            domain, subdomain, confidence = domain_from_keywords
+            parameters = self.extract_parameters(domain, normalized_text)
+
+            return ParsedIntent(
+                intent=domain,
+                confidence=confidence,
+                original_text=text,
+                normalized_text=normalized_text,
+                parameters=parameters
+            )
+
+        # Не распознано
+        self.logger.debug(f"Не удалось распознать намерение: '{text}'")
+        return None
+
+    def _check_patterns(self, text: str) -> Optional[Tuple[str, str, float]]:
+        """Проверка текста по regex-паттернам"""
+        for domain, pattern_list in self.patterns.items():
+            for pattern in pattern_list:
+                if re.search(pattern, text):
+                    self.logger.debug(f"Найден паттерн '{pattern}' для домена '{domain}'")
+                    return domain, None, 0.95  # Высокая уверенность для паттернов
+
+        return None
+
+    def _check_keywords(self, text: str) -> Optional[Tuple[str, str, float]]:
+        """Проверка текста по ключевым словам"""
+        found_domains = []
+
+        for domain, domain_config in self.domains.items():
+            for keyword in domain_config['keywords']:
+                if keyword in text:
+                    confidence = 0.9 if len(keyword) > 3 else 0.7
+                    found_domains.append((domain, None, confidence))
+                    self.logger.debug(f"Найдено ключевое слово '{keyword}' для домена '{domain}'")
+
+        if not found_domains:
+            return None
+
+        # Возвращаем домен с наивысшим приоритетом
+        found_domains.sort(key=lambda x: self.domains[x[0]]['priority'])
+        return found_domains[0]
+
+# Фабрика для создания парсера
+def create_fast_parser() -> FastIntentParser:
+    """Создание и настройка быстрого парсера"""
+    return FastIntentParser()

ml_classifier.py

@@ -0,0 +1,246 @@
+# 📄 src/core/intent_parser/ml_classifier.py
+import json
+import os
+import logging
+from typing import Dict, List, Optional, Any
+from dataclasses import dataclass
+from torch.quantization import quantize_dynamic
+import time
+
+# Импорты с обработкой ошибок
+print("Инициализация ML классификатора...")
+try:
+    import torch
+    import torch.nn.functional as F
+    from transformers import AutoTokenizer, AutoModelForSequenceClassification
+    ML_AVAILABLE = True
+except ImportError as e:
+    print(f"⚠️  ML библиотеки не установлены: {e}")
+    ML_AVAILABLE = False
+    torch = None
+    AutoTokenizer = None
+    AutoModelForSequenceClassification = None
+
+@dataclass
+class MLClassificationResult:
+    """Результат классификации ML моделью"""
+    intent: str
+    confidence: float
+    all_predictions: List[tuple]  # Список всех (интент, уверенность)
+    multi_label_predictions: Optional[List[tuple]] = None  # Интенты выше порога
+
+class MLIntentClassifier:
+    """
+    ML классификатор намерений на основе DistilBERT.
+    Поддерживает multi-label классификацию как в обученной модели.
+    """
+
+    def __init__(self, model_path: Optional[str] = None):
+        self.logger = logging.getLogger(__name__)
+        self.model = None
+        self.tokenizer = None
+        self.device = None
+        self.is_initialized = False
+
+        # Словарь интентов
+        self.intent_to_idx = {}
+        self.idx_to_intent = {}
+
+        # Настройки
+        self.confidence_threshold = 0.3
+        self.max_length = 128
+
+        # Путь к модели (по умолчанию из вашей структуры)
+        if model_path is None:
+            # Автоматически определяем путь в структуре проекта
+            base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+            base_dir = "C:/PycharmProjects/Ariel"
+            model_path = os.path.join(base_dir, "Data", "Models", "intent_classifier")
+
+        base_dir = "C:/PycharmProjects/Ariel"
+        model_path = os.path.join(base_dir, "Data", "models", "intent_classifier")
+
+        self.model_path = model_path
+        self._initialize_model()
+
+    def _initialize_model(self):
+        """Инициализация модели с обработкой ошибок"""
+        if not ML_AVAILABLE:
+            self.logger.warning("ML библиотеки не установлены. Использование заглушки.")
+            return
+
+        try:
+            # Проверяем существование директории
+            if not os.path.exists(self.model_path):
+                self.logger.error(f"Не найден файл: {self.model_path}")
+                self.logger.info("Проверьте, что вы распаковали архив в правильную папку")
+                return
+
+            # Проверяем наличие ключевых файлов
+            required_files = ['config.json']
+            weight_files = ['model.safetensors', 'pytorch_model.bin']
+
+            for file in required_files:
+                if not os.path.exists(os.path.join(self.model_path, file)):
+                    self.logger.error(f"Не найден файл: {os.path.join(self.model_path, file)}")
+                    return
+
+            # Проверяем наличие файла весов
+            has_weights = any(os.path.exists(os.path.join(self.model_path, wf)) for wf in weight_files)
+            if not has_weights:
+                self.logger.error(f"Не найден файл весов модели. Ожидается один из: {weight_files}")
+                self.logger.info(f"Файлы в директории: {os.listdir(self.model_path)}")
+                return
+
+            # Загружаем vocabulary интентов
+            vocab_path = os.path.join(self.model_path, "intent_vocab.json")
+            if os.path.exists(vocab_path):
+                with open(vocab_path, 'r', encoding='utf-8') as f:
+                    self.intent_to_idx = json.load(f)
+                # Преобразуем индексы в int если они строки
+                self.intent_to_idx = {k: int(v) for k, v in self.intent_to_idx.items()}
+                self.idx_to_intent = {v: k for k, v in self.intent_to_idx.items()}
+                self.logger.info(f"Загружен словарь интентов: {len(self.intent_to_idx)} классов")
+            else:
+                self.logger.warning("Файл intent_vocab.json не найден. Пытаюсь определить из config.json")
+                # Попробуем получить из конфига модели
+                pass
+
+            # Загружаем модель и токенизатор
+            self.logger.info(f"Загрузка модели из {self.model_path}...")
+            self.tokenizer = AutoTokenizer.from_pretrained(self.model_path, local_files_only=True)
+
+            # Загружаем модель с multi-label конфигурацией
+            self.model = AutoModelForSequenceClassification.from_pretrained(
+                self.model_path,
+                local_files_only=True,
+                problem_type="multi_label_classification"
+            )
+
+            self.model = quantize_dynamic(self.model, {torch.nn.Linear}, dtype=torch.qint8)
+
+
+
+            # Настройка устройства
+            self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+            self.model.to(self.device)
+            self.model.eval()
+
+            self.is_initialized = True
+            self.logger.info(f"✅ Модель загружена успешно!")
+            self.logger.info(f"   Устройство: {self.device}")
+            self.logger.info(f"   Классов: {len(self.intent_to_idx) if self.intent_to_idx else 'неизвестно'}")
+
+        except Exception as e:
+            self.logger.error(f"❌ Ошибка загрузки модели: {e}")
+            self.is_initialized = False
+
+    def predict(self, text: str, threshold: Optional[float] = None) -> MLClassificationResult:
+        """Предсказание интентов для текста (multi-label)"""
+        if not self.is_initialized:
+            self.logger.warning("Модель не инициализирована, возвращаем fallback")
+            return self._fallback_prediction(text)
+
+        try:
+            current_threshold = threshold if threshold is not None else self.confidence_threshold
+
+            # Токенизация
+            inputs = self.tokenizer(
+                text,
+                truncation=True,
+                padding='max_length',
+                max_length=self.max_length,
+                return_tensors="pt"
+            )
+
+            # Переносим на нужное устройство
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+
+            # Предсказание
+            with torch.no_grad():
+                outputs = self.model(**inputs)
+                # Для multi-label используем sigmoid
+                probabilities = torch.sigmoid(outputs.logits)
+
+            # Получаем numpy массив
+            probs = probabilities.cpu().numpy()[0]
+
+            # Собираем результаты
+            all_predictions = []
+            multi_label_predictions = []
+
+            for idx, prob in enumerate(probs):
+                if idx in self.idx_to_intent:
+                    intent_name = self.idx_to_intent[idx]
+                    confidence = float(prob)
+
+                    all_predictions.append((intent_name, confidence))
+
+                    if confidence >= current_threshold:
+                        multi_label_predictions.append((intent_name, confidence))
+
+            # Сортируем по уверенности
+            all_predictions.sort(key=lambda x: x[1], reverse=True)
+            multi_label_predictions.sort(key=lambda x: x[1], reverse=True)
+
+            # Определяем основной интент
+            main_intent = "unknown"
+            main_confidence = 0.0
+
+            if multi_label_predictions:
+                main_intent = multi_label_predictions[0][0]
+                main_confidence = multi_label_predictions[0][1]
+            elif all_predictions:
+                main_intent = all_predictions[0][0]
+                main_confidence = all_predictions[0][1]
+
+            return MLClassificationResult(
+                intent=main_intent,
+                confidence=main_confidence,
+                all_predictions=all_predictions,
+                multi_label_predictions=multi_label_predictions
+            )
+
+        except Exception as e:
+            self.logger.error(f"Ошибка предсказания: {e}")
+            return self._fallback_prediction(text)
+
+    def _fallback_prediction(self, text: str) -> MLClassificationResult:
+        """Заглушка при ошибках"""
+        return MLClassificationResult(
+            intent="unknown",
+            confidence=0.5,
+            all_predictions=[("unknown", 1.0)],
+            multi_label_predictions=[]
+        )
+
+    def get_model_info(self) -> Dict[str, Any]:
+        """Информация о модели"""
+        return {
+            "is_initialized": self.is_initialized,
+            "model_path": self.model_path,
+            "num_intents": len(self.intent_to_idx),
+            "intents": list(self.intent_to_idx.keys()) if self.intent_to_idx else [],
+            "confidence_threshold": self.confidence_threshold,
+            "device": str(self.device) if self.device else None
+        }
+
+def create_ml_classifier(model_path: Optional[str] = None) -> MLIntentClassifier:
+    """Фабричная функция для создания классификатора"""
+    return MLIntentClassifier(model_path)
+
+
+
+# Пример использования (раскомментировать):
+start = time.time()
+print("Загрузка модели, ожидайте...")
+classifier = create_ml_classifier("/Data/Models/intent_classifier")
+
+print("✅ Модель загружена! Тестируйте:")
+while True:
+    text = input("\nВведите текст: ")
+    if text.lower() == 'выход': break
+    result = classifier.predict(text)
+    print(f"Результат: {result.intent} ({result.confidence:.1%})")
+    for intent, conf in result.all_predictions[:3]:
+        print(f"  - {intent}: {conf:.1%}")