你将构建什么

本 notebook 使用 LlamaIndex 在 Qdrant 上实现生产风格的混合 RAG，专为多租户以及通过自定义分片进行横向扩展而设计。

• 混合搜索：稠密嵌入 + 稀疏 BM25，以获得更高的召回率和精确率。
• 多租户：使用 payload 过滤器和分片路由隔离租户。
• 自定义分片：让每个租户保持本地化，以提升性能和成本效率。
• 分布式 Qdrant：多节点部署并配置复制，以实现高可用性和吞吐量。

本 notebook 将介绍一个端到端的检索增强生成工作流，它使用 Qdrant 作为分布式混合搜索后端，并使用 LlamaIndex 作为编排层。你将构建一个结合稠密向量与稀疏信号的租户感知 RAG，使用过滤器按租户隔离数据，并使用自定义分片键对数据和查询进行路由，以实现扩展。

安装依赖

关于依赖项

• llama-index：用于摄取、索引和检索的编排层。
• llama-index-vector-stores-qdrant：支持混合搜索的 Qdrant 集成。
• fastembed：轻量且适合 CPU 的嵌入/稀疏模型

%pip install -U llama-index llama-index-vector-stores-qdrant fastembed

请确保已启动并运行分布式 Qdrant 集群。以下是一个 compose.yaml 文件：

services:
  qdrant_primary:
    image: "qdrant/qdrant:latest"
    ports:
      - "6333:6333"
    environment:
      QDRANT__CLUSTER__ENABLED: "true"
    command: ["./qdrant", "--uri", "http://qdrant_primary:6335"]
    restart: always
  qdrant_secondary:
    image: "qdrant/qdrant:latest"
    environment:
      QDRANT__CLUSTER__ENABLED: "true"
    command: ["./qdrant", "--bootstrap", "http://qdrant_primary:6335"]
    restart: always

导入和全局设置

设置与连接

• 嵌入：FastEmbedEmbedding('BAAI/bge-base-en-v1.5') 是一个紧凑、高质量的基线模型。
• 连接：QDRANT_URL 默认为 HTTP 端点；对于安全/云端环境，请设置 QDRANT_API_KEY。

import os

from qdrant_client import AsyncQdrantClient, QdrantClient
from qdrant_client import models

from llama_index.core import (
    Settings,
    VectorStoreIndex,
    Document,
    StorageContext,
)
from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.embeddings.fastembed import FastEmbedEmbedding

# Embeddings, small and fast
Settings.embed_model = FastEmbedEmbedding(model_name="BAAI/bge-base-en-v1.5")

# Qdrant connection, local by default, set QDRANT_URL and QDRANT_API_KEY for cloud
QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")

client: QdrantClient = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)
aclient: AsyncQdrantClient = AsyncQdrantClient(
    url=QDRANT_URL, api_key=QDRANT_API_KEY
)
COLLECTION = "hybrid_rag_multitenant_sharding_demo"

创建适用于分布式的集合

配置双向量模式（稠密 + 稀疏）

• 定义向量字段名：dense 用于嵌入，sparse 用于 BM25 风格信号。

• 稠密配置：

  • 在运行时通过探测 Settings.embed_model 确定嵌入维度（避免硬编码）。
  • 使用余弦距离衡量语义相似度。

• 稀疏配置：

  • 启用内存稀疏索引（on_disk=False）以支持混合评分。

• 这些设置建立了集合的双索引布局，稍后由 QdrantVectorStore 用于混合检索。

dense_vector_name = "dense"
dense_config = models.VectorParams(
    size=len(Settings.embed_model.get_text_embedding("probe")),
    distance=models.Distance.COSINE,
)
sparse_vector_name = "sparse"
sparse_config = models.SparseVectorParams(
    index=models.SparseIndexParams(on_disk=False)
)

分片键与选择器约定

• shard_keys：[‘tenant_a’, ‘tenant_b’] —— 与自定义分片一起使用的预定义分区，用于让每个租户保持本地化。
• payload_indexes：在 tenant_id 上建立关键字索引，以加速基于过滤器的查询。
• shard_key_selector_fn(tenant_id) -> tenant_id：返回同时用于写入和读取的分片键。

shard_keys = ["tenant_a", "tenant_b"]
payload_indexes = [
    {
        "field_name": "tenant_id",
        "field_schema": models.PayloadSchemaType.KEYWORD,
    }
]

def shard_key_selector_fn(tenant_id: str) -> models.ShardKeySelector:
    return tenant_id

初始化带自定义分片的 Qdrant 混合存储

此步骤会创建或连接到 COLLECTION 中指定的集合，并配置一个双向量混合存储：

• 混合搜索：enable_hybrid=True，并设置 dense_vector_name='dense' 和 sparse_vector_name='sparse'。

• 稠密配置：dense_config 使用余弦距离，并从 Settings.embed_model 推导维度大小。

• 稀疏配置：sparse_config 启用内存稀疏索引；fastembed_sparse_model='Qdrant/bm25' 提供 BM25 风格信号。

• 分布式拓扑：

  • sharding_method=Custom，并设置 shard_keys=['tenant_a','tenant_b']。
  • shard_key_selector_fn(tenant_id) -> tenant_id 同时路由写入和读取。
  • shard_number=6、replication_factor=2，以实现扩展和高可用性。

• Payload 索引：payload_indexes 加速对 tenant_id 的过滤。

幂等行为：如果集合不存在，向量存储会创建它，并在后续运行中复用它。

vector_store = QdrantVectorStore(
    collection_name=COLLECTION,
    client=client,
    aclient=aclient,
    dense_vector_name=dense_vector_name,
    sparse_vector_name=sparse_vector_name,
    enable_hybrid=True,
    dense_config=dense_config,
    sparse_config=sparse_config,
    fastembed_sparse_model="Qdrant/bm25",
    shard_number=6,
    sharding_method=models.ShardingMethod.CUSTOM,
    shard_key_selector_fn=shard_key_selector_fn,
    shard_keys=shard_keys,
    replication_factor=2,
    payload_indexes=payload_indexes,
)

准备多租户数据集

我们创建两个租户，每个租户包含少量文档。每个 Document 都带有 tenant_id、tags 和 doc_id。

数据集设计与可扩展性

我们模拟两个租户，每个租户包含几篇短文档。每个 Document 都带有：

• 用于隔离和分片路由的 tenant_id，
• 用于快速过滤和调试的 tags，
• 用于稠密/稀疏索引的 text 内容。

TENANT_DOCS: dict[str, list[Document]] = {
    "tenant_a": [
        Document(
            text="Solar panels reduce electricity bills and carbon footprint",
            metadata={"tenant_id": "tenant_a", "tags": ["energy", "solar"]},
        ),
        Document(
            text="Inverters convert DC power to AC for home appliances",
            metadata={"tenant_id": "tenant_a", "tags": ["energy", "hardware"]},
        ),
        Document(
            text="Net metering policies vary by region and utility provider",
            metadata={
                "tenant_id": "tenant_a",
                "tags": ["policy", "regulation"],
            },
        ),
    ],
    "tenant_b": [
        Document(
            text="Kubernetes orchestrates containers across a cluster",
            metadata={"tenant_id": "tenant_b", "tags": ["cloud", "k8s"]},
        ),
        Document(
            text="Service meshes add observability and traffic management",
            metadata={
                "tenant_id": "tenant_b",
                "tags": ["cloud", "networking"],
            },
        ),
        Document(
            text="Helm charts package and deploy Kubernetes applications",
            metadata={"tenant_id": "tenant_b", "tags": ["cloud", "devops"]},
        ),
    ],
}

使用分片键进行本地化摄取

这里我们使用当前启用的 Settings.embed_model 嵌入文本，然后使用 payload 和分片键更新插入每个点。这样可以让每个租户在集群中保持位于本地分片组内。

嵌入策略

• FastEmbed 让此演示对 CPU 友好。在生产环境中，请考虑使用服务化方案（例如 text-embedding-3-large 或内部模型），并缓存嵌入。
• 如果更改模型，请同步更新 dense_config.size，并考虑重建索引。
• 在 notebook 中避免每次运行都重新嵌入；请持久化或缓存，以加快迭代速度。

def create_dense_embeddings(docs: list[Document]) -> list[Document]:
    for doc in docs:
        doc.embedding = Settings.embed_model.get_text_embedding(doc.text)
    return docs

摄取流程与本地化保证

• 我们使用配置的嵌入模型（稠密）嵌入每个文档，并依赖向量存储构建稀疏表示。
• 写入使用 shard_identifier=tenant_id，确保文档位于预期的分片组中。

提示：对于大批量数据，优先使用异步摄取 API，并对文档进行分块，以控制背压。

for tenant_id, docs in TENANT_DOCS.items():
    docs = create_dense_embeddings(docs)
    await vector_store.async_add(docs, shard_identifier=tenant_id)

索引封装与可复用性

StorageContext.from_defaults(vector_store=vector_store) 将 Qdrant 集合绑定到 LlamaIndex 的 VectorStoreIndex，而无需重新摄取数据。

优势：

• 将同一个物理集合复用于多个检索器或查询流水线。
• 通过检索器配置（而非数据布局）切换检索模式（仅稠密、仅稀疏、混合）。
• 让摄取关注点（分片、复制）与应用查询逻辑解耦。

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_vector_store(
    vector_store, storage_context=storage_context
)

多租户检索

使用按租户限定的混合检索器，并让查询保持在分片本地。如果你希望在租户数据内部进一步过滤，也可以使用元数据过滤器。

混合模式的检索提示

• 设置 vector_store_query_mode=HYBRID 以结合稠密和稀疏检索。调整 similarity_top_k、sparse_top_k 和 hybrid_top_k。
• 传入 vector_store_kwargs=""shard_identifier": tenant_id"，让查询保持在租户的分片内。
• 在需要时添加元数据过滤器（例如针对 tenant_id 或 tags），进一步缩小候选范围。

from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.vector_stores.types import VectorStoreQueryMode

def create_retriever_for_tenant(tenant_id: str) -> VectorIndexRetriever:
    if tenant_id not in shard_keys:
        raise ValueError(
            f"Unknown tenant_id: {tenant_id}. Expected one of {shard_keys}"
        )
    return VectorIndexRetriever(
        index=index,
        vector_store_query_mode=VectorStoreQueryMode.HYBRID,
        similarity_top_k=5,
        sparse_top_k=5,
        hybrid_top_k=5,
        vector_store_kwargs={"shard_identifier": tenant_id},
    )

tenant_id = "tenant_b"
retriever = create_retriever_for_tenant(tenant_id)

query = "manage microservices traffic and observability"
results = retriever.retrieve(query)

print(f"Tenant: {tenant_id} | Query: {query}")
for i, r in enumerate(results, 1):
    meta = r.node.metadata
    print(
        f"{i}. score={r.score:.4f} | tags={meta.get('tags')} | text={r.node.get_content()}"
    )

Tenant: tenant_b | Query: manage microservices traffic and observability
1. score=4.6271 | tags=['cloud', 'networking'] | text=Service meshes add observability and traffic management
2. score=0.1213 | tags=['cloud', 'k8s'] | text=Kubernetes orchestrates containers across a cluster
3. score=0.0000 | tags=['cloud', 'devops'] | text=Helm charts package and deploy Kubernetes applications

解读结果

• 输出会显示混合分数、标签（元数据）以及匹配文本的片段。
• 通过切换 tenant_id 并观察结果仅来自该租户的文档，来验证租户隔离。