35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
"""Семантическая дедупликация атомов по эмбеддингам.
|
|
|
|
Наивный дедуп по content_hash ловит только точные повторы. LLM генерирует
|
|
вариации одного факта — их ловим по косинусной близости эмбеддингов в Qdrant.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from . import config
|
|
DEFAULT_THRESHOLD = float(getattr(config, "DEDUP_THRESHOLD", "0.90"))
|
|
|
|
|
|
def find_duplicate(text: str, threshold: float = DEFAULT_THRESHOLD, top_k: int = 3) -> int | None:
|
|
"""Найти существующий атом, близкий к text (косинус > threshold).
|
|
|
|
Возвращает atom_id дубликата или None.
|
|
"""
|
|
from . import qdrant_store
|
|
hits = qdrant_store.search(text, top_k=top_k)
|
|
if not hits:
|
|
return None
|
|
best = hits[0]
|
|
if best["score"] >= threshold:
|
|
return int(best["id"])
|
|
return None
|
|
|
|
|
|
def duplicate_score(text: str, top_k: int = 3) -> tuple[int | None, float]:
|
|
"""Вернуть (atom_id, score) лучшего совпадения (id=None если ниже порога)."""
|
|
from . import qdrant_store
|
|
hits = qdrant_store.search(text, top_k=top_k)
|
|
if not hits:
|
|
return None, 0.0
|
|
best = hits[0]
|
|
return int(best["id"]), best["score"]
|