【高级应用】Day9:向量数据库深入对比–Pinecone/Milvus/Weaviate/Chroma选型指南
章节导语
RAG的核心是向量检索,而向量检索的核心是向量数据库。但Pinecone、Milvus、Weaviate、Chroma……该选哪个?
每个向量数据库都有自己的特点:有的专注云原生,有的性能爆炸,有的免费开源。选错数据库,后续迁移成本巨大。
本文系统对比4大主流向量数据库,从原理到实战,帮你做出正确选择。
一、前置说明
1.1 学习路径
| 阶段 | 内容 |
|---|---|
| 基础 | 向量基础概念 |
| 进阶 | 向量数据库对比 |
1.2 读者需要的基础
- 向量概念:知道什么是Embedding
- Python基础:会安装库、调用API
- 数据库基础:了解CRUD操作
1.3 学习目标
学完本文,你将能够:
- 理解向量数据库的核心原理
- 对比4大主流向量数据库的优劣
- 根据场景选择最合适的数据库
- 快速上手至少2种向量数据库
二、向量数据库核心原理
2.1 什么是向量
向量是数据的数学表示。每个文本、图像都可以转换为一系列数字(向量)。
举例:
# 文本 "苹果" 的向量表示(简化示例)
# 实际向量维度通常是768、1536甚至更高
"苹果" → [0.12, -0.34, 0.78, 0.21, ...] # 1536维
# 相似的词,向量也相似
"苹果" → [0.12, -0.34, 0.78, 0.21, ...]
"水果" → [0.11, -0.32, 0.75, 0.20, ...] # 相似!
"手机" → [-0.45, 0.67, 0.12, 0.89, ...] # 不相似!
2.2 相似度计算
向量数据库通过计算向量之间的距离判断相似度。常见方法:
图1:向量相似度示意| 方法 | 公式 | 特点 |
|---|---|---|
| 余弦相似度 | cos(θ) | 关注方向,忽略大小 |
| 欧氏距离 | ||A-B|| | 关注绝对距离 |
| 点积 | A·B | 计算快,效果好 |
2.3 近似最近邻(ANN)
向量数据库不是精确查找,而是近似查找。因为精确查找在大规模数据上太慢。
索引算法:
- HNSW:图索引,速度快,内存占用高
- IVF:聚类索引,平衡精度和速度
- PQ:量化索引,内存占用低,精度稍差
三、四大向量数据库对比
3.1 概览
图2:四大向量数据库定位| 数据库 | 类型 | 优点 | 缺点 |
|---|---|---|---|
| Pinecone | 云原生 | 托管、免运维 | 付费、有 vendor lock-in |
| Milvus | 开源 | 功能强大、可私有部署 | 部署复杂 |
| Weaviate | 开源+云 | 多模态、内置向量化 | 文档待完善 |
| Chroma | 轻量开源 | 简单、适合入门 | 功能有限、不适合生产 |
3.2 选型决策树
需要快速启动、少量数据?
├── 是 → Chroma(简单上手)
└── 否
├── 需要云托管、不想运维?
│ ├── 是 → Pinecone
│ └── 否
│ ├── 需要多模态(文本+图片)支持?
│ │ ├── 是 → Weaviate
│ │ └── 否
│ │ ├── 数据量大、需要高可用?
│ │ │ ├── 是 → Milvus(集群模式)
│ │ │ └── 否 → Milvus(单机模式)
│ │ └── 其他 → 根据团队能力选择
四、环境配置
4.1 安装依赖
# 安装向量数据库客户端
pip install chromadb # Chroma
pip install pymilvus # Milvus
pip install weaviate-client # Weaviate
pip install pinecone-client # Pinecone
# 安装Embedding支持
pip install sentence-transformers
五、Chroma:入门首选
5.1 特点
- 简单:几行代码就能跑起来
- 轻量:适合学习和原型开发
- 本地:数据存在本地文件
不适合:生产环境、大规模数据、分布式场景
5.2 快速上手
import chromadb
from chromadb.config import Settings
class ChromaDemo:
"""Chroma演示:最简单的向量数据库
特点:
- 本地存储,一行配置
- API简洁,上手快
- 适合学习和原型
"""
def __init__(self, persist_dir="./chroma_db"):
# 创建客户端,指定持久化目录
self.client = chromadb.Client(Settings(
persist_directory=persist_dir,
anonymized_telemetry=False
))
print("✅ Chroma 客户端创建成功")
def create_collection(self, name="demo"):
"""创建集合(类似表)"""
self.collection = self.client.create_collection(name=name)
print(f"✅ 集合 '{name}' 创建成功")
def add_documents(self):
"""添加文档"""
documents = [
"人工智能是当今最热门的技术领域",
"机器学习是AI的核心技术之一",
"深度学习使用神经网络进行学习",
"Python是AI领域最常用的编程语言",
"GPT是一种大型语言模型"
]
# 生成ID
ids = [f"doc_{i}" for i in range(len(documents))]
# 添加到集合
self.collection.add(
documents=documents,
ids=ids
)
print(f"✅ 添加了 {len(documents)} 个文档")
def search(self, query, n=3):
"""检索"""
results = self.collection.query(
query_texts=[query],
n_results=n
)
print(f"\n🔍 查询:{query}")
print(f"找到 {len(results['documents'][0])} 个结果:")
for i, (doc, distance) in enumerate(zip(
results['documents'][0],
results['distances'][0]
), 1):
similarity = 1 - distance # 距离转相似度
print(f" {i}. [{similarity:.2f}] {doc}")
return results
# 使用
if __name__ == "__main__":
demo = ChromaDemo()
demo.create_collection()
demo.add_documents()
demo.search("AI和机器学习有什么关系?")
demo.search("Python在哪些领域用?")
5.3 运行结果
$ python chroma_demo.py
✅ Chroma 客户端创建成功
✅ 集合 'demo' 创建成功
✅ 添加了 5 个文档
🔍 查询:AI和机器学习有什么关系?
找到 3 个结果:
1. [0.95] 人工智能是当今最热门的技术领域
2. [0.89] 机器学习是AI的核心技术之一
3. [0.78] 深度学习使用神经网络进行学习
🔍 查询:Python在哪些领域用?
找到 3 个结果:
1. [0.92] Python是AI领域最常用的编程语言
2. [0.75] 机器学习是AI的核心技术之一
3. [0.71] 人工智能是当今最热门的技术领域
5.4 小结
Chroma适合快速原型和学习。正式项目建议用其他数据库。
六、Milvus:生产级开源
6.1 特点
- 高性能:支持十亿级向量检索
- 分布式:支持集群部署
- 多索引:HNSW、IVF、PQ等多种索引
- 开源:可私有部署
6.2 快速上手
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
class MilvusDemo:
"""Milvus演示:生产级向量数据库
注意:需要先安装并启动Milvus服务
docker run -d --name milvus-standalone \\
-p 19530:19530 \\
-p 19121:19121 \\
milvusdb/milvus:v2.3.3
"""
def __init__(self, host="localhost", port="19530"):
# 连接Milvus服务
connections.connect(host=host, port=port)
print(f"✅ Milvus 连接成功 ({host}:{port})")
self.collection_name = None
def create_collection(self, name="demo", dim=128):
"""创建集合
Milvus需要定义schema(结构)
"""
self.collection_name = name
# 定义字段
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="document", dtype=DataType.VARCHAR, max_length=500),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim)
]
schema = CollectionSchema(fields=fields, description="演示集合")
# 创建集合
self.collection = Collection(name=name, schema=schema)
print(f"✅ 集合 '{name}' 创建成功 (维度={dim})")
def create_index(self):
"""创建索引
HNSW索引:速度快,内存占用高
"""
index_params = {
"index_type": "HNSW",
"metric_type": "L2", # 欧氏距离
"params": {"M": 16, "efConstruction": 200}
}
self.collection.create_index(
field_name="embedding",
index_params=index_params
)
print("✅ 索引创建成功 (HNSW)")
def insert_data(self):
"""插入数据"""
# 模拟向量数据(实际用Embedding模型生成)
import random
documents = [
"人工智能是当今最热门的技术",
"机器学习是AI的核心技术",
"深度学习使用神经网络",
"Python是AI领域最常用的语言"
]
# 生成随机向量(实际应该用模型生成)
embeddings = [[random.random() for _ in range(128)] for _ in documents]
data = [documents, embeddings]
result = self.collection.insert(data)
self.collection.flush()
print(f"✅ 插入了 {len(documents)} 条数据")
return result
def search(self, query_vector, top_k=3):
"""检索"""
search_params = {"metric_type": "L2", "params": {"ef": 200}}
results = self.collection.search(
data=[query_vector],
anns_field="embedding",
param=search_params,
limit=top_k,
output_fields=["document"]
)
print(f"\n🔍 检索结果:")
for hits in results:
for hit in hits:
print(f" [{hit.distance:.4f}] {hit.entity.get('document')}")
return results
def drop_collection(self):
"""删除集合"""
if self.collection_name and utility.has_collection(self.collection_name):
utility.drop_collection(self.collection_name)
print(f"✅ 集合 '{self.collection_name}' 已删除")
# 使用(需要Milvus服务运行)
if __name__ == "__main__":
try:
demo = MilvusDemo()
demo.create_collection(dim=128)
demo.create_index()
demo.insert_data()
# 随机查询向量
import random
query = [random.random() for _ in range(128)]
demo.search(query)
demo.drop_collection()
except Exception as e:
print(f"⚠️ 需要Milvus服务运行:docker run -d --name milvus-standalone -p 19530:19530 milvusdb/milvus:v2.3.3")
print(f"错误:{e}")
6.3 适用场景
- 大规模数据:亿级向量
- 高并发:需要支持QPS
- 私有部署:数据不能上云
6.4 小结
Milvus是生产环境的首选开源方案,功能强大但部署相对复杂。
七、Pinecone:云原生托管
7.1 特点
图3:Pinecone云原生架构- 完全托管:不用管服务器
- 高可用:自动副本、自动恢复
- 按需扩展:从小到大自动扩容
- 云原生:AWS/GCP/Azure 都有
缺点:付费、数据需要上云
7.2 快速上手
from pinecone import Pinecone, ServerlessSpec
class PineconeDemo:
"""Pinecone演示:云原生向量数据库
前提:注册 pinecone.io 获取API Key
"""
def __init__(self, api_key="YOUR_API_KEY", environment="us-east-1"):
self.pc = Pinecone(api_key=api_key)
print("✅ Pinecone 客户端创建成功")
self.index = None
def create_index(self, name="demo", dim=1536):
"""创建索引"""
# 删除已存在的索引
if self.pc.list_indexes().names():
self.pc.delete_index(name)
# 创建新索引(Serverless规格,按使用付费)
spec = ServerlessSpec(
cloud="aws",
region=environment
)
self.pc.create_index(
name=name,
dimension=dim,
metric="cosine", # 余弦相似度
spec=spec
)
# 等待索引创建完成
while not self.pc.describe_index(name).status['ready']:
import time
time.sleep(1)
self.index = self.pc.Index(name)
print(f"✅ 索引 '{name}' 创建成功 (维度={dim})")
def upsert_vectors(self):
"""插入向量"""
vectors = [
{"id": "vec1", "values": [0.1] * 1536, "metadata": {"text": "人工智能是热门技术"}},
{"id": "vec2", "values": [0.2] * 1536, "metadata": {"text": "机器学习是AI的核心"}},
{"id": "vec3", "values": [0.3] * 1536, "metadata": {"text": "深度学习使用神经网络"}},
{"id": "vec4", "values": [0.4] * 1536, "metadata": {"text": "Python是常用编程语言"}},
]
self.index.upsert(vectors)
print(f"✅ 插入了 {len(vectors)} 条向量")
def query(self, vector, top_k=3):
"""查询"""
results = self.index.query(
vector=vector,
top_k=top_k,
include_metadata=True
)
print(f"\n🔍 查询结果:")
for match in results['matches']:
score = match['score']
text = match['metadata']['text']
print(f" [{score:.4f}] {text}")
return results
def delete_index(self, name="demo"):
"""删除索引"""
self.pc.delete_index(name)
print(f"✅ 索引 '{name}' 已删除")
# 使用
if __name__ == "__main__":
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("PINECONE_API_KEY")
if not api_key or api_key == "YOUR_API_KEY":
print("⚠️ 请设置 PINECONE_API_KEY 环境变量")
print(" 1. 注册 https://www.pinecone.io")
print(" 2. 获取 API Key")
print(" 3. 设置环境变量 export PINECONE_API_KEY=your_key")
else:
demo = PineconeDemo(api_key=api_key)
demo.create_index()
demo.upsert_vectors()
# 查询向量
query_vec = [0.15] * 1536
demo.query(query_vec)
demo.delete_index()
7.3 适用场景
- 快速启动:不想管理服务器
- 弹性需求:流量波动大
- 全球部署:多区域低延迟
7.4 小结
Pinecone适合不想运维、愿意付费的团队。
八、Weaviate:多模态先锋
8.1 特点
- 多模态:原生支持文本、图像、音视频
- 内置向量化:不用自己生成Embedding
- 图结构:支持关系查询
- 开源+云:可自托管也可用云服务
8.2 快速上手
import weaviate
class WeaviateDemo:
"""Weaviate演示:多模态向量数据库
支持:
- 文本(内置向量化)
- 图片(CLIP模型)
- 音视频
"""
def __init__(self):
# 连接Weaviate(本地或云端)
self.client = weaviate.Client("http://localhost:8080")
print("✅ Weaviate 客户端创建成功")
def create_schema(self):
"""创建Schema(类似表结构)"""
schema = {
"classes": [{
"class": "Article",
"description": "文章",
"vectorizer": "text2vec-transformers", # 内置向量化
"moduleConfig": {
"text2vec-transformers": {
"vectorizeClassName": False
}
},
"properties": [
{"name": "title", "dataType": ["text"]},
{"name": "content", "dataType": ["text"]},
{"name": "author", "dataType": ["text"]}
]
}]
}
# 删除已存在的类
if self.client.schema.exists("Article"):
self.client.schema.delete_class("Article")
self.client.schema.create(schema)
print("✅ Schema 创建成功")
def add_documents(self):
"""添加文档"""
documents = [
{"title": "AI简介", "content": "人工智能是当今最热门的技术领域"},
{"title": "机器学习", "content": "机器学习是AI的核心技术之一"},
{"title": "深度学习", "content": "深度学习使用神经网络进行学习"},
{"title": "Python", "content": "Python是AI领域最常用的编程语言"},
]
# 批量添加
with self.client.batch(batch_size=100) as batch:
for i, doc in enumerate(documents):
batch.add_data_object(
data_object=doc,
class_name="Article"
)
print(f"✅ 添加了 {len(documents)} 个文档")
def search(self, query, limit=3):
"""检索"""
results = self.client.query.get(
"Article",
["title", "content"]
).with_near_text({
"concepts": [query]
}).with_limit(limit).do()
print(f"\n🔍 查询:{query}")
articles = results['data']['Get']['Article']
print(f"找到 {len(articles)} 个结果:")
for art in articles:
print(f" - {art['title']}: {art['content'][:30]}...")
return results
def hybrid_search(self, query, limit=3):
"""混合检索:向量+关键词"""
results = self.client.query.get(
"Article",
["title", "content"]
).with_hybrid(query, limit=limit).do()
print(f"\n🔍 混合检索:{query}")
articles = results['data']['Get']['Article']
for art in articles:
print(f" - {art['title']}")
return results
def delete_all(self):
"""删除所有数据"""
self.client.schema.delete_all()
print("✅ 所有数据已删除")
# 使用(需要Weaviate服务运行)
if __name__ == "__main__":
try:
demo = WeaviateDemo()
demo.create_schema()
demo.add_documents()
demo.search("AI技术")
demo.search("编程语言")
demo.hybrid_search("神经网络 Python")
demo.delete_all()
except Exception as e:
print(f"⚠️ 需要Weaviate服务运行")
print(f" docker run -d -p 8080:8080 cr.weaviate.io/semitechnique/weaviate")
print(f"错误:{e}")
8.3 适用场景
- 多模态应用:同时处理文本+图片
- 图关系查询:需要关系数据支持
- 快速原型:不想自己处理向量化
8.4 小结
Weaviate适合多模态场景,内置向量化是亮点。
九、对比总结与选型
9.1 性能对比
| 数据库 | 100万向量 | 1000万向量 | 1亿向量 |
|---|---|---|---|
| Chroma | ✅ | ❌ | ❌ |
| Milvus | ✅ | ✅ | ✅ |
| Pinecone | ✅ | ✅ | ✅ |
| Weaviate | ✅ | ✅ | ⚠️ |
9.2 成本对比
| 数据库 | 起步价 | 规模成本 |
|---|---|---|
| Chroma | 免费 | 免费(受限于单机) |
| Milvus | 免费(自托管) | 云服务器成本 |
| Pinecone | $70/月 | 按向量数+查询计费 |
| Weaviate | 免费(自托管) | 云服务按需付费 |
9.3 选型决策
图4:向量数据库选型流程- 学习/原型 → Chroma
- 生产环境、量小 → Chroma(先),Pinecone(后期)
- 大规模、私有部署 → Milvus
- 不想运维、云优先 → Pinecone
- 多模态需求 → Weaviate
十、实战:RAG系统中的数据库选型
10.1 不同规模的RAG方案
# ============= 小规模RAG(<10万向量)=============
# 推荐:Chroma 或 Pinecone Starter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
vectorstore = Chroma(
embedding_function=OpenAIEmbeddings(),
persist_directory="./chroma_db"
)
# ============= 中等规模RAG(10万-1000万向量)=============
# 推荐:Pinecone 或 Milvus
from pinecone import Pinecone
pc = Pinecone(api_key="your-key")
index = pc.Index("your-index")
# ============= 大规模RAG(>1000万向量)=============
# 推荐:Milvus 集群
from pymilvus import connections, Collection
connections.connect(host="milvus-host", port="19530")
collection = Collection("your-collection")
collection.load()
# ============= 多模态RAG(文本+图片)=============
# 推荐:Weaviate
from weaviate import Client
client = Client("http://localhost:8080")
# 添加图片
client.batch.add_data_object(
data_object={"text": "一只猫的图片"},
class_name="Image",
vector={"image": image_vector} # CLIP生成
)
10.2 迁移方案
如果需要换数据库,可以用LangChain的迁移工具:
def migrate_vectorstore(source_db, target_db, batch_size=1000):
"""向量数据库迁移
从源数据库导出,重新插入目标数据库
"""
# 1. 从源数据库读取
source_docs = source_db.get_all()
# 2. 批量写入目标数据库
for i in range(0, len(source_docs), batch_size):
batch = source_docs[i:i+batch_size]
target_db.add_documents(batch)
print(f"迁移进度:{i+len(batch)}/{len(source_docs)}")
print("✅ 迁移完成")
十一、总结与练习
11.1 要点回顾
- Chroma:简单入门,适合学习,原型开发
- Milvus:功能强大,开源免费,生产级
- Pinecone:云托管,免运维,按需付费
- Weaviate:多模态支持,内置向量化
11.2 避坑指南
- 别用Chroma做生产:单机限制,数据量大了会崩溃
- Pinecone注意成本:按查询计费,量大时费用可观
- Milvus集群复杂:需要K8s经验
- 先评估数据量:选型前估算增长
11.3 延伸阅读
- 向量数据库对比报告:Benchmarks
- Milvus官方文档:milvus.io/docs
- Weaviate多模态:开发者文档
11.4 课后练习
基础题:用Chroma实现一个本地知识库问答。
进阶题:对比Chroma和Milvus的性能差异。
挑战题:设计一个多数据库路由系统,根据查询自动选择最合适的数据库。
扫码关注公众号
扫码添加QQ
【Prompt炼金术】Day8|模板库:拿来即用的实战模板集合
【Prompt炼金术】Day8|模板库:拿来即用的实战模板集合
【Prompt炼金术】Day7|思维链:让AI从”胡言乱语”到”有理有据”
【Prompt炼金术】Day6|高级参数:让AI输出稳定可控的秘诀