Расширенные возможности RAG в Milvus: гибридный поиск, HyDE, переписывание запросов и переранжирование.
Постройте пайплайн Milvus RAG пошагово: плотные и BM25-векторы, гибридный поиск RRF, фильтры метаданных, переписывание запросов, HyDE, декомпозиция и переранжирование с помощью кросс-кодера.
Источники: Beyond Semantic Search: The Complete Guide to Advanced RAG with Milvus | Sagar Patil
Создание системы RAG для демонстрации концепции занимает несколько минут: файл PDF разбивается на фрагменты, эти фрагменты обрабатываются с помощью модели генерации векторов, а полученные результаты передаются в большую языковую модель. Однако именно при работе в производственных условиях такая конфигурация, основанная исключительно на поиске векторов, сталкивается с трудностями. Настоящие пользователи не формулируют запросы так, как это делают оптимизированные движки обработки данных: они используют нечеткое формулирование, пропускают ключевые слова и объединяют несколько вопросов в один, из-за чего простой семантический поиск упускает большую часть информации. Когда поиск не дает результата, большой языковой модели приходится обобщать нерелевантный контекст, что приводит к косвенным иллюзиям или ответам, которые не помогают.
Чтобы преодолеть разрыв между базовым семантическим поиском и многоэтапной системой извлечения информации, необходимо использовать сложные решения. Приведенный ниже пример использует Milvus: сначала производится плотный поиск векторов, затем добавляется гибридный поиск, после чего применяются четыре техники (фильтрация метаданных, переписывание запросов, методы HyDE и разложение запроса) и в конце — переранжирование с использованием кросс-кодера. К каждому этапу прилагается рабочий код.
Почему существует RAG
Модели вроде GPT-4 обучаются на общих данных Интернета до определенной даты окончания обучения. Частные документы, внутренние вики и корпоративные правила остаются недоступными. Если спросить такую модель о корпоративной политике отпусков, она либо скажет, что не знает, либо придумает правдоподобный, но неверный ответ.
RAG (Retrieval-Augmented Generation) делит процесс на два шага:
- Поиск наиболее релевантных фрагментов из ваших документов с помощью семантического поиска.
- Генерация ответа путем передачи этих фрагментов в качестве контекста в LLM.
Общий алгоритм выглядит следующим образом:
User Question
│
▼
[Embed the question] → query vector
│
▼
[Search Vector DB] → top-K relevant document chunks
│
▼
[LLM prompt: "Given these passages, answer: {question}"]
│
▼
Accurate, Grounded Answer
Качество окончательного ответа зависит практически полностью от качества поиска, поэтому правильный поиск — это ключевой фактор.
Зачем нужна векторная база данных
Традиционные базы данных сравнивают точные значения, поэтому запрос WHERE text LIKE '%leave%' игнорирует документы с текстом «право на отпуск» или «медицинская нетрудоспособность», несмотря на одинаковое значение. Векторная база данных вместо этого хранит каждый документ в виде высокомерной числовой векторной структуры (эмбеддинга) и выполняет поиск по семантической близости, то есть определяет, насколько смысл запроса близок к смыслу каждого фрагмента текста. Milvus идеально подходит для этой задачи по следующим причинам:
- Это open source-решение, которое можно разместить на собственном сервере, что обеспечивает конфиденциальность данных.
- Milvus Lite работает из одного файла
.db, поэтому для разработки не требуется сервер.
Плотные и разреженные эмбеддинги
В данной системе используются два типа векторов, поэтому целесообразно ознакомиться с ними перед написанием кода. Плотный эмбеддинг — это вектор фиксированной длины, например, из 1024 чисел, генерируемый нейронной сетью. Все измерения содержат значения, и процесс обучения помогает модели располагать предложения с похожим значением рядом друг с другом:
"sick leave policy" → [0.12, -0.87, 0.34, 0.56, ...] (1024 numbers)
"medical absence entitlement" → [0.13, -0.85, 0.31, 0.54, ...] ← very close
"quarterly revenue target" → [0.91, 0.23, -0.67, 0.02, ...] ← far away
Плотные векторы хорошо передают смысл и синонимы. Их генерируют такие модели, как text-embedding-3-large от OpenAI. Редкий вектор — это огромный вектор, в котором большинство элементов равны нулю. Каждая ненулевая позиция соответствует одному слову из словаря, а её значение представляет собой степень важности этого слова, то есть вес в стиле TF-IDF:
"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}
BM25 — это классический алгоритм с низкой плотностью, превосходно справляющийся с точным поиском по ключевым словам. Он выявляет собственные имена, названия продуктов и технические термины, которые алгоритмы с высокой плотностью могут пропустить из-за отсутствия близких по смыслу аналогов.
Гибридный поиск использует оба подхода, сочетая их преимущества.
Настройка
Сначала установите необходимые зависимости:
pip install --upgrade pymilvus
pip install "pymilvus[model]"
pip install sentence-transformers
pip install langchain-text-splitters
pip install langchain-openai
pip install langchain-community
pip install scipy
pip install nltk
Затем импорты:
import uuid
from tqdm import tqdm
from pymilvus import (
MilvusClient, DataType,
AnnSearchRequest, RRFRanker
)
from pymilvus.model.sparse import BM25EmbeddingFunction
from pymilvus.model.sparse.bm25.tokenizers import build_default_analyzer
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
import scipy.sparse as sp
import re, json
import nltk
nltk.download('stopwords')
Храните все настраиваемые параметры вместе в верхней части файла:
PDF_PATH = "./data/sample_employee_handbook.pdf" # path of you document
COLLECTION_NAME = "rag_documents_hybrid"
MILVUS_DB_PATH = "./db/milvus_demo.db"
API_KEY = "sk-..."
EMBEDDING_MODEL = "text-embedding-3-large"
EMBEDDING_DIM = 1024
CHUNK_SIZE = 500
CHUNK_OVERLAP = 100
TOP_K = 5
Что касается размера блоков: 500 символов с 100-символьным перекрытием подходит в качестве отправной точки для простых документов. Нарративный текст лучше обрабатывается с более крупными блоками размером от 800 до 1000 символов, а объёмная техническая документация — с меньшими блоками от 200 до 300 символов. Перекрытие позволяет сохранять возможность нахождения предложений, обрывающихся на границе блока.
Процесс индексации
Индексация выполняется один раз или снова при изменении документов. Она преобразует сырые документы в векторы и загружает их в Milvus. Путь следующий:
PDF → Pages → Chunks → Dense Embeddings
→ Sparse Embeddings
→ Milvus Collection
Сначала инициализируйте модели и подключите их:
# Dense embedding model here we'll be using OpenAI's embedding model
embedding_obj = OpenAIEmbeddings(
model=EMBEDDING_MODEL,
api_key=API_KEY,
dimensions=EMBEDDING_DIM
)
# Milvus Lite - single file, no server needed
client = MilvusClient(MILVUS_DB_PATH)
print("Models and DB connection ready.")
Затем загрузите PDF в отдельный объект документа для каждой страницы:
# Load PDF — one Document object per page
loader = PyPDFLoader(PDF_PATH)
documents = loader.load()
print(f"Loaded {len(documents)} pages.")
# Split into overlapping chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks.")
RecursiveCharacterTextSplitter сначала делит текст на абзацы, затем — на новые строки, и только потом — на предложения, тем самым стараясь как можно дольше сохранять связанные предложения в одном блоке. Далее следуют эмбеддинги для обоих типов векторов:
texts = [doc.page_content for doc in chunks]
# Dense embeddings - one API call for the entire corpus
print("Generating dense embeddings...")
dense_embeddings = embedding_obj.embed_documents(texts)
print(f"Dense dimension: {len(dense_embeddings[0])}")
# Sparse embeddings - BM25 must be fit on YOUR corpus first
print("Fitting BM25 on corpus...")
analyzer = build_default_analyzer(language="en") # for this you will require nltk-stopwords
bm25_ef = BM25EmbeddingFunction(analyzer)
bm25_ef.fit(texts) # Builds vocabulary from your documents
sparse_embeddings = bm25_ef.encode_documents(texts)
print("Sparse embeddings generated.")
Вызов bm25_ef.fit(texts) имеет решающее значение. BM25 обучается словарю и частотам вхождения терминов на основе собственного корпуса данных, и именно поэтому он эффективен. Использование стандартной модели BM25 значительно снижает качество поиска по ключевым словам. А вот коллекция с её схемой и индексами:
# Drop and recreate for a clean state
if COLLECTION_NAME in client.list_collections():
client.drop_collection(COLLECTION_NAME)
# Define schema
schema = client.create_schema()
schema.add_field("id", DataType.VARCHAR, is_primary=True, max_length=100)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=EMBEDDING_DIM)
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
schema.add_field("text", DataType.VARCHAR, max_length=65535)
schema.add_field("page_number", DataType.INT64)
schema.add_field("source", DataType.VARCHAR, max_length=500)
schema.add_field("chunk_id", DataType.INT64)
# Create the collection
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)
# Build indexes separately
index_params = client.prepare_index_params()
index_params.add_index(
field_name="vector",
index_type="FLAT", # Exact search - swap to HNSW for production
metric_type="COSINE"
)
index_params.add_index(
field_name="sparse_vector",
index_type="SPARSE_INVERTED_INDEX",
metric_type="IP" # Inner Product is the only valid metric for sparse
)
client.create_index(collection_name=COLLECTION_NAME, index_params=index_params)
print("Collection and indexes created.")
Здесь важны два варианта индекса:
FLAT— для плотных векторов. Он выполняет точный поиск ближайших соседей, обеспечивая полную точность, но работает медленнее при больших объемах данных. Он подходит для разработки и наборов данных объемом менее примерно 100 тыс. элементов, в то время как в производственных условиях его следует заменить наHNSW.SPARSE_INVERTED_INDEX— для разреженных векторов. Это единственный допустимый тип индекса для разреженных данных в Milvus; он функционирует как классический обратный индекс, похожий на тот, который используют поисковые системы для индексации документов по словам.
Наконец, подготовьте записи и вставьте их:
def sparse_to_dict(s_emb) -> dict:
"""Convert a scipy sparse row into Milvus-compatible {index: value} dict."""
if sp.issparse(s_emb):
coo = s_emb.tocoo()
return {int(col): float(val) for col, val in zip(coo.col, coo.data)}
elif isinstance(s_emb, dict):
return s_emb
else:
return {int(i): float(v) for i, v in enumerate(s_emb) if v != 0.0}
# Build the records list
data = []
for idx, (chunk, d_emb) in enumerate(tqdm(zip(chunks, dense_embeddings), total=len(chunks))):
sparse_dict = sparse_to_dict(sparse_embeddings[idx])
if not sparse_dict:
print(f"Warning: empty sparse vector at chunk {idx}, skipping.")
continue
data.append({
"id": str(uuid.uuid4()),
"vector": d_emb,
"sparse_vector": sparse_dict,
"text": chunk.page_content,
"page_number": int(chunk.metadata.get("page", -1)),
"source": PDF_PATH,
"chunk_id": idx
})
# Insert into Milvus
res = client.insert(collection_name=COLLECTION_NAME, data=data)
print(f"Inserted {res['insert_count']} records.")
# Load into memory - required before any search operation
client.load_collection(COLLECTION_NAME)
print(f"Load state: {client.get_load_state(COLLECTION_NAME)}")
Помощник sparse_to_dict необходим, поскольку алгоритм BM25 генерирует разреженную матрицу из библиотеки scipy, тогда как Milvus требует обычного словаря Python вида {int: float}. Он справляется с двумерными разреженными строками, одномерными массивами и уже преобразованными словарями. Одна распространенная проблема: всегда следует вызывать client.load_collection() перед поиском. Индексы хранятся на диске в Milvus и загружаются в оперативную память по мере необходимости, поэтому без этого вызова поиск либо завершается с ошибкой, либо возвращает пустой результат.
Поиск с плотными данными как базовый вариант
После индексации данных самый простой способ поиска заключается в подстановке запроса и нахождении векторов ближайших документов с использованием косинусного сходства:
# ════════════════════════════════════════════════════════════
# Dense Vector Search
# ════════════════════════════════════════════════════════════
query = "What is the leave policy?"
# Step 1: Embed the query using the same model used at index time
query_dense_embedding = embedding_obj.embed_query(query)
# Step 2: Search
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_dense_embedding],
anns_field="vector",
search_param={"metric_type": "COSINE"},
limit=TOP_K,
output_fields=["text", "page_number", "source"]
)
# Step 3: Display results
for idx, hit in enumerate(results[0], start=1):
entity = hit["entity"]
print(f"Rank {idx} | Cosine Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
print(f" {entity['text'][:300]}\n")
При использовании метрики COSINE оценки варьируются от -1 до 1. Значение выше 0,7 обычно указывает на сильное семантическое совпадение, а лучшие результаты обычно находятся в диапазоне от 0,6 до 0,95. Для вопросов на естественном языке, ответы на которые находятся в близкой семантической области, достаточно плотного поиска — например, при запросе о объяснении льгот по отпуску по уходу за ребенком или о содержании кодекса поведения. Однако такой метод плохо справляется, когда в запросе присутствуют конкретные коды, аббревиатуры или собственные имена, которые модель эмбеддингов редко встречала.
Гибридный поиск с RRF
В гибридном поиске одновременно выполняются плотный поиск и разреженный поиск BM25, после чего оба ранжированных списка объединяются с помощью метода Reciprocal Rank Fusion (RRF). Идея проста: к оценке документа по RRF добавляются значения 1 / (k + rank) из каждого списка, причем k=60 служит константой сглаживания, поэтому документы, занимающие высокие позиции в обоих списках, оказываются в верхней части результата.
# ════════════════════════════════════════════════════════════
# Hybrid Search (Dense + Sparse)
# ════════════════════════════════════════════════════════════
query = "leave policy?"
# Dense query vector
query_dense = embedding_obj.embed_query(query)
# Sparse query vector - uses the same BM25 model fitted on the corpus
sparse_raw = bm25_ef.encode_queries([query])
sparse_dict = sparse_to_dict(sparse_raw[0])
print(f"Sparse query terms: {len(sparse_dict)}") # Should be > 0
# Build two separate ANN search requests
dense_req = AnnSearchRequest(
data=[query_dense],
anns_field="vector",
param={"metric_type": "COSINE"},
limit=TOP_K
)
sparse_req = AnnSearchRequest(
data=[sparse_dict],
anns_field="sparse_vector",
param={"metric_type": "IP"},
limit=TOP_K
)
# Execute hybrid search with RRF fusion
results = client.hybrid_search(
collection_name=COLLECTION_NAME,
reqs=[dense_req, sparse_req],
ranker=RRFRanker(k=60),
limit=TOP_K,
output_fields=["text", "page_number", "source"]
)
for idx, hit in enumerate(results[0], start=1):
entity = hit["entity"]
print(f"Rank {idx} | RRF Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
print(f" {entity['text'][:300]}\n")
О оценках: значение hit['distance'] из функции hybrid_search представляет собой суммированный балл ранжирования RRF, а не чистое значение косинусного сходства. Чем выше значение, тем лучше, хотя его диапазон не ограничивается значениями от 0 до 1, поэтому его не следует сравнивать с баллами косинусного сходства, полученными при использовании обычной функции search(). Для производственных систем типа RAG рекомендуется использовать гибридный способ поиска в качестве стандартного. Метод BM25 практически не увеличивает задержку и заметно улучшает точность воспроизведения ответов при запросах, содержащих множество ключевых слов.
Даже гибридный способ поиска может давать сбои в определённых ситуациях, и существуют четыре метода для устранения наиболее распространённых проблем.
Фильтрация метаданных
Проблема. Когда набор данных включает несколько документов, периодов или отделов, векторный поиск может возвращать значимые по смыслу фрагменты из неправильного источника — например, из устаревшей версии политики, руководства другого отдела или страниц, не относящихся к теме запроса.
Как это работает. Milvus применяет скалярное выражение фильтра перед векторным поиском, так что только соответствующие строки доходят до этапа ANN. Это не постфильтр. Фильтр сначала уменьшает количество кандидатов, что позволяет сохранить точность в отфильтрованном наборе.
# ════════════════════════════════════════════════════════════
# METADATA FILTERING
# ════════════════════════════════════════════════════════════
def search_with_metadata_filter(
client, collection_name, embedding_obj, bm25_ef,
query: str,
page_range: tuple = None,
source_file: str = None,
top_k: int = 5
):
filter_parts = []
if page_range:
lo, hi = page_range
filter_parts.append(f"page_number >= {lo} && page_number <= {hi}")
if source_file:
filter_parts.append(f'source == "{source_file}"')
filter_expr = " && ".join(filter_parts) if filter_parts else None
print(f"\n[Metadata Filter] Query : '{query}'")
print(f"[Metadata Filter] Filter: {filter_expr or 'None (unfiltered)'}")
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=query,
top_k=top_k,
filters=filter_expr
)
return results
# ── Run ──────────────────────────────────────────────────────
meta_results = search_with_metadata_filter(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
page_range = (1, 30),
source_file= None,
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nMETADATA-FILTERED RESULTS")
print("=" * 55)
if not meta_results or not meta_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(meta_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
Выражения фильтров следуют этой синтаксису:
'page_number >= 1 && page_number <= 30' # page range
'source == "hr_policy.pdf"' # exact source
'category in ["leave", "performance"]' # in a list
'source like "hr%"' # prefix match
Любая система RAG с поддержкой нескольких документов должна включать фильтрацию метаданных. Если пользователи работают в определённом диапазоне, например, только с руководством за 2024 год или только с финансовыми документами, фильтр должен автоматически формироваться на основе этого диапазона.
Переписывание запросов
Проблема. Пользователи формулируют запросы непринуждённо, например, «Сколько дней отпуска у меня есть?», тогда как в документах используется формальная терминология вроде «право на годовой отпуск» или «правила накопления дней отпуска». Плотный поиск частично сокращает эту пропасть, а хорошее переписывание значительно улучшает качество поиска и общую точность результатов.
Как это работает. Нейросеть генерирует несколько формальных вариантов переписанного запроса с акцентом на ключевые слова; каждый вариант обрабатывается отдельно в поиске, после чего система RRF объединяет полученные результаты и удаляет дубликаты.
# ════════════════════════════════════════════════════════════
# QUERY REWRITING
# ════════════════════════════════════════════════════════════
import re, json
REWRITE_PROMPT = """You are an expert at reformulating search queries to improve document retrieval.
Given a user query, produce {n} alternative search queries that:
- Use formal, document-style language
- Include relevant keywords and synonyms
- Cover different angles of the same question
User query: {query}
Respond ONLY with a JSON array of strings. Example:
["rewritten query 1", "rewritten query 2", "rewritten query 3"]"""
def rewrite_query(query: str, n: int = 3) -> list[str]:
prompt = REWRITE_PROMPT.format(query=query, n=n)
response = llm.invoke(prompt)
raw = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
try:
variants = json.loads(raw)
return [query] + variants # always keep the original
except json.JSONDecodeError:
print("Warning: Could not parse rewrites, using original query only.")
return [query]
def search_with_query_rewriting(
client, collection_name, embedding_obj, bm25_ef,
query: str,
n_rewrites: int = 3,
top_k: int = 5
):
variants = rewrite_query(query, n=n_rewrites)
print(f"\n[Query Rewriting] Original : '{query}'")
for i, v in enumerate(variants[1:], 1):
print(f"[Query Rewriting] Variant {i} : '{v}'")
seen_ids = {}
rank_scores = {}
for variant in variants:
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=variant,
top_k=top_k
)
if not results or not results[0]:
continue
for rank, hit in enumerate(results[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
return [merged]
# ── Run ──────────────────────────────────────────────────────
rewrite_results = search_with_query_rewriting(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
n_rewrites = 3,
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nQUERY-REWRITTEN RESULTS")
print("=" * 55)
if not rewrite_results or not rewrite_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(rewrite_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
Пример преобразования:
Input: "how many days off do I get?"
Output variants:
1. "annual leave entitlement number of days employee handbook"
2. "vacation days accrual policy full-time employee"
3. "paid time off PTO allowance per calendar year"
Каждая вариантная версия расширяет область поиска. Документ, не входящий в топ-5 для исходного запроса, может занять второе место при использовании третьей вариантной версии, а метод RRF выявляет документы, которые регулярно появляются в разных запросах.
HyDE (гипотетические вкладыши документов)
Проблема. Короткий вопрос и его ответ находятся в разных участках пространства вкладышей. Вопрос «Сколько дней болезни?» находится рядом с другими вопросами. Положение, гласящее, что сотрудники имеют право на 7 оплачиваемых дней больничного в год, находится рядом с другими положениями правил. Это расхождение снижает точность поиска.
Как это работает. Вместо того чтобы встраивать сам вопрос, большая языковая модель пишет гипотетический документ с ответом, и именно его встраивают. Вымышленный ответ оказывается рядом с настоящими документами с ответами, даже если в нем приведены неверные факты. Используется только его встраивание, а текст выбрасывается, поэтому выдуманные детали никогда не попадают к пользователю.
# ════════════════════════════════════════════════════════════
# HyDE (Hypothetical Document Embeddings)
# ════════════════════════════════════════════════════════════
HYDE_PROMPT = """You are a corporate policy document writer.
Write a 2-3 paragraph excerpt from an official HR policy or company document
that would DIRECTLY ANSWER the following question.
Write in formal document style. Do not mention the question itself.
Question: {query}
Document excerpt:"""
def generate_hypothetical_document(query: str) -> str:
response = llm.invoke(HYDE_PROMPT.format(query=query))
return response.content.strip()
def search_with_hyde(
client, collection_name, embedding_obj, bm25_ef,
query: str,
top_k: int = 5
):
hypothetical_doc = generate_hypothetical_document(query)
print(f"\n[HyDE] Query : '{query}'")
print(f"[HyDE] Hypothetical doc :\n {hypothetical_doc[:300]}...\n")
# Search using the hypothetical document's embedding
hyde_results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=hypothetical_doc, # embed the answer, not the question
top_k=top_k
)
# Also search with the original query and merge both via RRF
original_results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=query,
top_k=top_k
)
seen_ids = {}
rank_scores = {}
for result_set in [hyde_results, original_results]:
if not result_set or not result_set[0]:
continue
for rank, hit in enumerate(result_set[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
return [merged]
# ── Run ──────────────────────────────────────────────────────
hyde_results = search_with_hyde(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nHyDE RESULTS")
print("=" * 55)
if not hyde_results or not hyde_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(hyde_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
HyDE наиболее полезен для коротких, нечетких запросов, таких как процедура пересмотра зарплаты, с пробелами в словарном запасе между вопросами и документами, а также при очень неформальной формулировке. Он избыточен для точных запросов с большим количеством ключевых слов, где гибридный поиск уже показывает хорошие результаты, и когда важна задержка работы LLM, поскольку HyDE добавляет еще один вызов LLM на каждый запрос.
Разложение запроса
Проблема. Сложный, многокомпонентный вопрос невозможно ответить на основе одного отрывка текста. Вопросы о политике отпусков и о том, как оценка эффективности влияет на зарплату, касаются двух тем в двух разделах. Одиночный поиск по векторам либо сосредотачивается на одной теме и игнорирует другую, либо дает посредственные результаты для обеих.
Как это работает. LLM разделяет вопрос на атомарные подвопросы, каждый из которых ищется отдельно, затем RRF объединяет результаты, и в последнем вызове LLM получаются все релевантные фрагменты.
# ════════════════════════════════════════════════════════════
# QUERY DECOMPOSITION
# ════════════════════════════════════════════════════════════
DECOMPOSE_PROMPT = """You are an expert at breaking down complex questions for document retrieval.
Decompose the following question into 2-4 simple, self-contained sub-questions.
Each sub-question should target a single distinct piece of information.
Complex question: {query}
Respond ONLY with a JSON array of strings. Example:
["sub-question 1", "sub-question 2", "sub-question 3"]"""
def decompose_query(query: str) -> list[str]:
response = llm.invoke(DECOMPOSE_PROMPT.format(query=query))
raw = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
try:
return json.loads(raw)
except json.JSONDecodeError:
print("Warning: Could not parse decomposition, using original query.")
return [query]
def search_with_decomposition(
client, collection_name, embedding_obj, bm25_ef,
query: str,
top_k: int = 5
):
sub_questions = decompose_query(query)
print(f"\n[Decomposition] Original query : '{query}'")
for i, sq in enumerate(sub_questions, 1):
print(f"[Decomposition] Sub-question {i} : '{sq}'")
per_subquery_results = {}
seen_ids = {}
rank_scores = {}
for sq in sub_questions:
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=sq,
top_k=top_k
)
per_subquery_results[sq] = results
if not results or not results[0]:
continue
for rank, hit in enumerate(results[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
# Per sub-question breakdown
print("\n── Per Sub-question Results ──")
for sq, res in per_subquery_results.items():
print(f"\n SUB-QUERY: '{sq[:60]}'")
if res and res[0]:
for i, hit in enumerate(res[0], start=1):
print(f" {i}. Page {hit['entity']['page_number']} | Score {hit['distance']:.4f} | {hit['entity']['text'][:150]}")
return {"per_subquery": per_subquery_results, "merged": [merged]}
# ── Run ──────────────────────────────────────────────────────
decomp_results = search_with_decomposition(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy and how does it affect salary deductions?",
top_k = 5
)
# ── Print Merged Results ──────────────────────────────────────
print("\nDECOMPOSED — MERGED FINAL RESULTS")
print("=" * 55)
merged_hits = decomp_results["merged"]
if not merged_hits or not merged_hits[0]:
print("No results returned.")
else:
for idx, hit in enumerate(merged_hits[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
Пример разложения:
Input: "What is the leave policy and how does performance review affect salary?"
Sub-questions:
1. "What is the annual leave policy?"
2. "How many sick days are employees entitled to?"
3. "How does performance review affect salary?"
4. "What is the performance review schedule?"
Каждый подвопрос самостоятельно запрашивает необходимые фрагменты текста, и объединенный набор охватывает все аспекты исходного вопроса, чего не смогла бы сделать отдельная поисковая операция.
Переранжирование с использованием кросс-энкодеров
До сих пор все операции поиска выполнялись с использованием би-энкодеров: запрос и документы эмбеддингуются независимо, а сходство определяется с помощью скалярного произведения или косинусного расстояния. Этот метод быстр, но не является наиболее точным способом оценки релевантности. Кросс-энкодеры считывают запрос и один документ вместе как пару и оценивают степень их соответствия, учитывая всю их взаимодействие, что делает их гораздо более точными. Однако они снижают скорость обработки: слишком медленны для тысяч документов, но быстры при работе с небольшим набором из 20–50 кандидатов. Отсюда и такая схема: использовать би-энкодер для широкого поиска, а кросс-энкодер — для узкого переранжирования.
# ============================================================
# RERANKING WITH CROSS-ENCODER
# ============================================================
from sentence_transformers import CrossEncoder
# Huggingface: cross-encoder/ms-marco-MiniLM-L12-v2
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L12-v2")
query = "What is the leave policy?"
RETRIEVAL_K = 20 # fetch more than you need
FINAL_K = 5 # rerank down to this
# Step 1: Broad retrieval - fetch 20 candidates
query_dense = embedding_obj.embed_query(query)
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_dense],
anns_field="vector",
search_param={"metric_type": "COSINE"},
limit=RETRIEVAL_K,
output_fields=["text", "page_number", "source"]
)
hits = results[0]
print(f"Retrieved {len(hits)} candidates for reranking.")
# Step 2: Score each (query, chunk) pair with the cross-encoder
pairs = [[query, hit["entity"]["text"]] for hit in hits]
rerank_scores = cross_encoder.predict(pairs)
# Step 3: Sort by cross-encoder score
for hit, score in zip(hits, rerank_scores):
hit["rerank_score"] = float(score)
reranked = sorted(hits, key=lambda x: x["rerank_score"], reverse=True)[:FINAL_K]
# Step 4: Display
for idx, hit in enumerate(reranked, start=1):
entity = hit["entity"]
print(f"Rank {idx} | Rerank: {hit['rerank_score']:.4f} | Vector: {hit['distance']:.4f}")
print(f" Page {entity['page_number']}: {entity['text'][:300]}\n")
Оценки кросс-энкодеров представляют собой ненормализованные логиты, а не вероятности. Более высокое значение указывает на более сильную релевантность, поэтому 8.0 значительно превосходит 2.0, однако абсолютная шкала различается у разных моделей. Используемая модель ms-marco-MiniLM-L12-v2 — это мощный и легкий вариант, обученный на наборе данных MS MARCO для ранжирования текстов, и она хорошо справляется с запросами в формате вопросов к фрагментам документов.
Как сравниваются эти технологии
Разумный стандартный набор инструментов для производства включает:
- Гибридный поиск в качестве основного метода получения результатов, активного всегда.
- Фильтрацию метаданных каждый раз, когда можно определить диапазон поиска пользователя.
- Переранжирование с помощью кросс-энкодера для топ-20 кандидатов перед их передачей в LLM.
Эволюция методов выглядит следующим образом:
- Плотный поиск является отправной точкой: он быстрый, не зависит от ключевых слов и хорошо справляется со синонимами.
- Гибридный поиск представляет собой улучшенную версию, поскольку использование разреженных векторов BM25 надёжно повышает точность нахождения результатов с минимальным дополнительным уровнем сложности.
- Фильтрация, переписывание запроса, HyDE и декомпозиция каждый из них направлен на устранение определённого типа ошибок. Не следует применять их все одновременно без разбора. Выбирайте методы в зависимости от документов и шаблонов запросов, которые фактически используют пользователи.
- Переупорядочивание с использованием кросс-кодера почти всегда стоит добавить. Он работает локально, добавляет небольшую задержку при обработке небольшого набора кандидатов и надёжно улучшает качество окончательного результата.
В конечном итоге качество поиска определяет качество ответа. Глубокая нейронная сеть может быть настолько хороша, насколько хорош контекст, предоставленный ей, и каждая из описанных выше техник предназначена для того, чтобы сделать этот контекст более релевантным, полным и надежным. Все процессы выполняются локально в Milvus Lite, причем единственной внешней зависимостью для генерации эмбеддингов и выполнения операций является API OpenAI.