Practical notes: Beyond Semantic Search: Полное руководство по продвинутому 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

Создание пайплайна индексации

На этапе создания конвейера индексации необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на запутанный конвейер. Указывайте те фрагменты текста, которые легли в основу ответа. Без цитат операторы не смогут отличить вымысел от проблем с индексацией. На этапе создания конвейера индексации необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе от демо-режима к режиму sha.

красных средах.

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: гибридный поиск (плотный + разреженный)

На этапе улучшенного гибридного поиска 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: не храните ключи поставщика в репозитории, установите лимит токенов на сессию и сохраняйте записи данных рядом с фиксами для тестирования, чтобы последующие замены моделей оставались сопоставимыми.

Примечание по развертыванию 1 (c9664ffe2213): закрепите изображения, установите лимиты на количество запросов и проверьте изоляцию пользователей в тестовой среде перед более широким внедрением.

Примечание по развертыванию 2 (c9664ffe2213): закрепите изображения, установите лимиты на количество запросов и проверьте изоляцию пользователей в тестовой среде перед более широким внедрением.

Примечание к развертыванию 3 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 4 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 5 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 6 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 7 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 8 (c9664ffe2213): фиксируйте изображения, устанавливайте лимиты на запросы и проверяйте изоляцию тенантов на канарейке перед более широким развертыванием.

Примечание к развертыванию 9 (c9664ffe2213): фиксация изображений, установка лимитов на запросы и проверка изоляции тенантов на канарейке перед более широким внедрением.

Примечание к развертыванию 10 (c9664ffe2213): фиксация изображений, установка лимитов на запросы и проверка изоляции тенантов на канарейке перед более широким внедрением.