Розширені можливості RAG у Milvus: гібридний пошук, алгоритм HyDE, переписування запитів та їх повторне ранжування.
Поступово створіть потік обробки даних Milvus RAG: вектори типу dense та 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 ідеально підходить для цієї ролі з наступних причин:
- Він є відкритим кодом та може бути розміщений на власному сервері, що забезпечує конфіденційність даних.
- 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}. Вона справляється з 2D-розрідженими рядками, 1D-масивами та словниками, які вже були перетворені. Одна поширена проблема: завжди потрібно викликати 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 найкраще підходить для коротких, нечітких запитів, таких як процес перегляду заробітної плати, випадків розбіжностей у лексиконі між запитами та документами, а також для дуже неформального формулювання. Він є зайвим для точних запитів із великою кількістю ключових слів, де гібридний пошук вже ефективно працює, а також тоді, коли важлива затримка роботи ШІ, оскільки HyDE додає ще один виклик ШІ на кожен запит.
Розкладання запиту
Проблема. Складне, багатокомпонентне запитання не можна відповісти на основі одного уривку тексту. Запит про політику відпусток та вплив оцінки продуктивності на заробітну плату стосується двох тем у різних розділах. Одноразовий векторний пошук або зосереджується на одній темі та ігнорує іншу, або повертає посередні результати для обох.
Як це працює. ШІ розділяє запит на атомарні підзапити, кожен з яких шукається окремо, RRF об’єднує результати, а останній виклик ШІ отримує всі релевантні фрагменти.
# ════════════════════════════════════════════════════════════
# 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.