"""MinHash LSH для нечёткого поиска похожих документов. Позволяет быстро находить документы с похожим содержимым без точного сравнения всех пар. """ import logging from datasketch import MinHash, MinHashLSH logger = logging.getLogger(__name__) # Параметры MinHash LSH LSH_THRESHOLD = 0.5 # Минимальная схожесть для включения в результаты LSH_NUM_PERM = 128 # Количество хэш-функций (точность vs память) # Глобальный LSH индекс (in-memory) _lsh: MinHashLSH | None = None def get_lsh() -> MinHashLSH: """Получить или создать глобальный LSH индекс.""" global _lsh if _lsh is None: _lsh = MinHashLSH(threshold=LSH_THRESHOLD, num_perm=LSH_NUM_PERM) logger.info("MinHash LSH индекс создан") return _lsh def get_shingles(text: str, k: int = 3) -> set[str]: """ Получить k-слоговые шинглы из текста. Args: text: Исходный текст k: Размер шингла (количество слов) Returns: Множество шинглов """ words = text.lower().split() if len(words) < k: return {" ".join(words)} if words else set() return {" ".join(words[i : i + k]) for i in range(len(words) - k + 1)} def text_to_minhash(text: str) -> MinHash: """ Создать MinHash подпись для текста. Args: text: Исходный текст Returns: MinHash объект """ m = MinHash(num_perm=LSH_NUM_PERM) for shingle in get_shingles(text): m.update(shingle.encode("utf-8")) return m def add_to_lsh(doc_key: str, text: str) -> None: """ Добавить документ в LSH индекс. Args: doc_key: Уникальный ключ документа (например, "doc:{id}") text: Текст документа """ lsh = get_lsh() m = text_to_minhash(text) try: lsh.insert(doc_key, m) except ValueError: # Документ уже в индексе — игнорируем pass def find_similar(text: str) -> list[str]: """ Найти похожие документы в LSH индексе. Args: text: Текст для поиска похожих Returns: Список ключей похожих документов """ lsh = get_lsh() m = text_to_minhash(text) try: return lsh.query(m) except Exception as e: logger.warning(f"Ошибка запроса к MinHash LSH: {e}") return [] def compute_jaccard_minhash(text_a: str, text_b: str) -> float: """ Оценить схожесть двух текстов через MinHash Jaccard. Args: text_a: Первый текст text_b: Второй текст Returns: Оценка схожести Жаккара через MinHash """ m_a = text_to_minhash(text_a) m_b = text_to_minhash(text_b) return m_a.jaccard(m_b) def reset_lsh() -> None: """Сбросить LSH индекс (для тестов).""" global _lsh _lsh = None