Qdrant 向量库怎么导出重建？逐页保存向量和 payload 并核对 ID

需要把 Qdrant 中的向量记录导出到可检查的文件，可以用 scroll 分页取得 point ID、向量和 payload，再在新集合中按相同配置写回。关键是显式读取向量、正确推进分页游标，并对照重建前后的 ID 和内容；只导出 payload 不能恢复向量检索。

本文演示单个无名称的稠密向量字段，使用三条人工向量和虚构资料。它是记录内容的导出重建，不是完整服务快照，不包含分片、别名、权限、索引调参和全部运行配置。不会连接或修改实际业务向量库。

使用本地模式先验证流程

在独立 Python 环境安装 python -m pip install qdrant-client==1.19.1。Qdrant 官方客户端说明提供 QdrantClient(“:memory:”) 本地模式，可以不启动服务器就验证记录 API。本文本地示例只验证数据往返，不测量服务性能。

新建练习目录保存脚本；示例会覆盖这个目录的 points_export.jsonl。生产导出应先固定数据版本或暂停相应写入，避免分页期间数据变化产生缺失或重复；scroll 并不自动替你建立跨页一致性快照。

完整演示：两条一页，导出三条

import json
import math
from pathlib import Path
from qdrant_client import QdrantClient, models

client = QdrantClient(":memory:")
config = models.VectorParams(size=3, distance=models.Distance.COSINE)
client.create_collection("source_demo", vectors_config=config)
client.upsert("source_demo", points=[
    models.PointStruct(id=1, vector=[1.0, 0.0, 0.0], payload={"text": "示例保修条款", "version": 2}),
    models.PointStruct(id=2, vector=[0.0, 1.0, 0.0], payload={"text": "示例充电条款", "version": 2}),
    models.PointStruct(id=3, vector=[0.0, 0.0, 1.0], payload={"text": "示例包装说明", "version": 2}),
], wait=True)

def records(name):
    offset = None
    while True:
        page, next_offset = client.scroll(
            collection_name=name, limit=2, offset=offset,
            with_payload=True, with_vectors=True,
        )
        for point in page:
            if not isinstance(point.vector, list):
                raise ValueError("本文只处理单个无名称的稠密向量字段")
            yield {"id": point.id, "vector": point.vector, "payload": point.payload}
        if next_offset is None:
            break
        offset = next_offset

path = Path("points_export.jsonl")
with path.open("w", encoding="utf-8") as stream:
    for record in records("source_demo"):
        stream.write(json.dumps(record, ensure_ascii=False, allow_nan=False) + "\n")

client.create_collection("rebuilt_demo", vectors_config=config)
with path.open(encoding="utf-8") as stream:
    for line in stream:
        record = json.loads(line)
        vector = record["vector"]
        if len(vector) != 3 or not all(math.isfinite(value) for value in vector):
            raise ValueError("向量维度或数值不合法")
        client.upsert("rebuilt_demo", points=[models.PointStruct(**record)], wait=True)

original = {item["id"]: item for item in records("source_demo")}
restored = {item["id"]: item for item in records("rebuilt_demo")}
assert len(original) == 3 and original.keys() == restored.keys()
for point_id, before in original.items():
    after = restored[point_id]
    assert before["payload"] == after["payload"]
    assert all(math.isclose(x, y, abs_tol=1e-6)
               for x, y in zip(before["vector"], after["vector"]))
print({"exported": len(original), "rebuilt": len(restored), "ids": sorted(restored)})
client.close()

官方客户端实现可核对 scroll 的 offset、limit、with_vectors 和返回的下一页游标，以及 upsert 的 wait 参数。scroll 是按记录遍历，不进行相似度估计，所以适合导出；检索 top-k 不能代替遍历全库。

怎样确认没有漏页或丢字段

这个固定示例应导出 3 条、重建 3 条，ID 为 1、2、3。因为每页限制为 2，拿到三条能检验分页推进；各条 payload 与向量也必须逐项一致。仅数量相同并不能证明恢复正确，两个不同 ID 集合也可能恰好一样大。

真实资料还应保存集合的维度、距离函数、向量字段名称、所用嵌入模型及版本；本文在两边显式采用 size=3 与 COSINE。使用其他模型或维度时替换完整配置，不能将三维教学向量用于真实语义库。

从演示迁移到真实库时的边界

• 数据持续变化：为导出过程固定来源范围与版本，记录开始/结束时间，并用对应 ID 集合核验；需要一致性备份时评估官方快照方案。
• 命名向量或稀疏向量：本函数会拒绝不同结构，需要按真实 schema 导出和恢复全部字段，不能简单取字典第一项。
• 数据量变大：使用有界批量写入并保存进度，本文逐条 upsert 是方便检查的教学方式，不是吞吐优化建议。
• 恢复失败：在新集合处理并核对，不覆盖原集合。只有数据和检索验证都通过才考虑切换读取入口。

导出的 JSONL 含原文和元数据，应保持与来源库相同的访问范围；恢复授权字段不等于应用已经执行了正确授权。先跑本地三条记录，确认 ID、向量、payload 往返一致，再为自己的集合补齐配置和一致性要求。