Практичні поради: Векторні бази даних для виробничих систем RAG: індексування, гібридний пошук
Покрокове пояснення практичних порад: Векторні бази даних для виробничих систем RAG: індексування, гібридний пошук; контракти, перевірки та готові фрагменти коду для команд, які використовують цю схему.
Наведені нижче примітки описують практичний підхід до роботи з темою «Векторні бази даних для продакшн-систем RAG: індексування, гібридний пошук та масштабування процесу отримання даних». Основна увага приділяється контрактам, перевіркам та шаблонам коду замість мотиваційних аспектів. Під час роботи на етапі огляду спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код.
Векторна база даних — це алгоритм пошуку
База даних Vector Database працює найкраще, коли її розглядають як вимірювану поверхню. Зафіксуйте один ідеальний варіант роботи, один випадок збою та примітки щодо скасування змін перед розширенням обсягу. Документуйте як успішний, так і відновлювальний сценарії роботи разом. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Розділіть політику часткової обробки даних від політики їх пошуку. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.
Пошук точного найближчого сусіда: базовий підхід, якого не можна використовувати у масштабному форматі
Етап пошуку найближчого сусіда працює найкраще, якщо його розглядати як вимірювану поверхню. Запишіть один ідеальний приклад виконання, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед складними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Розділяйте політику часткового оброблення даних та політику їх пошуку. Зміна однієї з них не повинна змушувати переписувати іншу при зміні показників якості.
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: не включайте ключі постачальника до репозиторію, встановіть ліміт токенів на сеанс та зберігайте записи поруч із фікстурами для оцінки, щоб подальша заміна моделей залишалася порівнянною.
Для етапу 0 процесу зміцнення необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Призначте назви елементів, визначте критерії успіху та не допускайте беззвучного часткового завершення роботи.
Деталь зміцнення 0/864: вимірюйте час виконання, клас помилок та кількість витрачених токенів для цього етапу, а потім вирішуйте, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на індивідуальних спостереженнях.
Під час виконання першого етапу зауважень щодо посилення безпеки спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код.
Деталь посилення безпеки 1/864: вимірюйте час виконання, клас помилки та кількість витрачених токенів для цього зауваження, а потім вирішуйте, чи зберегти зміни, ґрунтуючись на фіксованому наборі питань, а не на індивідуальних спостереженнях.
Другий етап зауважень щодо посилення безпеки найкраще працює, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок невдачі та запис про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям коду перед складними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій.
Деталь посилення безпеки 2/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Для 3-го етапу запису щодо посилення безпеки визначте вхідні дані, відповідальну особу та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити цей крок з відомої точки контролю, не здогадуючись про прихований стан. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища до спільних.
Деталь посилення безпеки 3/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Під час виконання 4-го етапу додаткових заходів зпрочнення спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Документуйте як шлях успішної роботи, так і шлях відновлення одночасно. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої доробки.
Деталі заходу зпрочнення 4/864: вимірюйте час виконання, клас помилки та кількість витрачених токенів для цього етапу, а потім вирішуйте, чи залишити зміни, ґрунтуючись на фіксованому наборі критеріїв, а не на окремих випадках.
4-й етап додаткових заходів зпрочнення працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис виконання, один випадок невдачі та запис про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового завершення роботи.
Деталь посилення безпеки 5/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
На 6-му етапі запису щодо посилення безпеки необхідно визначити вхідні дані, відповідальну особу та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити цей етап з відомої точки контролю, не здогадуючись про прихований стан. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевірити, не читаючи весь код.
Деталь посилення безпеки 6/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Під час виконання 7-го етапу інструкцій з посилення безпеки спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Краще використовувати невеликі, перевірювані одиниці коду замість об’ємних скриптів. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на складну систему взаємозв’язків.
Деталь посилення безпеки 7/864: вимірюйте час виконання, клас помилки та кількість витрачених токенів для цього етапу, а потім вирішуйте, чи залишити зміни, ґрунтуючись на фіксованому наборі критеріїв, а не на індивідуальних спостереженнях.
7-й етап інструкцій з посилення безпеки працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний зразок виконання, один випадок невдачі та запис про скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним витратам під час переходу з демо-середовища до спільних середовищ.
Деталь посилення безпеки 8/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
На 9-му етапі запису щодо посилення безпеки визначте вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не здогадуючись про прихований стан. Документуйте як успішний, так і відновлювальний сценарії. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Деталь посилення безпеки 9/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Під час виконання етапу 10 з покращення безпеки спочатку запишіть угоду: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань.
Деталь 10/864 щодо покращення безпеки: вимірюйте час виконання, клас помилки та кількість витрачених ресурсів для цього етапу, а потім вирішуйте, чи залишити зміни, ґрунтуючись на фіксованому наборі критеріїв, а не на індивідуальних спостереженнях.
Етап 11 з покращення безпеки найкраще працює, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний зразок роботи, один випадок невдачі та запис про скасування змін перед розширенням обсягу роботи. Тримайте конфігурацію окремо від коду програми. Файли середовища, сховища секретних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код.
Деталь посилення безпеки 11/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
На 12-му етапі роботи з посиленням безпеки необхідно визначити вхідні дані, відповідальну особу та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити цей етап з відомої точки контролю, не здогадуючись про прихований стан. Краще використовувати невеликі, перевірювані одиниці коду замість об’ємних скриптів. Якщо етап провалюється, причина має вказувати на конкретну відповідальність, а не на складну систему взаємозв’язків.
Деталь посилення безпеки 12/864: виміряйте час виконання, клас помилки та кількість витрачених токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на окремих випадках.
Під час виконання етапу 13 інструкцій з посилення безпеки спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Поруч із функціональними результатами задокументуйте час виконання та витрати на токени або запити. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища до спільних.
Деталь посилення безпеки 13/864: виміряйте час виконання, клас помилки та витрати на токени для цієї інструкції, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі питань, а не на індивідуальних спостереженнях.
Етап 14 інструкцій з посилення безпеки найкраще працює, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис виконання, один випадок невдачі та запис про скасування змін перед розширенням обсягу роботи. Документуйте як успішний, так і відновлювальний сценарії. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Деталь посилення безпеки 14/864: виміряйте час обробки стіни, клас помилки та витрату токенів для цього запису, а потім вирішіть, чи залишити зміни, ґрунтуючись на фіксованому наборі запитань, а не на окремих випадках.