4.1 概述

向量数据库是专门用于存储、管理和查询高维向量的数据库系统。它们在嵌入向量应用中扮演着核心角色，提供高效的相似度搜索和向量检索能力。

4.1.1 核心功能

• 向量存储：高效存储高维向量数据
• 相似度搜索：快速找到最相似的向量
• 元数据管理：管理向量的相关属性信息
• 实时查询：支持毫秒级的向量检索

4.1.2 与传统数据库的区别

特性	传统数据库	向量数据库
数据类型	结构化数据	高维向量
查询方式	精确匹配	相似度搜索
索引结构	B树、哈希	HNSW、IVF
应用场景	业务数据	语义搜索、推荐

4.2 向量相似度搜索算法

1. 精确搜索 (Brute Force)

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class BruteForceSearch:
    def __init__(self):
        self.vectors = None
        self.metadata = None
    
    def add(self, vectors, metadata=None):
        """添加向量到数据库"""
        if self.vectors is None:
            self.vectors = vectors
            self.metadata = metadata if metadata is not None else [None] * len(vectors)
        else:
            self.vectors = np.vstack([self.vectors, vectors])
            if metadata is not None:
                self.metadata.extend(metadata)
    
    def search(self, query_vector, k=5, metric='cosine'):
        """暴力搜索最相似的k个向量"""
        if metric == 'cosine':
            similarities = cosine_similarity([query_vector], self.vectors)[0]
        elif metric == 'euclidean':
            similarities = -np.linalg.norm(self.vectors - query_vector, axis=1)
        
        # 获取top-k结果
        top_indices = np.argsort(similarities)[-k:][::-1]
        
        results = []
        for idx in top_indices:
            results.append({
                'index': idx,
                'similarity': similarities[idx],
                'metadata': self.metadata[idx]
            })
        
        return results

# 使用示例
search_engine = BruteForceSearch()

# 添加一些向量
vectors = np.random.rand(1000, 128)  # 1000个128维向量
metadata = [f"doc_{i}" for i in range(1000)]

search_engine.add(vectors, metadata)

# 搜索查询
query = np.random.rand(128)
results = search_engine.search(query, k=3)

for result in results:
    print(f"文档: {result['metadata']}, 相似度: {result['similarity']:.4f}")

2. 近似最近邻搜索 (ANN)

HNSW (Hierarchical Navigable Small World)

import hnswlib
import numpy as np

class HNSWIndex:
    def __init__(self, dim=128, max_elements=10000, ef_construction=200, M=16):
        self.dim = dim
        self.index = hnswlib.Index(space='cosine', dim=dim)
        
        # 初始化索引
        self.index.init_index(
            max_elements=max_elements,
            ef_construction=ef_construction,
            M=M
        )
        
        self.metadata = []
    
    def add(self, vectors, metadata=None):
        """添加向量到HNSW索引"""
        if metadata is None:
            metadata = [None] * len(vectors)
        
        # 添加向量
        self.index.add_items(vectors)
        self.metadata.extend(metadata)
    
    def search(self, query_vector, k=5, ef_search=50):
        """使用HNSW搜索"""
        # 设置搜索参数
        self.index.set_ef(ef_search)
        
        # 搜索
        labels, distances = self.index.knn_query(query_vector, k=k)
        
        results = []
        for label, distance in zip(labels[0], distances[0]):
            results.append({
                'index': int(label),
                'distance': float(distance),
                'similarity': 1 - float(distance),  # 转换为相似度
                'metadata': self.metadata[int(label)]
            })
        
        return results

# 使用示例
hnsw_index = HNSWIndex(dim=128, max_elements=10000)

# 添加向量
vectors = np.random.rand(5000, 128).astype(np.float32)
metadata = [f"item_{i}" for i in range(5000)]

hnsw_index.add(vectors, metadata)

# 搜索
query = np.random.rand(128).astype(np.float32)
results = hnsw_index.search(query, k=3)

for result in results:
    print(f"项目: {result['metadata']}, 相似度: {result['similarity']:.4f}")

4.3 主流向量数据库

4.3.1 Milvus

特点: 可能是最受欢迎的开源向量数据库之一，由 Zilliz 公司发起。功能非常全面，为生产环境设计，支持多种索引类型（如 IVF, HNSW），具备高可用、高扩展性和强大的标量字段过滤能力。生态系统非常成熟。

适用场景: 大规模、企业级的 AI 应用，对系统的稳定性、扩展性和功能完整性要求高的场景。

部署模式: 开源自托管，也有 Zilliz Cloud 提供的全托管服务。

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import numpy as np

# 连接Milvus服务器
connections.connect("default", host="localhost", port="19530")

# 定义集合结构
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512)
]

schema = CollectionSchema(fields, "text_collection")

# 创建集合
collection = Collection("text_collection", schema)

# 创建索引
index_params = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 200}
}

collection.create_index("vector", index_params)

# 插入数据
vectors = np.random.rand(1000, 128).astype(np.float32)
texts = [f"text_{i}" for i in range(1000)]
ids = list(range(1000))

collection.insert([ids, vectors, texts])

# 加载集合到内存
collection.load()

# 搜索
search_vectors = np.random.rand(1, 128).astype(np.float32)
results = collection.search(
    data=search_vectors,
    anns_field="vector",
    param={"ef": 50},
    limit=5,
    output_fields=["text"]
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 文本: {hit.entity.get('text')}, 距离: {hit.distance}")

4.3.2 Pinecone

特点: 虽然 Pinecone 以其托管服务闻名，但它最初也是作为一种专门的向量数据库技术出现的。它以极致的易用性和高性能著称，让开发者无需关心复杂的底层索引和集群管理。查询速度和性能优化做得非常好。

适用场景: 快速原型开发、对运维不熟悉但对性能要求高的团队、各种规模的 RAG 应用。

部署模式: 主要是全托管的云服务（Serverless）。

4.3.3 Weaviate

特点: 一个非常灵活的开源向量数据库。其一大亮点是内置了数据模块（Modules），可以与 OpenAI、Cohere、Hugging Face 等模型服务无缝集成，在数据导入时自动进行向量化，极大简化了开发流程。它支持 GraphQL API，并且具备强大的混合搜索（Hybrid Search）能力。

适用场景: 需要快速集成多种 AI 服务、重视混合搜索能力、希望简化数据处理流程的应用。

部署模式: 开源自托管，也有 Weaviate Cloud 提供的全托管服务。

import weaviate
import json

# 连接Weaviate
client = weaviate.Client("http://localhost:8080")

# 创建类（类似表）
class_obj = {
    "class": "Document",
    "vectorizer": "text2vec-openai",  # 使用OpenAI嵌入
    "moduleConfig": {
        "text2vec-openai": {
            "model": "ada",
            "type": "text"
        }
    }
}

client.schema.create_class(class_obj)

# 添加数据对象
data_object = {
    "content": "这是一个示例文档",
    "source": "example.pdf"
}

client.data_object.create(
    data_object=data_object,
    class_name="Document"
)

# 混合搜索（向量+关键词）
query = {
    "query": """
    {
        Get {
            Document(
                hybrid: {
                    query: "机器学习"
                },
                limit: 3
            ) {
                content
                source
                _additional {
                    score
                }
            }
        }
    }
    """
}

result = client.query.raw(query)
print(json.dumps(result, indent=2))

4.3.4 Qdrant

特点: 使用 Rust 语言开发，主打性能和内存安全。它的过滤查询（filtered search）性能非常出色，即使在应用复杂过滤条件的情况下，也能保持较高的搜索速度。提供了丰富的客户端库（Python, Go, Rust等）。

适用场景: 对搜索性能和过滤条件要求非常高的场景，重视资源效率和运行速度的应用。

部署模式: 开源自托管，也有 Qdrant Cloud 提供的全托管服务。

from qdrant_client import QdrantClient
import numpy as np

# 连接Qdrant
client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.recreate_collection(
    collection_name="documents",
    vectors_config={
        "size": 128,
        "distance": "Cosine"
    }
)

# 添加向量
vectors = np.random.rand(100, 128).tolist()
payloads = [
    {"text": f"document_{i}", "category": "tech" if i % 2 == 0 else "science"}
    for i in range(100)
]

client.upsert(
    collection_name="documents",
    points=[
        {
            "id": i,
            "vector": vectors[i],
            "payload": payloads[i]
        }
        for i in range(100)
    ]
)

# 带过滤条件的搜索
search_result = client.search(
    collection_name="documents",
    query_vector=np.random.rand(128).tolist(),
    query_filter={
        "must": [
            {
                "key": "category",
                "match": {"value": "tech"}
            }
        ]
    },
    limit=5
)

for hit in search_result:
    print(f"ID: {hit.id}, 文本: {hit.payload['text']}, 分数: {hit.score}")

4.3.5 Chroma DB

特点: 轻量级、对开发者友好，非常适合在本地开发和小型项目中使用。它被设计为"分析型"向量数据库，易于上手，可以作为 Python 库直接在应用中运行。随着发展，也逐渐支持了客户端/服务器模式。

适用场景: 个人项目、快速实验、Jupyter Notebook 中的数据分析、小型应用的后端。

import chromadb
from chromadb.utils import embedding_functions

# 初始化ChromaDB
client = chromadb.PersistentClient(path="./chroma_db")

# 创建集合
collection = client.create_collection(name="documents")

# 添加文档
documents = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络",
    "自然语言处理处理文本数据"
]

metadatas = [
    {"source": "doc1.pdf", "page": 1},
    {"source": "doc2.pdf", "page": 2},
    {"source": "doc3.pdf", "page": 3}
]

ids = ["doc1", "doc2", "doc3"]

collection.add(
    documents=documents,
    metadatas=metadatas,
    ids=ids
)

# 查询
query_text = "什么是神经网络"
results = collection.query(
    query_texts=[query_text],
    n_results=2
)

print(f"查询: {query_text}")
for doc, metadata in zip(results['documents'][0], results['metadatas'][0]):
    print(f"文档: {doc}")
    print(f"来源: {metadata['source']}")
    print("-" * 50)

4.3.6 FAISS (Facebook AI Similarity Search)

特点: 由Facebook AI Research开发的向量相似度搜索库，虽然严格来说不是完整的数据库系统，但它是业界最广泛使用的向量搜索工具之一。提供了多种高效的近似最近邻搜索算法，支持GPU加速，性能卓越。

适用场景: 学术研究、高性能向量搜索、需要与其他系统集成、对搜索算法有定制化需求的场景。

部署模式: 开源库，需要自行集成到应用中。

4.3.7 Elasticsearch (向量搜索)

特点: 虽然Elasticsearch primarily是全文搜索引擎，但现在的版本提供了强大的向量搜索能力。它可以将向量搜索与传统全文搜索结合，支持复杂的过滤和聚合操作，生态系统非常成熟。

适用场景: 需要同时支持文本搜索和向量搜索的应用、现有Elasticsearch用户的向量搜索需求、企业级搜索应用。

部署模式: 开源自托管，也有Elastic Cloud提供的云服务。

4.4 向量数据库应用场景

4.4.1 语义搜索

class SemanticSearchEngine:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db
    
    def add_documents(self, documents):
        """添加文档到搜索引擎"""
        # 生成嵌入向量
        embeddings = self.embedding_model.encode(documents)
        
        # 添加到向量数据库
        self.vector_db.add(embeddings, documents)
    
    def search(self, query, k=5):
        """搜索相关文档"""
        # 生成查询向量
        query_embedding = self.embedding_model.encode([query])[0]
        
        # 搜索相似文档
        results = self.vector_db.search(query_embedding, k=k)
        
        return results

# 使用示例
from sentence_transformers import SentenceTransformer

# 初始化
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
vector_db = HNSWIndex(dim=384)
search_engine = SemanticSearchEngine(embedding_model, vector_db)

# 添加文档
documents = [
    "机器学习算法可以自动从数据中学习模式",
    "深度学习使用多层神经网络处理复杂数据",
    "自然语言处理帮助计算机理解人类语言",
    "计算机视觉让机器能够理解和分析图像"
]

search_engine.add_documents(documents)

# 搜索
query = "如何让机器理解文本"
results = search_engine.search(query, k=2)

print(f"查询: {query}")
for result in results:
    print(f"相关文档: {result['metadata']}")
    print(f"相似度: {result['similarity']:.4f}")

4.4.2 推荐系统

class RecommendationSystem:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db
    
    def add_items(self, items):
        """添加推荐项目"""
        descriptions = [item['description'] for item in items]
        embeddings = self.embedding_model.encode(descriptions)
        
        self.vector_db.add(embeddings, items)
    
    def recommend(self, user_preference, k=5):
        """基于用户偏好推荐项目"""
        # 生成偏好向量
        preference_embedding = self.embedding_model.encode([user_preference])[0]
        
        # 搜索相似项目
        recommendations = self.vector_db.search(preference_embedding, k=k)
        
        return recommendations

# 使用示例
rec_system = RecommendationSystem(embedding_model, HNSWIndex(dim=384))

# 添加商品
products = [
    {'name': 'Python编程书', 'description': '学习Python编程的完整指南', 'price': 49.99},
    {'name': '机器学习课程', 'description': '深入了解机器学习算法和应用', 'price': 99.99},
    {'name': '数据分析工具', 'description': '用于数据分析和可视化的软件', 'price': 79.99}
]

rec_system.add_items(products)

# 生成推荐
user_preference = "我想学习人工智能编程"
recommendations = rec_system.recommend(user_preference, k=2)

print(f"用户偏好: {user_preference}")
for rec in recommendations:
    product = rec['metadata']
    print(f"推荐: {product['name']} - ${product['price']}")
    print(f"描述: {product['description']}")
    print(f"匹配度: {rec['similarity']:.4f}")

4.4.3 问答系统

class QASystem:
    def __init__(self, embedding_model, vector_db):
        self.embedding_model = embedding_model
        self.vector_db = vector_db
    
    def add_knowledge(self, qa_pairs):
        """添加问答知识库"""
        questions = [pair['question'] for pair in qa_pairs]
        embeddings = self.embedding_model.encode(questions)
        
        self.vector_db.add(embeddings, qa_pairs)
    
    def answer_question(self, question, k=3):
        """回答问题"""
        # 生成问题向量
        question_embedding = self.embedding_model.encode([question])[0]
        
        # 搜索相似问题
        similar_questions = self.vector_db.search(question_embedding, k=k)
        
        return similar_questions

# 使用示例
qa_system = QASystem(embedding_model, HNSWIndex(dim=384))

# 添加知识库
knowledge_base = [
    {'question': '什么是机器学习？', 'answer': '机器学习是人工智能的一个分支，让计算机从数据中自动学习。'},
    {'question': '深度学习和机器学习的区别？', 'answer': '深度学习是机器学习的子集，使用深层神经网络。'},
    {'question': '如何学习自然语言处理？', 'answer': '需要掌握语言学、机器学习和深度学习知识。'}
]

qa_system.add_knowledge(knowledge_base)

# 问答
user_question = "我想了解AI学习方法"
answers = qa_system.answer_question(user_question, k=2)

print(f"用户问题: {user_question}")
for answer in answers:
    qa_pair = answer['metadata']
    print(f"相关问题: {qa_pair['question']}")
    print(f"答案: {qa_pair['answer']}")
    print(f"相似度: {answer['similarity']:.4f}")
    print("-" * 50)