2.1 向量数据库基础 — AI知识库搭建全攻略

本节导读：深入理解向量数据库的核心概念、架构设计和工作原理，为AI知识库的构建奠定坚实的理论基础，掌握向量存储和检索的关键技术。

学习目标

• 理解向量数据库的基本概念和架构原理
• 掌握向量索引和检索的核心算法
• 了解主流向量数据库的技术特点
• 能够根据业务需求选择合适的向量数据库
• 具备向量数据库的性能优化能力

核心概念

向量数据库是专门用于存储、管理和检索高维向量数据的数据库系统。与传统关系型数据库不同，向量数据库专注于向量空间中的相似性计算，通过高效的向量索引算法实现快速检索。

向量数据库与传统数据库对比

向量数据库的核心组件

┌─────────────────────────────────────────┐
│            客户端接口层                 │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
│  │REST API │ │gRPC API │ │SDK接口  │   │
│  └─────────┘ └─────────┘ └─────────┘   │
└─────────────────────────────────────────┘
            ↑              ↑
┌─────────────────────────────────────────┐
│            查询处理层                   │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
│  │查询解析  │ │路由转发  │ │负载均衡  │   │
│  └─────────┘ └─────────┘ └─────────┘   │
└─────────────────────────────────────────┘
            ↑              ↑
┌─────────────────────────────────────────┐
│            索引管理层                   │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
│  │IVF索引   │ │HNSW索引  │ │其他索引  │   │
│  └─────────┘ └─────────┘ └─────────┘   │
└─────────────────────────────────────────┐
            ↑              ↑
┌─────────────────────────────────────────┐
│            存储层                      │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐   │
│  │向量存储  │ │元数据   │ │日志系统  │   │
│  └─────────┘ └─────────┘ └─────────┘   │
└─────────────────────────────────────────┘
​

环境准备 / 前置知识

基础数学知识

• 向量运算：点积、余弦相似度、欧氏距离
• 矩阵运算：矩阵乘法、特征值分解
• 概率统计：概率分布、统计分析
• 数据结构：树结构、图结构

技术栈要求

• 编程语言：Python 3.8+（推荐3.10+）
• 科学计算：NumPy、SciPy
• 机器学习：Scikit-learn
• 向量数据库：Milvus、FAISS、Chroma等

分步实战

步骤 1：理解向量空间和相似度计算

向量空间是向量数据库的理论基础，我们需要先理解向量空间的概念和相似度计算方法。

向量空间的数学基础

import numpy as np
from typing import List, Tuple

class VectorSpace:
    def __init__(self, dimension: int):
        self.dimension = dimension
        self.vectors = {}
    
    def add_vector(self, vector_id: str, vector: np.ndarray):
        """添加向量到向量空间"""
        if len(vector) != self.dimension:
            raise ValueError(f"向量维度必须是{self.dimension}")
        self.vectors[vector_id] = vector
    
    def euclidean_distance(self, v1: np.ndarray, v2: np.ndarray) -> float:
        """计算欧氏距离"""
        return np.linalg.norm(v1 - v2)
    
    def cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float:
        """计算余弦相似度"""
        dot_product = np.dot(v1, v2)
        norm_v1 = np.linalg.norm(v1)
        norm_v2 = np.linalg.norm(v2)
        return dot_product / (norm_v1 * norm_v2)
    
    def dot_product(self, v1: np.ndarray, v2: np.ndarray) -> float:
        """计算点积"""
        return np.dot(v1, v2)

# 使用示例
vs = VectorSpace(128)

# 添加向量
vector1 = np.random.rand(128)
vector2 = np.random.rand(128)
vs.add_vector("doc1", vector1)
vs.add_vector("doc2", vector2)

# 计算相似度
cos_sim = vs.cosine_similarity(vector1, vector2)
euclidean_dist = vs.euclidean_distance(vector1, vector2)
dot_product = vs.dot_product(vector1, vector2)

print(f"余弦相似度: {cos_sim:.4f}")
print(f"欧氏距离: {euclidean_dist:.4f}")
print(f"点积: {dot_product:.4f}")
​

相似度计算的应用场景

步骤 2：掌握向量索引技术

向量索引是向量数据库的核心技术，它决定了查询效率和准确性。我们主要学习几种主流的索引算法。

IVF（Inverted File Index）索引

import numpy as np
from typing import List, Tuple
from sklearn.cluster import KMeans
import faiss

class IVFIndex:
    def __init__(self, dimension: int, n_clusters: int):
        self.dimension = dimension
        self.n_clusters = n_clusters
        self.quantizer = KMeans(n_clusters=n_clusters, random_state=42)
        self.index = None
        self.cluster_centers = None
    
    def fit(self, vectors: np.ndarray):
        """训练IVF索引"""
        # 聚类，创建质心
        self.quantizer.fit(vectors)
        self.cluster_centers = self.quantizer.cluster_centers_
        
        # 创建FAISS索引
        quantizer = faiss.IndexFlatL2(self.dimension)
        self.index = faiss.IndexIVFFlat(quantizer, self.dimension, self.n_clusters)
        self.index.train(vectors)
        self.index.add(vectors)
    
    def search(self, query_vector: np.ndarray, k: int = 10, nprobe: int = 10):
        """搜索最相似的向量"""
        if self.index is None:
            raise ValueError("Index not trained")
        
        self.index.nprobe = nprobe
        distances, indices = self.index.search(np.array([query_vector]), k)
        return distances[0], indices[0]
    
    def get_cluster_info(self) -> dict:
        """获取聚类信息"""
        if self.cluster_centers is None:
            return {}
        
        return {
            "n_clusters": self.n_clusters,
            "cluster_centers": self.cluster_centers,
            "cluster_size": len(self.cluster_centers)
        }

# 使用示例
# 生成示例数据
vectors = np.random.rand(1000, 128).astype(np.float32)

# 创建IVF索引
ivf_index = IVFIndex(dimension=128, n_clusters=100)
ivf_index.fit(vectors)

# 搜索测试
query_vector = np.random.rand(128).astype(np.float32)
distances, indices = ivf_index.search(query_vector, k=5)

print("搜索结果:")
for i, (dist, idx) in enumerate(zip(distances, indices)):
    print(f"第{i+1}个: 距离={dist:.4f}, 索引={idx}")
​

HNSW（Hierarchical Navigable Small World）索引

import numpy as np
import faiss
from typing import List, Tuple

class HNSWIndex:
    def __init__(self, dimension: int, m: int = 16):
        self.dimension = dimension
        self.m = m
        self.index = None
    
    def fit(self, vectors: np.ndarray):
        """训练HNSW索引"""
        # 创建HNSW索引
        self.index = faiss.IndexHNSWFlat(self.dimension, self.m)
        
        # 训练并添加向量
        self.index.train(vectors)
        self.index.add(vectors)
    
    def search(self, query_vector: np.ndarray, k: int = 10, ef: int = 200):
        """搜索最相似的向量"""
        if self.index is None:
            raise ValueError("Index not trained")
        
        # 设置搜索参数
        self.index.hnsw.ef = ef
        
        distances, indices = self.index.search(np.array([query_vector]), k)
        return distances[0], indices[0]
    
    def save(self, file_path: str):
        """保存索引到文件"""
        if self.index is None:
            raise ValueError("Index not trained")
        faiss.write_index(self.index, file_path)
    
    def load(self, file_path: str):
        """从文件加载索引"""
        self.index = faiss.read_index(file_path)

# 使用示例
# 生成示例数据
vectors = np.random.rand(1000, 128).astype(np.float32)

# 创建HNSW索引
hnsw_index = HNSWIndex(dimension=128, m=16)
hnsw_index.fit(vectors)

# 搜索测试
query_vector = np.random.rand(128).astype(np.float32)
distances, indices = hnsw_index.search(query_vector, k=5)

print("HNSW搜索结果:")
for i, (dist, idx) in enumerate(zip(distances, indices)):
    print(f"第{i+1}个: 距离={dist:.4f}, 索引={idx}")

# 保存索引
hnsw_index.save("hnsw_index.faiss")
​

步骤 3：实现向量数据库基本功能

让我们实现一个简化的向量数据库，包含基本的存储、索引和检索功能。

自定义向量数据库实现

import numpy as np
from typing import List, Dict, Tuple, Optional
import pickle
import os
from dataclasses import dataclass
from sklearn.cluster import KMeans
import faiss

@dataclass
class VectorDocument:
    """文档向量数据结构"""
    doc_id: str
    vector: np.ndarray
    metadata: Dict[str, str]
    text: str

class SimpleVectorDB:
    def __init__(self, dimension: int, index_type: str = "IVF"):
        self.dimension = dimension
        self.index_type = index_type
        self.documents: Dict[str, VectorDocument] = {}
        self.vectors: np.ndarray = None
        self.index = None
        self.index_map: Dict[int, str] = {}  # 向量索引到文档ID的映射
    
    def add_document(self, doc_id: str, vector: np.ndarray, text: str, metadata: Dict[str, str] = None):
        """添加文档"""
        if len(vector) != self.dimension:
            raise ValueError(f"向量维度必须是{self.dimension}")
        
        if metadata is None:
            metadata = {}
        
        # 创建文档对象
        doc = VectorDocument(
            doc_id=doc_id,
            vector=vector.astype(np.float32),
            metadata=metadata,
            text=text
        )
        
        # 存储文档
        self.documents[doc_id] = doc
        
        # 重建索引
        self._rebuild_index()
    
    def add_documents(self, docs: List[VectorDocument]):
        """批量添加文档"""
        for doc in docs:
            self.documents[doc.doc_id] = doc
        
        self._rebuild_index()
    
    def _rebuild_index(self):
        """重建索引"""
        if len(self.documents) == 0:
            return
        
        # 准备向量数据
        self.vectors = np.array([doc.vector for doc in self.documents.values()])
        self.index_map = {i: doc_id for i, doc_id in enumerate(self.documents.keys())}
        
        # 根据类型创建索引
        if self.index_type == "IVF":
            self._create_ivf_index()
        elif self.index_type == "HNSW":
            self._create_hnsw_index()
        else:
            raise ValueError(f"不支持的索引类型: {self.index_type}")
    
    def _create_ivf_index(self):
        """创建IVF索引"""
        quantizer = faiss.IndexFlatL2(self.dimension)
        self.index = faiss.IndexIVFFlat(quantizer, self.dimension, min(100, len(self.documents)))
        self.index.train(self.vectors)
        self.index.add(self.vectors)
    
    def _create_hnsw_index(self):
        """创建HNSW索引"""
        self.index = faiss.IndexHNSWFlat(self.dimension, 16)
        self.index.train(self.vectors)
        self.index.add(self.vectors)
    
    def search(self, query_vector: np.ndarray, k: int = 10, 
               similarity_threshold: float = 0.5, 
               filter_metadata: Dict[str, str] = None) -> List[Tuple[str, float, str, Dict[str, str]]]:
        """搜索最相似的文档"""
        if self.index is None:
            return []
        
        # 执行搜索
        query_vector = query_vector.astype(np.float32).reshape(1, -1)
        distances, indices = self.index.search(query_vector, k)
        
        # 处理结果
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx == -1:  # 无效索引
                continue
            
            doc_id = self.index_map[idx]
            doc = self.documents[doc_id]
            
            # 计算相似度（这里假设距离越小相似度越高）
            similarity = 1.0 / (1.0 + dist)
            
            # 应用过滤条件
            if similarity < similarity_threshold:
                continue
            
            if filter_metadata:
                for key, value in filter_metadata.items():
                    if key not in doc.metadata or doc.metadata[key] != value:
                        break
                else:
                    results.append((doc_id, similarity, doc.text, doc.metadata))
            else:
                results.append((doc_id, similarity, doc.text, doc.metadata))
        
        return results
    
    def get_document(self, doc_id: str) -> Optional[VectorDocument]:
        """获取文档"""
        return self.documents.get(doc_id)
    
    def delete_document(self, doc_id: str):
        """删除文档"""
        if doc_id in self.documents:
            del self.documents[doc_id]
            self._rebuild_index()
    
    def update_document(self, doc_id: str, new_vector: Optional[np.ndarray] = None, 
                        new_text: Optional[str] = None, new_metadata: Optional[Dict[str, str]] = None):
        """更新文档"""
        if doc_id not in self.documents:
            raise ValueError(f"文档不存在: {doc_id}")
        
        doc = self.documents[doc_id]
        if new_vector is not None:
            doc.vector = new_vector.astype(np.float32)
        if new_text is not None:
            doc.text = new_text
        if new_metadata is not None:
            doc.metadata.update(new_metadata)
        
        self._rebuild_index()
    
    def save(self, directory: str):
        """保存数据库到文件"""
        os.makedirs(directory, exist_ok=True)
        
        # 保存文档
        docs_file = os.path.join(directory, "documents.pkl")
        with open(docs_file, 'wb') as f:
            pickle.dump(self.documents, f)
        
        # 保存索引
        if self.index is not None:
            index_file = os.path.join(directory, f"{self.index_type}_index.faiss")
            faiss.write_index(self.index, index_file)
    
    def load(self, directory: str):
        """从文件加载数据库"""
        # 加载文档
        docs_file = os.path.join(directory, "documents.pkl")
        with open(docs_file, 'rb') as f:
            self.documents = pickle.load(f)
        
        # 重建索引
        self._rebuild_index()
    
    def get_stats(self) -> Dict[str, any]:
        """获取数据库统计信息"""
        return {
            "total_documents": len(self.documents),
            "dimension": self.dimension,
            "index_type": self.index_type,
            "index_size": self.vectors.shape[0] if self.vectors is not None else 0
        }

# 使用示例
# 创建向量数据库
db = SimpleVectorDB(dimension=128, index_type="IVF")

# 添加文档
docs = [
    VectorDocument(
        doc_id="doc1",
        vector=np.random.rand(128),
        text="AI知识库是基于大语言模型和向量数据库的技术解决方案",
        metadata={"category": "AI", "length": "short"}
    ),
    VectorDocument(
        doc_id="doc2",
        vector=np.random.rand(128),
        text="向量数据库支持高效的语义搜索和相似度计算",
        metadata={"category": "Database", "length": "medium"}
    ),
    VectorDocument(
        doc_id="doc3",
        vector=np.random.rand(128),
        text="检索增强生成(RAG)技术通过外部知识库提高回答的准确性",
        metadata={"category": "RAG", "length": "long"}
    )
]

# 批量添加文档
db.add_documents(docs)

# 搜索测试
query_vector = np.random.rand(128)
results = db.search(query_vector, k=3, similarity_threshold=0.1)

print("搜索结果:")
for doc_id, similarity, text, metadata in results:
    print(f"文档ID: {doc_id}")
    print(f"相似度: {similarity:.4f}")
    print(f"文本: {text}")
    print(f"元数据: {metadata}")
    print("-" * 50)

# 获取统计信息
stats = db.get_stats()
print(f"数据库统计: {stats}")

# 保存数据库
db.save("./vector_db_example")
​

常见问题 FAQ

Q1：向量数据库与传统数据库有什么根本区别？

A：向量数据库与传统数据库在多个方面存在根本区别：

• 数据类型：传统数据库处理结构化数据，向量数据库处理高维向量
• 查询方式：传统数据库使用精确匹配和范围查询，向量数据库使用相似度搜索
• 索引结构：传统数据库使用B+树、哈希表等，向量数据库使用IVF、HNSW等
• 应用场景：传统数据库用于事务处理，向量数据库用于语义搜索和推荐

Q2：如何选择合适的向量索引算法？

A：选择向量索引算法需要考虑以下因素：

• 数据规模：小数据集可以使用HNSW，大数据集适合IVF
• 查询性能：HNSW查询速度快但构建成本高，IVF平衡性好
• 内存使用：IVF内存占用较少，HNSW需要更多内存
• 准确率要求：HNSW通常提供更好的搜索质量
• 硬件环境：考虑CPU和GPU资源分配

Q3：向量数据库的性能优化有哪些关键点？

A：性能优化的关键点包括：

• 索引参数调优：调整IVF的nprobe、HNSW的ef参数
• 数据分片：大规模数据采用分片策略
• 缓存机制：实现热点数据的缓存
• 并行处理：多线程并发查询
• 硬件加速：使用GPU加速计算

最佳实践与避坑

向量数据库选型最佳实践

1. 原型阶段：使用Chroma快速验证概念
2. 小规模部署：使用FAISS进行本地部署
3. 企业级应用：使用Milvus或Pinecone
4. 云原生架构：选择支持云原生的向量数据库

常见避坑指南

1. 维度灾难：避免使用过高维度的向量
2. 相似度阈值设置：根据应用场景调整相似度阈值
3. 索引重建：频繁更新数据时注意索引重建成本
4. 内存管理：大规模数据时注意内存使用
5. 查询优化：避免返回过多结果导致性能下降

本节小结

通过本节学习，我们深入理解了向量数据库的核心概念、架构原理和工作机制。掌握了向量空间、相似度计算、索引技术等基础知识，并学习了如何实现和优化向量数据库。

关键收获：

• 向量数据库专注于相似度搜索，与传统数据库有本质区别
• IVF和HNSW是两种主流的索引技术，各有优缺点
• 自定义向量数据库实现帮助我们理解底层原理
• 性能优化需要综合考虑索引参数、数据规模和硬件资源

下一节衔接：在理解向量数据库的基础上，下一节将深入讲解文档处理与向量化技术，包括如何将各种格式的文档转换为可检索的向量表示。

延伸阅读

• 官方文档：FAISS官方文档和技术报告
• 相关章节：本教程2.2节文档处理与向量化
• 技术博客：Milvus架构设计和最佳实践
• 参考资料：向量数据库技术综述论文

关键词：AI知识库搭建全攻略, 向量数据库, 向量索引, HNSW, IVF, 相似度搜索, 数据库架构, 技术原理
难度：进阶
预计阅读：45 分钟