章节导语

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 相似度计算

向量数据库通过计算向量之间的距离判断相似度。常见方法：

2.3 近似最近邻（ANN）

向量数据库不是精确查找，而是近似查找。因为精确查找在大规模数据上太慢。

索引算法：

• HNSW：图索引，速度快，内存占用高
• IVF：聚类索引，平衡精度和速度
• PQ：量化索引，内存占用低，精度稍差

三、四大向量数据库对比

3.1 概览

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 特点

• 完全托管：不用管服务器
• 高可用：自动副本、自动恢复
• 按需扩展：从小到大自动扩容
• 云原生：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 性能对比

9.2 成本对比

9.3 选型决策

• 学习/原型 → 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的性能差异。

挑战题：设计一个多数据库路由系统，根据查询自动选择最合适的数据库。