数据库

向量数据库完全指南：从嵌入到相似度搜索

向量数据库成为AI应用的基础设施。本文从原理到实战，完整介绍向量数据库的设计、选型和优化。

向量数据库：AI时代的新基础设施

如果说关系数据库是Web应用的基础设施，那么向量数据库就是AI应用的基础设施。从推荐系统到RAG再到AI搜索，向量数据库已经成为不可或缺的一环。但很多人对它的理解还停留在「能做向量搜索」这样的表面认识。

为什么需要向量数据库？

传统数据库是为结构化数据设计的：

  • SQL查询：精确匹配
  • 索引：B-tree、Hash等
  • 约束：主键、外键、唯一性

但AI时代的数据是向量化的：

  • 语义相似性：「好」和「不错」应该被视为相似
  • 高维空间：一个文本Embedding可能有1536维
  • 近似匹配：找到最相似的，而不是精确匹配

传统数据库根本不适合这种场景。向量数据库应运而生。

向量数据库的核心概念

1. Embedding 和向量表示

所有的数据都需要先转换为向量：

from openai import OpenAI

client = OpenAI()

def text_to_vector(text):
    """将文本转换为向量"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# 示例
vector1 = text_to_vector("Python是一种编程语言")
vector2 = text_to_vector("Python是编程的好工具")

# vector1 和 vector2 在语义上相似，所以向量相似度会很高
print(len(vector1))  # 1536维

关键特点：

  • 确定性：同样的文本生成相同的向量
  • 语义编码：相似的文本会产生相似的向量
  • 高维：通常是几百到几千维

2. 相似度计算

在向量空间中，有多种相似度指标：

import numpy as np

def cosine_similarity(v1, v2):
    """余弦相似度：最常用"""
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

def euclidean_distance(v1, v2):
    """欧几里得距离"""
    return np.linalg.norm(v1 - v2)

def manhattan_distance(v1, v2):
    """曼哈顿距离"""
    return np.sum(np.abs(v1 - v2))

# 测试
v1 = np.array([1, 0, 0])
v2 = np.array([0.5, 0.5, 0])
v3 = np.array([-1, 0, 0])

print(f"v1 vs v2 余弦相似度: {cosine_similarity(v1, v2):.3f}")  # 0.707
print(f"v1 vs v3 余弦相似度: {cosine_similarity(v1, v3):.3f}")  # -1.0

最常用的是余弦相似度，因为它度量的是向量方向而不是大小。

3. 索引结构

直接计算相似度的时间复杂度是O(n*d)，对大规模数据不可行。向量数据库使用特殊的索引加速查询：

所有向量
    ↓
[索引结构]
    ├─ HNSW (Hierarchical Navigable Small World)
    ├─ IVF (Inverted File)
    ├─ LSH (Locality Sensitive Hashing)
    └─ 其他
    ↓
最近邻向量 (毫秒级)

HNSW 是目前最流行的：

  • 基于图的结构
  • 多层导航
  • 查询快速且准确率高
  • Milvus、Weaviate、Pinecone都支持

实战：构建一个简单的向量搜索系统

使用 Pinecone

Pinecone是托管的向量数据库，最容易上手：

from pinecone import Pinecone
from openai import OpenAI

# 初始化
pc = Pinecone(api_key="your-api-key")
index = pc.Index("blog-index")
client = OpenAI()

def embed_text(text):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# 1. 存储文章
articles = [
    {"id": "1", "text": "Python是最流行的编程语言"},
    {"id": "2", "text": "JavaScript统治前端世界"},
    {"id": "3", "text": "Rust以内存安全著称"},
    {"id": "4", "text": "编程语言的选择取决于用途"},
]

vectors_to_upsert = []
for article in articles:
    vector = embed_text(article["text"])
    vectors_to_upsert.append((
        article["id"],
        vector,
        {"text": article["text"], "type": "article"}
    ))

index.upsert(vectors=vectors_to_upsert)

# 2. 搜索相似的文章
query = "什么编程语言最好"
query_vector = embed_text(query)

results = index.query(vector=query_vector, top_k=3, include_metadata=True)

for match in results["matches"]:
    print(f"相似度: {match['score']:.3f}")
    print(f"内容: {match['metadata']['text']}")
    print()

输出类似：

相似度: 0.856
内容: Python是最流行的编程语言

相似度: 0.743
内容: 编程语言的选择取决于用途

相似度: 0.621
内容: JavaScript统治前端世界

使用开源方案：Milvus

对于需要更多控制的场景，可以自己部署Milvus：

from pymilvus import (
    connections,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
)

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

# 定义Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),
]
schema = CollectionSchema(fields)
collection = Collection("articles", schema)

# 插入数据
data = [
    [1, 2, 3],  # id
    ["文本1", "文本2", "文本3"],  # content
    [vector1, vector2, vector3],  # embedding
]
collection.insert(data)

# 创建索引
index_params = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
    "params": {"efConstruction": 200, "M": 12},
}
collection.create_index("embedding", index_params)

# 搜索
query_results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},
    limit=3,
    output_fields=["content"],
)

for hits in query_results:
    for hit in hits:
        print(f"相似度: {hit.distance}, 内容: {hit.entity.get('content')}")

常见的向量数据库对比

数据库托管开源特点最适合
Pinecone✓✗全托管，开箱即用快速原型和小规模应用
Weaviate✓✓GraphQL API，多模态需要灵活查询的应用
Milvus✗✓高性能，支持大规模生产级应用
Qdrant✓✓Payload支持，性能好需要过滤的应用
Chroma✗✓轻量级，嵌入式开发和实验
Elasticsearch✗✓向量+全文搜索混合需要混合搜索

优化向量数据库性能

1. 选择合适的Embedding模型

# 维度越低，存储和查询越快，但信息丢失
# 维度越高，更准确，但成本增加

models = {
    "text-embedding-3-small": {"dim": 1536, "cost": "最低"},
    "text-embedding-3-large": {"dim": 3072, "cost": "较高"},
    "bge-small": {"dim": 384, "cost": "最低"},  # 开源
    "bge-large": {"dim": 1024, "cost": "低"},     # 开源
}

# 对大多数应用，1536维足够

2. 批量操作

# ❌ 低效：逐个插入
for i in range(1000):
    index.upsert(vectors=[(str(i), vector)])

# ✅ 高效：批量插入
batch_size = 100
for i in range(0, 1000, batch_size):
    batch = vectors[i:i+batch_size]
    index.upsert(vectors=batch)

3. 索引参数调优

# HNSW索引参数
index_params = {
    "M": 8,              # 图的度数，越大越准确但更慢
    "efConstruction": 200,  # 构建时的搜索深度
    "ef": 100,           # 查询时的搜索深度
}

# 平衡：M=8-16，efConstruction=100-300

4. 混合搜索

# 结合向量搜索和关键词过滤
results = index.query(
    vector=query_vector,
    top_k=10,
    filter={"category": "AI", "year": {"$gte": 2024}},
)

常见问题

Q: 向量数据库能完全替代关系数据库吗？

A: 不能。向量数据库专门处理语义搜索，但不适合需要ACID事务、复杂JOIN等操作的场景。应该和关系数据库配合使用。

Q: 每次都要重新Embedding吗？

A: 是的。如果Embedding模型更新或文本变化，需要重新计算。因此要将Embedding的模型版本记录下来。

Q: 如何处理Embedding的漂移？

A: Embedding模型会随时间更新（比如OpenAI发布了新版本），这会导致之前的Embedding不再适用。需要定期重新Embedding整个数据库。

总结

向量数据库是AI应用中最关键的基础设施：

  • RAG系统：用向量数据库存储文档，快速检索相关内容
  • 推荐系统：根据用户和物品的向量相似度推荐
  • AI搜索：超越关键词匹配的语义搜索
  • 聚类和分类：在向量空间中处理机器学习任务

掌握向量数据库已经成为AI开发者的必备技能。