14.09-Qdrant实践

要点

  • Qdrant 是 Rust 写的开源向量数据库——性能好、过滤能力强
  • 支持稠密向量 + 稀疏向量混合检索
  • 丰富的过滤语法，适合复杂的多租户和权限控制场景
  • 部署方式：Docker（开发）、Kubernetes（生产）、Qdrant Cloud（托管）
  • TypeScript SDK 完善，API 设计直观
  • 适合中等规模、需要复杂过滤和混合检索的场景

1. 部署

Docker（本地开发）

docker run -d -p 6333:6333 -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage \
  qdrant/qdrant
  • 6333：REST API
  • 6334：gRPC API

Docker Compose

# docker-compose.yml
services:
  qdrant:
    image: qdrant/qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - ./qdrant_storage:/qdrant/storage
    environment:
      - QDRANT__SERVICE__GRPC_PORT=6334

Qdrant Cloud（生产）

const client = new QdrantClient({
  url: 'https://your-cluster.qdrant.io',
  apiKey: process.env.QDRANT_API_KEY,
})

2. 创建 Collection

import { QdrantClient } from '@qdrant/js-client-rest'
 
const client = new QdrantClient({
  url: process.env.QDRANT_URL ?? 'http://localhost:6333',
  apiKey: process.env.QDRANT_API_KEY,
})
 
// 创建 collection
await client.createCollection('documents', {
  vectors: {
    size: 768,           // 向量维度，要和 embedding 模型匹配
    distance: 'Cosine',  // Cosine / Euclid / Dot
  },
  // 可选：稀疏向量（用于混合检索）
  sparse_vectors: {
    'text-sparse': {
      index: { on_disk: true },
    },
  },
  // 优化：payload 索引
  optimizers_config: {
    default_segment_number: 2,
    memmap_threshold: 20000,
  },
})
 
// 为常用的过滤字段创建索引（提升过滤查询性能）
await client.createPayloadIndex('documents', {
  field_name: 'document_id',
  field_schema: 'keyword',
})
 
await client.createPayloadIndex('documents', {
  field_name: 'user_id',
  field_schema: 'keyword',
})
 
await client.createPayloadIndex('documents', {
  field_name: 'created_at',
  field_schema: 'datetime',
})

3. 写入向量

单条写入

await client.upsert('documents', {
  wait: true,
  points: [
    {
      id: 'doc-1-chunk-0',
      vector: embedding,
      payload: {
        text: 'chunk 内容',
        document_id: 'doc-1',
        document_title: '文档标题',
        chunk_index: 0,
        user_id: 'user-123',
        tenant_id: 'tenant-456',
        created_at: new Date().toISOString(),
      },
    },
  ],
})

批量写入

export async function upsertChunks(
  collectionName: string,
  chunks: Array<{
    id: string
    vector: number[]
    payload: Record<string, unknown>
  }>
): Promise<void> {
  // Qdrant 单次批量写入限制约 100-1000 条（取决于 payload 大小）
  const BATCH_SIZE = 100
 
  for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
    const batch = chunks.slice(i, i + BATCH_SIZE)
 
    await client.upsert(collectionName, {
      wait: true,
      points: batch.map((chunk) => ({
        id: chunk.id,
        vector: chunk.vector,
        payload: chunk.payload,
      })),
    })
  }
}

带稀疏向量的写入（混合检索）

await client.upsert('documents', {
  wait: true,
  points: [
    {
      id: 'doc-1-chunk-0',
      vector: {
        dense: denseEmbedding,       // 768 维稠密向量
        'text-sparse': sparseVector, // 稀疏向量
      },
      payload: { text: '...', document_id: 'doc-1' },
    },
  ],
})

稀疏向量可以用 SPLADE、BM25 等方法生成。

4. 相似度查询

基础查询

const results = await client.query('documents', {
  query: queryVector,
  limit: 5,
  with_payload: true,
})
 
// results 结构
// {
//   points: [
//     { id: '...', version: 1, score: 0.92, payload: {...} },
//     ...
//   ]
// }

带过滤的查询

const results = await client.query('documents', {
  query: queryVector,
  limit: 5,
  filter: {
    must: [
      // AND 条件
      { key: 'tenant_id', match: { value: 'tenant-456' } },
      { key: 'created_at', range: { gte: '2024-01-01' } },
    ],
    should: [
      // OR 条件
      { key: 'document_id', match: { value: 'doc-1' } },
      { key: 'document_id', match: { value: 'doc-2' } },
    ],
    must_not: [
      // NOT 条件
      { key: 'status', match: { value: 'archived' } },
    ],
  },
})

过滤语法

// 关键词匹配
{ key: 'status', match: { value: 'active' } }
 
// 多值匹配（IN）
{ key: 'category', match: { any: ['tech', 'news'] } }
 
// 数值范围
{ key: 'score', range: { gte: 0.5, lte: 1.0 } }
 
// 日期范围
{ key: 'created_at', range: { gte: '2024-01-01', lt: '2025-01-01' } }
 
// 嵌套字段
{ key: 'metadata.author', match: { value: 'alice' } }
 
// 全文匹配（需要对字段建立全文索引）
{ key: 'content', match: { text: '关键词' } }

5. 混合检索

Qdrant 支持稠密 + 稀疏向量融合。

// 混合查询
const results = await client.query('documents', {
  prefetch: [
    // 稠密向量检索
    {
      query: denseQueryVector,
      using: 'dense',
      limit: 20,
    },
    // 稀疏向量检索
    {
      query: sparseQueryVector,
      using: 'text-sparse',
      limit: 20,
    },
  ],
  // 融合策略
  query: {
    fusion: 'rrf',  // Reciprocal Rank Fusion
  },
  limit: 5,
})

融合策略：

  • rrf（Reciprocal Rank Fusion）：按排名融合，不需要分数归一化
  • dbsf（Distributed Border Score Fusion）：按分数融合，需要分数在同一尺度

6. Scroll 遍历

当需要遍历所有数据时（如重建索引），用 scroll。

async function* scrollAll(collectionName: string) {
  let offset: string | undefined
 
  while (true) {
    const result = await client.scroll(collectionName, {
      limit: 100,
      offset,
      with_payload: true,
      with_vector: true,
    })
 
    for (const point of result.points) {
      yield point
    }
 
    if (!result.next_page_offset) break
    offset = result.next_page_offset
  }
}
 
// 使用
for await (const point of scrollAll('documents')) {
  console.log(point.id, point.score)
}

7. 多租户隔离

方案一：Payload 过滤

所有租户的数据在同一个 collection 里，用 tenant_id 字段过滤。

const results = await client.query('documents', {
  query: queryVector,
  filter: {
    must: [{ key: 'tenant_id', match: { value: user.tenantId } }],
  },
})

优点：管理简单。 缺点：依赖代码正确添加过滤条件。

方案二：Collection 隔离

每个租户一个独立的 collection。

const collectionName = `documents_${tenantId}`
 
// 首次使用时创建
const collections = await client.getCollections()
if (!collections.collections.find((c) => c.name === collectionName)) {
  await client.createCollection(collectionName, {
    vectors: { size: 768, distance: 'Cosine' },
  })
}
 
const results = await client.query(collectionName, { query: queryVector })

优点：物理隔离，更安全。 缺点：collection 数量多时管理复杂。

推荐

  • 租户数量少（< 100）→ Collection 隔离
  • 租户数量多 → Payload 过滤 + 严格权限检查

8. 监控

健康检查

// src/middleware/qdrant-health.ts
app.get('/health/qdrant', async (c) => {
  try {
    const result = await client.getCollections()
    return c.json({ status: 'ok', collections: result.collections.length })
  } catch (err) {
    return c.json({ status: 'error', error: String(err) }, 503)
  }
})

性能监控

Qdrant 暴露 Prometheus metrics 在 http://localhost:6333/metrics。

关键指标：

  • qdrant_rest_responses_total：请求总数
  • qdrant_rest_responses_duration_seconds：请求延迟
  • qdrant_collections_total：collection 数量
  • qdrant_points_count：向量总数

9. 备份和恢复

# 备份：Qdrant 提供 snapshot API
curl -X POST http://localhost:6333/collections/documents/snapshots
 
# 恢复
curl -X PUT http://localhost:6333/collections/documents/snapshots \
  -H "Content-Type: application/json" \
  -d '{"location": "file:///path/to/snapshot"}'

或者直接用 Docker volume 备份：

# 停止服务
docker compose stop qdrant
 
# 备份存储目录
tar -czf qdrant-backup.tar.gz ./qdrant_storage
 
# 恢复
tar -xzf qdrant-backup.tar.gz -C ./qdrant_storage
 
# 启动
docker compose start qdrant

10. 一个完整的 Qdrant Store 实现

// src/services/rag/qdrant-store.ts
import { QdrantClient } from '@qdrant/js-client-rest'
 
export class QdrantVectorStore {
  private client: QdrantClient
 
  constructor() {
    this.client = new QdrantClient({
      url: process.env.QDRANT_URL ?? 'http://localhost:6333',
      apiKey: process.env.QDRANT_API_KEY,
    })
  }
 
  async initCollection(name: string, dimensions: number): Promise<void> {
    const collections = await this.client.getCollections()
    if (!collections.collections.find((c) => c.name === name)) {
      await this.client.createCollection(name, {
        vectors: { size: dimensions, distance: 'Cosine' },
      })
    }
  }
 
  async upsertChunks(collectionName: string, chunks: ChunkWithEmbedding[]): Promise<void> {
    const BATCH_SIZE = 100
    for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
      const batch = chunks.slice(i, i + BATCH_SIZE)
      await this.client.upsert(collectionName, {
        wait: true,
        points: batch.map((chunk) => ({
          id: chunk.id,
          vector: chunk.embedding,
          payload: {
            text: chunk.content,
            ...chunk.metadata,
          },
        })),
      })
    }
  }
 
  async search(
    collectionName: string,
    queryVector: number[],
    options: {
      topK?: number
      tenantId?: string
      userId?: string
    } = {}
  ): Promise<SearchResult[]> {
    const { topK = 5, tenantId, userId } = options
 
    const filter: any = { must: [] }
    if (tenantId) filter.must.push({ key: 'tenant_id', match: { value: tenantId } })
    if (userId) filter.must.push({ key: 'user_id', match: { value: userId } })
 
    const results = await this.client.query(collectionName, {
      query: queryVector,
      limit: topK,
      filter: filter.must.length > 0 ? filter : undefined,
      with_payload: true,
    })
 
    return results.points.map((p) => ({
      id: String(p.id),
      score: p.score,
      content: p.payload?.text,
      metadata: p.payload ?? {},
    }))
  }
 
  async deleteDocument(collectionName: string, documentId: string): Promise<void> {
    await this.client.delete(collectionName, {
      filter: { must: [{ key: 'document_id', match: { value: documentId } }] },
    })
  }
}

总结

回顾这一节的要点：

  • Qdrant 是 Rust 写的开源向量数据库，性能和过滤能力领先
  • 部署：Docker（开发）、K8s（生产）、Qdrant Cloud（托管）
  • 丰富的过滤语法 + payload 索引提升过滤查询性能
  • 支持稠密 + 稀疏向量混合检索
  • 多租户隔离：payload 过滤（大规模）或 collection 隔离（小规模）
  • 监控：Prometheus metrics + 健康检查
  • 适合中等规模、需要复杂过滤的场景

下一篇讲 Milvus 实践——面向超大规模的向量数据库。