Sau 3 năm vận hành hệ thống semantic search cho nền tảng thương mại điện tử với hơn 50 triệu sản phẩm, tôi đã thử nghiệm gần như tất cả các vector database trên thị trường. Kết quả? Weaviate nổi lên như lựa chọn tối ưu nhờ kiến trúc modular, hỗ trợ hybrid search mạnh mẽ, và đặc biệt là khả năng tích hợp AI API với chi phí cực kỳ cạnh tranh khi sử dụng HolySheheep AI.
Bài viết này sẽ đưa bạn từ concept đến production-ready system với benchmark thực tế, tối ưu chi phí, và những best practices rút ra từ hàng nghìn giờ debugging thực chiến.
Tại Sao Weaviate Là Lựa Chọn Hàng Đầu Cho Semantic Search
Trước khi đi vào code, hãy hiểu tại sao Weaviate vượt trội trong các use case production:
- Hybrid Search tích hợp sẵn — kết hợp BM25 (keyword) + vector similarity trong một query duy nhất
- GraphQL API mạnh mẽ — linh hoạt hơn REST, hỗ trợ filtering phức tạp
- Module hóa kiến trúc — dễ dàng scale horizontal với Kubernetes
- Chi phí vận hành thấp — self-hosted miễn phí hoặc cloud với pricing cạnh tranh
Cài Đặt Môi Trường & Khởi Tạo Weaviate
Cài đặt qua Docker Compose
# docker-compose.yml cho môi trường development
version: '3.8'
services:
weaviate:
image: semitechnologies/weaviate:1.25.0
ports:
- "8080:8080"
environment:
QUERY_DEFAULTS_LIMIT: 25
AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
ENABLE_MODULES: 'text2vec-openai,ref2vec-centroid'
CLUSTER_HOSTNAME: 'node1'
OPENAI_APIKEY: '${OPENAI_APIKEY}'
volumes:
- weaviate_data:/var/lib/weaviate
volumes:
weaviate_data:# Khởi chạy Weaviate container
docker-compose up -d
Verify service đang chạy
curl -s http://localhost:8080/v1/meta | jq .Kết Nối Weaviate Với HolySheep AI
Trong production, bạn cần vectorize dữ liệu hiệu quả. HolySheep AI cung cấp embedding model với giá chỉ từ $0.42/MTok (DeepSeek V3.2), tiết kiệm 85%+ so với OpenAI. Kết nối qua Weaviate custom module:
# Cấu hình Weaviate sử dụng HolySheep cho embeddings
docker-compose.yml production config
version: '3.8'
services:
weaviate:
image: semitechnologies/weaviate:1.25.0
ports:
- "8080:8080"
environment:
QUERY_DEFAULTS_LIMIT: 25
AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'false'
AUTHENTICATION_APIKEY_ENABLED: 'true'
AUTHENTICATION_APIKEY_ALLOWED_KEYS: '${WEAVIATE_API_KEY}'
PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
ENABLE_MODULES: 'custom-holysheep-vectorizer'
CLUSTER_HOSTNAME: 'node1'
# Cấu hình custom vectorizer
CUSTOM_VECTORIZER_MODULE: 'custom-holysheep-vectorizer'
HOLYSHEEP_API_URL: 'https://api.holysheep.ai/v1'
HOLYSHEEP_API_KEY: '${HOLYSHEEP_API_KEY}'
# Performance tuning
GOMAXPROCS: '8'
PROMETHEUS_PORT: '9090'
volumes:
- weaviate_data:/var/lib/weaviate
deploy:
resources:
limits:
memory: 4G
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
volumes:
weaviate_dataSchema Design & Data Modeling
Schema design là yếu tố quyết định 70% performance của hệ thống. Dưới đây là pattern tôi đã áp dụng cho nền tảng e-commerce với 50M+ SKUs:
# schema_definitions.py
Schema tối ưu cho product search với hierarchical categories
import weaviate
client = weaviate.Client("http://localhost:8080")
Xóa schema cũ nếu tồn tại
if client.schema.exists("Product"):
client.schema.delete_class("Product")
if client.schema.exists("Category"):
client.schema.delete_class("Category")
Define Category schema với self-referencing
category_schema = {
"class": "Category",
"description": "Product category với hierarchical structure",
"vectorizer": "text2vec-transformers", # Local embedding
"moduleConfig": {
"text2vec-transformers": {
"vectorizeClassName": False,
" poolingMethod": "masked_mean",
"vectorizer": "sentence-transformers paraphrase-MiniLM-L6-v2"
}
},
"properties": [
{
"name": "name",
"dataType": ["text"],
"description": "Category name",
"moduleConfig": {
"text2vec-transformers": {
"skip": False,
"vectorizePropertyName": False
}
}
},
{
"name": "parent",
"dataType": ["Category"], # Self-reference
"description": "Parent category"
},
{
"name": "level",
"dataType": ["int"],
"description": "Hierarchy level (0 = root)"
}
],
"vectorIndexConfig": {
"distance": "cosine",
"efConstruction": 256, # Tuning for recall
"maxConnections": 64
}
}
Define Product schema với cross-references
product_schema = {
"class": "Product",
"description": "E-commerce product với full-text và vector search",
"vectorizer": "text2vec-transformers",
"moduleConfig": {
"text2vec-transformers": {
"vectorizeClassName": False
}
},
"properties": [
{"name": "name", "dataType": ["text"]},
{"name": "description", "dataType": ["text"]},
{"name": "brand", "dataType": ["text"]},
{"name": "sku", "dataType": ["text"]},
{"name": "price", "dataType": ["number"]},
{"name": "rating", "dataType": ["number"]},
{"name": "reviewCount", "dataType": ["int"]},
{"name": "inStock", "dataType": ["boolean"]},
{"name": "tags", "dataType": ["text[]"]},
{"name": "specifications", "dataType": ["object"]},
{"name": "category", "dataType": ["Category"]},
{
"name": "embedding",
"dataType": ["text"],
"moduleConfig": {
"text2vec-transformers": {
"skip": True # Dùng custom embedding từ HolySheep
}
}
}
],
"vectorIndexConfig": {
"distance": "cosine",
"efConstruction": 512,
"maxConnections": 128,
"ef": 256 # Search parameter
},
"invertedIndexConfig": {
"bm25": {
"k1": 1.5,
"b": 0.75
},
"cleanupIntervalSeconds": 60
}
}
Create schemas
client.schema.create_class(category_schema)
client.schema.create_class(product_schema)
print("✅ Schema created successfully")
print(f"Classes: {[c['class'] for c in client.schema.get()['classes']]}")Hybrid Search Implementation
Đây là phần core mà tôi đã optimize qua nhiều version. Hybrid search kết hợp keyword matching (BM25) với semantic similarity (vector), cho phép search "iPhone 15 pro case" và trả về cả sản phẩm chứa từ khóa chính xác lẫn sản phẩm liên quan về ngữ nghĩa.
# hybrid_search.py
Production-grade hybrid search implementation
import weaviate
import json
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchResult:
"""Standardized search result format"""
id: str
score: float
hybrid_score: float
bm25_score: float
vector_score: float
name: str
brand: str
price: float
category: str
in_stock: bool
highlights: Dict[str, List[str]]
class HybridSearchEngine:
"""
Production hybrid search engine với:
- Hybrid BM25 + Vector search
- Reranking (optionally với cross-encoder)
- Faceted filtering
- Performance tracking
"""
def __init__(
self,
weaviate_url: str = "http://localhost:8080",
api_key: Optional[str] = None,
use_holysheep: bool = True
):
self.client = weaviate.Client(weaviate_url)
self.use_holysheep = use_holysheep
if use_holysheep:
# Sử dụng HolySheep AI cho embedding với chi phí thấp
# Giá DeepSeek V3.2: $0.42/MTok (tiết kiệm 85%+)
self.embedding_endpoint = "https://api.holysheep.ai/v1"
# Lấy API key từ env hoặc secret manager
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
def search(
self,
query: str,
class_name: str = "Product",
limit: int = 20,
offset: int = 0,
filters: Optional[Dict] = None,
alpha: float = 0.5, # 0 = pure BM25, 1 = pure vector
boost_keywords: Optional[Dict[str, float]] = None,
return_properties: Optional[List[str]] = None,
include_vector: bool = False
) -> Tuple[List[SearchResult], Dict]:
"""
Execute hybrid search với optional filtering và boost.
Args:
query: Search query string
class_name: Weaviate class name
limit: Max results to return
offset: Pagination offset
filters: Weaviate where filter
alpha: Balance between BM25 (0) và vector (1)
boost_keywords: Dict of keyword -> boost factor
return_properties: Properties to include in response
include_vector: Include vector in response
Returns:
Tuple of (results, metadata)
"""
start_time = datetime.now()
# Build boost properties
if boost_keywords:
query_with_boost = self._apply_boost(query, boost_keywords)
else:
query_with_boost = query
# Build where filter
where_filter = None
if filters:
where_filter = self._build_where_filter(filters)
# Hybrid search query
search_params = {
"query": query_with_boost,
"alpha": alpha,
"limit": limit,
"offset": offset,
"class": class_name,
"returnProperties": return_properties or [
"name", "brand", "price", "inStock", "category"
],
"returnMetadata": ["score", "highlight"]
}
if where_filter:
search_params["where"] = where_filter
if include_vector:
search_params["returnMetadata"].append("vector")
try:
response = self.client.query.get(**search_params).do()
# Parse response
results = self._parse_response(
response, class_name, include_vector
)
# Track performance
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
metadata = {
"total_results": len(results),
"latency_ms": round(latency_ms, 2),
"query": query,
"alpha": alpha,
"filters_applied": bool(filters)
}
logger.info(
f"Search completed: {len(results)} results in {latency_ms:.2f}ms"
)
return results, metadata
except Exception as e:
logger.error(f"Search failed: {str(e)}")
raise
def _apply_boost(
self,
query: str,
boost: Dict[str, float]
) -> str:
"""
Apply keyword boosting sử dụng Weaviate syntax.
Format: (keyword_1 OR keyword_2)^boost_factor
"""
boosted_terms = []
for keyword, factor in boost.items():
if keyword.lower() in query.lower():
boosted_terms.append(f'({keyword})^{factor}')
return query + " " + " ".join(boosted_terms)
def _build_where_filter(self, filters: Dict) -> Dict:
"""Convert Python dict to Weaviate where filter"""
weaviate_filter = {"operator": "And"}
conditions = []
for field, value in filters.items():
if isinstance(value, dict):
# Range filter: {"gte": 100, "lte": 500}
if "gte" in value or "lte" in value:
op_list = []
if "gte" in value:
op_list.append({
"path": [field],
"operator": "GreaterThanEqual",
"valueNumber": value["gte"]
})
if "lte" in value:
op_list.append({
"path": [field],
"operator": "LessThanEqual",
"valueNumber": value["lte"]
})
conditions.extend(op_list)
else:
conditions.append({
"path": [field],
"operator": "Equal",
"valueBoolean" if isinstance(value, bool)
else "valueString" if isinstance(value, str)
else "valueNumber": value
})
weaviate_filter["operands"] = conditions
return weaviate_filter
def _parse_response(
self,
response: Dict,
class_name: str,
include_vector: bool
) -> List[SearchResult]:
"""Parse Weaviate response to SearchResult objects"""
results = []
data = response.get("data", {}).get("Get", {})
items = data.get(class_name, [])
for item in items:
metadata = item.get("_additional", {})
result = SearchResult(
id=metadata.get("id", ""),
score=metadata.get("score", 0),
hybrid_score=metadata.get("score", 0),
bm25_score=metadata.get("bm25", 0),
vector_score=metadata.get("vector", [0])[0] if include_vector else 0,
name=item.get("name", ""),
brand=item.get("brand", ""),
price=item.get("price", 0),
category=item.get("category", {}).get("name", ""),
in_stock=item.get("inStock", False),
highlights=metadata.get("highlight", {})
)
results.append(result)
return results
Usage example
if __name__ == "__main__":
engine = HybridSearchEngine(use_holysheep=True)
# Search với filters
results, meta = engine.search(
query="wireless headphones noise cancellation",
class_name="Product",
limit=20,
filters={
"price": {"gte": 50, "lte": 300},
"inStock": True,
"rating": {"gte": 4.0}
},
alpha=0.75 # 75% vector, 25% keyword
)
print(f"Query: {meta['query']}")
print(f"Results: {meta['total_results']}")
print(f"Latency: {meta['latency_ms']}ms")
for r in results[:5]:
print(f" - {r.name} | ${r.price} | Score: {r.hybrid_score:.3f}")GraphQL Query Thực Chiến
GraphQL API của Weaviate cho phép những query phức tạp mà REST không thể handle dễ dàng. Dưới đây là các pattern tôi sử dụng trong production:
# graphql_queries.py
Advanced GraphQL queries cho production use cases
import weaviate
from typing import List, Dict, Optional
import json
client = weaviate.Client("http://localhost:8080")
class ProductGraphQLQueries:
"""
Collection of production GraphQL queries cho e-commerce search.
"""
@staticmethod
def get_near_text_query(
concept: str,
limit: int = 10,
certainty: float = 0.7
) -> str:
"""
Semantic search bằng text proximity.
Dùng khi user input là description thay vì keywords.
"""
return f"""
{{
Get {{
Product(
nearText: {{
concepts: ["{concept}"]
certainty: {certainty}
}}
limit: {limit}
) {{
name
brand
price
description
_additional {{
certainty
distance
}}
}}
}}
}}
"""
@staticmethod
def get_hybrid_with_filter(
query: str,
category_id: str,
min_price: float,
max_price: float,
limit: int = 20
) -> str:
"""
Hybrid search với multi-filter.
Kết hợp semantic + keyword search với category và price range.
"""
return f"""
{{
Get {{
Product(
hybrid: {{
query: "{query}"
alpha: 0.7
}}
where: {{
operator: And
operands: [
{{
path: ["category", "id"]
operator: Equal
valueText: "{category_id}"
}}
{{
path: ["price"]
operator: GreaterThanEqual
valueNumber: {min_price}
}}
{{
path: ["price"]
operator: LessThanEqual
valueNumber: {max_price}
}}
{{
path: ["inStock"]
operator: Equal
valueBoolean: true
}}
]
}}
limit: {limit}
) {{
name
brand
sku
price
rating
reviewCount
tags
category {{
... on Category {{
name
path
}}
}}
_additional {{
score
explainScore
}}
}}
}}
}}
"""
@staticmethod
def get_near_object_similar(
reference_id: str,
limit: int = 10
) -> str:
"""
Find similar products dựa trên object reference.
Dùng cho "Products like this" feature.
"""
return f"""
{{
Get {{
Product(
nearObject: {{
id: "{reference_id}"
}}
limit: {limit}
) {{
name
brand
price
_additional {{
distance
}}
}}
}}
}}
"""
@staticmethod
def get_aggregated_search(
query: str,
group_by: str = "brand"
) -> str:
"""
Aggregated search với grouping.
Trả về count theo brand/category để build faceted search UI.
"""
return f"""
{{
Get {{
Product(
hybrid: {{
query: "{query}"
alpha: 0.6
}}
limit: 100
) {{
name
brand
_group: {group_by}
}}
}}
Aggregate {{
Product(
hybrid: {{
query: "{query}"
alpha: 0.6
}}
groupBy: "{group_by}"
) {{
groupedBy {{
value
prop
}}
count
price {{
minimum
maximum
average
}}
rating {{
average
}}
}}
}}
}}
"""
@staticmethod
def get_cross_reference_search(
product_id: str
) -> str:
"""
Search với cross-reference traversal.
Ví dụ: Tìm tất cả sản phẩm cùng category với một sản phẩm cụ thể.
"""
return f"""
{{
Get {{
Product(
where: {{
path: ["id"]
operator: Equal
valueText: "{product_id}"
}}
) {{
name
category {{
... on Category {{
name
_additional {{
id
}}
}}
}}
}}
}}
}}
# Sau đó query products cùng category:
# Get {{
# Product(
# nearObject: {{
# id: ""
# }}
# limit: 20
# ) {{
# name
# }}
# }}
"""
def execute_query(self, gql_query: str) -> Dict:
"""Execute GraphQL query và return results"""
result = client.query.raw(gql_query)
return result
def execute_batch_queries(
self,
queries: List[str]
) -> List[Dict]:
"""
Execute multiple queries trong một request sử dụng BATCH.
Giảm RTT và improve throughput.
"""
batch_query = " ".join(
f'_res{i}: ' + q.strip().strip('{{').strip('}}')
for i, q in enumerate(queries)
)
batch_gql = f"""
{{
Get {{
{batch_query}
}}
}}
"""
return client.query.raw(batch_gql)
Benchmark function
def benchmark_queries(num_runs: int = 100):
"""Benchmark different query types"""
import time
queries = ProductGraphQLQueries()
query_types = {
"near_text": queries.get_near_text_query("wireless headphones"),
"hybrid_filter": queries.get_hybrid_with_filter(
"noise cancellation",
"category-uuid",
50, 300
),
"near_object": queries.get_near_object_similar("product-uuid"),
}
results = {}
for name, query in query_types.items():
times = []
for _ in range(num_runs):
start = time.perf_counter()
queries.execute_query(query)
elapsed = (time.perf_counter() - start) * 1000
times.append(elapsed)
results[name] = {
"avg_ms": round(sum(times) / len(times), 2),
"p50_ms": round(sorted(times)[len(times)//2], 2),
"p95_ms": round(sorted(times)[int(len(times)*0.95)], 2),
"p99_ms": round(sorted(times)[int(len(times)*0.99)], 2),
}
return results
Run benchmark
if __name__ == "__main__":
print("Running GraphQL query benchmarks...")
bench_results = benchmark_queries(100)
print("\n📊 Benchmark Results (100 runs each):")
print("-" * 60)
for query_type, stats in bench_results.items():
print(f"{query_type}:")
print(f" Avg: {stats['avg_ms']:.2f}ms")
print(f" P50: {stats['p50_ms']:.2f}ms")
print(f" P95: {stats['p95_ms']:.2f}ms")
print(f" P99: {stats['p99_ms']:.2f}ms") Tối Ưu Hiệu Suất & Quản Lý Tài Nguyên
Vector Index Tuning
Weaviate sử dụng HNSW (Hierarchical Navigable Small World) cho vector indexing. Tuning đúng parameters có thể improve latency 3-5x:
# performance_tuning.py
Advanced performance tuning cho production workloads
import weaviate
from weaviate.classes.config import Configure, DataType, Property
import time
from typing import Dict, List
import statistics
class PerformanceTuner:
"""
Utility class để tune và benchmark Weaviate performance.
"""
# Benchmark configs
BENCHMARK_CONFIGS = {
"conservative": {
"efConstruction": 128,
"maxConnections": 16,
"ef": 64,
"description": "Lower memory, faster indexing"
},
"balanced": {
"efConstruction": 256,
"maxConnections": 64,
"ef": 128,
"description": "Good balance speed/accuracy"
},
"aggressive": {
"efConstruction": 512,
"maxConnections": 128,
"ef": 256,
"description": "Best recall, higher memory"
},
"production": {
"efConstruction": 512,
"maxConnections": 128,
"ef": 512,
"m": 32, # HNSW M parameter
"description": "Optimized for production workloads"
}
}
def __init__(self, client: weaviate.Client):
self.client = client
def benchmark_recall_vs_speed(
self,
class_name: str,
ground_truth: List[str],
query_vectors: List[List[float]],
configs: Dict = None
) -> Dict:
"""
Benchmark recall và speed cho different index configs.
Ground truth là list of correct result IDs.
"""
if configs is None:
configs = self.BENCHMARK_CONFIGS
results = {}
for config_name, config in configs.items():
print(f"\n🔧 Testing config: {config_name}")
print(f" Description: {config['description']}")
# Update index config
self._update_index_config(class_name, config)
# Benchmark queries
latencies = []
recalls = []
for i, (query_vec, truth) in enumerate(
zip(query_vectors, ground_truth)
):
start = time.perf_counter()
response = self.client.query.get(
class_name,
["id"]
).with_near_vector(
{"vector": query_vec}
).with_limit(20).do()
latency = (time.perf_counter() - start) * 1000
latencies.append(latency)
# Calculate recall
result_ids = [
item["id"] for item in
response["data"]["Get"][class_name]
]
recall = len(set(result_ids) & set(truth)) / len(truth)
recalls.append(recall)
results[config_name] = {
"config": config,
"avg_latency_ms": statistics.mean(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies)*0.95)],
"avg_recall": statistics.mean(recalls),
"config_description": config["description"]
}
print(f" Avg Latency: {results[config_name]['avg_latency_ms']:.2f}ms")
print(f" P95 Latency: {results[config_name]['p95_latency_ms']:.2f}ms")
print(f" Avg Recall: {results[config_name]['avg_recall']:.3f}")
return results
def _update_index_config(self, class_name: str, config: Dict):
"""Update vector index configuration"""
try:
# Update class configuration
self.client.schema.update_config(
class_name,
{"vectorIndexConfig": config}
)
except Exception as e:
print(f"⚠️ Config update failed (may need recreation): {e}")
def get_optimized_schema(
self,
class_name: str,
workload: str = "search" # "search", "recommendation", "deduplication"
) -> Dict:
"""
Get optimized schema configuration dựa trên workload type.
"""
base_properties = [
Property(name="name", data_type=DataType.TEXT),
Property(name="description", data_type=DataType.TEXT),
Property(name="price", data_type=DataType.NUMBER),
]
if workload == "search":
# Optimized cho search latency
return {
"class": class_name,
"vectorizer": "text2vec-transformers",
"properties": base_properties,
"vectorIndexConfig": Configure.VectorIndex.hnsw(
distance=Configure.VectorIndex.Distance.COSINE,
efConstruction=512,
ef=512,
maxConnections=128,
m=48
),
"invertedIndexConfig": Configure.InvertedIndexConfig(
bm25_k1=1.5,
bm25_b=0.75,
cleanupIntervalSeconds=60
)
}
elif workload == "recommendation":
# Optimized cho high recall
return {
"class": class_name,
"vectorizer": "text2vec-transformers",
"properties": base_properties,
"vectorIndexConfig": Configure.VectorIndex.hnsw(
distance=Configure.VectorIndex.Distance.COSINE,
efConstruction=512,
ef=1024, # Higher ef for better recall
maxConnections=64,
m=64
)
}
else:
raise ValueError(f"Unknown workload: {workload}")
def recommend_settings(self, data_stats: Dict) -> Dict:
"""
Recommend settings dựa trên data statistics.
Args:
data_stats: Dict với keys:
- num_objects: int
- avg_vector_dim: int
- query_qps: float
- p99_latency_target_ms: float
"""
num_objects = data_stats.get("num_objects", 1_000_000)
query_qps = data_stats.get("query_qps", 100)
target_p99 = data_stats.get("p99_latency_target_ms", 50)
# Estimate memory usage
vector_dim = data_stats.get("avg_vector_dim", 1536)
memory_per_vector = vector_dim * 4 # float32
estimated_memory_gb = (num_objects * memory_per_vector) / (1024**3)
# Calculate optimal ef
# Rule of thumb: ef ≈ target_p99_latency_in_ms * 2
optimal_ef = min(int(target_p99 * 3), 1024)
recommendations = {
"estimated_memory_gb": round(estimated_memory_gb, 2),
"recommended_ef": optimal_ef,
"recommended_m": min(64, max(16, num_objects // 100_000)),
"recommended_cache_size_mb": min(
2048, # Max 2GB cache
query_qps * optimal_ef * 8 # Rough estimate
),
"shards_recommendation": max(
1,
int(num_objects / 5_000_000) # 1 shard per 5M objects
)
}
return recommendations
Usage example
if __name__ == "__main__":
client = weaviate.Client("http://localhost:8080")
tuner = PerformanceTuner(client)
# Get recommendations cho 10M object dataset
data_stats = {