在这里插入图片描述

在 RAG（检索增强生成）场景中，非结构化数据（文本、图像等）的高效检索是核心需求。传统关系型数据库难以胜任，而向量数据库通过将数据转化为向量、基于相似度快速匹配，成为 RAG 的关键支撑。本文聚焦主流向量数据库 Chroma 与 Milvus，解析其在 RAG 中的实战应用与底层原理。

一、Chroma 实战与底层原理

1. Chroma 简介

定位：轻量级、易于使用的开源向量数据库。
特点：安装便捷、API简洁、支持多种编程语言，适合快速开发和原型验证。

2. Chroma 安装

对于Python开发者，通过pip命令即可轻松安装：

pip install chromadb

3. Chroma 基本操作

3.1 初始化客户端

内存模式（开发阶段快速测试）：

import chromadb
from chromadb.config import Settingsclient = chromadb.Client(Settings(persist_directory=None  # 数据不会持久化
))

client = chromadb.Client(Settings(persist_directory="./chroma_data"  # 数据持久化到该目录
))

3.2 创建集合（类似关系型数据库的表）：

collection = client.create_collection(name="my_collection")

3.3 添加数据

需提供文档（documents）、文档ID（ids），向量（embeddings）可选（不提供则用默认嵌入模型生成）。

documents = ["Chroma is a vector database","It is easy to use","Vector databases are useful for AI applications"
]
ids = ["doc1", "doc2", "doc3"]collection.add(documents=documents, ids=ids)

3.4 查询数据

不提供查询向量时，默认嵌入模型为查询文本生成向量。
结果包含相似文档ID、内容及相似度分数。

query = "What is Chroma?"
results = collection.query(query_texts=[query],n_results=2  # 返回最相似的2个结果
)print(results)

4. Chroma 底层原理

4.1 加速技术

批量处理向量数据，减少磁盘I/O操作，提高读写效率。
合理分配内存资源，将频繁访问的向量数据缓存到内存，加快查询响应。

4.2 支持的索引

Flat索引：存储原始向量，查询时计算与所有向量的距离，精度高但数据量大时速度慢，适合小规模数据。
HNSW（Hierarchical Navigable Small World）索引：基于图的近似最近邻搜索索引，构建多层导航图，在保证一定精度的前提下显著提高查询速度，适用于中大规模数据。

二、Milvus 实战与底层原理

1. Milvus 简介

定位：高性能、高可用的开源向量数据库，专为海量向量数据的存储、检索和分析设计。
特点：支持多种索引类型，能处理大规模向量数据，扩展性和容错性良好，适用于生产环境。

2. Milvus 安装

2.1 推荐使用Docker Compose，先安装Docker和Docker Compose。

2.2 下载配置文件：

wget https://github.com/milvus-io/milvus/releases/download/v2.3.4/milvus-standalone-docker-compose.yml -O docker-compose.yml

2.3 启动Milvus（服务在本地19530端口运行）：

docker-compose up -d

3. Milvus 基本操作

3.1 安装Python SDK：

pip install pymilvus

3.2 连接到Milvus：

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataTypeconnections.connect("default", host="localhost", port="19530")

3.3 创建集合

# 定义字段
id_field = FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True)
content_field = FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=512)
vector_field = FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=768)  # 向量维度为768# 定义集合schema
schema = CollectionSchema(fields=[id_field, content_field, vector_field], description="My collection")# 创建集合
collection = Collection(name="my_collection", schema=schema)

3.4 创建索引（以IVF_FLAT为例）：

index_params = {"index_type": "IVF_FLAT","metric_type": "L2",  # 欧氏距离"params": {"nlist": 128}  # 聚类数量
}collection.create_index(field_name="vector", index_params=index_params)

3.5 插入数据

import random# 生成示例数据
data = [{"content": "Milvus is a vector database", "vector": [random.random() for _ in range(768)]},{"content": "It is high performance", "vector": [random.random() for _ in range(768)]},{"content": "It is suitable for large - scale data", "vector": [random.random() for _ in range(768)]}
]# 转换数据格式
contents = [d["content"] for d in data]
vectors = [d["vector"] for d in data]# 插入数据
mr = collection.insert([contents, vectors])
print("Inserted ids:", mr.primary_keys)

3.6 加载集合并查询

collection.load()# 生成查询向量
query_vector = [random.random() for _ in range(768)]# 查询参数
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}# 执行查询
results = collection.search(data=[query_vector],anns_field="vector",param=search_params,limit=2,  # 返回最相似的2个结果output_fields=["content"]  # 返回content字段
)for result in results[0]:print(f"ID: {result.id}, Distance: {result.distance}, Content: {result.entity.get('content')}")