Векторный поиск и гибридный ранкинг с Typesense

schema = {
    "name": "pages",
    "fields": [
        {"name": "id", "type": "int32"},
        {"name": "slug", "type": "string"},
        {"name": "title", "type": "string", "locale": "ru"},
        {"name": "content", "type": "string", "locale": "ru"},
        {"name": "tags", "type": "string[]", "facet": True},
        {"name": "published_at", "type": "int64", "facet": True},
        {"name": "popularity", "type": "int32", "facet": True},
    ],
    "default_sorting_field": "published_at",
}

schema = {
    "name": "pages",
    "fields": [
        {"name": "id", "type": "int32"},
        {"name": "slug", "type": "string"},
        {"name": "title", "type": "string", "locale": "ru"},
        {"name": "content", "type": "string", "locale": "ru"},
        {"name": "tags", "type": "string[]", "facet": True},
        {"name": "published_at", "type": "int64", "facet": True},
        {"name": "popularity", "type": "int32", "facet": True},

        # новое поле для вектора
        {
            "name": "embedding",
            "type": "float[]",
            "num_dim": 768, # длина вектора
        },
    ],
    "default_sorting_field": "published_at",
}

pip install sentence-transformers

# embeddings.py
from typing import List
from sentence_transformers import SentenceTransformer

# Модель подбирай под свои нужды (мультиязычная, русская и т.д.)
MODEL_NAME = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"

_model = None


def get_model() -> SentenceTransformer:
    global _model
    if _model is None:
        _model = SentenceTransformer(MODEL_NAME)
    return _model


def embed_texts(texts: List[str]) -> List[List[float]]:
    """
    Возвращает список векторов float[] такой же длины,
    как входной список строк.
    """
    model = get_model()
    emb = model.encode(texts, batch_size=32, show_progress_bar=False)
    # преобразуем в обычные Python-списки для JSON
    return [e.tolist() for e in emb]

# mapper.py
from bs4 import BeautifulSoup
from datetime import datetime
from embeddings import embed_texts # новая функция

def clean_html(html: str) -> str:
    soup = BeautifulSoup(html or "", "html.parser")
    text = soup.get_text(" ", strip=True)
    return text

def make_snippet(text: str, max_len: int = 1500) -> str:
    if len(text) <= max_len:
        return text
    return text[:max_len] + "…"

def row_to_typesense_doc(row) -> dict:
    """
    Преобразует строку MySQL в документ Typesense с эмбеддингом.
    row: dict с полями, как в SELECT.
    """
    content_raw = row["content"]
    text = clean_html(content_raw)
    snippet = make_snippet(text)

    published_at_dt: datetime = row["published_at"]
    published_ts = int(published_at_dt.timestamp())

    tags_str = row.get("tags") or ""
    tags = [t.strip() for t in tags_str.split(",") if t.strip()]

    # базовый документ
    doc = {
        "id": row["id"],
        "slug": row["slug"],
        "title": row["title"],
        "content": snippet,
        "tags": tags,
        "published_at": published_ts,
        "popularity": row["popularity"],
        # embedding добавим чуть ниже
    }

    # Текст для эмбеддинга: заголовок + сниппет
    embed_source = f'{row["title"]}. {snippet}'
    # embed_texts ждёт список строк → вернёт список векторов
    embedding = embed_texts([embed_source])[0]
    doc["embedding"] = embedding

    return doc

# ts_vector_search.py
from typing import Dict, Any, List
from ts_client_search import client as ts_client # search-only ключ :contentReference[oaicite:12]{index=12}
from embeddings import embed_texts

COLLECTION_NAME = "pages"


def build_vector_query(embedding: List[float], k: int = 20) -> str:
    """
    Формирует строку для параметра vector_query.
    Пример формата: "embedding:([0.1,0.2,...], k:20)".
    """
    # аккуратно конвертим список float -> строку
    vec_str = ",".join(str(x) for x in embedding)
    return f"embedding:([{vec_str}], k:{k})"


def semantic_search(q: str, k: int = 20, per_page: int = 20) -> Dict[str, Any]:
    """
    Чистый семантический поиск (без текстового q).
    Возвращает raw-ответ Typesense.
    """
    q = (q or "").strip()
    if not q:
        raise ValueError("Пустой запрос для semantic_search")

    query_vec = embed_texts([q])[0]
    vector_query = build_vector_query(query_vec, k=k)

    search_params = {
        "q": "*", # текстовый поиск не используем, только вектор
        "query_by": "title,content",
        "page": 1,
        "per_page": per_page,
        "vector_query": vector_query,
        # можно исключить поле embedding из ответа, чтобы не гонять лишнее
        "exclude_fields": "embedding",
    }

    result = ts_client.collections[COLLECTION_NAME].documents.search(search_params)
    return result

def hybrid_search(q: str, page: int = 1, per_page: int = 10, k: int = 50) -> Dict[str, Any]:
    """
    Гибридный поиск: обычный текстовый + векторный.
    """
    q = (q or "").strip()
    if not q:
        # для пустого запроса гибрид не нужен
        raise ValueError("Пустой запрос для hybrid_search")

    query_vec = embed_texts([q])[0]
    vector_query = build_vector_query(query_vec, k=k)

    search_params = {
        "q": q,
        "query_by": "title,content",
        "page": page,
        "per_page": per_page,
        "vector_query": vector_query,
        "exclude_fields": "embedding",
        # остальные параметры ты можешь подкрутить под свою версию Typesense,
        # см. раздел Hybrid Search в официальной доке.
    }

    result = ts_client.collections[COLLECTION_NAME].documents.search(search_params)
    return result

# app.py (фрагмент: обновлённая логика поиска)
from flask import Flask, request, render_template
from ts_client_search import client as ts_client
from embeddings import embed_texts
from ts_vector_search import build_vector_query # из предыдущего примера

app = Flask(__name__)
COLLECTION_NAME = "pages"


@app.route("/search")
def search():
    q = (request.args.get("q") or "").strip()
    page_arg = request.args.get("page", "1")
    tag = (request.args.get("tag") or "").strip()
    sort = (request.args.get("sort") or "").strip()

    try:
        page = int(page_arg)
        if page < 1:
            page = 1
    except ValueError:
        page = 1

    per_page = 10

    search_params = {
        "q": q or "*",
        "query_by": "title,content",
        "page": page,
        "per_page": per_page,
        "highlight_full_fields": "title,content",
        "highlight_start_tag": "<mark>",
        "highlight_end_tag": "</mark>",
        "facet_by": "tags",
        "max_facet_values": 50,
    }

    filter_clauses = []
    if tag:
        filter_clauses.append(f"tags:=[{tag}]")
    if filter_clauses:
        search_params["filter_by"] = " && ".join(filter_clauses)

    if sort == "new":
        search_params["sort_by"] = "published_at:desc"
    elif sort == "popular":
        search_params["sort_by"] = "popularity:desc"

    # --- пробуем добавить векторный сигнал ---
    use_hybrid = bool(q) # только если есть текстовый запрос

    if use_hybrid:
        try:
            query_vec = embed_texts([q])[0]
            vector_query = build_vector_query(query_vec, k=50)
            search_params["vector_query"] = vector_query
            search_params["exclude_fields"] = "embedding"
        except Exception as e:
            # если что-то пошло не так — просто логируем и продолжаем без вектора
            app.logger.warning("Hybrid search disabled due to error: %s", e)

    result = ts_client.collections[COLLECTION_NAME].documents.search(search_params)

    hits = []
    for hit in result.get("hits", []):
        doc = hit.get("document", {})
        highlights = hit.get("highlights", [])
        hits.append({
            "id": doc.get("id"),
            "slug": doc.get("slug"),
            "title": doc.get("title"),
            "snippet": doc.get("content"),
            "highlights": highlights,
        })

    found = result.get("found", 0)
    facet_counts = result.get("facet_counts", [])
    pages_total = (found + per_page - 1) // per_page if found else 0

    facets = {}
    for facet in facet_counts:
        field_name = facet.get("field_name")
        counts = facet.get("counts", [])
        facets[field_name] = [
            {"value": c.get("value"), "count": c.get("count")}
            for c in counts
        ]

    return render_template(
        "search.html",
        q=q,
        tag=tag,
        sort=sort,
        hits=hits,
        found=found,
        page=page,
        pages_total=pages_total,
        per_page=per_page,
        facets=facets,
    )