Пашыраны RAG на Milvus: гібрыдны пошук, HyDE, перапісванне запытаў і пераранжаванне
Пакрокавы RAG-канвеер на Milvus: шчыльныя і BM25-вектары, гібрыдны пошук з RRF, фільтры па метаданых, перапісванне запытаў, HyDE, дэкампазіцыя і пераранжаванне крос-энкодэрам.
Крыніцы: Beyond Semantic Search: The Complete Guide to Advanced RAG with Milvus | Sagar Patil
Сістэма доведэння концепціі RAG працюе за кілька хвілін: даній PDF роздыляецца на частыні, ці частыні апрацоўваюцца за допомогою мадэлі вбудовування, а рэзультаты перэдаюцца у LLM. Проблемі вінікаюць під час фазі віробніцтва, колі потрыбна рэалізація лішэ пошуку за векторамі. Справжні корыстувачы не формулююць запіты так, як гэта робляць оптымізовані двігуні пошуку даных. Яны вікорыстовуюць нечыткі формулювання, пропускаюць ключові слова і об’еднуюць кілька запітань у одын, праз што працэс пошуку не можа точно відтворыты суць запіту. У разі невдачы пошуку LLM змушэній узагальнюваты нерэлевантній кантэкст, што прызводыць до появі хібніх адказаў.
Шчоб подолаты розбіжносты між базовім семантычнім пошуком і багатоетапною сістэмою отрымання даных, потрыбні додаткові елементы. У наведэному ніжчэ прыклады вікорыстовуецца 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— для ўпакоўаных вектараў. Ён выконвае точны пошук найбліжэйшага суседа, ёсць абоўсумна точны, але медленей пры вялікых масштабах. Ён падходзіць для разработкі і для набораў даных, якія складаюцца з прыблізна 100K частак, а ў працэсе выкорыстання яго трэба заменіць на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 і завантажуюцца ў RAM за патрэбай, таму без гэтага вызову пошук можа выдаць адказ пра адмову або павернуць порожнія рэзультаты.
Пошук у ўплотненым фармаце як базовая мадэль
Калі даныя індэксаваны, найпростышы спосаб пошуку — гэта включэнне запіту і знаходжэнне вектараў дакументаў, якія знаходзяцца найбліжэй, на адной падставе косай схожасці:
# ════════════════════════════════════════════════════════════
# 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 года або толькі з фінансавымі дакументамі, фільтр павінен автаматычна стварацца на адпаведныя даныя.
Перапісва запітаў
Проблема. Корыстуначы пішучы запіты вжываюць неформальную мову, напрыклад «Скількі дзён выплывае мне?», тады як у дакументамах вжываецца формальная тэрмінологія, такая як «права на годовыя канікулы» або «палітры накоплення канікулярных дзён». Інтэнсыўныя пошукавыя алгорытмы часткова заполняюць гэты прыем, а правільнае перапісва значна павышае канцэнтрацыю рэзультаатаў і загальную ўспаміннасць.
Як гэта працуе. LLM стварае калькі формальных, наполненых ключоўымі словамі варыянтаў перапісванага запіта, кожны варыянт прабуваецца за дапамогою самастоятэльнага пошуку, а 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 (Hypothetical Document Embeddings)
Проблема. Короткі запіт і яго адпаведна адказ знаходзяцца ў разных частках простору імбеддынга. Запіт „Скількі дзён хворобы?“ знаходзіцца поблізу іншых запітоў. Такса правіла, якая гаворыць, што працавальнікі магуць атрымваць 7 дзён оплачванай хворобовай, знаходзіцца поблізу іншых положэнняў правіла. Гэтыя разлікі зменшаюць тачнасць.
Як гэта працюе. У зямест на тое, каб уключыць сама запітанне, LLM пішае гіпотэтычны дакумент з адказам, і самэў гэты дакумент уключаецца. Выдуманы адказ з’являецца поблізу рэальных дакументаў з адказамі, нават якшо ў яму некалькі фактов не правыя. Выварачваецца толькі яго уключэнне, а тэкст проста выкідаецца, таму выдуманыя дэталі ніколі не доходзяць да корыстніка.
# ════════════════════════════════════════════════════════════
# 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, адтуды ўжо ў кожны запіт дадаецца адна дапамога 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 і дэкомпазіцыя кожны нацелены на адзін спосаб абэктавання. Не застаўляйце іх усіх слепа. Выбірайце ў залежнасці ад дакументаў і патэранаў запытанняў, якія насправдę ствараюць карыстальнікаў.
- Пераранжаванне за дапамогою крос-энкодэра практычна завжды варта дадаць. Ён працуе локальна, даўае мінімальную затрэмтванасць для невялікага набору кандыдатаў і надзеямо павышае кантэкст на выходзе.
У канцы якосьць адчытнай інфармацыі вялікое значэнне для якосці адказу. LLM можа быць настолькі ж хорашым, насколькі кантэкст, які ёй даёма, і кожная з вышэўказаных тэхнік існуе для таго, каб гэты кантэкст быў болей рэлевантным, полным і надзеяным. Усё працюе локальна на Milvus Lite, пры чым ўнікальной зовнішняй залежнасцю для стварэння эмбедзінгаў і генеравання ёсць API OpenAI.