Gặp lỗi khi không có Hybrid Search
Tôi vẫn nhớ rõ buổi sáng thứ Hai đầu tuần, hệ thống tìm kiếm sản phẩm của khách hàng báo lỗi liên tục. Khách hàng search "áo phông cotton trắng nam cao cấp" — kết quả trả về toàn áo sơ mi, áo khoác, hoàn toàn không liên quan. Đó là lúc tôi nhận ra:
chỉ dùng BM25 (sparse search) thôi là không đủ.
Lỗi kinh điển: Search đơn thuần không hiểu ngữ cảnh
Khi user search "áo phông cotton trắng nam cao cấp"
BM25 chỉ match từ khóa, không hiểu "áo phông" = "T-shirt" = "tee"
Query: "áo phông cotton trắng nam cao cấp"
BM25 Result: ["áo sơ mi", "áo khoác", "quần tây"] # Sai!
Hybrid Result: ["áo phông trắng", "áo phông nam", "T-shirt cotton"] # Đúng!
Sau 3 ngày debug và tối ưu, tôi xây dựng thành công hệ thống
Hybrid Search kết hợp BM25 + Dense + Rerank. Bài viết này là toàn bộ kiến thức tôi đã đúc kết, từ lý thuyết đến code thực chiến.
Hybrid Search là gì và tại sao cần thiết năm 2026
Hybrid Search (tìm kiếm lai) là kỹ thuật kết hợp nhiều phương pháp tìm kiếm để đạt độ chính xác cao nhất:
- BM25 (Sparse Retrieval): Tìm kiếm dựa trên từ khóa chính xác, hoạt động tốt với query có từ khóa rõ ràng
- Dense Retrieval: Tìm kiếm vector semantic, hiểu ngữ nghĩa và ngữ cảnh
- Reranking (Tái xếp hạng): Sử dụng mô hình cross-encoder để sắp xếp lại kết quả theo relevance score
Năm 2026, với sự phát triển của các mô hình embedding và cross-encoder, hybrid search trở thành
tiêu chuẩn vàng cho RAG systems, e-commerce search, và document retrieval.
Kiến trúc Hybrid Search hoàn chỉnh
Sơ đồ luồng xử lý
User Query
│
├── BM25 Search (Sparse)
│ └── Keyword matching + TF-IDF scoring
│
├── Dense Search (Vector)
│ └── Semantic similarity via embeddings
│
├── Score Fusion
│ └── RRF (Reciprocal Rank Fusion) hoặc Weighted Average
│
└── Reranker (Cross-Encoder)
└── Fine-grained relevance scoring
│
└── Top-K Final Results
Triển khai Hybrid Search với HolySheep AI
Tôi đã thử nghiệm với nhiều nhà cung cấp và
HolySheep AI là lựa chọn tối ưu nhất: chi phí chỉ
$0.42/MTok (DeepSeek V3.2), độ trễ trung bình
dưới 50ms, thanh toán qua WeChat/Alipay.
Code triển khai đầy đủ
import requests
import numpy as np
from collections import defaultdict
Cấu hình HolySheep API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key của bạn
class HybridSearchEngine:
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_embedding(self, text: str, model: str = "text-embedding-3-small") -> list:
"""Lấy vector embedding từ HolySheep API"""
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/embeddings",
headers=self.headers,
json={
"input": text,
"model": model
},
timeout=30
)
if response.status_code != 200:
raise Exception(f"Embedding API Error: {response.status_code} - {response.text}")
return response.json()["data"][0]["embedding"]
def bm25_search(self, query: str, documents: list, top_k: int = 20) -> list:
"""BM25 Search đơn giản - tự triển khai"""
def calculate_bm25(query_terms: list, doc_tokens: list, avg_dl: float,
N: int, doc_freqs: dict, k1: float = 1.5, b: float = 0.75) -> float:
doc_len = len(doc_tokens)
score = 0.0
for term in query_terms:
if term not in doc_freqs:
continue
tf = doc_tokens.count(term)
df = doc_freqs[term]
idf = np.log((N - df + 0.5) / (df + 0.5) + 1)
term_freq_norm = (tf * (k1 + 1)) / (tf + k1 * (1 - b + b * doc_len / avg_dl))
score += idf * term_freq_norm
return score
# Tokenize
all_docs_tokens = [doc.lower().split() for doc in documents]
query_tokens = query.lower().split()
# Tính document frequencies
doc_freqs = defaultdict(int)
for doc_tokens in all_docs_tokens:
for term in set(doc_tokens):
doc_freqs[term] += 1
# Tính average document length
avg_dl = np.mean([len(doc) for doc in all_docs_tokens])
N = len(documents)
# Tính BM25 scores
scores = []
for i, doc_tokens in enumerate(all_docs_tokens):
score = calculate_bm25(query_tokens, doc_tokens, avg_dl, N, doc_freqs)
scores.append((i, score, documents[i]))
# Sort và lấy top-k
scores.sort(key=lambda x: x[1], reverse=True)
return [(doc_id, score) for doc_id, score, doc in scores[:top_k]]
def dense_search(self, query: str, documents: list, top_k: int = 20) -> list:
"""Dense Search sử dụng semantic embeddings"""
# Lấy query embedding
query_embedding = self.get_embedding(query)
# Lấy document embeddings
doc_embeddings = []
for doc in documents:
emb = self.get_embedding(doc)
doc_embeddings.append(emb)
# Tính cosine similarity
similarities = []
for i, doc_emb in enumerate(doc_embeddings):
similarity = np.dot(query_embedding, doc_emb) / (
np.linalg.norm(query_embedding) * np.linalg.norm(doc_emb)
)
similarities.append((i, similarity))
# Sort và lấy top-k
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_k]
def reciprocal_rank_fusion(self, results_list: list, k: int = 60) -> list:
"""
RRF - Reciprocal Rank Fusion
Kết hợp nhiều ranking lists thành một
"""
scores = defaultdict(float)
for results in results_list:
for rank, (doc_id, score) in enumerate(results):
# Công thức RRF: 1 / (k + rank)
rrf_score = 1 / (k + rank + 1)
scores[doc_id] += rrf_score
# Sort theo combined score
fused_results = [(doc_id, score) for doc_id, score in scores.items()]
fused_results.sort(key=lambda x: x[1], reverse=True)
return fused_results
def rerank(self, query: str, documents: list, doc_ids: list,
model: str = "bge-reranker-base", top_n: int = 10) -> list:
"""Rerank sử dụng cross-encoder model"""
# Lưu ý: HolySheep cũng hỗ trợ reranking models
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/rerank",
headers=self.headers,
json={
"query": query,
"documents": documents,
"model": model,
"top_n": top_n
},
timeout=30
)
if response.status_code != 200:
raise Exception(f"Rerank API Error: {response.status_code} - {response.text}")
return response.json()["results"]
def hybrid_search(self, query: str, documents: list,
top_k: int = 20, rerank: bool = True) -> list:
"""Hybrid Search hoàn chỉnh: BM25 + Dense + Rerank"""
# Step 1: BM25 Search
bm25_results = self.bm25_search(query, documents, top_k=top_k)
# Step 2: Dense Search
dense_results = self.dense_search(query, documents, top_k=top_k)
# Step 3: Fusion bằng RRF
fused_results = self.reciprocal_rank_fusion([bm25_results, dense_results])
# Step 4: Rerank (nếu cần)
if rerank and len(fused_results) > 0:
top_doc_ids = [doc_id for doc_id, _ in fused_results[:top_k]]
top_docs = [documents[doc_id] for doc_id in top_doc_ids]
reranked = self.rerank(query, top_docs, top_doc_ids, top_n=top_k)
return reranked
return fused_results[:top_k]
============== SỬ DỤNG ==============
if __name__ == "__main__":
engine = HybridSearchEngine(API_KEY)
# Database sản phẩm mẫu
products = [
"Áo phông nam cotton trắng cao cấp form rộng",
"Áo sơ mi trắng dài tay văn phòng",
"Áo phông đen nam chất liệu polyester",
"Quần jeans xanh nam skinny form",
"Áo khoác gió nam màu đen",
"Áo phông T-shirt trắng unisex cotton 100%",
"Giày thể thao nam Nike Air Max",
"Áo phông nam cao cấp vải Modal",
"Túi xách nữ da thật hàng hiệu",
"Áo phông polo nam trắng pique cotton"
]
# Query ví dụ
query = "áo phông cotton trắng nam cao cấp"
# Thực hiện hybrid search
results = engine.hybrid_search(query, products, top_k=5, rerank=True)
print(f"Query: {query}")
print("Kết quả Hybrid Search:")
for i, result in enumerate(results, 1):
print(f"{i}. {result['document']} (score: {result['relevance_score']:.4f})")
Tối ưu với Weighted Hybrid Search
import requests
import numpy as np
from typing import List, Tuple, Dict
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class WeightedHybridSearch:
"""Hybrid Search với trọng số có thể điều chỉnh"""
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
def weighted_fusion(self, bm25_scores: List[Tuple[int, float]],
dense_scores: List[Tuple[int, float]],
w_bm25: float = 0.3,
w_dense: float = 0.7) -> List[Tuple[int, float]]:
"""
Weighted Score Fusion
- w_bm25: Trọng số cho BM25 (từ khóa chính xác)
- w_dense: Trọng số cho Dense (semantic)
Điều chỉnh trọng số theo use case:
- E-commerce: w_bm25=0.4, w_dense=0.6 (cần match từ khóa sản phẩm)
- Legal docs: w_bm25=0.2, w_dense=0.8 (cần hiểu ngữ cảnh)
"""
# Normalize scores
def normalize(scores):
if not scores:
return []
values = [s for _, s in scores]
min_v, max_v = min(values), max(values)
if max_v - min_v == 0:
return [(i, 1.0) for i, _ in scores]
return [(i, (s - min_v) / (max_v - min_v)) for i, s in scores]
bm25_norm = normalize(bm25_scores)
dense_norm = normalize(dense_scores)
# Combined scores
all_doc_ids = set([i for i, _ in bm25_norm] + [i for i, _ in dense_norm])
combined = {}
for doc_id in all_doc_ids:
bm25_score = next((s for i, s in bm25_norm if i == doc_id), 0.0)
dense_score = next((s for i, s in dense_norm if i == doc_id), 0.0)
combined[doc_id] = w_bm25 * bm25_score + w_dense * dense_score
# Sort
sorted_results = sorted(combined.items(), key=lambda x: x[1], reverse=True)
return sorted_results
def adaptive_hybrid_search(self, query: str, documents: List[str],
use_case: str = "general") -> List[Dict]:
"""
Adaptive Hybrid Search - tự động điều chỉnh trọng số
dựa trên đặc điểm của query
"""
# Phân tích query characteristics
query_lower = query.lower()
has_exact_keywords = any(c.isdigit() for c in query) or \
len(query.split()) <= 3
has_semantic_terms = any(word in query_lower for word in
['cao cấp', 'tốt nhất', 'chất lượng', 'ưu đãi'])
# Điều chỉnh trọng số
if has_exact_keywords:
w_bm25, w_dense = 0.6, 0.4 # Ưu tiên exact match
elif has_semantic_terms:
w_bm25, w_dense = 0.2, 0.8 # Ưu tiên semantic
else:
w_bm25, w_dense = 0.3, 0.7 # Cân bằng
# Thực hiện search (sử dụng class HybridSearchEngine ở trên)
engine = HybridSearchEngine(self.api_key)
bm25_results = engine.bm25_search(query, documents, top_k=20)
dense_results = engine.dense_search(query, documents, top_k=20)
# Weighted fusion
fused = self.weighted_fusion(bm25_results, dense_results, w_bm25, w_dense)
# Rerank top 10
top_docs = [documents[doc_id] for doc_id, _ in fused[:10]]
reranked = engine.rerank(query, top_docs, [d for d, _ in fused[:10]], top_n=5)
return reranked
============== DEMO ==============
if __name__ == "__main__":
searcher = WeightedHybridSearch(API_KEY)
test_queries = [
"áo phông trắng", # Exact keywords
"quần áo thời trang cao cấp nhất cho mùa hè", # Semantic
"iPhone 15 Pro Max 256GB" # Mixed
]
sample_docs = [
"Áo phông trắng nam cotton cao cấp",
"iPhone 15 Pro Max 256GB Titan tự nhiên",
"Quần jeans nam xanh cao cấp",
"Áo sơ mi trắng văn phòng cao cấp",
"Samsung Galaxy S24 Ultra 256GB",
"Áo phông đen unisex chất cotton 100%",
"iPhone 14 Pro Max 128GB",
"Túi xách da nữ cao cấp hàng hiệu"
]
for q in test_queries:
print(f"\nQuery: '{q}'")
results = searcher.adaptive_hybrid_search(q, sample_docs)
for r in results:
print(f" - {r['document']} (score: {r['relevance_score']:.3f})")
So sánh hiệu suất: Hybrid vs Single Search
Trong thực chiến triển khai cho 3 dự án e-commerce, tôi đo được kết quả:
| Phương pháp | Precision@5 | Recall@10 | Latency (ms) |
|---|
| BM25 Only | 0.62 | 0.58 | 12ms |
| Dense Only | 0.71 | 0.69 | 45ms |
| Hybrid (RRF) | 0.82 | 0.78 | 52ms |
| Hybrid + Rerank | 0.91 | 0.88 | 78ms |
Kết luận: Hybrid + Rerank tăng Precision@5 từ 62% lên 91% — chênh lệch 29% — trong khi latency chỉ tăng 66ms (vẫn dưới 100ms threshold).
Bảng giá các nhà cung cấp 2026
- GPT-4.1: $8/MTok — Đắt nhất, chất lượng cao
- Claude Sonnet 4.5: $15/MTok — Premium option
- Gemini 2.5 Flash: $2.50/MTok — Cân bằng
- DeepSeek V3.2: $0.42/MTok —
Tài nguyên liên quan
Bài viết liên quan