名称: qdrant-向量搜索描述: 用于RAG和语义搜索的高性能向量相似性搜索引擎。适用于构建需要快速最近邻搜索、混合搜索与过滤或具有Rust性能的可扩展向量存储的生产RAG系统。版本: 1.0.0 作者: Orchestra Research 许可证: MIT 标签: [RAG, 向量搜索, Qdrant, 语义搜索, 嵌入, 相似性搜索, HNSW, 生产, 分布式] 依赖项: [qdrant-client>=1.12.0]

Qdrant - 向量相似性搜索引擎

用Rust编写的高性能向量数据库，专为生产RAG和语义搜索设计。

何时使用Qdrant

使用Qdrant当:

构建需要低延迟的生产RAG系统
需要混合搜索（向量 + 元数据过滤）
要求通过分片/复制进行水平扩展
希望本地部署并完全控制数据
需要每条记录的多向量存储（稠密 + 稀疏）
构建实时推荐系统

关键特性:

Rust驱动: 内存安全，高性能
丰富过滤: 在搜索期间按任何有效载荷字段过滤
多向量支持: 每个点支持稠密、稀疏、多稠密向量
量化: 标量、乘积、二进制量化以优化内存效率
分布式: Raft共识、分片、复制
REST + gRPC: 两个API具有完全功能对等

使用替代方案当:

Chroma: 简单设置，嵌入式用例
FAISS: 最大原始速度，研究/批处理
Pinecone: 完全托管，零运维偏好
Weaviate: GraphQL偏好，内置向量化器

快速开始

安装

# Python客户端
pip install qdrant-client

# Docker（推荐用于开发）
docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant

# Docker带持久化存储
docker run -p 6333:6333 -p 6334:6334 \
    -v $(pwd)/qdrant_storage:/qdrant/storage \
    qdrant/qdrant

基本用法

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

# 连接到Qdrant
client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# 插入带有效载荷的向量
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 384维向量
            payload={"title": "文档 1", "category": "技术"}
        ),
        PointStruct(
            id=2,
            vector=[0.3, 0.4, ...],
            payload={"title": "文档 2", "category": "科学"}
        )
    ]
)

# 带过滤的搜索
results = client.search(
    collection_name="documents",
    query_vector=[0.15, 0.25, ...],
    query_filter={
        "must": [{"key": "category", "match": {"value": "技术"}}]
    },
    limit=10
)

for point in results:
    print(f"ID: {point.id}, 分数: {point.score}, 有效载荷: {point.payload}")

核心概念

点 - 基本数据单元

from qdrant_client.models import PointStruct

# 点 = ID + 向量(s) + 有效载荷
point = PointStruct(
    id=123,                              # 整数或UUID字符串
    vector=[0.1, 0.2, 0.3, ...],        # 稠密向量
    payload={                            # 任意JSON元数据
        "title": "文档标题",
        "category": "技术",
        "timestamp": 1699900000,
        "tags": ["python", "ml"]
    }
)

# 批量上插（推荐）
client.upsert(
    collection_name="documents",
    points=[point1, point2, point3],
    wait=True  # 等待索引
)

集合 - 向量容器

from qdrant_client.models import VectorParams, Distance, HnswConfigDiff

# 创建带HNSW配置的集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=384,                        # 向量维度
        distance=Distance.COSINE         # COSINE, EUCLID, DOT, MANHATTAN
    ),
    hnsw_config=HnswConfigDiff(
        m=16,                            # 每个节点的连接数（默认16）
        ef_construct=100,                # 构建时精度（默认100）
        full_scan_threshold=10000        # 低于此值切换到暴力搜索
    ),
    on_disk_payload=True                 # 在磁盘上存储有效载荷
)

# 集合信息
info = client.get_collection("documents")
print(f"点数: {info.points_count}, 向量数: {info.vectors_count}")

距离度量

度量	用例	范围
`COSINE`	文本嵌入，归一化向量	0 到 2
`EUCLID`	空间数据，图像特征	0 到 ∞
`DOT`	推荐，非归一化	-∞ 到 ∞
`MANHATTAN`	稀疏特征，离散数据	0 到 ∞

搜索操作

基本搜索

# 简单最近邻搜索
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=10,
    with_payload=True,
    with_vectors=False  # 不返回向量（更快）
)

过滤搜索

from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

# 复杂过滤
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="技术")),
            FieldCondition(key="timestamp", range=Range(gte=1699000000))
        ],
        must_not=[
            FieldCondition(key="status", match=MatchValue(value="已归档"))
        ]
    ),
    limit=10
)

# 简写过滤语法
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter={
        "must": [
            {"key": "category", "match": {"value": "技术"}},
            {"key": "price", "range": {"gte": 10, "lte": 100}}
        ]
    },
    limit=10
)

批量搜索

from qdrant_client.models import SearchRequest

# 单个请求中的多个查询
results = client.search_batch(
    collection_name="documents",
    requests=[
        SearchRequest(vector=[0.1, ...], limit=5),
        SearchRequest(vector=[0.2, ...], limit=5, filter={"must": [...]}),
        SearchRequest(vector=[0.3, ...], limit=10)
    ]
)

RAG集成

使用sentence-transformers

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

# 初始化
encoder = SentenceTransformer("all-MiniLM-L6-v2")
client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# 索引文档
documents = [
    {"id": 1, "text": "Python是一种编程语言", "source": "wiki"},
    {"id": 2, "text": "机器学习使用算法", "source": "教科书"},
]

points = [
    PointStruct(
        id=doc["id"],
        vector=encoder.encode(doc["text"]).tolist(),
        payload={"text": doc["text"], "source": doc["source"]}
    )
    for doc in documents
]
client.upsert(collection_name="knowledge_base", points=points)

# RAG检索
def retrieve(query: str, top_k: int = 5) -> list[dict]:
    query_vector = encoder.encode(query).tolist()
    results = client.search(
        collection_name="knowledge_base",
        query_vector=query_vector,
        limit=top_k
    )
    return [{"text": r.payload["text"], "score": r.score} for r in results]

# 在RAG管道中使用
context = retrieve("什么是Python？")
prompt = f"上下文: {context}

问题: 什么是Python？"

使用LangChain

from langchain_community.vectorstores import Qdrant
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Qdrant.from_documents(documents, embeddings, url="http://localhost:6333", collection_name="docs")
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

使用LlamaIndex

from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.core import VectorStoreIndex, StorageContext

vector_store = QdrantVectorStore(client=client, collection_name="llama_docs")
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(documents, storage_context=storage_context)
query_engine = index.as_query_engine()

多向量支持

命名向量（不同嵌入模型）

from qdrant_client.models import VectorParams, Distance

# 带多种向量类型的集合
client.create_collection(
    collection_name="hybrid_search",
    vectors_config={
        "dense": VectorParams(size=384, distance=Distance.COSINE),
        "sparse": VectorParams(size=30000, distance=Distance.DOT)
    }
)

# 插入带命名向量的点
client.upsert(
    collection_name="hybrid_search",
    points=[
        PointStruct(
            id=1,
            vector={
                "dense": dense_embedding,
                "sparse": sparse_embedding
            },
            payload={"text": "文档文本"}
        )
    ]
)

# 搜索特定向量
results = client.search(
    collection_name="hybrid_search",
    query_vector=("dense", query_dense),  # 指定哪个向量
    limit=10
)

稀疏向量（BM25, SPLADE）

from qdrant_client.models import SparseVectorParams, SparseIndexParams, SparseVector

# 带稀疏向量的集合
client.create_collection(
    collection_name="sparse_search",
    vectors_config={},
    sparse_vectors_config={"text": SparseVectorParams(index=SparseIndexParams(on_disk=False))}
)

# 插入稀疏向量
client.upsert(
    collection_name="sparse_search",
    points=[PointStruct(id=1, vector={"text": SparseVector(indices=[1, 5, 100], values=[0.5, 0.8, 0.2])}, payload={"text": "文档"})]
)

量化（内存优化）

from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType

# 标量化（4倍内存减少）
client.create_collection(
    collection_name="quantized",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(
            type=ScalarType.INT8,
            quantile=0.99,        # 裁剪异常值
            always_ram=True      # 在RAM中保持量化
        )
    )
)

# 带重排的搜索
results = client.search(
    collection_name="quantized",
    query_vector=query,
    search_params={"quantization": {"rescore": True}},  # 重排顶部结果
    limit=10
)

有效载荷索引

from qdrant_client.models import PayloadSchemaType

# 为更快过滤创建有效载荷索引
client.create_payload_index(
    collection_name="documents",
    field_name="category",
    field_schema=PayloadSchemaType.KEYWORD
)

client.create_payload_index(
    collection_name="documents",
    field_name="timestamp",
    field_schema=PayloadSchemaType.INTEGER
)

# 索引类型: KEYWORD, INTEGER, FLOAT, GEO, TEXT（全文）, BOOL

生产部署

Qdrant云

from qdrant_client import QdrantClient

# 连接到Qdrant云
client = QdrantClient(
    url="https://your-cluster.cloud.qdrant.io",
    api_key="your-api-key"
)

性能调优

# 优化搜索速度（更高召回率）
client.update_collection(
    collection_name="documents",
    hnsw_config=HnswConfigDiff(ef_construct=200, m=32)
)

# 优化索引速度（批量加载）
client.update_collection(
    collection_name="documents",
    optimizer_config={"indexing_threshold": 20000}
)

最佳实践

批量操作 - 使用批量上插/搜索以提高效率
有效载荷索引 - 索引过滤中使用的字段
量化 - 对于大型集合（>100万向量）启用
分片 - 用于集合>1000万向量
磁盘存储 - 启用on_disk_payload用于大型有效载荷
连接池 - 重用客户端实例

常见问题

带过滤的搜索慢:

# 为过滤字段创建有效载荷索引
client.create_payload_index(
    collection_name="docs",
    field_name="category",
    field_schema=PayloadSchemaType.KEYWORD
)

内存不足:

# 启用量化和磁盘存储
client.create_collection(
    collection_name="large_collection",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(...),
    on_disk_payload=True
)

连接问题:

# 使用超时和重试
client = QdrantClient(
    host="localhost",
    port=6333,
    timeout=30,
    prefer_grpc=True  # gRCP以获取更好性能
)

参考

高级用法 - 分布式模式，混合搜索，推荐系统
故障排除 - 常见问题，调试，性能调优

资源

GitHub: https://github.com/qdrant/qdrant (2.2万+星)
文档: https://qdrant.tech/documentation/
Python客户端: https://github.com/qdrant/qdrant-client
云服务: https://cloud.qdrant.io
版本: 1.12.0+
许可证: Apache 2.0