Семантический поиск: embeddings и чанки на Python

Короткие выводы: семантический поиск сравнивает смысл текстов через векторы, а не совпадение слов.

Чем embedding отличается от обычного индекса#

Короткие выводы: embedding кодирует смысл; keyword-индекс кодирует слова. Вместе они покрывают больше запросов.

Как устроен пайплайн от текста до ответа#

1. 1
   Из документа извлекают полный текст.
2. 2
   Текст режут на чанки — небольшие фрагменты.
3. 3
   Для каждого чанка считают embedding.
4. 4
   В базу пишут: document_id, chunk_order, content, embedding.
5. 5
   При запросе считают embedding запроса, находят ближайшие чанки и отдают их в LLM или в UI.

Text
Копировать
документ → full_text → chunks → embeddings → таблица чанков
                                              ↓
запрос → embedding запроса → поиск ближайших → релевантные фрагменты

Короткие выводы: ищут не «документ в среднем», а конкретный фрагмент с ответом.

Зачем резать текст на чанки#

Короткие выводы: чанки дают разрешение поиска. Среднее по всей книге ≠ поиск по нужному куску.

Резать ли строго по абзацам#

• •
  резать по структуре: заголовки → абзацы → предложения;
• •
  держать размер roughly 200–500 токенов;
• •
  давать overlap 10–20%, чтобы не рвать связку между соседями;
• •
  для Markdown/HTML сначала резать по заголовкам, потом дробить крупные секции.

Короткие выводы: абзац — предпочтительная граница. Потолок по токенам и overlap обязательны.

Как хранить данные в базе#

SQL
Копировать
CREATE TABLE documents (
  id          TEXT PRIMARY KEY,
  title       TEXT NOT NULL,
  full_text   TEXT NOT NULL,
  created_at  TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE document_chunks (
  id           BIGSERIAL PRIMARY KEY,
  document_id  TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
  chunk_order  INT  NOT NULL,
  content      TEXT NOT NULL,
  embedding    VECTOR(1536) NOT NULL,  -- размерность зависит от модели
  UNIQUE (document_id, chunk_order)
);

CREATE INDEX ON document_chunks
  USING ivfflat (embedding vector_cosine_ops);

Короткие выводы: документ хранит полный текст; поиск работает по таблице чанков с векторами.

Практика на Python: от текста до поиска#

Bash
Копировать
pip install openai numpy langchain-text-splitters

Шаг 1. Подготовить документы#

Python
Копировать
DOCUMENTS = [
    {
        "id": "doc_refund",
        "title": "Возврат средств",
        "full_text": (
            "Возврат оформляют в личном кабинете в разделе «Платежи».\n\n"
            "Срок возврата — 14 дней с даты оплаты. Деньги приходят на ту же карту.\n\n"
            "Частичный возврат доступен, если товар ещё не отгрузили."
        ),
    },
    {
        "id": "doc_ship",
        "title": "Доставка",
        "full_text": (
            "Доставка занимает 3–5 рабочих дней по России.\n\n"
            "Международная доставка занимает до 21 дня.\n\n"
            "Трек-номер появляется в кабинете после передачи заказа курьеру."
        ),
    },
]

Шаг 2. Разрезать на чанки#

Python
Копировать
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,       # ориентир в символах; в проде лучше считать токены
    chunk_overlap=60,     # ~15%
    separators=["\n\n", "\n", ". ", " ", ""],
)

def make_chunks(doc: dict) -> list[dict]:
    parts = splitter.split_text(doc["full_text"])
    return [
        {
            "document_id": doc["id"],
            "chunk_order": i,
            "content": part,
        }
        for i, part in enumerate(parts)
    ]

Шаг 3. Посчитать embeddings#

Python
Копировать
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
MODEL = "text-embedding-3-small"


def embed_texts(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(model=MODEL, input=texts)
    # API возвращает объекты с index — сортируем на всякий случай
    sorted_data = sorted(response.data, key=lambda x: x.index)
    return [item.embedding for item in sorted_data]

Шаг 4. Сохранить в память как «база»#

Python
Копировать
import numpy as np

store: list[dict] = []

for doc in DOCUMENTS:
    chunks = make_chunks(doc)
    vectors = embed_texts([c["content"] for c in chunks])
    for chunk, vector in zip(chunks, vectors):
        store.append({**chunk, "embedding": np.array(vector, dtype=np.float64)})

Шаг 5. Найти ближайшие чанки по запросу#

Python
Копировать
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


def search(query: str, top_k: int = 3) -> list[dict]:
    query_vec = np.array(embed_texts([query])[0], dtype=np.float64)
    scored = []
    for row in store:
        score = cosine_similarity(query_vec, row["embedding"])
        scored.append({**row, "score": score})
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]


results = search("как вернуть деньги за заказ")
for r in results:
    print(f"{r['score']:.3f} | {r['document_id']}#{r['chunk_order']}")
    print(r["content"])
    print("---")

Шаг 6. Отдать фрагменты в LLM (RAG)#

Python
Копировать
def build_context(query: str, top_k: int = 3) -> str:
    hits = search(query, top_k=top_k)
    blocks = [
        f"[{h['document_id']}#{h['chunk_order']}]\n{h['content']}"
        for h in hits
    ]
    return "\n\n".join(blocks)


# context = build_context("срок возврата средств")
# дальше: system + context + вопрос пользователя → chat completions

Короткие выводы: минимальный контур — chunk → embed → store → cosine search → контекст для ответа.

Частая ошибка: один вектор на весь документ#

Python
Копировать
def mean_pool(vectors: list[np.ndarray]) -> np.ndarray:
    stacked = np.stack([v / np.linalg.norm(v) for v in vectors])
    mean = stacked.mean(axis=0)
    return mean / np.linalg.norm(mean)


# чанки одной «книги» про разные темы
book_chunks = [
    "Глава про доставку: сроки 3–5 дней по России.",
    "Глава про возврат: оформить заявку в кабинете за 14 дней.",
    "Глава про бонусы: кешбэк 5% за повторную покупку.",
]
book_vectors = [np.array(v) for v in embed_texts(book_chunks)]
book_mean = mean_pool(book_vectors)

query = "как оформить возврат"
q = np.array(embed_texts([query])[0])

print("score к среднему по книге:", cosine_similarity(q, book_mean))
print("score к чанку про возврат:", cosine_similarity(q, book_vectors[1]))
print("score к чанку про доставку:", cosine_similarity(q, book_vectors[0]))

Короткие выводы: для индексации источников хранят отдельный вектор на каждый чанк. Mean pooling оставляют для коротких сущностей с лимитом входа.

Что проверить перед продакшеном#

• •
  Одна и та же embedding-модель для документов и запросов.
• •
  Размерность вектора в схеме совпадает с моделью.
• •
  При обновлении документа чанки пересобирают целиком.
• •
  Размер чанка держат в диапазоне 200–500 токенов, overlap — 10–20%.
• •
  Для оценки качества готовят набор «вопрос → ожидаемый фрагмент» и меряют recall@k.
• •
  Keyword-поиск оставляют рядом: артикулы, ID, точные имена лучше ловит BM25.

• •
  OpenAI Embeddings guide
• •
  LangChain Text Splitters
• •
  Chunking strategies overview
• •
  RAG pipeline study: chunk size, overlap, hybrid
• •
  pgvector

Короткие выводы: продакшен упирается в переиндексацию, единую модель и измерение recall, а не в «магическое» поле embedding на документе.

Краткие итоги#