Практычныя нарады: Базы дадзеных вектораў для прымэння RAG у практыцы: індексаванне, гібрыдны пошук.
Практычныя нарады: Базы дадзеных вектораў для прымэння RAG у практыцы: індексаванне, гібрыдны пошук; контракты, перакананні та шаблоны коду для команд, якія викорыстоўваюць гэты падход.
Наступныя прыміткі паказваюць практычны шлях для розумэння „Баз дадзейна вектараў для RAG у прымэнні: індексаванне, гібрыдны пошук і масштабаванне процеса выкарыстоўвання дадзейнаў“. Акцэнт ставіцца на контракты, перакананняя і месцы для коду, які можна легка адразу застаўіць, а не на мотывацыйныя аспекты. Калі працуеце над стадзіяй агляду, спачатку запісайце контракт: неабходныя вхідныя даны, сігнал успеху і тое, што выканаецца у разе частковага нявыпання задачы. Такі список контроля дапамагае заліцварыць пазнейшыя змены ў кодзе. Храніце настройкі парадульна ад коду прыемлівача. Файлы сяродавішча, хранальнікі секрэтных дадзейнаў і флагі функцыйяй должны знаходзіцца ў аднам месцы, куды аператары можаць адбавіць аудыт без неабходнасці чытаць весь код.
База дадзейнаў вектараў — гэта алгорытм пошуку
База дадзеных Vector работае наяўней, калі яе спрыяваць як мерыемую паверхню. Зафіксавайце адны ідеальны прыклад, адну ситуацыю неудачы і прыметкі па адвярненню змян пры расшырэнні масштаба. Дакументавайце як шлях успеху, так і шлях вяснавання. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є часткай продукту, а не наступным этапам дапрацоўкі. Аддзельна ставьце правілы частковай обработкі дадзеных і правілы ўзяць іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказатэлі якосці.
Пошук точнае найбліжэйшага суседа: базовы падход, які нельга викорыстоўваць у масштабных системах
Этап пошуку найбліжчага суседа працюе наякша, калі яго розглядаць як вимерную паверхню. Запісаўце адна ідеальная транскрыпцыя, адзін прыклад неудачы і прыметку па анулюванні змян перш чым расширваць масштаб. Валіце маленькія, тэставаныя елементы замест велічзючых скрыптав. Калі якісьць крок не выходзіць, прычына неудачы павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Раздзеляйце правілы часткавання дадзеных і правілы ўзяць іх. Змена адных не павінна вымагаць перапісвання іншых, калі змянююцца паказнікі якасці.
import faiss
import numpy as np
from typing import Tuple
def build_exact_index(
embeddings: np.ndarray,
use_cosine: bool = True
) -> faiss.IndexFlatIP:
"""
Build a FAISS flat index for exact nearest neighbour search.
embeddings: (N, D) float32 array.
use_cosine: If True, normalises a copy of the embeddings and uses inner
product (equivalent to cosine similarity). The caller's array
is not mutated.
Returns a FAISS flat index. Benchmark latency against your corpus and
latency SLO before deciding whether ANN indexing is necessary.
"""
dimension = embeddings.shape[1]
if use_cosine:
# Copy before normalising to avoid mutating the caller's array.
embeddings_copy = embeddings.astype(np.float32).copy()
faiss.normalize_L2(embeddings_copy)
index = faiss.IndexFlatIP(dimension)
index.add(embeddings_copy)
return index
else:
index = faiss.IndexFlatL2(dimension)
index.add(embeddings.astype(np.float32).copy())
return index
def search_exact(
index: faiss.IndexFlatIP,
query_vector: np.ndarray,
top_k: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Search the flat index. Returns (distances, indices).
query_vector must already be normalised if the index was built with
normalised embeddings.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
distances, indices = index.search(query, top_k)
return distances[0], indices[0]
HNSW: Чаму ён так популярны у прымэтных пошукавых системах
Этап HNSW Why It Is працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксавце адны ідеальны прыклад, адну справу з бягункамі і запіс пра відкатаванне перш чым расширваць масштаб. Разглядзіце этап як кантракт межа вхіднымі даннымі і перакананымі выходамі. Дайце назвы артыфактам, задаце критэрыі успеху і не прабуйце прыймаць часткова завершэння без паведамлення. Раздзеліце політіку часткавання данных ад політіки ўзяць іх. Змена аднай з яных не должна вымагаць перапісвання другой, калі зменяюцца паказнікі якосці. Этап HNSW Why It Is працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксавце адны ідеальны прыклад, адну справу з бягункамі і запіс пра відкатаванне перш чым расширваць масштаб. Зберагачыце настройкі параду ўнутры коду прыемлена. Файлы сераўіса, храненні секрэтных данных і флагі функций должны знаходзіцца ў аднам месцы, куда аператары можуць аудытуваць іх без неабяжнага чытання всіх дадзеных.
Як будуецца граф
Для стадіі «Як ствараецца граф» неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крэтыяры завершэння пры перадзеі коду. Аперацыйныя працавнікі павінны магчымае перазапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія перакрыцці і обробка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Указваць трэба тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аперацыйныя працавнікі не зможуць адразнаць галюцинацыю ад прасоў у індэксаванні.
Параметры, якія вызначаюць баланс межы запамятоввання і затрымкі
Для параметра, які вызначаюць стадію, неабяжна ўважна прызначэнне вхідных дадзеных, адпаведальнага за крок і крэтарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі павінны магчымае перазапускаць крок з вядомай точкі контролю, не прабуючы спадарацца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выканаецца, прычына неудачы павінна вказываць на адзін конкрэтны элемент, а не на заплутаны ланцюг задач. Прытамульвайце тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аперацыйныя працавнікі не зможуць адразліць галюцинацыю ад працявання з непаштоўнымі дадзенымі.
import faiss
import numpy as np
from typing import Tuple
def build_hnsw_index(
embeddings: np.ndarray,
m: int = 32,
ef_construction: int = 200,
ef_search: int = 100,
use_cosine: bool = True
) -> faiss.IndexHNSWFlat:
"""
Build a FAISS HNSW index for approximate nearest neighbour search.
m: Graph connectivity parameter. Higher = better recall potential, more memory.
Starting range for banking policy corpora: 16 to 32. Benchmark your corpus.
ef_construction: Candidates explored during index build. Higher = better graph quality.
One-time cost at index build; does not affect query latency.
ef_search: Candidates explored at query time. Controls recall-latency trade-off.
Can be changed without rebuilding. Starting range: 50 to 200.
use_cosine: Normalise embeddings and use inner product (cosine similarity).
Note: FAISS HNSW does not support GPU acceleration. For GPU-accelerated ANN,
use IndexIVFPQ variants.
"""
dimension = embeddings.shape[1]
# Copy before normalising to avoid mutating the caller's array.
embeddings_to_index = embeddings.astype(np.float32).copy()
if use_cosine:
faiss.normalize_L2(embeddings_to_index)
index = faiss.IndexHNSWFlat(dimension, m, faiss.METRIC_INNER_PRODUCT)
else:
index = faiss.IndexHNSWFlat(dimension, m, faiss.METRIC_L2)
index.hnsw.efConstruction = ef_construction
index.hnsw.efSearch = ef_search
index.add(embeddings_to_index)
return index
def search_hnsw(
index: faiss.IndexHNSWFlat,
query_vector: np.ndarray,
top_k: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Search the HNSW index. Returns (scores, indices).
query_vector must be normalised if the index was built with normalised embeddings.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
scores, indices = index.search(query, top_k)
return scores[0], indices[0]
Трэбаванні памяці HNSW
Для стадіі «Тэрабатка патрэбнасцей памяці HNSW» неабходна ўзначэння вхідных дадзеных, адпаведальнага за этап і крэатарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі должны магчымае перзапускаць этап з вядомай точкі контролю, не падозрываючы прыхованы стан. Спрыяйце цій стадіі як даговору межа вхіднымі дадзенымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, узначьце крэатарыі успеху і адмовіцеся ад беззвучнага частковага завершэння. Цітуйце тыя часткі, якія фактычна лежалі в основе адпаведнай адказы. Без цітаў аперацыйныя працавнікі не зможуць адразніць галюцинацыю ад прасоў у індэксаванні. Для стадіі «Тэрабатка патрэбнасцей памяці HNSW» неабходна ўзначэння вхідных дадзеных, адпаведальнага за этап і крэатарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі должны магчымае перзапускаць этап з вядомай точкі контролю, не падозрываючы прыхованы стан. Зберагачыце настройкі за межамі коду прыемленае. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, якое працавнікі можуць аудытаваць, не чытаючы весь код.
IVF: Адверсный індексавання файлаў для развертання з обмежанымі ресурсамі памяці
Калі працюеце над этапам адверснага індексавання файлаў IVF, спачатку запісайце умовы: неабходныя вхідныя даны, сигнал успеху і тое, што відбываецца у разы частковага невыпання. Такі список контролю дапамагае залишыцца чэстным пад час пазнейшых змян у кодзе. Документавайце як шлях успеху, так і шлях вярнення до нормы. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є частью продукту, а не пазнейшым дапрацоўкам. Перад налаштаваннем запитаў пераканайцеся ў рэверсі на фіксаваным наборе запитаў. Частая зміна запитоў рэдка калі вярнее слабкую систему пошуку.
import faiss
import numpy as np
from typing import Tuple
def build_ivf_index(
embeddings: np.ndarray,
nlist: int = 1024,
nprobe: int = 64,
use_cosine: bool = True
) -> faiss.IndexIVFFlat:
"""
Build a FAISS IVF flat index.
nlist: Number of Voronoi cells. A common starting heuristic is sqrt(N),
where N is corpus size. For 100K vectors: 300-1000. For 1M: 1024-4096.
Validate empirically.
nprobe: Number of cells searched at query time. Higher = better recall, slower.
Set based on your recall benchmark results.
use_cosine: Use inner product on normalised vectors.
Requires training on a representative sample before adding vectors.
"""
dimension = embeddings.shape[1]
embeddings_to_index = embeddings.astype(np.float32).copy()
if use_cosine:
faiss.normalize_L2(embeddings_to_index)
quantiser = faiss.IndexFlatIP(dimension)
index = faiss.IndexIVFFlat(quantiser, dimension, nlist, faiss.METRIC_INNER_PRODUCT)
else:
quantiser = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFFlat(quantiser, dimension, nlist)
# Use a random representative sample for training. Using the first N records
# risks training on a non-representative slice if the corpus is ordered by
# date, jurisdiction, or document type.
n_available = len(embeddings_to_index)
desired_training_size = min(n_available, 40 * nlist)
rng = np.random.default_rng(seed=42)
training_indices = rng.choice(n_available, size=desired_training_size, replace=False)
training_sample = embeddings_to_index[training_indices]
index.train(training_sample)
index.nprobe = nprobe
index.add(embeddings_to_index)
return index
IVF з квантызацыяй продукту
Калі працюеце над стадзіяй квантызацыі продукту ў рамках тэхналогіі IVF, спачатку запісайце контракт: неабходныя вхідныя даны, сигнал успеху і тое, што выканаецца у разы частковага нявыполнення. Такі список пераканальвае застаўляцца чыстымі падчас пазнейшых зменаў коду. Валіце маленькія, тэставаныя елементы замест вялікіх скрыптав. Калі якісь крок не выйшае, прычына нявыполнення павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцужок задач. Перад налаштаваннем запитоў пераканайцеся ў рэверсіі на фіксаваным наборе запитаў. Частая змена запітоў рэдка калі вярнайша слабкую эфектыўнасць адзысквання інформацыі.
import faiss
import numpy as np
from typing import Tuple
def build_ivfpq_index(
embeddings: np.ndarray,
nlist: int = 1024,
m_subvectors: int = 8,
bits_per_code: int = 8,
nprobe: int = 64
) -> Tuple[faiss.IndexIVFPQ, faiss.IndexFlatIP]:
"""
Build a FAISS IVF-PQ index paired with a flat index for exact re-scoring.
m_subvectors: Number of sub-vectors. Must divide dimension evenly.
For 1536 dimensions: m=8 (192 dims each), m=16 (96 dims each).
Select based on the storage-recall trade-off for your corpus.
bits_per_code: Bits per sub-vector code. 8 bits = 256 centroids per sub-vector.
Lower bits = smaller code, larger recall degradation.
Returns (pq_index, flat_index).
Use pq_index to retrieve top-N candidates cheaply; use flat_index to re-score
those candidates with full float32 precision.
"""
dimension = embeddings.shape[1]
assert dimension % m_subvectors == 0, (
f"Dimension {dimension} must be divisible by m_subvectors {m_subvectors}"
)
norm_embeddings = embeddings.astype(np.float32).copy()
faiss.normalize_L2(norm_embeddings)
# Compressed IVF-PQ index for broad retrieval
quantiser = faiss.IndexFlatIP(dimension)
pq_index = faiss.IndexIVFPQ(
quantiser, dimension, nlist, m_subvectors, bits_per_code,
faiss.METRIC_INNER_PRODUCT
)
training_size = min(len(norm_embeddings), 50 * nlist)
rng = np.random.default_rng(seed=42)
training_indices = rng.choice(len(norm_embeddings), size=training_size, replace=False)
pq_index.train(norm_embeddings[training_indices])
pq_index.nprobe = nprobe
pq_index.add(norm_embeddings)
# Flat index for exact re-scoring of PQ candidates
flat_index = faiss.IndexFlatIP(dimension)
flat_index.add(norm_embeddings)
return pq_index, flat_index
def two_stage_search(
pq_index: faiss.IndexIVFPQ,
flat_index: faiss.IndexFlatIP,
query_vector: np.ndarray,
top_k: int = 10,
candidate_multiplier: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Two-stage retrieval: broad PQ candidate recall followed by exact flat re-scoring.
Stage 1: IVF-PQ retrieves top_k * candidate_multiplier candidates cheaply.
Stage 2: The flat index re-scores those candidates with full float32 precision.
The flat index must have been built with the same normalised embeddings added
in the same corpus order so that IVF-PQ indices align to flat index positions.
candidate_multiplier: Higher values improve recall at higher latency cost.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
n_candidates = top_k * candidate_multiplier
_, candidate_indices = pq_index.search(query, n_candidates)
valid_mask = candidate_indices[0] >= 0
valid_candidates = candidate_indices[0][valid_mask]
if len(valid_candidates) == 0:
return np.array([]), np.array([])
# Reconstruct candidate vectors from the flat index and score them exactly.
candidate_vectors = np.zeros(
(len(valid_candidates), flat_index.d), dtype=np.float32
)
for i, idx in enumerate(valid_candidates):
flat_index.reconstruct(int(idx), candidate_vectors[i])
exact_scores = (candidate_vectors @ query.T).flatten()
reranked_order = np.argsort(exact_scores)[::-1][:top_k]
final_indices = valid_candidates[reranked_order]
final_scores = exact_scores[reranked_order]
return final_scores, final_indices
Порэванне баз данных вектараў для 2026 года
Калі працюеце над стадзіяй парабарэнтавання баз данных вектораў, спачатку запісайце угоду: неабходныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Спрэчвайце гэтую стадзію як угоду межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаць правіла пераканання успеху і не прымайце часткова завершэння без паведамлення. Змяроўваце рэкалі на фіксаванай сэтке запытаў прычым налаштаванні прапаза. Частая змена прапаза рэдка калі вялікі эфект на слабую систему пошуку.
FAISS
FAISS-этап работае найкраща, калі яго спрыяваць як до меры можна. Зафіксавайце адны ідеальны прыклад, адну ситуацыю неудачы і прыметкі па поверненню да пачатковага стану пры расшырэнні масштаба. Документавайце як шлях успеху, так і шлях вярнення да нормальнага стану разам. Перапрыятыя спробы, людзкі контроль і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Раздзеляйце правілы часткавання дадзеных і правілы ўзяць іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказатэлі якосці.
pgvector
Этап pgvector работае наяўней, калі яго спрыяваць як мерыемую паверхню. Зафіксавайце адна «золатая» транскрыпцыю, адин прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны процес. Раздзеліце правіла часткавання дадзеных ад правіл ўтрымання іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказателі якосці.
-- Enable the pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;
-- Policy chunk table with vector and structured metadata
CREATE TABLE policy_chunks (
chunk_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
document_id TEXT NOT NULL,
document_version TEXT NOT NULL,
policy_id TEXT,
jurisdiction TEXT,
effective_date DATE,
section TEXT,
content_type TEXT NOT NULL,
chunk_text TEXT NOT NULL,
classification TEXT NOT NULL DEFAULT 'INTERNAL',
permitted_roles TEXT[] NOT NULL DEFAULT '{}',
embedding_model TEXT NOT NULL,
embedding vector(1536),
indexed_at TIMESTAMPTZ DEFAULT NOW()
);
-- HNSW index for cosine similarity retrieval
CREATE INDEX ON policy_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- Partial index for jurisdiction-scoped retrieval (common query pattern)
CREATE INDEX ON policy_chunks
USING hnsw (embedding vector_cosine_ops)
WHERE jurisdiction = 'EU';
-- Standard indexes for metadata filter columns
CREATE INDEX ON policy_chunks (policy_id);
CREATE INDEX ON policy_chunks (jurisdiction);
CREATE INDEX ON policy_chunks (classification);
CREATE INDEX ON policy_chunks (effective_date);
import psycopg2
import numpy as np
from typing import List, Dict, Optional
def search_policy_chunks(
query_embedding: List[float],
jurisdiction: Optional[str] = None,
classification_ceiling: str = "INTERNAL",
permitted_role: Optional[str] = None,
top_k: int = 10,
ef_search: int = 100,
connection_string: str = "postgresql://user:password@localhost:5432/rag_db"
) -> List[Dict]:
"""
Retrieve policy chunks from pgvector with jurisdiction and access filtering.
ef_search: Controls the HNSW recall-latency trade-off for this session.
Set per-session; does not require index rebuild.
"""
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute(f"SET hnsw.ef_search = {ef_search};")
classification_levels = {"PUBLIC": 0, "INTERNAL": 1, "CONFIDENTIAL": 2}
max_level = classification_levels.get(classification_ceiling, 1)
permitted_classifications = [
k for k, v in classification_levels.items() if v <= max_level
]
filters = ["classification = ANY(%s)"]
params: List = [permitted_classifications]
if jurisdiction:
filters.append("jurisdiction = %s")
params.append(jurisdiction)
if permitted_role:
filters.append("%s = ANY(permitted_roles) OR cardinality(permitted_roles) = 0")
params.append(permitted_role)
where_clause = " AND ".join(filters)
embedding_str = "[" + ",".join(str(x) for x in query_embedding) + "]"
query = f"""
SELECT
chunk_id,
document_id,
document_version,
policy_id,
jurisdiction,
effective_date,
section,
content_type,
chunk_text,
1 - (embedding <=> %s::vector) AS cosine_similarity
FROM policy_chunks
WHERE {where_clause}
ORDER BY embedding <=> %s::vector
LIMIT %s;
"""
params_with_embedding = [embedding_str] + params + [embedding_str, top_k]
cur.execute(query, params_with_embedding)
rows = cur.fetchall()
columns = [
"chunk_id", "document_id", "document_version", "policy_id",
"jurisdiction", "effective_date", "section", "content_type",
"chunk_text", "cosine_similarity"
]
results = [dict(zip(columns, row)) for row in rows]
cur.close()
conn.close()
return results
Qdrant
Этап Qdrant працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксуйце адны ідеальны прыклад, адну справу з бягамі та прыметку па вярнэнне да пачатковага стану пры расшырэнні масштаба. Разглядзайце этап як кантракт межа вхіднымі даннымі та пераканаленымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаце критэрыі успеху та не падзейцеся частым, непূরным выкананням задач. Раздзеліце політыку часткавання данных ад політіки ўзяць іх. Змена аднай з яных не должна вымагаць перапісвання другой, калі зменяюцца паказнікі якосці. Этап Qdrant працюе найкраща, калі яго розглядаць як вимерную паверхню. Зафіксуйце адны ідеальны прыклад, адну справу з бягамі та прыметку па вярнэнне да пачатковага стану пры расшырэнні масштаба. Зберагачыце настройкі пазначкай ад коду прыемліка. Файлы сераўіса, хранільнікі секрэтных данных та флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можуць аудытаваць іх без неабяжнага чытання всіх дадзеных.
from qdrant_client import QdrantClient
from qdrant_client.models import (
VectorParams, Distance, HnswConfigDiff,
PointStruct, Filter, FieldCondition, MatchValue, MatchAny,
SparseVectorParams, SparseIndexParams, SparseVector
)
from typing import List, Dict, Optional
client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "banking_policy"
DENSE_VECTOR_NAME = "dense"
SPARSE_VECTOR_NAME = "sparse"
def create_policy_collection(
dimension: int = 1536,
m: int = 16,
ef_construction: int = 200
) -> None:
"""
Create a Qdrant collection configured for both dense and sparse vectors.
"""
client.recreate_collection(
collection_name=COLLECTION_NAME,
vectors_config={
DENSE_VECTOR_NAME: VectorParams(
size=dimension,
distance=Distance.COSINE,
hnsw_config=HnswConfigDiff(
m=m,
ef_construct=ef_construction,
full_scan_threshold=10000
)
)
},
sparse_vectors_config={
SPARSE_VECTOR_NAME: SparseVectorParams(
index=SparseIndexParams(on_disk=False)
)
}
)
def upsert_policy_chunks(chunks: List[Dict]) -> None:
"""
Index policy chunks with dense vectors, sparse vectors, and metadata payloads.
Each chunk dict must contain:
chunk_id, dense_vector, sparse_indices, sparse_values,
chunk_text, document_id, document_version, policy_id,
jurisdiction, effective_date, content_type, classification,
permitted_roles
"""
points = [
PointStruct(
id=chunk["chunk_id"],
vector={
DENSE_VECTOR_NAME: chunk["dense_vector"],
SPARSE_VECTOR_NAME: SparseVector(
indices=chunk["sparse_indices"],
values=chunk["sparse_values"]
)
},
payload={
"chunk_text": chunk["chunk_text"],
"document_id": chunk["document_id"],
"document_version": chunk["document_version"],
"policy_id": chunk.get("policy_id"),
"jurisdiction": chunk.get("jurisdiction"),
"effective_date": chunk.get("effective_date"),
"content_type": chunk["content_type"],
"classification": chunk["classification"],
"permitted_roles": chunk.get("permitted_roles", []),
"status": "active"
}
)
for chunk in chunks
]
client.upsert(collection_name=COLLECTION_NAME, points=points)
def search_dense_filtered(
dense_query: List[float],
jurisdiction: Optional[str] = None,
permitted_classifications: List[str] = None,
top_k: int = 10,
score_threshold: float = 0.3
) -> List[Dict]:
"""
Dense vector search with integrated payload filtering.
Filtering is applied inside the HNSW graph traversal, not as a post-filter.
"""
if permitted_classifications is None:
permitted_classifications = ["PUBLIC", "INTERNAL"]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if jurisdiction:
must_conditions.append(
FieldCondition(key="jurisdiction", match=MatchValue(value=jurisdiction))
)
search_filter = Filter(must=must_conditions)
results = client.search(
collection_name=COLLECTION_NAME,
query_vector=(DENSE_VECTOR_NAME, dense_query),
query_filter=search_filter,
limit=top_k,
score_threshold=score_threshold,
with_payload=True
)
return [
{"chunk_id": hit.id, "score": hit.score, **hit.payload}
for hit in results
]
def search_hybrid_qdrant(
dense_query: List[float],
sparse_query_indices: List[int],
sparse_query_values: List[float],
jurisdiction: Optional[str] = None,
permitted_classifications: List[str] = None,
top_k: int = 10
) -> List[Dict]:
"""
Hybrid search using both dense and sparse vectors with access-control filtering.
This uses Qdrant's native prefetch-and-fuse API. Both the dense and sparse
signals contribute to retrieval. The Fusion.RRF strategy applies Reciprocal
Rank Fusion internally.
"""
from qdrant_client.models import Prefetch, FusionQuery, Fusion
if permitted_classifications is None:
permitted_classifications = ["PUBLIC", "INTERNAL"]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if jurisdiction:
must_conditions.append(
FieldCondition(key="jurisdiction", match=MatchValue(value=jurisdiction))
)
search_filter = Filter(must=must_conditions)
results = client.query_points(
collection_name=COLLECTION_NAME,
prefetch=[
Prefetch(
query=dense_query,
using=DENSE_VECTOR_NAME,
limit=top_k * 5,
filter=search_filter
),
Prefetch(
query=SparseVector(
indices=sparse_query_indices,
values=sparse_query_values
),
using=SPARSE_VECTOR_NAME,
limit=top_k * 5,
filter=search_filter
),
],
query=FusionQuery(fusion=Fusion.RRF),
limit=top_k,
with_payload=True
)
return [
{"chunk_id": hit.id, "score": hit.score, **hit.payload}
for hit in results.points
]
Weaviate
Для стадіі Weaviate неабяжна практыка адзначыць вхідныя данні, адпаведальнага за крок і крэтырыя для завершэння пры змены коду. Аперацыйныя працавнікі должны магчымасць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабяжна аддактуваць дакументацыю як для стандартнага, так і для альтернатывнага падходу. Практыка перапрыбутків, людзкага контролю і обработкі некоректных паведамленняў є часткаю продукту, а не чымсь, што дадаецца пазней. Паказваць часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аперацыйныя працавнікі не зможуць адразніць галюцинацыю ад прасоў у індэксаванні.
Milvus
Для стадіі Milvus неабяжна практыка адзначыць вхідныя данні, адпаведальнага за крок і крэтырыя для завершэння пры змены коду. Аперацыйныя працавнікі должны магчымаюць перзапускаць крок з вядомага пункта контролю, не прымуячы заўважэнняў на аб’екты, якія залишаюцца незрозумелымі. Лепей выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выконваецца, прычына нехарактэрыстыкі должна вказываць на адзін конкрэтны элемент, а не на заплутаны ланцуг задач. Неабяжна цітаваць тые часткі тексту, якія фактычна служылі падставай для адпаведнага адказу. Без цітатаў аперацыйныя працавнікі не можуць разлічыць галюцинацію ад працягу праз прычыны, зв’язаныя з індэксаванням.
ChromaDB
Для стадіі ChromaDB, перш чым зменяць код, неабходна ясная ваказка пра вхідныя даны, адміністратара крока і крэтэрыяў завершэння. Аператары должны магчымае запускать крок з вядомай точкі контролю, не спрабоўваючы здогадвацца пра схованы стан. Спрыяйце таму, каб гэтая стадія была схожая на контракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, вакажце крэтэрыяў успеху і не прабоўваце прыймаць часткова завершаныя рэзультаты без падтверджэння. Указвайце тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адпаведзі. Без цых цытатаў аператары не зможаць розлічыць галюцинацію ад працэсу індексавання. Для стадіі ChromaDB, перш чым зменяць код, неабходна ясная ваказка пра вхідныя даны, адміністратара крока і крэтэрыяў завершэння. Аператары должны магчымае запускать крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Храніце настройкі праза код аплікацыі. Файлы сераўнавання, сховішчы секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаючы весь код.
Pinecone
Калі працуеце з этапам Pinecone, спачатку запісайце контракт: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Документавайце як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є частью продукту, а не пазнейшым дапрацоўкам. Змяроўвайце ступень вярнення даных на фіксаваным наборе запитаў прычым регулюванні падказак. Частае змена падказак рэдка калі вярнее слабкага механізма пошуку.
Гібрыдны пошук: спалучэнне ўжоўткнутага і розрывчатага пошуку
Калі працуеце над стадзіяй «Спалучэнне гібрантных пошукоў з ўжыццем ўсёя інфармацыі», спачатку запішыце умовы: неабходныя даны, сігнал успеху і тое, што выканаецца у разы частковага нявыпалення. Такі список контроля дапамагае заліцьваты змяны ў кодзе пазнейша. Валіце маленькія, тэставаныя елементы замест большых скрыптаў. Калі якісь крок не выйшоў, прычына нявыпалення павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Перад налаштаваннем запитоў пераканайцеся, як працуея функцыя воспамінання на адзначанай групе запытанняў. Частае зміненне запитоў рэдка калі вярна выправляе слабкія аспекты пошуку.
Спалучэнне рангаў на адной основе з вагамі
Калі працуеце над стадзіяю фузыі зважаных взаўнэчных рангоў, спачатку запісайце кантракт: неабходныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заставіць пазнейшыя змены коду быць чыстымі. Спрэчвайце гэтую стадзію як кантракт межа вхіднымі данымі і перакананымі выходамі. Дайце назвы артыфактам, задаць правіла пераканання успеху і не падзеўляйцеся частковым завершэнням без паведамлення. Змяроўваце рівень запам’ятовання на фіксаванай сэтцы пытанняў прычым регулюванні запрошэнняў. Частае змены запрошэнняў рэдка калі вылечваюць слабкую структуру адзысквання інформаціі. Калі працуеце над стадзіяю фузыі зважаных взаўнэчных рангоў, спачатку запісайце кантракт: неабходныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контроля дапамагае заставіць пазнейшыя змены коду быць чыстымі. Зберагаюце настройкі праза код аплікацыі. Файлы сераўні, хранільнікі секрэтных дадзенняў і флагі функций павінны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабяжнага чытання всіх элементаў.
from typing import List, Dict, Tuple
from collections import defaultdict
def weighted_reciprocal_rank_fusion(
dense_results: List[Tuple[str, float]],
sparse_results: List[Tuple[str, float]],
k: int = 60,
dense_weight: float = 0.6,
sparse_weight: float = 0.4
) -> List[Tuple[str, float]]:
"""
Fuse dense vector search results with sparse BM25 results using weighted RRF.
dense_results: List of (chunk_id, dense_score) sorted by dense score descending.
sparse_results: List of (chunk_id, sparse_score) sorted by sparse score descending.
k: RRF constant. Higher k reduces the influence of top-ranked documents.
Conventional default: 60.
dense_weight / sparse_weight: Relative weights. Tune against your evaluation set.
For corpora with high-precision identifier queries, increase sparse_weight.
Returns fused list of (chunk_id, rrf_score) sorted by rrf_score descending.
"""
rrf_scores: Dict[str, float] = defaultdict(float)
for rank, (chunk_id, _) in enumerate(dense_results, start=1):
rrf_scores[chunk_id] += dense_weight * (1.0 / (k + rank))
for rank, (chunk_id, _) in enumerate(sparse_results, start=1):
rrf_scores[chunk_id] += sparse_weight * (1.0 / (k + rank))
return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
Настройка ўплотненага-розрэджанага вагі
Этап настройкі ўплотненага-розрэджанага вагі працюе найкраща, калі яго розглядаць як вимерную плошчу. Перад расширэнням масштаба зафіксавайце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану. Дакументавайце як успішны, так і вярнучыся шляхы. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є частью продукту, а не наступным етапам доработкі. Раздзеліце політыку частковай обработкі дадзеных ад політыки ўзяць дадзеныя. Змена адной з іх не должна вымагаць перапісву другой, калі зменяюцыся показнікі якосці.
Фільтрацыя метадаў: масштаб узяць дадзеныя і межы автарызаціі
Этап фільтрацыі метадаў і выкарыстоўвання рэзультатаў працюе наякша, калі яго розглядаць як мерыемую структуру. Зафіксавайце адну ідеальную версію, адзін прыклад неудачы і прыметку па адкату перш чым расширваць сферу дзеяння. Валідзіруйце маленькія, тэставаныя елементы замест большых скрыптаў. Калі які-небудзь крок не выйшаў, прычына неудачы павінна вказываць на адную конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Раздзеліце правілы часткавай обробкі дадзеных ад правіл выкарыстоўвання рэзультатаў. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказнікі якосці.
from typing import List, Dict, Optional
from enum import Enum
class ClassificationLevel(Enum):
PUBLIC = 0
INTERNAL = 1
CONFIDENTIAL = 2
def build_access_filter(
user_classification_ceiling: str,
user_jurisdiction: Optional[str] = None,
user_roles: Optional[List[str]] = None
) -> Dict:
"""
Build a Qdrant-compatible filter dict enforcing access control rules.
user_classification_ceiling: Highest classification the user can see.
user_jurisdiction: If set, restrict to chunks applicable to that jurisdiction.
user_roles: If set, restrict to chunks permitted for those roles.
Integrate with your identity provider at request time, not at index time.
This filter represents one layer of the authorisation model; it does not
replace identity verification, audit logging, tenant isolation, or
downstream response controls.
"""
ceiling = ClassificationLevel[user_classification_ceiling].value
permitted = [
level.name
for level in ClassificationLevel
if level.value <= ceiling
]
must_conditions = [
{"key": "classification", "match": {"any": permitted}},
{"key": "status", "match": {"value": "active"}}
]
if user_jurisdiction:
must_conditions.append(
{"key": "jurisdiction", "match": {"value": user_jurisdiction}}
)
if user_roles:
# Chunks with empty permitted_roles are accessible to all roles.
must_conditions.append({
"should": [
{"key": "permitted_roles", "match": {"any": user_roles}},
{"is_empty": {"key": "permitted_roles"}}
]
})
return {"must": must_conditions}
Інкрементальная індексацыя: дадаванне новых дакументаў без перабудовы
Этап дадзення новых элементаў у інкрементным індексаванні працюе наяўней, калі яго спрыяваць як мерыемую структуру. Зберагчыце адну ідеальную копію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Спрыяйце гэтам этапу як кантракту межа вхіднымі дадзеннямі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, задаць критэрыя успеху і адмовіцеся ад тыхняй частковай роботы без паведамлення. Раздзеліце правілы часткавага абрабатвання дадзеных і правілы ўтрыманні іх. Змена адных не павинна вымагаць перапісвання іншых, калі змянююцца паказнікі якосці. Этап дадзення новых элементаў у інкрементным індексаванні працюе наяўней, калі яго спрыяваць як мерыемую структуру. Зберагчыце адну ідеальную копію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Зберагчыце настройкі параду ўнутры коду прыемліцеля. Файлы сераўнавання, хранілішча секрэтных дадзеных і флагі функций павинны знаходзіцца ў аднам месцы, куда аператары можаць адбавляць аудыт без неабходнасці чытання всей структуры.
import logging
from typing import List, Dict
from datetime import datetime
logger = logging.getLogger(__name__)
class IncrementalIndexManager:
"""
Manages incremental updates to a Qdrant collection using soft deletion.
Production pattern:
1. New chunks are inserted immediately with status='active'.
2. Superseded chunks are marked status='deleted' (soft delete).
3. Retrieval filters exclude deleted chunks without graph rebuild.
4. Full rebuild is triggered on schedule or when deleted fraction exceeds threshold.
"""
def __init__(self, qdrant_client, collection_name: str):
self.client = qdrant_client
self.collection = collection_name
self.deleted_threshold = 0.15 # Rebuild when 15% of index is soft-deleted
def upsert_policy_version(
self,
new_chunks: List[Dict],
superseded_chunk_ids: List[str],
policy_id: str,
new_version: str
) -> Dict:
"""
Insert new policy version chunks and soft-delete superseded ones.
"""
if superseded_chunk_ids:
self.client.set_payload(
collection_name=self.collection,
payload={
"status": "deleted",
"deleted_at": datetime.now().isoformat(),
"superseded_by_version": new_version
},
points=superseded_chunk_ids
)
logger.info(
f"Soft-deleted {len(superseded_chunk_ids)} chunks "
f"from policy {policy_id}, superseded by version {new_version}"
)
from qdrant_client.models import PointStruct
points = [
PointStruct(
id=chunk["chunk_id"],
vector={"dense": chunk["dense_vector"]},
payload={
**{k: v for k, v in chunk.items()
if k not in ("chunk_id", "dense_vector")},
"status": "active",
"indexed_at": datetime.now().isoformat()
}
)
for chunk in new_chunks
]
self.client.upsert(collection_name=self.collection, points=points)
logger.info(
f"Inserted {len(new_chunks)} chunks for policy {policy_id} version {new_version}"
)
return {
"inserted": len(new_chunks),
"soft_deleted": len(superseded_chunk_ids),
"policy_id": policy_id,
"new_version": new_version
}
def should_rebuild(self) -> bool:
"""Check whether the fraction of soft-deleted vectors justifies a full rebuild."""
from qdrant_client.models import Filter, FieldCondition, MatchValue
total = self.client.get_collection(self.collection).vectors_count
deleted_filter = Filter(must=[
FieldCondition(key="status", match=MatchValue(value="deleted"))
])
deleted_count = self.client.count(
collection_name=self.collection,
count_filter=deleted_filter
).count
fraction = deleted_count / total if total > 0 else 0
logger.info(
f"Index health: {deleted_count}/{total} soft-deleted ({fraction:.1%})"
)
return fraction >= self.deleted_threshold
Змены версій модэля
У стадії змян версій модэлю Embedding неабходна прадзефінаванне вхідных дадзеных, адпраўніка крока і крэтарыяў завершэння працы перад змянай коду. Аператары должны магчымае запускіць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна аддзеіставіць дакументацыю як для стандартнага, так і для альтернатывнага падхоўду. Практыка перапрыбуткі, людзкія пераказы і обработка некоректных паведамленняў є часткай продукту, а не дадатковым элементам пасля завершэння розработы. Калі наступны крок — це запуск коду або вызов інструмента, лепш выкарыстоўваць структураваныя выходныя данні з пераканальванням схемы, чым вольныя тэкстовыя апісанні.
Шардаванне і реплікацыя: масштабаванне за межы адной вузловай структуры
Для стадіі шардавання і масштабавання за допамою реплікацыі неабходна прадзеяванне вхідных дадзей, абонента данага крока і крэтарыяў выходу пры перадзеяванні коду. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не падозрываючы прыхованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшае, прычына неудачы павінна вказываць на адну конкрэтную абавязку, а не на заплутаны процес. Наводзіце тыя часткі тексту, якія фактычна лежалі в основе адпаведнай адказы. Без цых цітатаў аператары не зможуць разлічыць галюцинацію ад працягу індэксавання.
Стратэгіі шардавання
Для стадіі стратэгій шардавання неабяжна ўзначыць вхідныя даны, адпаведальнага за выкананне крока і крэтырыя завершэння прычыні змены коду. Аперацыйныя працавнікі павінны магчымаецца перзапускаць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрацавляйце з гэтай стадіяй як з кантрактом межаў вхідных дадзеных і паверыльных выходных рэзультатаў. Даць назвы артыфактам, узначыць крэтырыя успеху і адмовіцца ад беззвучнага частковага завершэння. Цітаваць тыя часткі, якія фактычна лежалі в основе адпаведнай адпаведзі. Без цітатаў аперацыйныя працавнікі не зможуць адразніць галюцинацыю ад працягу індэксавання. Для стадіі стратэгій шардавання неабяжна ўзначыць вхідныя даны, адпаведальнага за выкананне крока і крэтырыя завершэння прычыні змены коду. Аперацыйныя працавнікі павінны магчымаецца перзапускаць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Храніце настройкі праза код аплікацыі. Файлы сераўіса, сховішчы секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, якое працавнікі можуць аудытаваць, не чытаючы весь граф.
Планаванне прыемкі
Калі вы працуеце над стадзіяй планавання прыемкі, спачатку запісайце умовы контракту: неабяжлівыя даны, сигнал успеху і тое, што будзе ў разы частковага абякання. Такі список контроля дапамагае заліцьваты змяны коду пазнейша. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія етапы перагляду та обробка некоректных паведамленняў є частью продукту, а не елементамі пазнейшай доработкі. Перад налаштаванням запитоў пераканайцеся, што система правільна адпавядае на фіксованы набор запитаў. Частая зміна запитоў рэдка калі вярнее слабкую якасць адпаведзенняў.
from dataclasses import dataclass
@dataclass
class VectorIndexCapacityPlan:
"""
Illustrative capacity model for HNSW vector indexes.
All figures are approximations for planning purposes.
Benchmark against your actual implementation and workload.
"""
n_vectors: int
dimension: int
hnsw_m: int = 16
replication_factor: int = 2
avg_payload_bytes: int = 2048
memory_headroom_factor: float = 1.5
def vector_storage_gb(self) -> float:
return (self.n_vectors * self.dimension * 4) / (1024 ** 3)
def hnsw_graph_gb(self) -> float:
# Approximate; actual graph overhead varies by implementation and configuration.
return (self.n_vectors * self.hnsw_m * 2 * 8) / (1024 ** 3)
def payload_storage_gb(self) -> float:
return (self.n_vectors * self.avg_payload_bytes) / (1024 ** 3)
def total_index_gb(self) -> float:
return self.vector_storage_gb() + self.hnsw_graph_gb() + self.payload_storage_gb()
def memory_per_replica_gb(self) -> float:
return self.total_index_gb() * self.memory_headroom_factor
def total_cluster_memory_gb(self) -> float:
# Each replica holds a full copy of the index.
return self.memory_per_replica_gb() * self.replication_factor
def report(self) -> str:
return (
f"Illustrative capacity model — {self.n_vectors:,} vectors at {self.dimension}d:\n"
f" Vector storage (approx): {self.vector_storage_gb():.2f} GB\n"
f" HNSW graph estimate: {self.hnsw_graph_gb():.2f} GB\n"
f" Payload storage (approx): {self.payload_storage_gb():.2f} GB\n"
f" Index footprint before overhead: {self.total_index_gb():.2f} GB\n"
f" Per-replica memory + headroom: {self.memory_per_replica_gb():.2f} GB\n"
f" Total cluster memory (approx): {self.total_cluster_memory_gb():.2f} GB\n"
f" ({self.replication_factor} replicas, each holding a full copy)\n"
f" Treat these as planning estimates, not deployment guarantees.\n"
f" Benchmark against your implementation before provisioning."
)
# Illustrative example: banking policy corpus
plan = VectorIndexCapacityPlan(
n_vectors=500_000,
dimension=1536,
hnsw_m=16,
replication_factor=2,
avg_payload_bytes=2048,
memory_headroom_factor=1.5
)
print(plan.report())
Полны цікл адпаведзення
Калі працюеце над стадзіяй «The Complete Retrieval Pipeline», спачатку запісайце контракт: неабяжлівыя вхідныя даны, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Валіце маленькія, тэставаныя елементы замест вялікіх скрыптав. Калі якась ступеня нявыпана, прычына нявыпання павінна вказваць на адну адпаведальнасць, а не на заплутаны процес. Перад налаштаваннем прамптаў пераканайцеся ў рэкалі на фіксаваным наборе запытанняў. Частае змена прамптаў рэдка калі вярнуе слабую эфектыўнасць адзысквання інформаціі.
User Query
↓
Query Embedding (dense + sparse)
↓
Metadata / Authorisation Constraints
↓
Dense Vector Retrieval (filtered HNSW)
+
Sparse Retrieval (BM25)
↓
Weighted RRF Fusion
↓
Candidate Documents with Provenance Metadata
↓
[Part 7: Reranking and Context Assembly]
↓
LLM Generation
import logging
import re
from typing import List, Dict, Optional
from dataclasses import dataclass
from collections import defaultdict
logger = logging.getLogger(__name__)
@dataclass
class RetrievalConfig:
dense_candidate_pool: int = 50
sparse_candidate_pool: int = 50
rrf_k: int = 60
dense_weight: float = 0.6
sparse_weight: float = 0.4
final_top_k: int = 10
score_threshold: float = 0.2
class BankingPolicyRetriever:
"""
Production retrieval pipeline for a regulated banking policy corpus.
Combines dense vector search, BM25 sparse retrieval, access filtering,
and weighted RRF score fusion.
Retrieval ends at the fused candidate list. Reranking and context assembly
are handled in Part 7.
"""
def __init__(
self,
qdrant_client,
collection_name: str,
embedding_pipeline,
bm25_index,
chunk_store: Dict[str, Dict],
config: Optional[RetrievalConfig] = None
):
self.client = qdrant_client
self.collection = collection_name
self.embedder = embedding_pipeline
self.bm25 = bm25_index
self.chunk_store = chunk_store
self.config = config or RetrievalConfig()
def retrieve(
self,
query: str,
user_classification_ceiling: str = "INTERNAL",
user_jurisdiction: Optional[str] = None,
user_roles: Optional[List[str]] = None
) -> List[Dict]:
"""
Full hybrid retrieval with access control.
Returns top-k chunks with provenance metadata, access-filtered
for the requesting user's classification ceiling and jurisdiction.
"""
from qdrant_client.models import Filter, FieldCondition, MatchValue, MatchAny
query_embedding = self.embedder.embed_query(query)
classification_levels = {"PUBLIC": 0, "INTERNAL": 1, "CONFIDENTIAL": 2}
ceiling = classification_levels.get(user_classification_ceiling, 1)
permitted_classifications = [
k for k, v in classification_levels.items() if v <= ceiling
]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if user_jurisdiction:
must_conditions.append(
FieldCondition(
key="jurisdiction",
match=MatchValue(value=user_jurisdiction)
)
)
access_filter = Filter(must=must_conditions)
# Dense vector search with integrated access filtering
dense_hits = self.client.search(
collection_name=self.collection,
query_vector=("dense", query_embedding),
query_filter=access_filter,
limit=self.config.dense_candidate_pool,
score_threshold=self.config.score_threshold,
with_payload=True
)
dense_results = [(hit.id, hit.score) for hit in dense_hits]
# Sparse BM25 retrieval with post-retrieval access filtering
tokens = re.findall(r'\b\w+\b', query.lower())
bm25_scores = self.bm25.get_scores(tokens)
sparse_ranked = sorted(enumerate(bm25_scores), key=lambda x: x[1], reverse=True)
chunk_ids = list(self.chunk_store.keys())
sparse_results = []
for corpus_idx, score in sparse_ranked:
if score <= 0 or len(sparse_results) >= self.config.sparse_candidate_pool:
break
chunk_id = chunk_ids[corpus_idx]
chunk_meta = self.chunk_store.get(chunk_id, {})
if chunk_meta.get("classification") not in permitted_classifications:
continue
if user_jurisdiction and chunk_meta.get("jurisdiction") != user_jurisdiction:
continue
if chunk_meta.get("status") != "active":
continue
sparse_results.append((chunk_id, score))
# Weighted RRF fusion
rrf_scores: Dict[str, float] = defaultdict(float)
k = self.config.rrf_k
for rank, (chunk_id, _) in enumerate(dense_results, start=1):
rrf_scores[chunk_id] += self.config.dense_weight * (1.0 / (k + rank))
for rank, (chunk_id, _) in enumerate(sparse_results, start=1):
rrf_scores[chunk_id] += self.config.sparse_weight * (1.0 / (k + rank))
fused = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
top_chunk_ids = [cid for cid, _ in fused[:self.config.final_top_k]]
# Assemble results with provenance metadata
results = []
for chunk_id in top_chunk_ids:
chunk_data = self.chunk_store.get(chunk_id, {})
results.append({
"chunk_id": chunk_id,
"rrf_score": rrf_scores[chunk_id],
**chunk_data
})
logger.info(
f"Retrieval complete: query={query[:60]!r}, "
f"dense_candidates={len(dense_results)}, "
f"sparse_candidates={len(sparse_results)}, "
f"final_results={len(results)}"
)
return results
Візуабельнасць: Што меркаваць у прыменэнні
Калі працюеце над стадзіяй «Што мераваць для абсарвабельнасці», спачатку запісайце контракт: неабяжлівыя данні, сигнал успеху і тое, што выходзіць у разе частковага абякання. Такій чарткі служыць для таго, каб пазначальныя змены коду былі чыстымі. Спрэцаваўце гэтую стадзію як контракт межа даннімі і перакананымі выходамі. Дайце назвы артыфактам, задаце правілы пераканання успеху і адмовіцеся ад тыхоўскага частковага завершэння. Перад налаштаваннем запытаў змерьце рэкалі на фіксаванай сэтке запытаў. Рэдка калі змены запытаў вылечваюць слабую систему аднаходжэння інформацыі. Калі працюеце над стадзіяй «Што мераваць для абсарвабельнасці», спачатку запісайце контракт: неабяжлівыя данні, сигнал успеху і тое, што выходзіць у разе частковага абякання. Такій чарткі служыць для таго, каб пазначальныя змены коду былі чыстымі. Зберагаце настройкі за межамі коду прыемленае. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабяжлівага чытання всей структуры.
import time
import logging
from typing import Callable, TypeVar, Any
from functools import wraps
logger = logging.getLogger(__name__)
F = TypeVar("F", bound=Callable[..., Any])
def retrieval_instrumented(func: F) -> F:
"""
Decorator that adds structured latency logging and empty-result alerting
to retrieval functions. Wrap your primary retrieve() method in production.
"""
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = None
error = None
try:
result = func(*args, **kwargs)
return result
except Exception as e:
error = str(e)
raise
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
n_results = len(result) if result is not None else 0
log_payload = {
"function": func.__name__,
"latency_ms": round(elapsed_ms, 2),
"n_results": n_results,
"error": error
}
query = kwargs.get("query", args[1] if len(args) > 1 else None)
if query:
log_payload["query_prefix"] = str(query)[:80]
if error:
logger.error("retrieval_error", extra=log_payload)
elif n_results == 0:
logger.warning("retrieval_empty_result", extra=log_payload)
elif elapsed_ms > 500:
logger.warning("retrieval_high_latency", extra=log_payload)
else:
logger.info("retrieval_success", extra=log_payload)
return wrapper # type: ignore
Верніцеся да прыяву кредыту у розмярзе 12 мільйонаў еўра
Этап «Верніцеся да EUR» работае наякша, калі яго спрыяваць як мерыемую структуру. Запісаўце адна ідеальная ситуацыя, адзін прыклад неудачы і запіс пра вярненне да пачатковага стану перш чым расширваць масштабы. Дакументаваўце як успішны, так і вярнучыся парадоксы разам. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Раздзеліце правілы часткавання інфармацыі ад правілаў яе выявлення. Змена ў одных не павінна прымусваць перапісванне іншых, калі змянююцца паказнікі якосці.
Перш чым перейсці да параджання: чарткі для прыемлівай працы
Этап «Перш чым пераходзіць да выканання» працюе наяўней, калі яго розглядаць як мерыямую паверхню. Запісаце адна «золатая» версія, адин прыклад неудачы і прыметку па вярнэнню да поперадней версіі пры расшырэнні масштаба. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны процес. Раздзеляйце правілы часткавання і правілы выявлення. Змена адных не должна прыводзіць да перапісвання іншых, калі змянююцца паказатэлі якосці.
Што будзе далей
Этап «Што будзе далей» працюе найэфектывней, калі яго спрыяваць як меравальную плошчу. Зберагачыце адны ідеальны прыклад, адзін кейс неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Спрыяйце гэтам этапу як кантракту межа вхіднымі даннымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, задаце критэрыя успеху і адмовіцеся ад тыхняй частковай роботы без паведамлення. Раздзеліце політыку часткавага апранкавання дадзеных ад політыки ўтрымання іх. Змена адной з яных не павінна вымагаць перапісвання другой, калі зменяюцыся паказнікі якосці. Этап «Што будзе далей» працюе найэфектывней, калі яго спрыяваць як меравальную плошчу. Зберагачыце адны ідеальны прыклад, адзін кейс неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Зберагачыце настройкі параду ўнутры коду прыемліцеля. Файлы сяродавішняе сераўісу, хранілішчы секрэтных дадзеных і флагі функцый павінны знаходзіцца ў адном месцы, якое аператары можаць пераглядаць без неабяжнага чытання всіх дадзеных.
Чэк-ліст для эксплуатацыі
У стадії перагляду канцэлярскіх пунктав пры змěне коду неабходна ясная ваказка па інпутах, адміністратары ўзела і крэтэрыях завершэння. Аперацыйныя працавнікі павінны магчымае перадзванаць этап з вядомай точкі контролю, не прабуючы спадарацца пра схованы стан.
Запісваюць час выконання і вартасць токеноў або запытав пры якіх-небудзь функцыйнальных рэзультатах. Відразліва візуабільнасць вартасцей запобегае неспадзяваным рахункам, калі процес пераходзіць з дэмовай среды ў спяльнаваныя сераўеры.
Прыкладзіце часткі тексту, якія фактычна сталі падставай для адпаведнай адказу. Без ціх цитатаў аперацыйныя працавнікі не зможуць адразніць галюцинацыю ад працягу індэксавання.
Следзіце за вартасцю і затрымкамі разам з якасцю. Адказ, які ў меру гіршы, але коштае у 10 разоў меней, можа быць правильным выборам для прыемнай среды.
Фіксуйце версіі залежнасцяў і запісваюце хеш-значэнне зображэння, якое выканало дэму. Возможнасць перадзванення результатаў важлівей, чым традыцыйныя знання.
Лепшыя маленькія, тэставаныя елементы, чым велікія скрыпты. Калі якісь крок не выйшае, адказнасць за гэта патрабуецца толькі ад однаго элемента, а не ад усіяўшайся ланцужні.
Перад апраноўкай стака неабходна заморазіць версіі, зафіксаваць «золаты» транскрыпт для критычнага шляху і паверыцца ў крокі для вярнення да пачатковага стану. У спільных средах трэба насталявати ліміты частоты запытоў, пераканацца ў правільнасці належнасці ресурсаў і вызначыць чыстаго адпаведальнага за зміны секрэтных дадзенняў. Лепшая ўпорядкованасць і надзеянае функцыонаванне, чым крэатывныя, але еднакратныя дамыслы.
Прымітка для fa68a70d815a: не трэба кластыць ключі прадастоўнікаў у репазітары, выставіць ліміт токена на адну сесію і зберагаць транскрыпты разам з фіксатрамі для ацэнкі, ўпрымку, каб пазнейшыя замены модэляў заставаліся порównанымі.
Для стадіі 0 пры практыцы зміцнення неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры зміне коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Спрыяйце цій стадіі як даговору межаў вхідных даных і перакананых выходных рэзультатаў. Даць назвы артыфактам, узначыць перакананні ў успеху і адмовіцца ад беззвучнага частковага завершэння.
Дзеянні зміцнення 0/864: вымерыць час выканання, класію каштоўкаў і выкарыстоўванне токенав для гэтай прыметкі, а пасля — вырашыць, чы рашыцца застаўіць змяну, ствараючыся на адной фіксованай сэтцы пытанняў, а не на асобістых спазыраннях.
Калі працуеце над першым этапам зміцнення, спачатку запісайте умовы кантракта: неабяцковыя даны, сігнал успеху і тое, што выходзіць на частыя неудачы. Такі список дапамагае заліцварваць пазнейшыя змены коду. Зберагаюце настройкі пазней ад коду прыемлі. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адбавіць аудыт без неабяцковага чытання всіх элементаў.
Дзялённе зміцнення 1/864: вымерыце час выканання, класію памылак і колькасць викорыстоўваных токенав для гэтага пункту, а потым выберыце, чы робіць змену на адной пазначанай базе, а не на аснове індывідуальных спостарэнняў.
Этап зміцнення 2 працюе лепей, калі яго спрыямаць як меравальную плошчу. Запісайце адны ідеальны прыклад работы, адзін кейс неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Валідзіце маленькія, тэставаныя елементы замест большых скрыптав. Калі якісь крок не выйшае, неудача должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны процес.
Дзеянне паўжасткі 2/864: звярніце увагу на час выканання, класы памылак і колькасць токенаў, выкорыстаных для гэтага запісу, а пасля, на аднойчынай базе фіксаваных пытанняў, а не на індывідуальных прыкладах, выявіце, чы рэшыцца застаўіць змяну.
Для 3-й стадзіі паўжасткі запісу з’явіце вхідныя даны, адпаведальнага за крок і критэрыяы завершэння пры перамены коду. Аперацыйныя працавнікі должны магчыма было перазваляць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Запісвайце час выканання і колькасць токенаў або запытак палягліва да рэзультатаў функцыянальнай працы. Візуабельнасць костоў з самага пачатку запобегае неспакойным рахункам, калі процес пераходзіць з дэмаверсіі ў спяльныя среды.
Дзеянне паўжасткі 3/864: звярніце увагу на час выканання, класы памылак і колькасць токенаў, выкорыстаных для гэтага запісу, а пасля, на аднойчынай базе фіксаваных пытанняў, а не на індывідуальных прыкладах, выявіце, чы рэшыцца застаўіць змяну.
Калі працуеце над 4-й стадзіяю прыемкі з павышэння безпекі, спачатку запісайце угоду: неабяжлівыя данні, сігнал успеху і тое, што выходзіць пад частковы нявыплэн. Такі список контроля дапамагае заставіць пазнейшыя змены коду быць чыстымі.
Документавайце як «шчаслівы» шлях, так і шлях вяснавання. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў ёсць частью продукту, а не пазнейшым дапрацоўкам.
Дзялей 4/864 прыемкі з павышэння безпекі: вымерайце час выканання, класыя ошибкі і витрату токенав для гэтай прыемкі, а пасля выберайце, чы робіць змены на адной фіксаванай сэтке пытанняў, а не на адной лічбе прыкладаў.
4-я стадзія прыемкі з павышэння безпекі работае лепей, калі яе спрыямаць як вымеральную плошчу. Запісайце адну «золатую» транскрыпцыю, адны прыклад нявыплэну і прыемку для абраткаў перад расшырэнням масштаба. Спрыяйце гэтай стадзіі як угоды межа даннімі і перакананымі выходамі. Дайце назвы артыфактам, задаце перакананні успеху і адмовіцеся ад тыхоўскага частковага завершэння.
Дзеянне паўжасткі 5/864: звярніце увагу на час выканання, класы памылак і витрату токенаў для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на індывідуальных прыкладах, выявіце, чы рэшыцца застаўіць змяну.
Для 6-й стадзіі паўжасткі запісу перад змянай коду неабходна визначыць вхідныя даны, адпаведальнага за крок і критэрыя завершэння. Аперацыяныя працавнікі должны магчымае перадзваначыць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Канфігурацыю трэба зберагчы за межамі коду прыемленае, таму што файлы сяродавішча, хранілішча секрэтных дадзенняў і флагі функцый належаць у аднам месца, якое працавнікі можу аудытаваць, не чытаючы весь граф.
Дзеянне паўжасткі 6/864: звярніце увагу на час выканання, класы памылак і витрату токенаў для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на індывідуальных прыкладах, выявіце, чы рэшыцца застаўіць змяну.
Калі працуеце над 7-ю стадзіяй забезпечэння безпекі, спачатку запісайце умовы кантракта: неабяжлівыя даны, сігнал успеху і тое, што выходзіць на частым неудачам. Такі список контроля дапамагае залічыць пазнейшыя змены ў кодзе чыстымі. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, неудача павінна вказваць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач.
Дзеянне 7/864 забезпечэння безпекі: замерайце час выканання, класы каштоўкаў і витрату токенав для гэтай стадзіі, а пасля выберайце, чы робіць змены на адной фіксаванай сэтке пытанняў, а не на адной толькі прымітцы.
8-я стадзія забезпечэння безпекі працуе лепей, калі яе спрыяваць як меравальную плошчу. Запісайце адны ідеальны прыклад роботы, адзін кейс неудачы і прыміткі па адвярненню змэн, прычым не расшыряйце сферу дзеяння. Запісвайце часы выканання і вартасць токенав або запыткаў разам з функцыйнальнымі рэзултатамі. Відкрытая інформацыя пра вартасці запобегае неспакойным рашчыткам, калі працэс пераходзіць з дэмаверсіі ў спяльныя среды.
Дзеянне паўжчання 8/864: звярніце увагу на час выканання, класыя ошибак і колькасць токенаў, якія былі выкарыстаны для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на асобістых спазырэннях, выявіце, чы хацяце застаўіць змены.
Для 9-го этапу паўжчання неабходна перад змянай коду чытко апісаць вхідныя даны, адпаведальнага за выкананне крока і критэрыяы завершэння. Аперацыйныя працавнікі должны магчымае перадзвануць гэты крок з вядомага пункта контролю, не прымуджаючыся здагадвацца пра схованы стан. Неабходна адночасна задокументаваць шлях успеху і шлях вярнення да нормальнага стану. Практыка павтарэння спроб, людзкія контрольныя пункты і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней.
Дзеянне паўжчання 9/864: звярніце увагу на час выканання, класыя ошибак і колькасць токенаў, якія былі выкарыстаны для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на асобістых спазырэннях, выявіце, чы хацяце застаўіць змены.
Калі працуеце над 10-ю стадзіяй ударожэння, спачатку запісайце контракт: неабяжлівыя данні, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список контроля дапамагае заліцьварыць пазнейшыя змены ў кодзе. Спрыймайце гэтую стадзію як контракт межа даннімі і перакананымі выходамі. Дайце назву артыфактам, задаце критэрыя успеху і адмовіцеся ад мовчанкавага частковага завершэння.
Дзеянні ўдарожэння 10/864: звярніце увагу на час выканання, класы каштоўкаў і витрату токенав для гэтай змены, а пасля вырашыце, чы робіць яе, стаўячыся на адну і тую ж сэтку пытанняў, а не на асобістыя спазыркі.
11-я стадзія ударожэння працюе лепей, калі яе спрыймаюць як меравальную плошчу. Запісайце адна ідеалная транскрыпцыя, адзін прыклад нявыпання і змест запісу про анулюванне, перш чым расширваць масштаб. Зберагайце настройкі параду ўнутры коду прыемленае. Файлы сяродавішча, хранілішчы секрэтных дадзенняў і флагі функций павінны знаходзіцца ў адном месцы, куды аператары можаць аудытаваць іх, не чытаючы весь граф.
Дзеянне паўжчання 11/864: звярніце увагу на час выканання, клас памылак і колькасць викорыстоўваных токенаў для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на асобістых спазыраннях, выявіце, чы рэшацца застаўляць змяну.
Для 12-го этапа паўжчання неабходна з’явіць вхідныя даны, абавесцяўца крока і критэрыі завершэння пры перамены коду. Аперацыйныя працавнікі павінны магчымае перадзвігнуць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшоў, прычына неудачы павінна вказываць на адную абавесцяўку, а не на заплутаны ланцужок задач.
Дзеянне паўжчання 12/864: звярніце увагу на час выканання, клас памылак і колькасць викорыстоўваных токенаў для гэтага запісу, а пасля, на аднойчынай базе фіксаванага набору пытанняў, а не на асобістых спазыраннях, выявіце, чы рэшацца застаўляць змяну.
Калі працуеце над стадзіяй 13 з адаптавання системы, спачатку запісайце умовы кантракту: неабяжлівыя даны, сігнал успеху і тое, што выходзіць пад частковыя неудачы. Такі список дапамагае заліцварваць будучыя змены ў кодзе. Запісвайце час выканання задачы, а таксама вартасць токенаў чы роезпытакоў пад функцыйнальнымі рэзултатамі. Відразы вартасці з самага пачатку запобегае неспакойным рахункам, калі система пераходзіць з дэмаверсіі ў спяльныя среды.
Дакладнасць адаптавання 13/864: замеры часу выканання, класаў паказакоў і витрачання токенаў для гэтай стадзіі, пасля чаго прымкніце рашэнне пра тое, чы трэба застаўіць змену на адной пазначанай базе даказваў, а не на аснове індывідуальных спазырэнняў.
Стадзія 14 з адаптавання работае лепей, калі яе спрыямаць як меравальную плошчу. Запісвайце адну ідеальную транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Дакументавайце як успішны, так і вярнэнчы паты. Перапрыбуткі, людзкія контралі і обработка неканальных паказакоў є часткай продукту, а не чымсь, што дадаецца пазней.
Дзеянне паўжчання 14/864: звярніце увагу на час обработкі, клас памылак і колькасць выкарыстоўваных токенав для гэтага зьязку, а пасля выберыце, чы робіць змяну на аднойчы заданай сэткі пытанняў, а не на аднойчы прымітцы.