Initial commit: VKUS vitrina MVP (frontend + backend).
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
"""
|
||||
Временный локальный RAG-слой внутри backend.
|
||||
|
||||
Целевая архитектура выносит RAG на платформу ИИ (Docling + BERT + VectorDB).
|
||||
Для тестирования витрины используется упрощённый индекс TF-IDF по папке Manuals/.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import re
|
||||
import zipfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
SUPPORTED_SUFFIXES = {".txt", ".md", ".docx"}
|
||||
|
||||
TOKEN_RE = re.compile(r"[a-zA-Zа-яА-ЯёЁ0-9]{3,}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class DocumentChunk:
|
||||
id: str
|
||||
text: str
|
||||
source_title: str
|
||||
source_path: str
|
||||
system: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class SearchHit:
|
||||
chunk: DocumentChunk
|
||||
score: float
|
||||
|
||||
|
||||
def tokenize(text: str) -> list[str]:
|
||||
return [t.lower() for t in TOKEN_RE.findall(text)]
|
||||
|
||||
|
||||
def detect_system_from_text(text: str) -> str | None:
|
||||
lower = text.lower()
|
||||
rules = [
|
||||
(r"meridium|меридиум", "Meridium 4"),
|
||||
(r"есуид", "ЕСУИД"),
|
||||
(r"\b1[сc]\b", "1С"),
|
||||
(r"starlims|старлимс", "StarLims"),
|
||||
(r"лимс|lims", "ЛИМС"),
|
||||
(r"вкус|vkus", "ВКУС"),
|
||||
(r"vpn|впн", "Общее"),
|
||||
]
|
||||
for pattern, name in rules:
|
||||
if re.search(pattern, lower):
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
class LocalRagIndex:
|
||||
def __init__(self) -> None:
|
||||
self.chunks: list[DocumentChunk] = []
|
||||
self._idf: dict[str, float] = {}
|
||||
self._tf_vectors: list[dict[str, float]] = []
|
||||
|
||||
@property
|
||||
def is_ready(self) -> bool:
|
||||
return len(self.chunks) > 0
|
||||
|
||||
def build(self, documents_path: Path, chunk_size: int = 450) -> int:
|
||||
self.chunks.clear()
|
||||
self._idf.clear()
|
||||
self._tf_vectors.clear()
|
||||
|
||||
if not documents_path.exists():
|
||||
return 0
|
||||
|
||||
chunk_idx = 0
|
||||
for file_path in sorted(documents_path.rglob("*")):
|
||||
if not file_path.is_file():
|
||||
continue
|
||||
if file_path.suffix.lower() not in SUPPORTED_SUFFIXES:
|
||||
continue
|
||||
|
||||
try:
|
||||
content = _read_document_text(file_path)
|
||||
except OSError:
|
||||
continue
|
||||
|
||||
if not content:
|
||||
continue
|
||||
|
||||
rel_path = str(file_path.relative_to(documents_path.parent))
|
||||
title = file_path.stem
|
||||
system = detect_system_from_text(content) or detect_system_from_text(str(file_path))
|
||||
|
||||
for piece in _split_into_chunks(content, chunk_size):
|
||||
self.chunks.append(
|
||||
DocumentChunk(
|
||||
id=f"chunk-{chunk_idx}",
|
||||
text=piece,
|
||||
source_title=title,
|
||||
source_path=rel_path,
|
||||
system=system,
|
||||
)
|
||||
)
|
||||
chunk_idx += 1
|
||||
|
||||
self._build_tfidf()
|
||||
return len(self.chunks)
|
||||
|
||||
def _build_tfidf(self) -> None:
|
||||
doc_freq: dict[str, int] = {}
|
||||
self._tf_vectors = []
|
||||
|
||||
for chunk in self.chunks:
|
||||
tokens = tokenize(chunk.text)
|
||||
if not tokens:
|
||||
self._tf_vectors.append({})
|
||||
continue
|
||||
tf: dict[str, float] = {}
|
||||
for token in tokens:
|
||||
tf[token] = tf.get(token, 0.0) + 1.0
|
||||
for token in tf:
|
||||
tf[token] /= len(tokens)
|
||||
self._tf_vectors.append(tf)
|
||||
for token in set(tokens):
|
||||
doc_freq[token] = doc_freq.get(token, 0) + 1
|
||||
|
||||
n_docs = max(len(self.chunks), 1)
|
||||
self._idf = {token: math.log((1 + n_docs) / (1 + freq)) + 1 for token, freq in doc_freq.items()}
|
||||
|
||||
def search(self, query: str, top_k: int = 5, min_score: float = 0.0) -> list[SearchHit]:
|
||||
if not self.chunks:
|
||||
return []
|
||||
|
||||
q_tokens = tokenize(query)
|
||||
if not q_tokens:
|
||||
return []
|
||||
|
||||
q_tf: dict[str, float] = {}
|
||||
for token in q_tokens:
|
||||
q_tf[token] = q_tf.get(token, 0.0) + 1.0
|
||||
for token in q_tf:
|
||||
q_tf[token] /= len(q_tokens)
|
||||
q_vec = {t: q_tf[t] * self._idf.get(t, 0.0) for t in q_tf}
|
||||
|
||||
hits: list[SearchHit] = []
|
||||
for chunk, tf_vec in zip(self.chunks, self._tf_vectors):
|
||||
score = _cosine(q_vec, {t: tf_vec[t] * self._idf.get(t, 0.0) for t in tf_vec})
|
||||
if score >= min_score:
|
||||
hits.append(SearchHit(chunk=chunk, score=score))
|
||||
|
||||
hits.sort(key=lambda h: h.score, reverse=True)
|
||||
return hits[:top_k]
|
||||
|
||||
|
||||
def _read_document_text(file_path: Path) -> str:
|
||||
suffix = file_path.suffix.lower()
|
||||
if suffix in {".txt", ".md"}:
|
||||
return file_path.read_text(encoding="utf-8", errors="ignore").strip()
|
||||
if suffix == ".docx":
|
||||
return _extract_docx_text(file_path)
|
||||
return ""
|
||||
|
||||
|
||||
def _extract_docx_text(file_path: Path) -> str:
|
||||
with zipfile.ZipFile(file_path) as archive:
|
||||
xml = archive.read("word/document.xml").decode("utf-8", errors="ignore")
|
||||
|
||||
text = re.sub(r"</w:p>", "\n\n", xml)
|
||||
text = re.sub(r"<[^>]+>", " ", text)
|
||||
text = re.sub(r"[ \t]+", " ", text)
|
||||
return re.sub(r"\n{3,}", "\n\n", text).strip()
|
||||
|
||||
|
||||
def _split_into_chunks(text: str, chunk_size: int) -> list[str]:
|
||||
paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
|
||||
if not paragraphs:
|
||||
return []
|
||||
|
||||
chunks: list[str] = []
|
||||
current = ""
|
||||
|
||||
for para in paragraphs:
|
||||
if len(para) > chunk_size:
|
||||
if current:
|
||||
chunks.append(current.strip())
|
||||
current = ""
|
||||
for i in range(0, len(para), chunk_size):
|
||||
chunks.append(para[i : i + chunk_size].strip())
|
||||
continue
|
||||
|
||||
candidate = f"{current}\n\n{para}".strip() if current else para
|
||||
if len(candidate) <= chunk_size:
|
||||
current = candidate
|
||||
else:
|
||||
if current:
|
||||
chunks.append(current.strip())
|
||||
current = para
|
||||
|
||||
if current:
|
||||
chunks.append(current.strip())
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
def _cosine(a: dict[str, float], b: dict[str, float]) -> float:
|
||||
if not a or not b:
|
||||
return 0.0
|
||||
common = set(a) & set(b)
|
||||
dot = sum(a[t] * b[t] for t in common)
|
||||
norm_a = math.sqrt(sum(v * v for v in a.values()))
|
||||
norm_b = math.sqrt(sum(v * v for v in b.values()))
|
||||
if norm_a == 0 or norm_b == 0:
|
||||
return 0.0
|
||||
return dot / (norm_a * norm_b)
|
||||
Reference in New Issue
Block a user