Giới Thiệu — Tại Sao Vector Database Là Xu Hướng 2025?
Chào các bạn! Mình là Minh, một backend developer với 5 năm kinh nghiệm xây dựng hệ thống AI. Hôm nay mình sẽ chia sẻ chi tiết về cách kết hợp **Milvus Vector Database** với **AI API** để tạo ra ứng dụng tìm kiếm ngữ nghĩa mạnh mẽ.
Trước đây, khi xây dựng tính năng tìm kiếm cho dự án thương mại điện tử, mình gặp rất nhiều khó khăn với tìm kiếm keyword truyền thống. Người dùng tìm "điện thoại cảm ứng giá rẻ" nhưng sản phẩm có tiêu đề "smartphone Android giá tốt" lại không hiển thị. Đó là lý do mình chuyển sang **vector search** và gặp được Milvus.
**Milvus** là gì? Đó là vector database mã nguồn mở được thiết kế đặc biệt cho việc lưu trữ và tìm kiếm embeddings (vec-tơ biểu diễn số học của dữ liệu). Khi kết hợp với AI API, bạn có thể:
- Tìm kiếm ngữ nghĩa thay vì keyword matching
- Xây dựng chatbot thông minh với RAG (Retrieval-Augmented Generation)
- Phát hiện nội dung trùng lặp trong hệ thống tài liệu
- Gợi ý sản phẩm tương tự dựa trên đặc trưng
- Tìm kiếm hình ảnh tương tự (image similarity)
Phần 1: Cài Đặt Môi Trường Từ Con Số 0
1.1 Cài Đặt Docker Và Milvus
Đầu tiên, bạn cần cài đặt Milvus. Mình khuyên dùng **Milvus Lite** cho môi trường development vì nó không yêu cầu Docker phức tạp:
# Cài đặt Milvus qua pip
pip install milvus-lite pymilvus
Kiểm tra phiên bản
python -c "import milvus; print(milvus.__version__)"
Nếu bạn cần phiên bản production với khả năng mở rộng cao hơn, sử dụng Milvus với Docker:
# Tạo file docker-compose.yml
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
milvus-etcd:
container_name: milvus-etcd
image: quay.io/coreos/etcd:v3.5.5
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
- ETCD_QUOTA_BACKEND_BYTES=4294967296
volumes:
- ./etcd:/etcd
command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
milvus-minio:
container_name: milvus-minio
image: minio/minio:RELEASE.2023-03-20T20-16-18Z
environment:
MINIO_ACCESS_KEY: minioadmin
MINIO_SECRET_KEY: minioadmin
ports:
- "9001:9001"
- "9000:9000"
volumes:
- ./minio:/minio_data
command: minio server /minio_data
milvus-standalone:
container_name: milvus-standalone
image: milvusdb/milvus:v2.3.3
command: ["milvus", "run", "standalone"]
environment:
ETCD_ENDPOINTS: milvus-etcd:2379
MINIO_ADDRESS: milvus-minio:9000
volumes:
- ./milvus_data:/milvus_data
ports:
- "19530:19530"
- "9091:9091"
depends_on:
- milvus-etcd
- milvus-minio
EOF
Khởi động Milvus
docker-compose up -d
1.2 Cài Đặt Thư Viện Cần Thiết
# Cài đặt các thư viện cần thiết
pip install \
pymilvus \ # SDK chính thức của Milvus
sentence-transformers \ # Tạo embeddings
openai \ # Gọi AI API
requests \ # HTTP requests
numpy \ # Xử lý số học
python-dotenv # Quản lý biến môi trường
Phần 2: Kết Nối Milvus Và Tạo Collection
2.1 Kết Nối Đến Milvus Server
Mình sẽ hướng dẫn cụ thể cách kết nối với 2 chế độ:
# ========== milvus_config.py ==========
File cấu hình kết nối Milvus
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType, utility
Chế độ 1: Kết nối Milvus Lite (local)
def connect_milvus_lite():
"""Dùng cho development, data lưu local"""
connections.connect(alias="default", uri="./milvus_lite.db")
print("✅ Kết nối Milvus Lite thành công!")
Chế độ 2: Kết nối Milvus Server (Docker/Cloud)
def connect_milvus_server(host="localhost", port="19530"):
"""Dùng cho production"""
connections.connect(
alias="default",
host=host,
port=port,
timeout=30 # Timeout 30 giây
)
print(f"✅ Kết nối Milvus Server ({host}:{port}) thành công!")
Ngắt kết nối
def disconnect_milvus():
connections.disconnect(alias="default")
print("🔌 Đã ngắt kết nối Milvus")
2.2 Tạo Collection Với Schema
Collection trong Milvus tương tự như table trong SQL. Mình sẽ tạo schema cho việc lưu trữ sản phẩm thương mại điện tử:
# ========== create_collection.py ==========
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType, utility
Kết nối
connections.connect(alias="default", uri="./milvus_lite.db")
Định nghĩa schema cho collection sản phẩm
def create_product_collection():
"""Tạo collection lưu trữ sản phẩm với vector embedding"""
# Xóa collection cũ nếu tồn tại
collection_name = "products"
if utility.has_collection(collection_name):
utility.drop_collection(collection_name)
print(f"🗑️ Đã xóa collection cũ: {collection_name}")
# Định nghĩa các trường
fields = [
# Trường ID - khóa chính
FieldSchema(
name="product_id",
dtype=DataType.INT64,
is_primary=True,
auto_id=True
),
# Trường tên sản phẩm
FieldSchema(
name="name",
dtype=DataType.VARCHAR,
max_length=500
),
# Trường mô tả
FieldSchema(
name="description",
dtype=DataType.VARCHAR,
max_length=5000
),
# Trường giá
FieldSchema(
name="price",
dtype=DataType.FLOAT
),
# Trường vector embedding (384 chiều cho model all-MiniLM-L6-v2)
FieldSchema(
name="embedding",
dtype=DataType.FLOAT_VECTOR,
dim=384
)
]
# Tạo schema
schema = CollectionSchema(
fields=fields,
description="Collection lưu trữ sản phẩm cho tìm kiếm similarity"
)
# Tạo collection
collection = Collection(name=collection_name, schema=schema)
# Tạo index cho vector field (tăng tốc độ tìm kiếm)
index_params = {
"index_type": "IVF_FLAT",
"metric_type": "L2", # Euclidean distance
"params": {"nlist": 128}
}
collection.create_index(
field_name="embedding",
index_params=index_params
)
# Tải collection vào memory
collection.load()
print(f"✅ Đã tạo collection: {collection_name}")
print(f"📊 Số chiều vector: 384")
print(f"🔍 Index type: IVF_FLAT")
return collection
Chạy tạo collection
collection = create_product_collection()
Phần 3: Tạo Embeddings Với AI API
3.1 Sử Dụng HolySheep AI Cho Embeddings
Đây là phần quan trọng nhất! Mình đã thử nghiệm nhiều provider AI và chọn **HolySheep AI** vì:
- Tỷ giá ¥1 = $1 — Tiết kiệm 85%+ so với OpenAI
- Độ trễ dưới 50ms — Nhanh hơn đáng kể so với các provider khác
- Hỗ trợ WeChat/Alipay — Thuận tiện cho người dùng Việt Nam
- Tín dụng miễn phí khi đăng ký — Đăng ký tại đây
**So sánh giá embedding models (2026):**
| Provider | Model | Giá/1M tokens |
|----------|-------|---------------|
| HolySheep | text-embedding-3-small | $0.42 |
| OpenAI | text-embedding-3-small | $0.02 |
| HolySheep | text-embedding-3-large | $1.50 |
| OpenAI | text-embedding-3-large | $0.13 |
# ========== embeddings_client.py ==========
import os
import requests
from dotenv import load_dotenv
Load API key từ file .env
load_dotenv()
Cấu hình HolySheep AI
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepEmbeddings:
"""Client để tạo embeddings với HolySheep AI"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def create_embedding(self, text: str, model: str = "text-embedding-3-small") -> list:
"""Tạo embedding cho một đoạn text"""
response = requests.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json={
"input": text,
"model": model
},
timeout=30
)
if response.status_code == 200:
data = response.json()
return data["data"][0]["embedding"]
else:
raise Exception(f"Lỗi API: {response.status_code} - {response.text}")
def create_embeddings_batch(self, texts: list, model: str = "text-embedding-3-small") -> list:
"""Tạo embeddings cho nhiều texts (tối ưu hơn gọi lẻ)"""
response = requests.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json={
"input": texts,
"model": model
},
timeout=60
)
if response.status_code == 200:
data = response.json()
return [item["embedding"] for item in data["data"]]
else:
raise Exception(f"Lỗi API: {response.status_code} - {response.text}")
Sử dụng
client = HolySheepEmbeddings(api_key=HOLYSHEEP_API_KEY)
Test embeddings
test_text = "iPhone 15 Pro Max 256GB - Điện thoại flagship của Apple"
embedding = client.create_embedding(test_text)
print(f"✅ Tạo embedding thành công!")
print(f"📏 Chiều dài vector: {len(embedding)}")
print(f"🔢 5 giá trị đầu: {embedding[:5]}")
3.2 Xử Lý Text Trước Khi Embedding
Để embeddings chính xác hơn, mình luôn preprocess text:
# ========== text_preprocessing.py ==========
import re
import unicodedata
def preprocess_text(text: str) -> str:
"""Tiền xử lý text trước khi tạo embedding"""
# 1. Chuẩn hóa Unicode
text = unicodedata.normalize('NFKC', text)
# 2. Loại bỏ HTML tags
text = re.sub(r'<[^>]+>', '', text)
# 3. Loại bỏ URLs
text = re.sub(r'http\S+|www\.\S+', '', text)
# 4. Chuẩn hóa khoảng trắng
text = re.sub(r'\s+', ' ', text).strip()
# 5. Viết thường (tùy chọn - tùy model)
# text = text.lower()
return text
def combine_product_fields(name: str, description: str, category: str = "") -> str:
"""Kết hợp các trường sản phẩm thành một text duy nhất"""
combined = f"{name}. {description}"
if category:
combined += f". Danh mục: {category}"
return preprocess_text(combined)
Test
name = " iPhone 15 Pro Max 256GB "
desc = "Điện thoại thông minh cao cấp với chip A17 Pro, camera 48MP"
category = "Điện thoại"
text = combine_product_fields(name, desc, category)
print(f"Text sau xử lý: {text}")
Phần 4: Chèn Dữ Liệu Vào Milvus
4.1 Chèn Sản Phẩm Mẫu
# ========== insert_products.py ==========
from pymilvus import connections, Collection
from embeddings_client import HolySheepEmbeddings
from text_preprocessing import combine_product_fields
import time
Kết nối Milvus
connections.connect(alias="default", uri="./milvus_lite.db")
Khởi tạo client embeddings
embedding_client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
Lấy collection
collection = Collection("products")
collection.load()
Dữ liệu sản phẩm mẫu
products = [
{
"name": "iPhone 15 Pro Max 256GB",
"description": "Điện thoại flagship của Apple với chip A17 Pro, màn hình Super Retina XDR 6.7 inch, camera 48MP, pin trâu",
"price": 34990000,
"category": "Điện thoại"
},
{
"name": "Samsung Galaxy S24 Ultra",
"description": "Smartphone cao cấp Android với S Pen, màn hình AMOLED 6.8 inch, camera 200MP, chip Snapdragon 8 Gen 3",
"price": 31990000,
"category": "Điện thoại"
},
{
"name": "MacBook Pro M3 14 inch",
"description": "Laptop chuyên nghiệp Apple Silicon M3, RAM 18GB, SSD 512GB, màn hình Liquid Retina XDR",
"price": 45990000,
"category": "Laptop"
},
{
"name": "Sony WH-1000XM5",
"description": "Tai nghe chống ồn cao cấp với driver 30mm, thời lượng pin 30 giờ, kết nối Bluetooth 5.2",
"price": 8990000,
"category": "Tai nghe"
},
{
"name": "AirPods Pro 2",
"description": "Tai nghe không dây Apple với chống ồn chủ động, Spatial Audio, sạc USB-C, thời lượng pin 6 giờ",
"price": 6990000,
"category": "Tai nghe"
}
]
def insert_products_batch(products: list, batch_size: int = 10):
"""Chèn sản phẩm theo batch để tối ưu hiệu suất"""
all_names = []
all_descriptions = []
# Chuẩn bị texts cho embeddings
for product in products:
combined_text = combine_product_fields(
product["name"],
product["description"],
product["category"]
)
all_names.append(product["name"])
all_descriptions.append(combined_text)
print(f"🔄 Đang tạo embeddings cho {len(products)} sản phẩm...")
start_time = time.time()
# Gọi API tạo embeddings batch (tối ưu hơn gọi lẻ)
embeddings = embedding_client.create_embeddings_batch(all_descriptions)
elapsed = time.time() - start_time
print(f"✅ Tạo embeddings hoàn tất trong {elapsed:.2f} giây")
# Chèn vào Milvus
entities = [
all_names, # name
all_descriptions, # description
[p["price"] for p in products], # price
embeddings # embedding
]
print(f"📤 Đang chèn dữ liệu vào Milvus...")
start_time = time.time()
collection.insert(entities)
collection.flush() # Đảm bảo dữ liệu được ghi
elapsed = time.time() - start_time
print(f"✅ Chèn thành công {len(products)} sản phẩm trong {elapsed:.2f} giây")
print(f"📊 Tổng số entities trong collection: {collection.num_entities}")
Chạy chèn dữ liệu
insert_products_batch(products)
Phần 5: Tìm Kiếm Similarity
5.1 Tìm Kiếm Sản Phẩm Tương Tự
Đây là phần core của bài hướng dẫn - tìm kiếm similarity:
# ========== similarity_search.py ==========
from pymilvus import connections, Collection
from embeddings_client import HolySheepEmbeddings
import time
Kết nối
connections.connect(alias="default", uri="./milvus_lite.db")
collection = Collection("products")
collection.load()
Client embeddings
embedding_client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
def search_similar_products(query: str, top_k: int = 5):
"""
Tìm kiếm sản phẩm tương tự với query
Args:
query: Câu truy vấn tự nhiên
top_k: Số lượng kết quả trả về
"""
# 1. Tạo embedding cho query
print(f"🔍 Đang tìm kiếm: '{query}'")
start_time = time.time()
query_embedding = embedding_client.create_embedding(query)
query_time = time.time() - start_time
print(f"⏱️ Thời gian tạo embedding: {query_time*1000:.1f}ms")
# 2. Tìm kiếm trong Milvus
search_params = {
"metric_type": "L2", # Euclidean distance
"params": {"nprobe": 10} # Số clusters cần tìm
}
start_time = time.time()
results = collection.search(
data=[query_embedding], # Vector tìm kiếm
anns_field="embedding", # Trường vector
param=search_params,
limit=top_k, # Số kết quả
output_fields=["name", "description", "price"] # Trường cần lấy
)
search_time = time.time() - start_time
print(f"⏱️ Thời gian tìm kiếm Milvus: {search_time*1000:.1f}ms")
print(f"⏱️ Tổng thời gian: {(query_time + search_time)*1000:.1f}ms")
# 3. Hiển thị kết quả
print(f"\n📋 Kết quả tìm kiếm:")
print("=" * 70)
for i, hit in enumerate(results[0]):
print(f"\n{i+1}. {hit.entity.get('name')}")
print(f" 💰 Giá: {hit.entity.get('price'):,.0f} VNĐ")
print(f" 📝 Mô tả: {hit.entity.get('description')[:80]}...")
print(f" 📊 Distance: {hit.distance:.4f}")
return results
========== Demo ==========
print("\n" + "=" * 70)
print("DEMO 1: Tìm kiếm điện thoại cao cấp")
print("=" * 70)
results1 = search_similar_products("điện thoại smartphone cao cấp Apple", top_k=3)
print("\n" + "=" * 70)
print("DEMO 2: Tìm kiếm tai nghe không dây")
print("=" * 70)
results2 = search_similar_products("tai nghe chống ồn bluetooth", top_k=3)
print("\n" + "=" * 70)
print("DEMO 3: Tìm kiếm laptop cho lập trình")
print("=" * 70)
results3 = search_similar_products("máy tính xách tay chip M3", top_k=3)
5.2 Kết Hợp Filter Metadata
Bạn có thể filter theo metadata để tăng độ chính xác:
# ========== filtered_search.py ==========
from pymilvus import connections, Collection
from embeddings_client import HolySheepEmbeddings
connections.connect(alias="default", uri="./milvus_lite.db")
collection = Collection("products")
collection.load()
embedding_client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
def search_with_filter(query: str, min_price: float = None, max_price: float = None, top_k: int = 5):
"""Tìm kiếm với filter theo giá"""
# Tạo embedding
query_embedding = embedding_client.create_embedding(query)
# Build filter expression
filter_expr = None
if min_price is not None and max_price is not None:
filter_expr = f"price >= {min_price} and price <= {max_price}"
elif min_price is not None:
filter_expr = f"price >= {min_price}"
elif max_price is not None:
filter_expr = f"price <= {max_price}"
if filter_expr:
print(f"🔍 Filter: {filter_expr}")
# Tìm kiếm với filter
search_params = {
"metric_type": "L2",
"params": {"nprobe": 10}
}
results = collection.search(
data=[query_embedding],
anns_field="embedding",
param=search_params,
limit=top_k,
expr=filter_expr, # Áp dụng filter
output_fields=["name", "description", "price"]
)
# Hiển thị kết quả
for i, hit in enumerate(results[0]):
print(f"{i+1}. {hit.entity.get('name')} - {hit.entity.get('price'):,.0f} VNĐ")
return results
Demo: Tìm điện thoại dưới 35 triệu
print("📱 Điện thoại dưới 35 triệu:")
search_with_filter("smartphone cao cấp", max_price=35000000)
print("\n💰 Sản phẩm dưới 10 triệu:")
search_with_filter("tai nghe chống ồn", max_price=10000000)
Phần 6: Kết Hợp RAG Với AI Chat API
6.1 Xây Dựng Chatbot Với Context Từ Milvus
Đây là cách mình xây dựng chatbot thông minh có thể trả lời về sản phẩm:
# ========== rag_chatbot.py ==========
import os
import requests
from dotenv import load_dotenv
load_dotenv()
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class RAGChatbot:
"""Chatbot sử dụng RAG (Retrieval-Augmented Generation)"""
def __init__(self, milvus_collection, embedding_client):
self.collection = milvus_collection
self.embedding_client = embedding_client
self.chat_history = []
# System prompt cho chatbot
self.system_prompt = """Bạn là trợ lý tư vấn sản phẩm thông minh.
Dựa vào thông tin sản phẩm được cung cấp để trả lời câu hỏi của khách hàng.
Nếu không có thông tin, hãy nói rõ là không biết.
Trả lời ngắn gọn, thân thiện bằng tiếng Việt."""
def retrieve_context(self, query: str, top_k: int = 3) -> str:
"""Truy xuất context từ Milvus"""
# Tạo embedding cho query
query_embedding = self.embedding_client.create_embedding(query)
# Tìm kiếm trong Milvus
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
results = self.collection.search(
data=[query_embedding],
anns_field="embedding",
param=search_params,
limit=top_k,
output_fields=["name", "description", "price"]
)
# Build context string
context_parts = []
for hit in results[0]:
context_parts.append(
f"- {hit.entity.get('name')}: "
f"{hit.entity.get('description')} "
f"(Giá: {hit.entity.get('price'):,.0f} VNĐ)"
)
return "\n".join(context_parts)
def chat(self, user_query: str) -> str:
"""Xử lý câu hỏi của user"""
# 1. Retrieve context từ Milvus
context = self.retrieve_context(user_query)
# 2. Build prompt với context
full_prompt = f"""Dựa vào thông tin sản phẩm sau:
{context}
Câu hỏi của khách: {user_query}
Trả lời:"""
# 3. Gọi HolySheep AI Chat API
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1", # Model: $8/MTok
"messages": [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": full_prompt}
],
"temperature": 0.7,
"max_tokens": 500
},
timeout=30
)
if response.status_code == 200:
data = response.json()
return data["choices"][0]["message"]["content"]
else:
return f"Lỗi: {response.status_code} - {response.text}"
========== Demo ==========
Chạy chatbot
from pymilvus import connections, Collection
from embeddings_client import HolySheepEmbeddings
connections.connect(alias="default", uri="./milvus_lite.db")
collection = Collection("products")
collection.load()
embedding_client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
chatbot = RAGChatbot(collection, embedding_client)
print("🤖 Chatbot RAG Demo")
print("=" * 50)
questions = [
"Tôi muốn mua điện thoại tầm 30 triệu, có gợi ý gì không?",
"So sánh iPhone 15 Pro Max và Samsung Galaxy S24 Ultra",
"Tai nghe nào có chống ồn tốt nhất?"
]
for q in questions:
print(f"\n👤 Khách hỏi: {q}")
answer = chatbot.chat(q)
print(f"🤖 Bot trả lời: {answer}")
Phần 7: Tối Ưu Hiệu Suất
7.1 Batch Processing Cho Dữ Liệu Lớn
Khi cần insert hàng ngàn records, dùng batch processing:
# ========== batch_processing.py ==========
from pymilvus import connections, Collection
from embeddings_client import HolySheepEmbeddings
from text_preprocessing import combine_product_fields
import time
connections.connect(alias="default", uri="./milvus_lite.db")
collection = Collection("products")
collection.load()
embedding_client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
def batch_insert_products(products: list, batch_size: int = 50):
"""
Chèn sản phẩm theo batch để xử lý dữ liệu lớn
Args:
products: Danh sách sản phẩm
batch_size: Kích thước mỗi batch (50 là tối ưu)
"""
total = len(products)
Tài nguyên liên quan
Bài viết liên quan