Đọc bài viết này, bạn sẽ hiểu cách chọn thuật toán vector index phù hợp cho ứng dụng AI của mình. Là một kỹ sư đã triển khai RAG (Retrieval-Augmented Generation) cho hơn 20 dự án thực tế, tôi sẽ giải thích khác biệt giữa HNSW, IVF và DiskANN bằng ngôn ngữ đơn giản nhất, kèm theo code mẫu có thể chạy ngay và phân tích chi phí chi tiết.

向量索引 là gì? Tại sao cần nó?

Trước khi so sánh các thuật toán, chúng ta cần hiểu bản chất vấn đề. Khi bạn xây dựng ứng dụng AI như chatbot tài liệu, hệ thống tìm kiếm thông minh, hay recommendation engine, bạn cần lưu trữ hàng triệu vector (đại diện số của văn bản, hình ảnh, âm thanh).

Vấn đề cốt lõi: Tìm kiếm tuyến tính 1 triệu vector mất 1 giây. Tìm kiếm với vector index chỉ mất dưới 50ms. Đó là lý do vector index tồn tại.

Ba thuật toán phổ biến nhất

1. HNSW (Hierarchical Navigable Small World)

HNSW hoạt động như một hệ thống đường cao tốc nhiều tầng. Tầng trên có ít nút hơn nhưng kết nối xa hơn, tầng dưới có nhiều nút hơn và kết nối gần hơn. Khi tìm kiếm, thuật toán bắt đầu từ tầng cao nhất, nhanh chóng thu hẹp phạm vi, rồi tìm chi tiết ở tầng thấp hơn.

# Ví dụ: Khởi tạo HNSW Index với FAISS
import numpy as np


Tạo 10,000 vector 128 chiều (fake data)

np.random.seed(42)
dimension = 128
nb_vectors = 10000
vectors = np.random.rand(nb_vectors, dimension).astype('float32')


Import FAISS - thư viện vector index phổ biến nhất

import faiss


Tạo HNSW index


M = số kết nối tối đa mỗi nút (thường 16-64)


efConstruction = kích thước danh sách candidate khi build (thường 100-400)

hnsw_index = faiss.IndexHNSWFlat(dimension, M=32, efConstruction=200)


Thêm vectors vào index

print(f"Đang thêm {nb_vectors} vectors vào HNSW index...")
hnsw_index.add(vectors)
print(f"HNSW index đã sẵn sàng! Số vectors: {hnsw_index.ntotal}")


Cấu hình tham số tìm kiếm


efSearch = kích thước danh sách duyệt khi tìm kiếm (thường 50-500)


Giá trị cao hơn = kết quả chính xác hơn nhưng chậm hơn

hnsw_index.hnsw.efSearch = 100


Tìm kiếm 5 vector gần nhất

query_vector = np.random.rand(1, dimension).astype('float32')
k = 5
distances, indices = hnsw_index.search(query_vector, k)

print(f"\nKết quả tìm kiếm:")
print(f"- Indices: {indices[0]}")
print(f"- Distances: {distances[0]}")

2. IVF (Inverted File Index)

IVF hoạt động như thư viện có ngăn kéo. Thay vì tìm kiếm toàn bộ sách, bạn biết sách đó thuộc ngăn nào (cluster), rồi chỉ tìm trong ngăn đó. Thuật toán chia không gian vector thành N clusters, mỗi vector thuộc một cluster và được lập chỉ mục trong danh sách ngược (inverted list).

# Ví dụ: Khởi tạo IVF Index với FAISS
import faiss
import numpy as np


Dữ liệu vector

dimension = 128
nb_vectors = 50000
vectors = np.random.rand(nb_vectors, dimension).astype('float32')


Tạo IVF index với 1000 clusters


nlist = số lượng clusters (thường sqrt(nb_vectors) đến 4*sqrt(nb_vectors))

nlist = 1000
quantizer = faiss.IndexFlatIP(dimension)  # Inner Product (cosine similarity)
ivf_index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT)


Phải train index trước khi thêm vectors


Training giúp tạo ra các centroids tối ưu

print("Đang train IVF index...")
ivf_index.train(vectors)
print("Train hoàn tất!")


Thêm vectors

print(f"Đang thêm {nb_vectors} vectors...")
ivf_index.add(vectors)


Cấu hình nprobe - số clusters cần tìm


nprobe cao = chính xác hơn, chậm hơn

ivf_index.nprobe = 50  # Tìm trong 50 clusters gần nhất


Tìm kiếm

query_vector = np.random.rand(1, dimension).astype('float32')
k = 5
distances, indices = ivf_index.search(query_vector, k)

print(f"\nKết quả IVF:")
print(f"- Indices: {indices[0]}")
print(f"- Distances: {distances[0]}")
print(f"- Số clusters đã tìm: {ivf_index.nprobe}")

3. DiskANN (Disk-based ANN)

DiskANN được thiết kế cho dữ liệu lớn không thể lưu trong RAM. Thuật toán lưu index trên đĩa cứng (SSD) và chỉ đọc phần cần thiết vào bộ nhớ. Đây là lựa chọn tốt nhất khi bạn có hàng tỷ vectors hoặc ngân sách hạn chế không thể mua đủ RAM.

# Ví dụ: Sử dụng DiskANN với DiskANNPlus (Microsoft)

Cài đặt: pip install diskannpy


import numpy as np


Dữ liệu mẫu

dimension = 128
nb_vectors = 100000
vectors = np.random.rand(nb_vectors, dimension).astype('float32')


Import DiskANN SDK

try:
    import diskannpy as dap
    
    # Build index trên đĩa
    index_path = "./hnsw_graph.index"
    
    print("Đang build DiskANN index...")
    dap.build_disk_index(
        data=vectors,
        index_path=index_path,
        complexity=64,        # Độ phức tạp search
        graph_degree=32,      # Bậc đồ thị
        search_memory_max=0.2,  # 20% RAM cho search
        build_memory_max=0.4,   # 40% RAM cho build
    )
    print(f"Index đã lưu tại: {index_path}")
    
    # Load và tìm kiếm
    index = dap.DiskIndex(index_path)
    
    query = np.random.rand(1, dimension).astype('float32')
    k = 10
    distances, indices = index.search(query, k)
    
    print(f"\nKết quả DiskANN:")
    print(f"- Top {k} vectors gần nhất: {indices[0]}")
    
except ImportError:
    print("DiskANN SDK chưa cài đặt. Cài đặt với: pip install diskannpy")

So sánh chi tiết: HNSW vs IVF vs DiskANN

Tiêu chí HNSW IVF DiskANN
Nguyên lý hoạt động Đồ thị phân cấp nhiều tầng Phân cụm + danh sách ngược Đồ thị + lưu trữ đĩa
Yêu cầu RAM Cao (toàn bộ index) Trung bình (quantizer + lists) Thấp (chỉ search beam)
Tốc độ tìm kiếm Rất nhanh (1-10ms) Nhanh (10-50ms) Chậm hơn (20-100ms)
Độ chính xác (Recall) Rất cao (95-99%) Trung bình-cao (80-95%) Cao (90-97%)
Build time Chậm (giờ) Nhanh (phút) Trung bình (30-60 phút)
Kích thước index Lớn nhất Nhỏ hơn HNSW Nhỏ nhất
Quy mô dữ liệu 10 triệu - 1 tỷ vectors 1 triệu - 100 triệu vectors 100 triệu - 10 tỷ vectors
Độ phức tạp cấu hình Thấp Trung bình Cao

Phù hợp / không phù hợp với ai

HNSW - Phù hợp khi:

HNSW - Không phù hợp khi:

IVF - Phù hợp khi:

IVF - Không phù hợp khi:

DiskANN - Phù hợp khi:

DiskANN - Không phù hợp khi:

Giá và ROI

Khi chọn thuật toán vector index, bạn cần tính toán TCO (Total Cost of Ownership) bao gồm chi phí compute, storage và operational.

Thuật toán RAM/Storage/1M vectors Chi phí hàng tháng (1M vectors) Chi phí hàng tháng (100M vectors)
HNSW ~2GB RAM $20-50 $2,000-5,000
IVF ~1.5GB RAM $15-40 $1,500-4,000
DiskANN ~100MB RAM + 500MB SSD $5-15 $500-1,500

So sánh với HolySheep AI

Nếu bạn sử dụng HolySheep AI để deploy RAG application, chi phí vector search được tối ưu hóa sẵn với các thuật toán tốt nhất. Với mức giá từ $2.50/MTok cho embedding model Gemini 2.5 Flash, bạn tiết kiệm đến 85% chi phí so với OpenAI.

Kinh nghiệm thực chiến từ 20+ dự án RAG

Qua hơn 20 dự án triển khai RAG cho doanh nghiệp Việt Nam, tôi đã rút ra những bài học quý giá:

Dự án 1: Chatbot hỏi đáp pháp luật - 500K tài liệu pháp luật. Ban đầu dùng IVF vì tiết kiệm RAM, nhưng recall chỉ đạt 78% khiến câu trả lời sai. Chuyển sang HNSW với efSearch=200, recall lên 96%, latency 8ms. Đáng đầu tư.

Dự án 2: Recommendation engine e-commerce - 50 triệu sản phẩm. Dùng DiskANN vì không đủ ngân sách RAM. Latency 45ms với SSD NVMe, chấp nhận được. Tiết kiệm $3,000/tháng so với HNSW.

Dự án 3: Tìm kiếm ảnh y tế - 10 triệu ảnh X-ray. Dùng HNSW kết hợp PQ (Product Quantization) để giảm 70% RAM. Recall 94%, latency 15ms.

Hướng dẫn tích hợp với HolySheep AI

HolySheep AI cung cấp API chuẩn hóa cho vector search, hỗ trợ cả ba thuật toán với cấu hình tối ưu sẵn. Bạn không cần quản lý infrastructure, chỉ cần gọi API.

# Tích hợp HolySheep AI cho Vector Search và RAG

Base URL: https://api.holysheep.ai/v1


API Key: YOUR_HOLYSHEEP_API_KEY


import requests
import numpy as np


Cấu hình HolySheep AI

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Lấy từ https://www.holysheep.ai/register

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}


============================================


Bước 1: Tạo Collection với index strategy


============================================

def create_vector_collection():
    """
    Tạo collection với index type tùy chọn
    HNSW: best for speed, IVF: balanced, DiskANN: large scale
    """
    payload = {
        "name": "my_documents",
        "dimension": 1536,  # OpenAI ada-002 dimension
        "index_type": "hnsw",  # hnsw | ivf | diskann
        "metric_type": "cosine",  # cosine | euclidean | dot_product
        "hnsw_params": {
            "m": 16,
            "ef_construction": 200,
            "ef_search": 100
        }
    }
    
    response = requests.post(
        f"{BASE_URL}/collections",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        data = response.json()
        print(f"✅ Collection '{data['name']}' đã tạo thành công!")
        print(f"   Index type: {data['index_type']}")
        return data['id']
    else:
        print(f"❌ Lỗi: {response.status_code}")
        print(response.json())
        return None


============================================


Bước 2: Encode văn bản thành vectors


============================================

def encode_text(text):
    """Sử dụng HolySheep embedding API để encode văn bản"""
    payload = {
        "model": "embedding-3-large",  # 3072 dimensions
        "input": text
    }
    
    response = requests.post(
        f"{BASE_URL}/embeddings",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        return response.json()["embedding"]
    else:
        raise Exception(f"Embedding failed: {response.status_code}")


============================================


Bước 3: Thêm documents vào collection


============================================

def add_documents(collection_id, documents):
    """Thêm documents với vectors đã encode"""
    embeddings = [encode_text(doc["content"]) for doc in documents]
    
    payload = {
        "documents": [
            {
                "id": doc["id"],
                "content": doc["content"],
                "metadata": doc.get("metadata", {})
            }
            for doc in documents
        ],
        "embeddings": embeddings
    }
    
    response = requests.post(
        f"{BASE_URL}/collections/{collection_id}/documents",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        result = response.json()
        print(f"✅ Đã thêm {result['added_count']} documents")
        return result
    else:
        raise Exception(f"Add documents failed: {response.text}")


============================================


Bước 4: Vector similarity search


============================================

def search_similar(collection_id, query, top_k=5):
    """Tìm kiếm documents tương tự"""
    query_embedding = encode_text(query)
    
    payload = {
        "query_vector": query_embedding,
        "top_k": top_k,
        "include_embeddings": False  # Tiết kiệm bandwidth
    }
    
    response = requests.post(
        f"{BASE_URL}/collections/{collection_id}/search",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        return response.json()["results"]
    else:
        raise Exception(f"Search failed: {response.text}")


============================================


Ví dụ sử dụng đầy đủ


============================================

if __name__ == "__main__":
    # Tạo collection
    collection_id = create_vector_collection()
    
    if collection_id:
        # Thêm sample documents
        docs = [
            {
                "id": "doc1",
                "content": "Hướng dẫn cài đặt Python trên Windows",
                "metadata": {"category": "programming", "difficulty": "beginner"}
            },
            {
                "id": "doc2",
                "content": "Tối ưu hóa PostgreSQL cho production",
                "metadata": {"category": "database", "difficulty": "advanced"}
            },
            {
                "id": "doc3",
                "content": "Giới thiệu về Machine Learning cơ bản",
                "metadata": {"category": "ml", "difficulty": "intermediate"}
            }
        ]
        
        add_documents(collection_id, docs)
        
        # Tìm kiếm
        results = search_similar(collection_id, "Python tutorial for beginners")
        
        print("\n🔍 Kết quả tìm kiếm:")
        for i, result in enumerate(results, 1):
            print(f"  {i}. {result['content']}")
            print(f"     Score: {result['score']:.4f}")
            print(f"     ID: {result['id']}")
# RAG Pipeline hoàn chỉnh với HolySheep AI

Kết hợp vector search + LLM generation


import requests
import json

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

def rag_query(collection_id, user_question, model="deepseek-v3.2"):
    """
    RAG query: Tìm kiếm context + Generate answer
    """
    # Bước 1: Tìm kiếm documents liên quan
    search_response = requests.post(
        f"{BASE_URL}/collections/{collection_id}/search",
        headers=headers,
        json={
            "query_vector": encode_text(user_question),
            "top_k": 3,
            "min_score": 0.7  # Chỉ lấy kết quả có similarity > 0.7
        }
    )
    
    results = search_response.json()["results"]
    
    if not results:
        return {
            "answer": "Không tìm thấy thông tin liên quan trong cơ sở dữ liệu.",
            "sources": []
        }
    
    # Bước 2: Build context từ kết quả tìm kiếm
    context_parts = []
    for result in results:
        context_parts.append(f"[Document {result['id']}]: {result['content']}")
    
    context = "\n\n".join(context_parts)
    
    # Bước 3: Gọi LLM với context
    prompt = f"""Dựa trên các documents sau, trả lời câu hỏi của người dùng.

Documents:
{context}

Câu hỏi: {user_question}

Trả lời (bằng tiếng Việt):"""
    
    # Sử dụng DeepSeek V3.2 - giá chỉ $0.42/MTok
    llm_response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": "Bạn là trợ lý AI hữu ích, trả lời dựa trên context được cung cấp."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,  # Low temperature cho factual answers
            "max_tokens": 1000
        }
    )
    
    llm_data = llm_response.json()
    
    return {
        "answer": llm_data["choices"][0]["message"]["content"],
        "sources": [r["id"] for r in results],
        "usage": llm_data.get("usage", {})
    }


Ví dụ sử dụng

if __name__ == "__main__":
    collection_id = "my_documents"
    
    question = "Làm thế nào để cài đặt Python?"
    result = rag_query(collection_id, question)
    
    print("=" * 50)
    print("Câu hỏi:", question)
    print("=" * 50)
    print("\nTrả lời:", result["answer"])
    print("\nNguồn tham khảo:", result["sources"])
    
    # Chi phí ước tính
    if result.get("usage"):
        tokens = result["usage"].get("total_tokens", 0)
        cost = tokens / 1_000_000 * 0.42  # DeepSeek V3.2 pricing
        print(f"\n💰 Chi phí LLM: ${cost:.6f}")
        print(f"   Tokens sử dụng: {tokens}")

Lỗi thường gặp và cách khắc phục

Lỗi 1: "OutOfMemoryError" khi build HNSW index

Mô tả: Khi thêm quá nhiều vectors vào HNSW, process bị kill do hết RAM.

# Cách khắc phục 1: Sử dụng Product Quantization (PQ) để giảm kích thước vector
import faiss
import numpy as np

dimension = 1536  # OpenAI embedding dimension
nb_vectors = 10000000  # 10 triệu vectors


Thay vì lưu vector đầy đủ 1536 float32 (~6KB/vector)


PQ mã hóa thành 128 bytes/vector (tiết kiệm 98% RAM)



Build PQ index với 64 subquantizers, mỗi subquantizer 256 codewords

pq = faiss.ProductQuantizer(dimension, 64, 8)  # 64*8bits = 512bits = 64bytes


Tạo index với PQ

index = faiss.IndexIVFPQ(pq, dimension, 65536, 8, 8)

                         |     |           |  |  |


                    quantizer  ncentroids nbits nsubq



Train và add như bình thường


vectors_train phải có ít nhất 100 * ncentroids vectors

vectors = np.random.rand(nb_vectors, dimension).astype('float32')
vectors_train = vectors[:1000000].astype('float32')  # 1M cho training

index.train(vectors_train)
index.add(vectors)

print(f"Index đã build với PQ!")
print(f"Kích thước RAM giảm từ {nb_vectors * dimension * 4 / 1e9:.1f}GB xuống còn ~{nb_vectors * 64 / 1e9:.1f}GB")

Lỗi 2: Recall thấp với IVF index

Mô tả: IVF trả về kết quả không chính xác, recall chỉ 60-70%.

# Cách khắc phục: Tăng nprobe và sử dụng refine
import faiss
import numpy as np


Tạo IVF index

dimension = 128
nb_vectors = 50000
vectors = np.random.rand(nb_vectors, dimension).astype('float32')

nlist = 500
quantizer = faiss.IndexFlatIP(dimension)
ivf_index = faiss.IndexIVFFlat(quantizer, dimension, nlist)


Training

ivf_index.train(vectors)
ivf_index.add(vectors)


============================================


Khắc phục 1: Tăng nprobe (mặc định = 1)


============================================

print("Testing với nprobe khác nhau...")

for nprobe in [1, 10, 50, 100, nlist]:
    ivf_index.nprobe = nprobe
    
    query = vectors[0:1]
    distances, indices = ivf_index.search(query, 10)
    
    # Tính recall (indices[0][0] phải là 0 vì query từ chính dataset)
    recall_at_1 = 1 if indices[0][0] == 0 else 0
    
    print(f"  nprobe={nprobe:4d} -> recall@1 = {recall_at_1}")


============================================


Khắc phục 2: Sử dụng refine để cải thiện recall


============================================


Refine lấy thêm candidates và re-rank

refine_index = faiss.IndexRefineFlat(ivf_index)
refine_index.k_factor = 4  # Lấy 4x candidates

query = vectors[0:1]
distances, indices = refine_index.search(query, 10)
print(f"\nVới refine: recall@1 = {1 if indices[0][0] == 0 else 0}")

Lỗi 3: DiskANN latency cao bất thường

Mô tả: DiskANN chậm hơn dự kiến, có thể lên đến 500ms.

# Cách khắc phục: Tối ưu SSD và search parameters
import numpy as np


============================================


Khắc phục 1: Đảm bảo sử dụng SSD NVMe, không phải HDD


============================================

import shutil

def check_disk_type(path):
    """Kiểm tra loại ổ đĩa"""
    try:

Tài nguyên liên quan
Bài viết liên quan