Практычныя прытамулкі: За межамі семантичнага пошуку: Повныя інструкцыі па прымітнах RAG
Практычныя прытамулкі: За межами семантичнага пошуку: Повныя інструкцыі па прымітнах RAG: контракты, перакананні та слоты для коду для команд, якія викорыстоўваюць гэты патэрн.
У гэтым карыце парадоксальнае рашчыранне шляху ад сыр'ёў да рабочай системы для кніги «Beyond Semantic Search: The Complete Guide to Advanced RAG with Milvus» ад яе автара. Акцэнт ставіцца на практычныя крокі, чыстае перакананне і код, які можна проста дадзіць у репазітарый без неабяснення меты. У стадіўцы «Аптаварыс» неабходна з'явіць вхідныя даны, адпаведальнага за крок і критэрыя завершэння прычым перад зменай коду. Аперацыйныя працавнікі должны магчымае перадзеўсці крок з вядомага пункта контролю, не спрабоўваючы з'ясаваць схованы стан. Запісваюцца час выконання і вартасць токенав або запытаў разам з функцыйнальнымі рэзултатамі. Відразувыя даны пра вартасці запобегаюць неспакойным рахункам, калі шлях пераходзіць з дэмавайнага режыма ў спакульнаныя сераўысы.
Што такое RAG і чаму ён існуе?
Калі працюеце над тэмай «Што такое RAG і які ў яго етапы», спачатку запісайце умовы працы: неабяжныя даны, сігнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае заліцвачыць змяны ў кодзе пазнейша. Храніце настройкі за межамі коду прыемліка. Файлы сераўіса, базы секрэтных даных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаюць адбавіць аудыт без неабяжнага чытання всіх элементаў системы. Перад налаштаваннем запитоў пераканайцеся, як працуе алгорытм воспамінання на фіксованай сэтке запытаў. Частае зміненне запытаў рэдка калі вярна выправляе слабыя аспекты процесу пошуку.
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
Ролі векторной базы дадзеных
Калі працюеце над «Роллю стэджу», спачатку запісайце контракт: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага неяксамоства. Такі список пераканаецца дапамагае залічваць змяны ў кодзе чыста.
Розумэнне эмбеддзінаў: ўжыткавыя і рэдкасныя
Калі працуеце над раздзелам «Understanding Embeddings Dense», спачатку запісайце умовы викорыстоўвання: неабходныя даны, сігнал успеху і тое, што выходзіць па частковай нявыполненасці. Такі список дапамагае залічыць пазнейшыя змены ў кодзе.
Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, прычына нявыполненасці павінна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач.
Перад налаштаваннем запитоў пераканайцеся ў рэверсі на фіксаваным наборе запитанняў. Частыя змены запитоў рэдка калі вядуць да павышэння якасці выкарыстоўвання дадзеных.
Калі працуеце над раздзелам «Understanding Embeddings Dense», спачатку запісайце умовы викорыстоўвання: неабходныя даны, сігнал успеху і тое, што выходзіць па частковай нявыполненасці. Такі список дапамагае залічыць пазнейшыя змены ў кодзе.
Запісвайце час выконання і вартасьць токенав або запыта разам з функцыйнальнымі рэзултатамі. Відразувыя даны пра вартасьці запобегаюць неспакойным рахункам, калі праця пераходзіць з дэмавайнага режыма ў спяльныя сераўеры.
Жыцкае падмешчанне
Этап жыцкага падмешчання працюе наявнасцю краща, калі яго розглядаць як вимерную паверхню. Запісаўце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Зберагаюце настройкі праза код прыемліка. Файлы сяродавішча, хранілішча секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, куды аператары можаць адбавляць без неабяжнага чытання всіх дадзеных. Раздзеляйце правілы часткавання дадзеных ад правілаў ўзяць іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцца паказнікі якосці.
"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
Разрэджанае падмешчанне (BM25)
Этап Sparse Embeddings BM25 працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксавце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Дакументавце як шлях успеху, так і шлях вярнэння. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў ёсць частью продукту, а не пасляднім дапрацоўкам. Раздзеліце правілы частковай обработкі даных ад правілаў ўзяць даныя. Змена аднаго з іх не должна вымагаць перапісвання другога, калі зменяюцца паказнікі якосці.
"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}
Чаму вам патрэбны оба
Этап «Прычыны, чаму патрэбны оба» работае найкраща, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідеальны прыемлівы результат, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перад расшырэнням масштаба. Валіце маленькія, тэставальныя элементы замест большых скрыптов. Калі якісь крок не выйшоў, неудача должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцужок задач. Раздзеляйце правілы часткавання інфармацыі ад правіл яе выявлення. Змена ў одных не павинна прымусваць перапісванне іншых, калі змянююцца паказнікі якосці. Этап «Прычыны, чаму патрэбны оба» работае найкраща, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідеальны прыемлівы результат, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перад расшырэнням масштаба. Запісвайце часы выконання і косты токеноў або запытаў разам з функцыйнальнымі рэзултатамі. Відразувыя даны пра косты запобегаюць неспакою, калі процес пераходзіць з дэмавайшанняў у спяльныя сераўы.
Настройка проекту і залежнасці
У стадії налагоджэння проекта і вызначэння залежнасцей неабяжна практычна вказаць інпуты, адміністратара крока та критэрыя завершэння пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не падозрываючы прыхованы стан. Конфігурацыю трэба знаходзіць за межамі коду прыемленае. Файлы сераўнавання, хранільнікі секрэтных дадзеных і флагі функцыйяў должны быць аднароджаны ў аднам месца, якое аператары можаць пераглядаць, не чытаючы весь ланцуг задач. Неабяжна цітаваць тые часткі, якія фактычна лежалі в основе адпаведнай адказы. Без цітатаў аператары не можаць разлічыць галюцинацыю ад працягу індэксавання.
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
Стварэнне канвею індэксавання
Для стадіі стварэння каналу індексавання неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выконваецца, прычына нехарактэрства павінна вказываць на адзін конкрэтны элемент, а не на заплутаны канал. Прыцісніце цитаты з тых частак, якія фактычна падтрымлівалі адпаведны адказ. Без цитатаў аператары не можуць разлічыць галюцинацію ад працягу індексавання. Для стадіі стварэння каналу індексавання неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Запісвайце час выконання і вартасьць токеноў або запытаў разам з функцыйнальнымі рэзультатамі. Відразлівасьць вартасцей з самага пачатку запобегае неспакою, калі процес пераходзіць з дэмаверсіі на іншы режым.
чырвоныя среды.PDF → Pages → Chunks → Dense Embeddings
→ Sparse Embeddings
→ Milvus Collection
Шаг 1 і 2: Ініцыялізацыя модэляў і з’ёднанне
Калі працуеце над стадіяй Шаг 1 і 2 – ініцыялізацыя, спачатку запісайте умовы викорыстання: неабяжлівыя даннэ, сигнал успеху і тое, што выходзіць на падзею частковага невыпання. Такі список контролю дапамагае заліцварыць пазнейшыя змены ў кодзе. Зберагаюце настройкі паза кодам прыемліка. Файлы среды, храненні секрэтных данных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можу працаваць без неабяжлівага чытання всей структуры. Кэшуйце стабільныя інструкцыі системы і схемы інструментаў. Перадзесланне ідэнтычных прамаўляючых частак є распашчытным факторам витрачання ресурсаў.
# 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.")
Шаг 3 і 4: Завантажэнне і разбіўка дакумента
Калі працуеце над стадзіяй 3 і 4 «Завантажэнне», спачатку запісайце умовы контракту: неабяжлівыя даны, сигнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае заліцварыць пазнейшыя змены ў кодзе. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія перакрыцця і обробка некоректных паведамленняў є часткай продукту, а не элементамі пазнейшага дапрацоўкі. Перад налаштаваннем запитаў пераканайцеся ў рэверсіі на фіксаваным наборе запитаў. Частая зміна запитоў рэдка калі вярнайце слабкую эфектыўнась запошуку.
# 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.")
Шаг 5: Стварыць оба типу імбеддзінга
Калі працуеце над этапам 5 «Стварыць оба», спачатку запісайте умовы працы: неабяжлівыя данні, сигнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае заліцвачыць пазнейшыя змены ў кодзе. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выканаецца, прычына нявыполнення павінна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся, наскількі эфектыва система адпаведзядае на фіксаваны набор запитаў. Частыя змены запитоў рэдка калі вядуць да павышэння якосці адпаведзенняў.
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.")
Калі працуеце над этапам 5 «Стварыць оба», спачатку запісайте умовы працы: неабяжлівыя данні, сигнал успеху і тое, што выканаецца у разе частковага нявыполнення. Такі список контроля дапамагае заліцвачыць пазнейшыя змены ў кодзе. Запісвайце час выканання задач і кост токеноў або запитаў разам з функцыональнымі рэзултатамі. Відразувыя даны пра косты запобегаюць неспакойным рахункам, калі праця пераходзіць з дэмаверсіі ў спакульнаныя сераўысы.
Шаг 6: Стварэнне калекцыі з схемай і індексамі
Шаг 6 «Стварэнне» будзе эфектываў, якш тлумачыць яго як меравальную площыну. Зафіксавайце адна ідеальная транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Зберагаюце настройкі паза кодам прыемліка. Файлы сераўнавання, хранілішчы секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, куда аператары можуць адбавляць контроль без неабяжнага чытання всіх дадзеных. Раздзеляйце політыку часткавання дадзеных ад політыки ўтрымання іх. Змена адной з яных не должна вымагаць перапісвання другой, калі зменяюцыся паказнікі якосці.
# 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.")
Шаг 7 і 8: Падготовка записаў і ўведэнне
Этап падзеў 7 і 8 «Падготовка» работае наякша, калі яго спрыяваць як мерыемую паверхню. Зафіксавайце адна «золатая» транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Дакументавайце як шлях успеху, так і шлях вярнэння. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсць частью продукту, а не пасляднім дапрацоўкам. Раздзеліце політыку частакавання дадзеных ад політыкі ўтрымання іх. Змена адной з яных не должна вымагаць перапісвання другой, калі зменяюцыся паказнікі якосці.
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)}")
Базавы RAG: Цяжкая векторная пашуковая сістэма
Этап Basic RAG Dense Vector працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксавце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць сферу дзеяння. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны ланцюг задач. Раздзеляйце правілы часткавання дадзеных і правілы ўзяць дадзеныя. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцца паказнікі якасці. Этап Basic RAG Dense Vector працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксавце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць сферу дзеяння. Запісвуйце часы выканення і косты токэнаў або запитаў разам з функцыональнымі рэзултатамі. Відразувыя даны пра косты запобегаюць неспакоўным рахункам, калі процес пераходзіць з дэмовай среды ў спакульную.
# ════════════════════════════════════════════════════════════
# 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")
Лепшы RAG: Гібрыдны пошук (Dense + Sparse)
Для стадіі кращага гібрыднага пошуку RAG неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры зміне коду. Аперацыйныя працавнікі должны магчыма ўвайсці крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Конфігурацыю трэба залічыць паза кодам прыкладнага програму. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое працавнікі можуць пераглядаць, не чытаючы весь структураны код. Паказваць трэба тыя часткі тексту, якія фактычна ляглі в основу адпаведнай адказы. Без цых цітатаў працавнікі не зможуць адразніць галюцинацыю ад працягу праз складанне індэкса.
# ════════════════════════════════════════════════════════════
# 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")
Развітыя тэхнікі RAG — чатыры методы выкарыстоўвання дадзеных
Для стадіі Advanced RAG Four Retrieval неабяжна прадзеўкаваць вхідныя даны, адпаведальнага за этап і крэтыя выходу пры перадзеўкаванні коду. Аператары должны магчыма было перзапусціць этап з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вяснавання проблем. Перапрыбуткі, людзкія пераказы і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Паказваць часткі тексту, якія фактычна сталі падставай для адпаведнай адказы. Без ціх цитатаў аператары не можуць разлічыць галюцинацію ад прасоўкі ў індэксаванні.
Фільтраўванне метадаў
Для стадіі фільтрацыі мета-даных неабяжна прадзефінаваць вхідныя даны, адпраўніка крока і крэтыяры завершэння пры перадзмене коду. Аперацыйныя працавнікі должны магчымае перайсці на выкананне крока з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест абмежлівых скрыптав. Калі крок не выканаецца, прычына неудачы должна вказываць на адну конкрэтную адпраўніцу, а не на заплутаны ланцоўкі задач. Наводзіце тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аперацыйныя працавнікі не зможуць адразніць галюцинацыю ад працягу індэксавання. Для стадіі фільтрацыі мета-даных неабяжна прадзефінаваць вхідныя даны, адпраўніка крока і крэтыяры завершэння пры перадзмене коду. Аперацыйныя працавнікі должны магчымае перайсці на выкананне крока з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Запісвайце час выканання, а таксама кост токенав або запытаў разам з функцыйнаімі рэзультатамі. Відразуваяя відомасць костаў запобегае неспакойным рахункам, калі процес пераходзіць з дэмовай среды ў спяльнаныя сераўеры.
# ════════════════════════════════════════════════════════════
# 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
Перапісвацьце запитоў
Калі працюяте над стадіяй перапісвацьця запитоў, спачатку запішыце умовы викорыстання: неабходныя даны, сігнал успеху та тое, што вядзець да частковага невыпання. Такі список дапамагае заліцьваты змяны ў кодзе пазнейша. Храніце настройкі праза код аплікацыі. Файлы сераўнавання, базы секрэтных дадзеных та флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можу пераглядаць іх без неабходнасці чытання всей структуры. Перад налаштаваннем запрошэнняяў пераканайцеся ў рівні відтворэння дадзеных на фіксаванай групе запытанняў. Частае змінэ запрошэнняяў рэдка калі выправляе слабкія аспекты пошуку.
# ════════════════════════════════════════════════════════════
# 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"
HyDE — Гіпотэтычныя вбудованні дакументаў
Калі працюеце над стадзіям укладання гіпотэтычных дакументаў HyDE, спачатку запісайце умовы кантракту: неабходныя даны, сігнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заліцьварыць пазнейшыя змены ў кодзе. Дакументавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія перакрыцчы і обробка некоректных паведамленняў є часткай продукту, а не пазнейшым дапрацоўкам. Змяркуйце рэтард на фіксаваным наборы запитаў прычыну налаштавання підказак. Частае змены підказак рэдка калі вярнуюць слабкую эфектыўнасць пошуку.
# ════════════════════════════════════════════════════════════
# 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]}")
Разбіўка запита
Калі працюеце над стадзіяй разбівання запитаў, спачатку запішыце умовы вярбунка: неабходныя данні, сигнал успеху і тое, што выканаецца у разе частковага невыпалення. Такі список контролю дапамагае залічыць пазнейшыя змены ў кодзе. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выпанаецца, невыпаленне павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся ў рівень вярнага адлучэння даных на фіксаванай сэтцы запитаў. Частыя змены запитоў рэдка калі вядуць да павышэння якасці адлучэння даных. Калі працюеце над стадзіяй разбівання запитаў, спачатку запішыце умовы вярбунка: неабходныя данні, сигнал успеху і тое, што выканаецца у разе частковага невыпалення. Такі список контролю дапамагае залічыць пазнейшыя змены ў кодзе. Рэгіструйце час выканання і вартасьць токеноў або запитаў разам з функцыональнымі рэзултатамі. Відразувыя данні пра вартасьці запобегаюць неспакойным рахункам, калі праця пераходзіць з дэмавай версіі ў спяльныя сераўы.
# ════════════════════════════════════════════════════════════
# 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?"
Перыякорэнаванне за дапамою Cross-Encoder
Этап перыякорэнавання за дапамою Cross-Encoder працуе наяўней, калі яго розглядаюць як меркаваную плошчу. Перад расшырэнням масштаба зафіксавайце адна ідеальная транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да поперадньага стану. Храніце настройкі пазырочна ад коду прыемліцеля. Файлы сераўіса, хранальнікі секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, куда аператары можуць адбавляць аудыт без неабяжнага чытання всіх дадзеных. Раздзеляйце правілы часткавання дадзеных ад правілаў ўзяць іх. Змена аднаго з яных не павінна вымагаць перапісву другога, калі зменяюцыся паказнікі якосці.
# ============================================================
# 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")
Як поручваюцься гэтыя тэхнікі
Этап «Як порівнюваюцься тэхнікі» працуе найэфектыўней, калі яго розглядаць як вимерную плошчу. Зафіксавайце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра відкатанне перш чым расширваць масштабы. Дакументавайце як шлях успеху, так і шлях вяснавання разам. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсць частью продукту, а не наступным етапам дорабкі. Раздзеліце політыку частковай обработкі даных ад політыки ўзяць даныя. Змена адной з іх не должна прыводзіць да перапісвання другой, калі змянююцыся показатэлі якосці.
Вывык
Этап Заключэння працюе найкраща, калі яго спрыяваць як до мерыям падлеглую сфэру. Зберагчыце адны ідеальны прыклад роботы, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць масштабы. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны ланцужок дзеянняў. Раздзеляйце правілы часткавага апрантавання інфармацыі з правіламі яе выявлення. Змена адных не павінна прымусваць перапісванне іншых, калі змянююцца паказнікі якосці. Этап Заключэння працюе найкраща, калі яго спрыяваць як до мерыям падлеглую сфэру. Зберагчыце адны ідеальны прыклад роботы, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць масштабы. Запісвайце час выканання і кост токэнаў або запытаў разам з функцыйнымі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмавайнага режыма ў спяльныя сераўры.
Чэк-ліст для эксплуатацыі
Этап перагляду канцэлекту праблемы працюе наяўней, калі яго спрыяваць як мерыемую плошчу. Зафіксавайце адну ідеальную версію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць масштабы.
Спрыяйце гэтаму этапу як кантракту межа вхіднымі дадзеннямі і паверынутымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаце критэрыя успеху і адмовіцеся ад тыхоўскага частковага завершэння.
Раздзеліце правілы часткавага апрантавання дадзенняў ад правіл яхо выявлення. Змена аднаго з іх не павінна вымагаць перапісвання другога, калі зменяюцыся паказнікі якосці.
Калі дозволяе бюджет, дадзіце тэст на працэсаванне критычнага маршруту ў системе CI з викорыстанням фіксатываў, а не рэальных платных API.
Запісвайце часы выканання і кост токенаў або запытаў разам з функцыйнаімі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі маршрут пераходзіць з дэмовай среды ў спакульнаныя сераўы.
Неабяжнае правіле падзелкі зместу на часткі аднаэ самае правіле выявлення. Змена аднаго з іх не павінна прымусваць перапісванне другага, калі зменяюцца паказнікі якосці.
Перш чым пераводзіць систему на новы рэжым, заморажаваюць версіі, фіксуюць ідеальны транскрыпт для критычнага маршруту і паверачваюць способы адката. У спільных средах неабяжны ліміты частоты запыткаў, пераканальнія перагледы і чысткі власнік для змены секрэтных даных. Лепшая надзеяная надзейнасць чым хітрыя експерыментальныя прыклады.
Запіс для пакета c9664ffe2213: не кладзіце ключы падаёжніка ў репазітарый, задаце верхнюю межу токена на кожную сесію і зберагаце транскрыпты рыхтоўкаў праз адныя з фіксатываючых элементаў, каб празмены модэляў застаўаліся порównаннімы.
Запіс па адкладанню 1 (c9664ffe2213): фіксаваце зображэнні, задаце бюджеты запыткаў і пераканальваце ізоляцію тэнанта на прыкладны экземпляры раней чым пераводзіць на шырэйшую аплывку.
Запіс па адкладанню 2 (c9664ffe2213): фіксаваце зображэнні, задаце бюджеты запыткаў і пераканальваце ізоляцію тэнанта на прыкладны экземпляры раней чым пераводзіць на шырэйшую аплывку.
Запіс працэўвання 3 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 4 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 5 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 6 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 7 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 8 (c9664ffe2213): фіксавацыя зображэнняў, заданне лімітаваў на запыты і перакананне ў ізоляцыі корыстнікаў на прыкладнай сістэме canary перад шырэйшым расповсюджэнням.
Запіс працэўвання 9 (c9664ffe2213): фіксацыя зображэнняў, заданне лімітавых значэнь для запытаў і перакананне ў ізоляцыі корыстнікаў на прыкладным экземпляре прычаму да шырэйшага распраўлення.
Запіс працэўвання 10 (c9664ffe2213): фіксацыя зображэнняў, заданне лімітавых значэнь для запытаў і перакананне ў ізоляцыі корыстнікаў на прыкладным экземпляре прычаму да шырэйшага распраўлення.