qdrant-vector-search

Qdrant - 向量相似度搜索引擎

使用 Rust 编写的高性能向量数据库，适用于生产级 RAG 和语义搜索。

何时使用 Qdrant

在以下情况下使用 Qdrant：

• 构建需要低延迟的生产级 RAG 系统
• 需要混合搜索（向量 + 元数据筛选）
• 需要通过分片／副本实现水平扩展
• 希望在本地部署并完全掌控数据
• 需要在每条记录中存储多个向量（稠密 + 稀疏）
• 构建实时推荐系统

主要功能：

• 由 Rust 驱动：内存安全，性能高
• 丰富的筛选功能：搜索时可按任意负载字段筛选
• 多种向量：每个点支持稠密、稀疏和多稠密向量
• 量化：支持标量、乘积和二进制量化，提升内存效率
• 分布式：Raft 共识、分片、副本
• REST + gRPC：两种 API 均提供完整功能

以下情况建议使用替代方案：

• Chroma：设置更简单，适用于嵌入式场景
• FAISS：追求极致原始速度，适用于研究／批处理
• Pinecone：偏好完全托管且无需运维
• Weaviate：偏好 GraphQL 或内置向量化器

快速入门

安装

# Python client
pip install qdrant-client

# Docker (recommended for development)
docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant

# Docker with persistent storage
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

# Connect to Qdrant
client = QdrantClient(host="localhost", port=6333)

# Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# Insert vectors with payload
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 384-dim vector
            payload={"title": "Doc 1", "category": "tech"}
        ),
        PointStruct(
            id=2,
            vector=[0.3, 0.4, ...],
            payload={"title": "Doc 2", "category": "science"}
        )
    ]
)

# Search with filtering
results = client.search(
    collection_name="documents",
    query_vector=[0.15, 0.25, ...],
    query_filter={
        "must": [{"key": "category", "match": {"value": "tech"}}]
    },
    limit=10
)

for point in results:
    print(f"ID: {point.id}, Score: {point.score}, Payload: {point.payload}")

核心概念

点（Point）- 基本数据单元

from qdrant_client.models import PointStruct

# Point = ID + Vector(s) + Payload
point = PointStruct(
    id=123,                              # Integer or UUID string
    vector=[0.1, 0.2, 0.3, ...],        # Dense vector
    payload={                            # Arbitrary JSON metadata
        "title": "Document title",
        "category": "tech",
        "timestamp": 1699900000,
        "tags": ["python", "ml"]
    }
)

# Batch upsert (recommended)
client.upsert(
    collection_name="documents",
    points=[point1, point2, point3],
    wait=True  # Wait for indexing
)

集合（Collection）- 向量容器

from qdrant_client.models import VectorParams, Distance, HnswConfigDiff

# Create with HNSW configuration
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=384,                        # Vector dimensions
        distance=Distance.COSINE         # COSINE, EUCLID, DOT, MANHATTAN
    ),
    hnsw_config=HnswConfigDiff(
        m=16,                            # Connections per node (default 16)
        ef_construct=100,                # Build-time accuracy (default 100)
        full_scan_threshold=10000        # Switch to brute force below this
    ),
    on_disk_payload=True                 # Store payload on disk
)

# Collection info
info = client.get_collection("documents")
print(f"Points: {info.points_count}, Vectors: {info.vectors_count}")

距离度量

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

搜索操作

基本搜索

# Simple nearest neighbor search
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=10,
    with_payload=True,
    with_vectors=False  # Don't return vectors (faster)
)

筛选搜索

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

# Complex filtering
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="tech")),
            FieldCondition(key="timestamp", range=Range(gte=1699000000))
        ],
        must_not=[
            FieldCondition(key="status", match=MatchValue(value="archived"))
        ]
    ),
    limit=10
)

# Shorthand filter syntax
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter={
        "must": [
            {"key": "category", "match": {"value": "tech"}},
            {"key": "price", "range": {"gte": 10, "lte": 100}}
        ]
    },
    limit=10
)

批量搜索

from qdrant_client.models import SearchRequest

# Multiple queries in one request
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 集成

详情、代码示例和参数表：references/rag-integration.md。在适用此步骤时阅读。

多向量支持

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

from qdrant_client.models import VectorParams, Distance

# Collection with multiple vector types
client.create_collection(
    collection_name="hybrid_search",
    vectors_config={
        "dense": VectorParams(size=384, distance=Distance.COSINE),
        "sparse": VectorParams(size=30000, distance=Distance.DOT)
    }
)

# Insert with named vectors
client.upsert(
    collection_name="hybrid_search",
    points=[
        PointStruct(
            id=1,
            vector={
                "dense": dense_embedding,
                "sparse": sparse_embedding
            },
            payload={"text": "document text"}
        )
    ]
)

# Search specific vector
results = client.search(
    collection_name="hybrid_search",
    query_vector=("dense", query_dense),  # Specify which vector
    limit=10
)

稀疏向量（BM25、SPLADE）

from qdrant_client.models import SparseVectorParams, SparseIndexParams, SparseVector

# Collection with sparse vectors
client.create_collection(
    collection_name="sparse_search",
    vectors_config={},
    sparse_vectors_config={"text": SparseVectorParams(index=SparseIndexParams(on_disk=False))}
)

# Insert sparse vector
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": "document"})]
)

量化（内存优化）

from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType

# Scalar quantization (4x memory reduction)
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,        # Clip outliers
            always_ram=True      # Keep quantized in RAM
        )
    )
)

# Search with rescoring
results = client.search(
    collection_name="quantized",
    query_vector=query,
    search_params={"quantization": {"rescore": True}},  # Rescore top results
    limit=10
)

负载索引

from qdrant_client.models import PayloadSchemaType

# Create payload index for faster filtering
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
)

# Index types: KEYWORD, INTEGER, FLOAT, GEO, TEXT (full-text), BOOL

生产环境部署

Qdrant Cloud

from qdrant_client import QdrantClient

# Connect to Qdrant Cloud
client = QdrantClient(
    url="https://your-cluster.cloud.qdrant.io",
    api_key="your-api-key"
)

性能调优

# Optimize for search speed (higher recall)
client.update_collection(
    collection_name="documents",
    hnsw_config=HnswConfigDiff(ef_construct=200, m=32)
)

# Optimize for indexing speed (bulk loads)
client.update_collection(
    collection_name="documents",
    optimizer_config={"indexing_threshold": 20000}
)

最佳实践

1. 批量操作 - 使用批量 upsert／搜索以提高效率
2. 负载索引 - 为筛选中使用的字段建立索引
3. 量化 - 对大型集合（>1M 个向量）启用
4. 分片 - 对包含 >10M 个向量的集合启用
5. 磁盘存储 - 对大型负载启用 on_disk_payload
6. 连接池 - 重用客户端实例

常见问题

筛选搜索速度慢：

# Create payload index for filtered fields
client.create_payload_index(
    collection_name="docs",
    field_name="category",
    field_schema=PayloadSchemaType.KEYWORD
)

内存不足：

# Enable quantization and on-disk storage
client.create_collection(
    collection_name="large_collection",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(...),
    on_disk_payload=True
)

连接问题：

# Use timeout and retry
client = QdrantClient(
    host="localhost",
    port=6333,
    timeout=30,
    prefer_grpc=True  # gRPC for better performance
)

参考资料

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

资源

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

Agent 操作流程

1. 检查环境。 确认语料、嵌入模型和向量存储的版本，以及索引的存放位置。
2. 明确输入。 根据数据或用户确认格式、标识符和参数。任何会改变结果的值，都应询问而非猜测。
3. 先运行小规模版本。 为一个小型子集建立索引，并使用已知的问题—段落对测试检索效果。
4. 使用上述说明和参考资料执行完整任务。
5. 验证结果。 在带标签的数据集上衡量 recall@k；答案只能引用检索到的段落。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及还有哪些不确定之处。

如果发生这种情况	请这样做
检索返回了无关片段	重新检查分块方式，添加 BM25/混合搜索，或先重新排序，再考虑更改生成器。
已安装版本中缺少这些说明里的函数、标志或端点	查看已安装版本自己的文档（help()、--help、官方文档），进行调整并告知用户。切勿编造 API。
必需的输入、标识符或参数含糊不清	询问用户，或在运行前明确说明假设。

完整性规则

• 切勿捏造结果、参数、标识符、引文或统计数据。如果无法运行或验证某项内容，请如实说明。
• 答案必须来自检索到的文本；如果语料中没有答案，请说明。
• 请将此处特定于版本的详细信息视为可能已过时：对照已安装版本的官方文档进行确认。
• 对于会产生费用、占用共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，必须先征求许可。

相关技能

• pinecone：介绍如何通过 Python 客户端以及 LangChain 和 LlamaIndex 集成使用 Pinecone（一种托管式无服务器向量数据库）。
• faiss：Facebook 推出的库，用于高效地搜索和聚类稠密向量。
• chroma：面向 AI 应用的开源嵌入数据库。