# Qdrant 开发使用指南

> 面向 Python 开发者。所有代码基于 `qdrant-client 1.19.x`，可直接复制进项目。
> 完整可运行封装见 [`dev/qdrant_helper.py`](./dev/qdrant_helper.py)；部署与运维见 [`Qdrant生产运维手册.md`](./Qdrant生产运维手册.md)
>
> 安装：`pip install "qdrant-client>=1.19" fastembed`

---

## 第 1 章 客户端初始化

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

```python
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 双向量

```python
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 参数选择的判断依据

| 参数 | 怎么定 | 定错的后果 |
|---|---|---|
| `size` | **严格等于** embedding 模型输出维度 | 写小了不报错但结果全乱（最隐蔽的故障） |
| `distance` | 文本语义用 `COSINE`；已归一化向量可用 `DOT` | 混用会导致排序完全错乱 |
| `on_disk` | 数据量 > 内存时设 `True` | 全内存超容量会被 OOM Killer 干掉 |
| `m` | 16 起步，召回不够再调大 | 调太大内存和建索引时间暴涨 |
| `replication_factor` | 单机 1；集群 ≥2 才有高可用 | 集群设 1 = 挂一节点丢数据 |

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

| 场景 | 配置 |
|---|---|
| 单一 embedding 模型 | 单向量即可（`VectorParams` 直接传） |
| 混合检索（dense + sparse）| 命名向量：`{"dense": ..., "sparse": ...}` ★ 你的场景 |
| 多模型融合 / ColBERT 多向量 | 命名向量 + `multivector_config` |

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

---

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

### 3.1 铁律

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

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

### 3.2 索引类型选择表

| 字段用途 | `field_schema` | 示例 |
|---|---|---|
| 精确匹配的分类/枚举 | `keyword` | `chunk_type`、`doc_id`、`user_id`、`tags` |
| 数值范围过滤 | `integer` / `float` | `chapter_no`、`score`、`token_count` |
| 时间范围过滤 | `datetime` | `ingested_at`、`effective_date` |
| 全文检索（分词匹配） | `text` | `title`、`content` |
| 布尔 | `bool` | `is_active`、`has_code` |
| 地理位置 | `geo` | 门店坐标 |

### 3.3 建索引代码

```python
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（最重要的一条）

```python
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 批量写入

```python
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="")
```

| 参数 | 建议 | 说明 |
|---|---|---|
| `batch_size` | 100~1000 | 再大收益递减，内存峰值上升 |
| `wait` | 导入期 False，末批 True | False 时返回成功 ≠ 已可检索 |
| 并发 | 2~4 路 | 过高会互相争抢 CPU |

### 4.3 导入期的两个提速技巧

```python
# ① 导入前：把索引阈值调到极大，避免边导边反复重建索引（可快数倍）
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 更新与删除

```python
# 只更新 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 完整实现

```python
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 三种检索模式对比

| 模式 | 代码要点 | 何时用 |
|---|---|---|
| 纯稠密 | 单向量 `query=vec` | 口语化提问、概念性检索 |
| **混合（推荐）** | `prefetch` 双路 + `FusionQuery(RRF)` | ★ 默认用它。精确词与语义都要 |
| 加权融合 | `FusionQuery(fusion=DBSF)` + 权重 | 需要人工指定两路重要性时 |

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

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

```python
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 结果组装的正确姿势

```python
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 章 过滤语法速查

```python
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 两种模式

| 模式 | 做法 | 适合 | 注意 |
|---|---|---|---|
| **payload 隔离（推荐）** | 单集合 + `user_id` 字段过滤 + keyword 索引 | 租户数多、每租户数据量小 | ★ 过滤条件绝不能漏，漏一次就是数据泄露 |
| 每租户一集合 | 集合名带租户 ID | 租户数少、需要独立配额/备份 | 集合数上千后元数据管理成本高 |

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

### 7.2 实现要点

```python
# ① 写入时强制带上 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 章 分页、批量与统计

```python
# 分页遍历全部数据（做评测集、做导出都用它）
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 切换

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

```python
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) 旧集合保留只读一段时间（追溯"当时系统是按哪版回答的"），确认无误后再删
```

**业务代码只需要认别名**：

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

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

```python
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（开发侧）

```python
# 创建快照（返回快照名，文件落在服务端 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 章 异步客户端（高并发场景）

```python
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 集成要点**：

```python
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 个高频坑（按踩到概率排序）

| # | 坑 | 症状 | 正解 |
|---|---|---|---|
| 1 | **过滤字段没建 payload 索引** | 加了 filter 后查询暴慢 | filter 用到的字段全部建索引 |
| 2 | **`size` 与模型维度不一致** | 结果为空或完全错乱，**不报错** | 严格对齐；变更需重建集合 |
| 3 | 用命名向量但查询没写 `using` | 报错或结果异常 | 查询时显式指定通道名 |
| 4 | 点 ID 用自增序号 | 重导后数据翻倍 | `uuid5(user_id + content_hash)` |
| 5 | 导入后没调回 `indexing_threshold` | 检索一直慢、CPU 高 | 导完调回并等 `indexed_vectors_count` 追平 |
| 6 | 多租户漏写 `user_id` 过滤 | **数据泄露** | 封装层强制注入 + 断言 |
| 7 | 融合查询把 filter 放在顶层 | 过滤不生效或行为不一致 | 过滤放 `prefetch` 里 |
| 8 | `with_payload=True` 不裁剪字段 | 带宽与延迟浪费 | 用 `payload_selector` 只取需要的字段 |
| 9 | 忘了 `wait=True` 就立刻查询 | 查不到刚写的数据 | 末批用 `wait=True` |
| 10 | 客户端 `timeout` 是默认 5s | 批量操作超时 | 显式设 30~60s |
| 11 | 以为删数据内存立刻释放 | 内存不降 | 等优化器 vacuum，或临时调小 `deleted_threshold` |
| 12 | 用本地嵌入式模式上线 | 数据格式不兼容、>2 万点告警 | 生产必须 server 模式 |
| 13 | alias 切换后旧集合立刻删除 | 无法回溯、无法快速回滚 | 旧集合保留只读一段时间 |
| 14 | filter 里值类型与 payload 不一致 | 静默返回空 | 写入与过滤类型统一 |
| 15 | 同时传 `hnsw_ef` 和 `exact=True` | 参数无效、白花性能 | 二选一 |
| 16 | 每请求新建客户端 | 连接池失效、延迟高 | 启动时建一次，全局复用 |
| 17 | `/metrics` 抓取没带 key | Prometheus 401 | 用只读 key 配 `authorization.credentials` |
| 18 | 只调稀疏向量不调稠密 | 语义检索能力缺失 | 两条腿都要（混合检索） |
| 19 | 单条 upsert 循环写 | 慢到不可用 | 批量 100~1000/次 |
| 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 已放宽
[ ] 密钥从环境变量读取
[ ] 有别名机制，版本切换零停机
[ ] 有快照脚本，且验证过恢复流程
```

---

**配套文件**

| 文件 | 用途 |
|---|---|
| [`dev/qdrant_helper.py`](./dev/qdrant_helper.py) | 上述所有能力的可运行封装，带 `--selftest` 自检 |
| [`Qdrant生产运维手册.md`](./Qdrant生产运维手册.md) | 部署、配置、备份、监控、故障排查 |
| [`deploy/production.yaml`](./deploy/production.yaml) | 服务端生产配置 |