向量数据库选型与实战指南：5分钟上手 Milvus，打造智能语义搜索

向量数据库选型与实战指南：5分钟上手 Milvus，打造智能语义搜索

引言

随着大语言模型（LLM）和生成式 AI 的爆发，非结构化数据的处理需求急剧攀升。文本、图像、音频等数据无法再用传统的关键词匹配进行高效检索，向量数据库应运而生，成为支撑语义搜索、推荐系统、知识库问答等场景的核心基础设施。

目前主流的向量数据库有 Milvus、Pinecone、Weaviate、Chroma、Qdrant 等。面对琳琅满目的选项，开发者该如何选择？本文将从核心概念出发，对比不同产品的适用场景，并以Milvus为例，提供一套完整可运行的代码示例，帮助你快速从零搭建一个语义搜索引擎。

一、核心概念：从向量到近似最近邻

1. 什么是向量嵌入（Embedding）

向量嵌入是通过深度学习模型将非结构化数据（如一段文字、一张图片）映射到高维空间中的固定长度浮点数组。在这个空间中，语义相近的对象距离更近。例如：

• “苹果”和“水果”的嵌入向量余弦相似度很高；
• “苹果”和“手机”可能距离较远（除非上下文是苹果公司）。

2. 相似度度量

常用的相似度（距离）函数包括：

• 欧氏距离（L2）：直线距离，值越小越相似；
• 余弦相似度：夹角余弦，只关注方向，对向量长度不敏感，常用于文本嵌入；
• 内积（IP）：用于最大内积搜索，适合归一化后的向量。

3. 近似最近邻搜索（ANN）

如果每次查询都对全量向量进行暴力计算，性能会随数据量增长而急剧下降。ANN 算法通过构建索引，在牺牲少量精度的前提下大幅提升检索速度。常见索引类型：

• IVF_FLAT：基于聚类的倒排索引，用 nlist 个聚类中心加速；
• HNSW：分层可导航小世界图，内存占用高但查询极快；
• PQ（乘积量化）：压缩向量以节省内存，适合超大规模。
• DiskANN：支持磁盘索引，用于十亿级向量。

二、主流向量数据库选型速览

本文选择 Milvus，因为它功能全面、社区活跃，并且支持本地单机模式，非常适合学习与开发。

三、实战：用 Milvus 构建语义搜索（完整代码）

3.1 环境准备

确保本机已安装 Docker，然后拉取并启动 Milvus Standalone：

# 下载 docker-compose 文件 wget https://github.com/milvus-io/milvus/releases/download/v2.3.4/milvus-standalone-docker-compose.yml -O docker-compose.yml # 启动 Milvus docker compose up -d

等待容器启动后，Milvus 会在localhost:19530提供 gRPC 服务。

安装 Python 依赖：

pip install pymilvus sentence-transformers

• pymilvus：Milvus 官方 Python 客户端；
• sentence-transformers：生成文本嵌入的轻量模型。

3.2 生成文本嵌入

我们使用all-MiniLM-L6-v2模型，它能把任意文本转换为 384 维向量。

from sentence_transformers import SentenceTransformer # 加载模型（首次运行会下载，约 80 MB） model = SentenceTransformer('all-MiniLM-L6-v2') # 测试生成嵌入 texts = ["深度学习改变了人工智能", "今天天气真好", "苹果发布了新款手机"] embeddings = model.encode(texts) print(f"向量维度：{embeddings.shape[1]}")

3.3 连接 Milvus 并创建集合

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType # 1. 连接 Milvus connections.connect(host="localhost", port="19530") # 2. 定义集合 Schema fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384), # 384维向量 FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512) # 原始文本 ] schema = CollectionSchema(fields, description="语义搜索演示") # 3. 创建集合（如果已存在同名集合，先删除） collection_name = "demo_semantic_search" try: Collection(collection_name).drop() except: pass collection = Collection(name=collection_name, schema=schema) print("集合创建成功！")

3.4 插入数据并构建索引

准备一批示例句子，生成嵌入后批量插入。

# 示例语料 documents = [ "Milvus 是一款开源的向量数据库", "向量数据库主要用于相似性检索", "今天食堂的午餐非常美味", "Python 是最受欢迎的编程语言之一", "语义搜索能够理解用户意图", "Milvus 支持多种索引类型", "Docker 简化了应用部署流程", "推荐系统常用协同过滤和向量检索" ] # 生成嵌入 doc_embeddings = model.encode(documents) # 准备插入数据（列表形式） entities = [ doc_embeddings.tolist(), # 向量列 documents # text 列 ] # 插入 insert_result = collection.insert(entities) print(f"插入 {collection.num_entities} 条数据") # 为 embedding 列创建索引 index_params = { "index_type": "IVF_FLAT", "metric_type": "COSINE", # 余弦相似度 "params": {"nlist": 4} # 聚类数，数据量小时可适当调小 } collection.create_index(field_name="embedding", index_params=index_params) print("索引创建完成")

3.5 执行语义搜索

编写一个搜索函数，输入查询文本，返回最相似的 top_k 条结果。

def semantic_search(query, top_k=3): # 将集合加载到内存（仅需执行一次） collection.load() # 生成查询向量 query_embedding = model.encode([query]).tolist() # 搜索参数 search_params = {"metric_type": "COSINE", "params": {"nprobe": 2}} # 执行搜索 results = collection.search( data=query_embedding, anns_field="embedding", param=search_params, limit=top_k, output_fields=["text"] # 返回原始文本 ) # 输出结果 for hits in results: for hit in hits: print(f"相似度: {hit.distance:.4f}, 内容: {hit.entity.get('text')}") # 测试搜索 print("查询: 向量数据库") semantic_search("向量数据库") print("\n查询: 编程语言") semantic_search("编程语言")

预期输出（近似）：

查询: 向量数据库 相似度: 0.9999, 内容: Milvus 是一款开源的向量数据库 相似度: 0.9123, 内容: 向量数据库主要用于相似性检索 相似度: 0.8567, 内容: Milvus 支持多种索引类型 查询: 编程语言 相似度: 0.9876, 内容: Python 是最受欢迎的编程语言之一 相似度: 0.7543, 内容: Docker 简化了应用部署流程 相似度: 0.6234, 内容: 推荐系统常用协同过滤和向量检索

可以看到，搜索“向量数据库”准确命中了相关文档，而“编程语言”也与 Python 强关联，体现了语义理解能力。

3.6 完整可运行脚本

将上述步骤整合为一个 Python 脚本semantic_search_demo.py，可以直接运行（确保 Milvus 已启动）：

```python
from sentence_transformers import SentenceTransformer
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

连接 Milvus

connections.connect(host="localhost", port="19530")

加载模型

model = SentenceTransformer('all-MiniLM-L6-v2')

定义集合

collection_name = "demo_semantic_search"
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512)
]
schema = CollectionSchema(fields, description="语义搜索演示")

重建集合

try: Collection(collection_name).drop()
except: pass
collection = Collection(name=collection_name, schema=schema)

插入数据

documents = [
"Milvus 是一款开源的向量数据库",
"向量数据库主要用于相似性检索",
"今天食堂的午餐非常美味",
"Python 是最受欢迎的编程语言之一",
"语义搜索能够理解用户意图",
"Milvus 支持多种索引类型",
"Docker 简化了应用部署流程",
"推荐系统常用协同过滤和向量检索"
]
doc_embeddings = model.encode(documents)
collection.insert([doc_embeddings.tolist(), documents])

创建索引

index_params = {"index_type": "IVF_FLAT", "metric_type": "COSINE", "params": {"nlist": 4}}
collection.create_index(field_name="embedding", index_params=index_params)
collection.load()

搜索函数

def search(query, k=3):
vec = model.encode([query]).tolist()
params = {"metric_type": "COSINE", "params": {"nprobe": 2}}
res = collection.search(vec, "embedding", params, limit=k, output_fields=["text"])
for hits in res:
for hit in hits:
print(f"[{hit.distance:.4f}] {hit.entity.get('text')}")

ifname== "main":
print("