Qdrant 开发使用指南

面向 Python 开发者。所有代码基于 qdrant-client 1.19.x，可直接复制进项目。

完整可运行封装见 dev/qdrant_helper.py；部署与运维见 Qdrant生产运维手册.md

安装：pip install "qdrant-client>=1.19" fastembed

第 1 章 客户端初始化

三种模式，配置差异只在初始化那一行：

import os
from qdrant_client import QdrantClient

# ---------- ① 本地嵌入式：只用于开发验证和评测，不要用在生产 ----------
# 纯 Python + SQLite 实现，数据格式与 server 版不兼容，>2 万点会告警
client = QdrantClient(path="./qdrant_local")

# ---------- ② 自建服务（生产）----------
client = QdrantClient(
    url=os.environ.get("QDRANT_URL", "http://127.0.0.1:6333"),
    api_key=os.environ["QDRANT_API_KEY"],     # 密钥永远从环境变量来，不写进代码
    timeout=30,                                # 默认 5s 对批量导入偏小
    prefer_grpc=False,                         # True 走 6334，吞吐更高（生产建议开）
)

# ---------- ③ Qdrant Cloud ----------
# client = QdrantClient(url="https://xxx.cloud.qdrant.io:6333",
#                       api_key=..., cloud_inference=True)   # 开了可用云端推理生成向量

三个初始化要点：

1. timeout 一定要显式设置。默认 5 秒，批量 upsert 或快照操作很容易超时；
2. prefer_grpc=True 是生产默认选择，REST 和 gRPC 端口都要在防火墙放行；
3. 本地嵌入式与 server 的切换要显式配置，别让代码"碰巧"连上错误的目标——用环境变量驱动，并在启动日志里打印实际连接目标。

第 2 章 建集合（生产写法）

2.1 完整示例：dense + sparse 双向量

from qdrant_client import QdrantClient, models

def create_kb_collection(client, name: str, dense_dim: int = 1024):
    """创建知识库集合：dense + sparse 双通道 + int8 量化 + 生产级优化器配置"""
    if client.collection_exists(name):
        return

    client.create_collection(
        collection_name=name,

        # ---- 命名向量：dense 管语义，sparse 管精确词（"第三十九条"、"hreflang"）----
        vectors_config={
            "dense": models.VectorParams(
                size=dense_dim,                        # ★ 必须严格等于模型输出维度
                distance=models.Distance.COSINE,       # 文本语义用 COSINE
                on_disk=False,                         # 热数据放内存
            ),
        },
        sparse_vectors_config={
            "sparse": models.SparseVectorParams(
                index=models.SparseIndexParams(on_disk=False),
                modifier=models.Modifier.IDF,          # 让 BM25 式加权更规范，建议开
            ),
        },

        # ---- HNSW 参数 ----
        hnsw_config=models.HnswConfigDiff(
            m=16,                     # 每点连接数，16 是通用最优起点
            ef_construct=100,         # 建索引搜索宽度，质量与速度的平衡点
            full_scan_threshold=10000,# 小于此规模直接暴力扫（更准）
        ),

        # ---- 量化：内存不够的第一选择，省 75% 且召回损失通常 <1% ----
        quantization_config=models.ScalarQuantization(
            scalar=models.ScalarQuantizationConfig(
                type=models.ScalarType.INT8,
                quantile=0.99,        # 剔掉 1% 离群值再定标，避免量化区间被拉扁
                always_ram=True,      # 量化数据常驻内存，原始向量可留磁盘
            ),
        ),

        # ---- 优化器：控制索引何时建、段何时合并 ----
        optimizers_config=models.OptimizersConfigDiff(
            default_segment_number=2,     # 段数 ≈ 并行搜索粒度，2~4 核保持 2
            indexing_threshold=20000,     # 未索引数据到 20MB 触发建索引
            memmap_threshold=50000,       # 段超 50MB 改用 mmap，把冷数据交给 OS
            deleted_threshold=0.2,        # 垃圾点占 20% 触发 vacuum 回收内存
        ),

        # ---- 分片与副本：单机固定 1；集群再调 ----
        shard_number=1,
        replication_factor=1,
        write_consistency_factor=1,
    )

2.2 参数选择的判断依据

2.3 命名向量 vs 单向量：什么时候需要多向量

注意：一旦用了命名向量，查询时必须指定 using="dense" 之类的通道名，否则报错。这是从单向量迁过来最常见的坑。

第 3 章 Payload 索引（决定检索速度的关键）

3.1 铁律

所有会出现在 filter 里的字段，都必须在 Qdrant 里建索引。

不建索引的 filter 在 Qdrant 是全量扫描——100 万条数据的集合上，一个没索引的 must 条件能让查询从 15ms 变成 900ms。这是自建向量库最常见的性能问题，比向量算法本身重要得多。

3.2 索引类型选择表

3.3 建索引代码

from qdrant_client import models

def ensure_indexes(client, collection: str):
    """建集合后立刻建 payload 索引。幂等：已存在会覆盖，不会报错。"""
    indexes = {
        # 过滤主力字段
        "user_id":      models.PayloadSchemaType.KEYWORD,     # 多租户隔离，必建
        "doc_id":       models.PayloadSchemaType.KEYWORD,     # 按文档过滤
        "doc_type":     models.PayloadSchemaType.KEYWORD,
        "chunk_type":   models.PayloadSchemaType.KEYWORD,     # guide/code/table
        "kb_version":   models.PayloadSchemaType.KEYWORD,     # 版本路由，必建
        "effective_status": models.PayloadSchemaType.KEYWORD,
        "tags":         models.PayloadSchemaType.KEYWORD,     # 数组也支持
        "chapter_no":   models.PayloadSchemaType.INTEGER,
        "ingested_at":  models.PayloadSchemaType.DATETIME,    # 时间过滤"近三个月"
        "effective_date": models.PayloadSchemaType.DATETIME,
        "has_code":     models.PayloadSchemaType.BOOL,
    }
    for field, schema in indexes.items():
        client.create_payload_index(collection_name=collection,
                                    field_name=field, field_schema=schema)

设计原则：payload 字段 = 将来可能进 filter 的维度。宁可在设计时多列几个，也别等线上查询慢了再补——大集合上建索引会触发重建，只能在低峰期做。

容量提醒：每个索引都常驻内存。10 个 keyword 索引在千万级数据上不是小数目，别什么都往上加。

第 4 章 写入

4.1 幂等点 ID（最重要的一条）

import hashlib, uuid

def make_point_id(*parts: str) -> str:
    """用内容哈希做点 ID —— 同一内容重复导入永远落在同一个点，天然幂等"""
    raw = "|".join(parts)
    # 方式一：uuid5（标准 UUID 格式，Qdrant 接受）
    return str(uuid.uuid5(uuid.NAMESPACE_URL, raw))
    # 方式二：sha256 前 32 位（更短，但要是合法的 uint64 或 UUID 格式）
    # return int(hashlib.sha256(raw.encode()).hexdigest()[:16], 16)

为什么必须这么做：知识库更新时你会重新跑导入脚本。如果点 ID 是自增序号，第二次导入会产生一整套重复数据，检索结果里同一段内容出现两次，还白占一倍内存。用 uuid5(user_id + content_hash) 后，重导就是覆盖。

4.2 批量写入

def upsert_batched(client, collection: str, points: list, batch_size: int = 256,
                   final_wait: bool = True | False):
    """分批写入。导入期 wait=False 提吞吐，最后一批 wait=True 确保落盘。"""
    total = len(points)
    for i in range(0, total, batch_size):
        batch = points[i:i + batch_size]
        # 最后一批用 wait=True，确保全部落盘后再返回
        is_last = i + batch_size >= total
        client.upsert(collection_name=collection, points=batch,
                      wait=(final_wait if is_last else False))
        print(f"\r写入 {min(i + batch_size, total)}/{total}", end="")

4.3 导入期的两个提速技巧

# ① 导入前：把索引阈值调到极大，避免边导边反复重建索引（可快数倍）
client.update_collection(collection_name=coll,
    optimizers_config=models.OptimizersConfigDiff(indexing_threshold=1_000_000))

# ② 导入完成后：调回正常值，触发索引构建
client.update_collection(collection_name=coll,
    optimizers_config=models.OptimizersConfigDiff(indexing_threshold=20_000))

# ③ 等待索引完成 —— 这一步不做，导入后的检索会一直慢
import time
while True:
    info = client.get_collection(coll)
    pending = (info.points_count or 0) - (info.indexed_vectors_count or 0)
    if pending <= 0:
        break
    print(f"待索引 {pending} 条 ...")
    time.sleep(5)

4.4 更新与删除

# 只更新 payload（不改向量）—— 版本号刷新、状态变更等场景
client.set_payload(collection_name=coll, payload={"effective_status": "已修订"},
                   points=[point_id])

# 按 ID 删
from qdrant_client import models
client.delete(collection_name=coll,
              points_selector=models.PointIdsList(points=[id1, id2]))

# 按条件批量删（如删掉整个 v1 版本）
client.delete(collection_name=coll,
              points_selector=models.FilterSelector(
                  filter=models.Filter(must=[
                      models.FieldCondition(key="kb_version",
                                            match=models.MatchValue(value="2021"))
                  ])))

# 统计（验证删除结果）
client.count(collection_name=coll, exact=True).count

删除后内存不立刻释放：Qdrant 靠优化器 vacuum 回收，要等垃圾点占比达到 deleted_threshold（默认 20%）。这是设计如此。急着回收就临时调小阈值。

第 5 章 检索（四层结构）

这是全文最重要的一章。生产级检索永远是这四层叠加：

① filter 预过滤（缩小候选集）
      ↓
② prefetch 多路召回（dense 语义 + sparse 精确词）
      ↓
③ fusion 融合（RRF，无需调权重）
      ↓
④ 按需重排（cross-encoder，精度优先时加）

5.1 完整实现

from qdrant_client import QdrantClient, models

def hybrid_search(client, collection: str, query_text: str,
                  dense_vector: list[float], sparse_vector: models.SparseVector,
                  user_id: str | None = None, chunk_types: list[str] | None = None,
                  top_k: int = 5, prefetch_k: int = 20):
    """
    混合检索：过滤 → 双路召回 → RRF 融合
    query_text / sparse_vector 由 bge-m3 一并产出（一个模型同时给稠密和稀疏）
    """
    # ---------- ① 构造过滤条件 ----------
    must = []
    if user_id:                                    # 多租户隔离：★ 绝不能漏
        must.append(models.FieldCondition(key="user_id",
                                          match=models.MatchValue(value=user_id)))
    if chunk_types:                                # 只要代码块（按类型收窄候选集）
        must.append(models.FieldCondition(key="chunk_type",
                                          match=models.MatchAny(any=chunk_types)))
    query_filter = models.Filter(must=must) if must else None

    # ---------- ② 双路召回 ----------
    prefetch = [
        models.Prefetch(query=dense_vector, using="dense",
                        limit=prefetch_k, filter=query_filter),
        models.Prefetch(query=sparse_vector, using="sparse",
                        limit=prefetch_k, filter=query_filter),
    ]

    # ---------- ③ RRF 融合（免调权重，两路各自排名取倒数求和）----------
    resp = client.query_points(
        collection_name=collection,
        prefetch=prefetch,
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        limit=top_k,
        with_payload=True,
        with_vectors=False,
        # 查询参数：ef 调大提召回，但边际递减且增加延迟
        params=models.SearchParams(hnsw_ef=128, exact=False),
    )
    return resp.points

注意：融合查询不要同时传 query_filter 和 prefetch 的 filter——过滤条件放在 prefetch 里才生效（这是 Qdrant Query API 的一个易错点），顶层 filter 在融合场景下行为不一致。

5.2 三种检索模式对比

实测数据支撑：在你那份 SEO 手册上，纯稠密（弱模型 512 维）Recall@5 = 90%，两个失败用例全是代码块。补语义摘要后到 100%。稀疏通道则解决另一类问题：用户直接问"第三十九条怎么规定的"，纯向量会把第三十九条和第四十九条当近义。

5.3 重排（rerank）：最后一公里的精度

def rerank_with_cross_encoder(question: str, hits: list, top_n: int = 3):
    """cross-encoder 精排：把 (query, doc) 成对送进模型打相关性分
    比向量检索准得多，代价是慢（几十到几百毫秒），所以只对 top20 内的候选做"""
    from sentence_transformers import CrossEncoder
    model = CrossEncoder("BAAI/bge-reranker-v2-m3")   # 全局加载一次，别每请求加载
    pairs = [(question, h.payload.get("content") or h.payload.get("title", ""))
             for h in hits]
    scores = model.predict(pairs)
    ranked = sorted(zip(hits, scores), key=lambda x: -x[1])
    return [(h, float(s)) for h, s in ranked[:top_n]]

用不用重排的判断：召回已经够（R@5 达标）但前 3 条排序不对 → 加重排；召回本身就不够 → 先修分块和混合检索，重排救不了。

5.4 结果组装的正确姿势

def format_hits(hits) -> str:
    """把检索结果拼成给 LLM 的上下文 —— 必须带溯源信息"""
    blocks = []
    for i, h in enumerate(hits, 1):
        p = h.payload or {}
        blocks.append(
            f"[{i}] 《{p.get('doc_title','')}》 "
            f"{p.get('breadcrumb','')}\n"
            f"{p.get('content','')}\n"
            f"来源: {p.get('source_url','')} (score={h.score:.3f})"
        )
    return "\n\n---\n\n".join(blocks)

为什么一定要带 breadcrumb 和 source_url：LLM 答完能标出处，用户能核对原文。这也把"和模型记忆混着编"的幻觉压下去——有出处可查的答案，用户会去查。

第 6 章 过滤语法速查

from qdrant_client import models as m

# 精确匹配
m.FieldCondition(key="doc_type", match=m.MatchValue(value="行政法规"))

# 多值任选其一（枚举多选）
m.FieldCondition(key="chunk_type", match=m.MatchAny(any=["code", "table"]))

# 排除
m.FieldCondition(key="tags", match=m.MatchExcept(**{"except": ["过期"]}))

# 数值范围
m.FieldCondition(key="chapter_no", range=m.Range(gte=3, lte=5))

# 时间范围（ISO 字符串，用 DatetimeRange）
m.FieldCondition(key="ingested_at",
                 range=m.DatetimeRange(gte="2026-01-01T00:00:00Z"))

# 布尔
m.FieldCondition(key="has_code", match=m.MatchValue(value=True))

# 字符串前缀（1.19 新增，keyword 索引开 prefix=true 后可用）
m.FieldCondition(key="source_url", match=m.MatchText(text="https://qdrant."))  # 注意这是分词
# 前缀匹配需在建索引时置 {"type":"keyword","prefix":true}

# 嵌套字段（payload 里有 dict）
m.FieldCondition(key="meta.author", match=m.MatchValue(value="张三"))

# 组合
m.Filter(
    must=[...],          # AND
    should=[...],        # OR（满足任一）
    must_not=[...],      # NOT
)

两个常见错误：

1. 类型不匹配：payload 里 chapter_no 存的是字符串 "3"，filter 用 MatchValue(value=3)（整数）→ 永远匹配不上且不报错。写入时统一类型，并建对应类型的索引；
2. 字段名写错：chunk_type 写成 chunkType → 静默返回空。用 scroll 抽样看一眼真实结构再写 filter。

第 7 章 多租户

7.1 两种模式

1.16 起还有 tiered multitenancy，把大租户拆到独立分片、小租户共享分片，是前两种的折中。

7.2 实现要点

# ① 写入时强制带上 user_id
point = models.PointStruct(
    id=make_point_id(user_id, content_hash),
    vector={"dense": dense, "sparse": sparse},
    payload={..., "user_id": user_id},
)

# ② 查询时强制过滤 —— 用包装函数保证"不可能漏"
def search_for_user(client, collection, user_id, **kwargs):
    assert user_id, "user_id 不能为空，否则会跨租户召回"
    return hybrid_search(client, collection, user_id=user_id, **kwargs)

# ③ 生产加固：用 Filter 里的 must 而非 should
#    should 是"满足任一"，如果没有其他 must 条件，它等于不过滤！

红线：多租户场景下，租户过滤一旦失效就是数据泄露事故。别依赖"记得写 filter"，而是在客户端封装层强制注入（上面的 search_for_user），并加断言。

第 8 章 分页、批量与统计

# 分页遍历全部数据（做评测集、做导出都用它）
def iter_all_points(client, collection, page_size=256):
    offset = None
    while True:
        points, offset = client.scroll(
            collection_name=collection, limit=page_size, offset=offset,
            with_payload=True, with_vectors=False,
        )
        yield from points
        if offset is None:      # None 表示到底了
            break

# 精确计数（默认是近似值，做校验要用 exact）
client.count(collection_name=collection, exact=True).count

# 带过滤的计数
client.count(collection_name=collection, count_filter=models.Filter(
    must=[models.FieldCondition(key="kb_version",
                                match=models.MatchValue(value="2026"))]
), exact=True).count

# 分面统计（1.19 大幅提速）：看各类型各有多少条
client.facet(collection_name=collection, key="chunk_type", limit=20)

大集合上遍历要慢：全量 scroll 千万级数据会很慢。生产做评测集时抽样即可（1.19 新增的 slice filter 支持取稳定子集，非常适合做可复现的抽样评测）。

第 9 章 版本化与 alias 切换

这是"知识库按版本更新"的标准做法，业务代码零改动。

def switch_kb_version(client, new_collection: str, alias: str = "kb_current"):
    """把别名原子性地切到新集合。切换期间无停机。"""
    # 一次请求里：删旧别名 + 建向新集合的别名（原子操作，不存在中间态）
    client.update_collection_aliases(change_aliases_operations=[
        models.DeleteAliasOperation(
            delete_alias=models.DeleteAlias(alias_name=alias)),
        models.CreateAliasOperation(
            create_alias=models.CreateAlias(
                collection_name=new_collection, alias_name=alias)),
    ])
    print(f"alias {alias} → {new_collection}")


# ---------- 年度大版本更新流程 ----------
# 1) 用 build_chunks.py 生成 2026 版语料（新 chunk_id / 新 content_hash）
# 2) 建新集合 kb_chunks_v2026，导入
# 3) 跑 20 题评测集，Recall@5 达标（≥90%）
# 4) switch_kb_version(client, "kb_chunks_v2026")
# 5) 旧集合保留只读一段时间（追溯"当时系统是按哪版回答的"），确认无误后再删

业务代码只需要认别名：

client.query_points(collection_name="kb_current", ...)   # 永远指向当前版本

增量补丁（不动库的情况）：年内只改了几份文档时，不要重建集合。按 content_hash 比对出变更块，upsert 覆盖即可（幂等点 ID 保证不产生重复）。

def incremental_update(client, collection, new_chunks: list[dict]):
    """只写入内容有变化的块：content_hash 变了才 upsert"""
    existing = {}
    for p in iter_all_points(client, collection):
        existing[p.payload["chunk_id"]] = p.payload.get("content_hash")

    changed = [c for c in new_chunks
               if existing.get(c["chunk_id"]) != c["metadata"]["content_hash"]]
    print(f"总块 {len(new_chunks)}，需更新 {len(changed)}")
    # 变更的 upsert，被删除的块按 chunk_id 差集删掉

第 10 章 快照 API（开发侧）

# 创建快照（返回快照名，文件落在服务端 snapshots_path）
snap = client.create_snapshot(collection_name="kb_current", wait=True)
print(snap.name)

# 列出 / 删除
for s in client.list_snapshots(collection_name="kb_current"):
    print(s.name, s.creation_time, s.size)

client.delete_snapshot(collection_name="kb_current", snapshot_name="xxx.snapshot")

# 恢复（可恢复到新集合名，先验证再切换 —— 比直接覆盖安全得多）
client.recover_snapshot(collection_name="kb_chunks_v2_verify",
                        location="file:///var/lib/qdrant/snapshots/xxx.snapshot",
                        wait=True)

开发侧的两个建议：

1. 升级前、大版本切换前，用代码自动打快照（别手动记），快照名带日期和用途标签；
2. 恢复先恢复到临时集合，核对 points_count 和抽检几条数据无误，再做 alias 切换。

第 11 章 异步客户端（高并发场景）

from qdrant_client import AsyncQdrantClient, models

async def main():
    client = AsyncQdrantClient(url="http://127.0.0.1:6333", api_key="...")

    # 并发多查询：批量场景下比串行快数倍
    import asyncio
    queries = [{"dense": d, "sparse": s, "user": u} for d, s, u in batch]
    tasks = [search_one(client, q) for q in queries]
    results = await asyncio.gather(*tasks)

    await client.close()

什么时候需要异步：FastAPI 服务里单请求内要发多个检索（比如分别查知识库和长期记忆），或者做批量离线任务。单个检索用同步客户端完全够——异步不会让单次查询变快。

FastAPI 集成要点：

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # 客户端在启动时创建一次并复用（内部有连接池，不要每请求新建）
    app.state.qdrant = QdrantClient(url=settings.qdrant_url,
                                    api_key=settings.qdrant_key, timeout=30)
    yield
    app.state.qdrant.close()

第 12 章 20 个高频坑（按踩到概率排序）

第 13 章 上线前性能检查清单

【集合】
[ ] size 与 embedding 维度严格一致
[ ] distance 与模型匹配（文本用 COSINE）
[ ] 内存不够时已开 int8 量化
[ ] HNSW m/ef_construct 已按数据量设定

【索引】
[ ] filter 用到的每个字段都建了 payload 索引
[ ] 索引类型与字段类型匹配（keyword/integer/datetime/bool）
[ ] 索引字段数量已审视（每个都吃内存）

【数据】
[ ] 点 ID 幂等（uuid5 / 内容哈希）
[ ] 批量写入 100~1000/批，末批 wait=True
[ ] 导入后 indexing_threshold 已调回，indexed_vectors_count ≈ points_count
[ ] 总数与语料块数一致（count exact 校验）

【检索】
[ ] 混合检索（dense + sparse + RRF）已开
[ ] 查询时按需收窄候选集（chunk_type / 版本过滤）
[ ] 需要时加了 cross-encoder 重排
[ ] 评测集 Recall@5 ≥ 90%，且每次改动后回归

【工程】
[ ] 客户端全局复用，timeout 已放宽
[ ] 密钥从环境变量读取
[ ] 有别名机制，版本切换零停机
[ ] 有快照脚本，且验证过恢复流程

配套文件