Create app.py

app.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+Главный скрипт для определения намерений пользователя.
+Использует двухступенчатую систему: сначала быстрый парсер на правилах,
+при его неудаче — ML-классификатор (DistilBERT).
+"""
+
+import logging
+import sys
+import time
+from typing import Dict, Any
+
+from fast_parser import create_fast_parser, ParsedIntent
+from ml_classifier import create_ml_classifier, MLClassificationResult
+
+
+def setup_logging(level: int = logging.INFO) -> None:
+    """Настройка логирования."""
+    logging.basicConfig(
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        level=level,
+        handlers=[logging.StreamHandler(sys.stdout)]
+    )
+
+
+def parse_intent(
+    text: str,
+    fast_parser,
+    ml_classifier
+) -> Dict[str, Any]:
+    # Шаг 1: быстрый парсер на правилах
+    fast_result: ParsedIntent = fast_parser.parse(text)
+    if fast_result is not None:
+        logging.info(f"Fast parser сработал: intent={fast_result.intent}, "
+                     f"confidence={fast_result.confidence:.2f}")
+        return {
+            "source": "fast_parser",
+            "intent": fast_result.intent,
+            "confidence": fast_result.confidence,
+            "parameters": fast_result.parameters,
+            "original_text": fast_result.original_text,
+            "normalized_text": fast_result.normalized_text,
+        }
+
+    # Шаг 2: ML-классификатор (если доступен)
+    if ml_classifier.is_initialized:
+        ml_result: MLClassificationResult = ml_classifier.predict(text)
+        logging.info(f"ML классификатор сработал: intent={ml_result.intent}, "
+                     f"confidence={ml_result.confidence:.2f}")
+        return {
+            "source": "ml_classifier",
+            "intent": ml_result.intent,
+            "confidence": ml_result.confidence,
+            "all_predictions": ml_result.all_predictions,
+            "multi_label_predictions": ml_result.multi_label_predictions,
+        }
+    else:
+        logging.warning("ML классификатор недоступен, возвращаем unknown")
+        return {
+            "source": "none",
+            "intent": "unknown",
+            "confidence": 0.0,
+            "error": "Ни один парсер не смог обработать запрос",
+        }
+
+
+def main() -> None:
+    """Основная функция: инициализация парсеров и интерактивный цикл."""
+    setup_logging(logging.INFO)
+    # Для более детальной трассировки можно включить DEBUG:
+    # logging.getLogger().setLevel(logging.DEBUG)
+
+    logging.info("Инициализация парсеров...")
+
+    # FastParser всегда доступен
+    fast_parser = create_fast_parser()
+    logging.info("Fast parser готов.")
+
+    # MLClassifier с замером времени загрузки
+    logging.info("Начинаю загрузку ML модели...")
+    ml_start = time.time()
+    ml_classifier = create_ml_classifier()  # путь к модели берётся по умолчанию
+    ml_end = time.time()
+    load_time = ml_end - ml_start
+
+    if ml_classifier.is_initialized:
+        logging.info(f"ML классификатор успешно загружен за {load_time:.2f} сек.")
+        model_info = ml_classifier.get_model_info()
+        logging.info(f"Модель: устройство {model_info['device']}, "
+                     f"интентов: {model_info['num_intents']}")
+    else:
+        logging.warning(f"ML классификатор не загружен (потрачено {load_time:.2f} сек). "
+                        "Будет использован только fast parser.")
+
+    print("Введите ваш запрос (или 'выход' / 'exit' / 'quit' для завершения):")
+
+    while True:
+        try:
+            user_input = input("\n> ").strip()
+            if not user_input:
+                continue
+            if user_input.lower() in ("выход", "exit", "quit"):
+                print("До свидания!")
+                break
+
+            result = parse_intent(user_input, fast_parser, ml_classifier)
+
+            # Вывод результатов
+            print("\nРезультат разбора:")
+            print(f"  Источник: {result['source']}")
+            print(f"  Интент: {result['intent']}")
+            print(f"  Уверенность: {result['confidence']:.2%}")
+
+            if "parameters" in result and result["parameters"]:
+                print(f"  Параметры: {result['parameters']}")
+
+            if "all_predictions" in result and result["all_predictions"]:
+                print("  Все предсказания ML (top-5):")
+                for intent, conf in result["all_predictions"][:5]:
+                    print(f"    - {intent}: {conf:.2%}")
+
+        except KeyboardInterrupt:
+            print("\nЗавершение работы по Ctrl+C.")
+            break
+        except Exception as e:
+            logging.exception("Необработанная ошибка в цикле обработки")
+            print(f"Произошла ошибка: {e}")
+
+
+if __name__ == "__main__":
+    main()