基于多索引表架构的大规模向量检索-对象存储(OSS)-阿里云帮助中心

随着 RAG 和语义搜索需求的增长，向量检索系统常面临两类挑战：

多租户隔离：SaaS 服务商为大量企业客户提供知识库，或企业内部各部门拥有独立知识库，要求数据严格隔离。
超大规模数据：单索引数据量达到千万甚至亿级时，检索延迟显著增加，难以满足实时性要求。

OSS 向量 Bucket 支持同一账号在同一地域创建大量向量索引（Index）。通过多索引架构，可以按租户或业务维度拆分数据，兼顾隔离性和检索性能。

多索引架构的优势

数据隔离：不同租户或业务的数据存放在独立索引中，从底层避免跨租户数据泄露。
检索提速：将大表拆分为多张小表，缩小单次检索范围；配合并发检索多个索引再合并结果，可有效降低总响应耗时。
运维灵活：不同索引可独立配置维度、模型、相似度算法。删除某个租户的数据只需删除对应索引，无需逐条过滤删除。

通过 CLI 按租户导入数据

oss-vectors-embed CLI 工具支持将指定文件写入指定索引，实现按租户或业务维度的定向导入。

安装方式请参见使用OSS Vectors Embed CLI工具写入和检索向量数据。

开始前，请确保满足以下条件：

已配置环境变量 OSS_ACCESS_KEY_ID、OSS_ACCESS_KEY_SECRET 和 DASHSCOPE_API_KEY。
已创建向量 Bucket 和各租户对应的向量索引。

将以下示例中的占位符替换为实际值：

占位符	说明
`<your-account-id>`	阿里云账号 ID
`<your-vector-bucket>`	向量 Bucket 名称

按租户写入不同索引

将不同租户的数据写入各自独立的索引，实现数据隔离。

# 将租户 A 的文档写入租户 A 的索引
oss-vectors-embed \
  --account-id "<your-account-id>" \
  --vectors-region cn-hangzhou \
  put \
  --vector-bucket-name "<your-vector-bucket>" \
  --index-name "tenantcompanya" \
  --model-id text-embedding-v4 \
  --text-value "租户A的知识库文档内容" \
  --key "doc_001" \
  --metadata '{"tenant": "company_a", "category": "faq"}'

# 将租户 B 的文档写入租户 B 的索引
oss-vectors-embed \
  --account-id "<your-account-id>" \
  --vectors-region cn-hangzhou \
  put \
  --vector-bucket-name "<your-vector-bucket>" \
  --index-name "tenantcompanyb" \
  --model-id text-embedding-v4 \
  --text-value "租户B的知识库文档内容" \
  --key "doc_001" \
  --metadata '{"tenant": "company_b", "category": "manual"}'

按租户定向检索

检索时只查询目标租户的索引，天然实现数据隔离。

# 仅在租户 A 的索引中检索
oss-vectors-embed \
  --account-id "<your-account-id>" \
  --vectors-region cn-hangzhou \
  query \
  --vector-bucket-name "<your-vector-bucket>" \
  --index-name "tenantcompanya" \
  --model-id text-embedding-v4 \
  --text-value "常见问题" \
  --top-k 5 \
  --return-metadata

通过 SDK 构建多索引架构

Python SDK

开始前请安装 alibabacloud-oss-v2 SDK：

pip install alibabacloud-oss-v2

确保已配置环境变量 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET。

创建多租户索引

以租户 ID 为后缀命名索引，批量创建独立的向量索引。

import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors

ACCOUNT_ID = "<your-account-id>"
REGION = "cn-hangzhou"
BUCKET = "<your-vector-bucket>"


def create_vector_client():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = REGION
    cfg.account_id = ACCOUNT_ID
    return oss_vectors.Client(cfg)


client = create_vector_client()

# 批量为租户创建索引
tenant_ids = ["companya", "companyb", "companyc"]

for tenant_id in tenant_ids:
    index_name = f"tenant{tenant_id}"
    result = client.put_vector_index(oss_vectors.models.PutVectorIndexRequest(
        bucket=BUCKET,
        index_name=index_name,
        dimension=1024,
        data_type="float32",
        distance_metric="cosine",
    ))
    print(f"索引 {index_name} 创建完成，status_code={result.status_code}")

运行后输出：

索引 tenantcompanya 创建完成，status_code=200
索引 tenantcompanyb 创建完成，status_code=200
索引 tenantcompanyc 创建完成，status_code=200

按租户写入数据

将不同租户的数据写入各自的索引。

import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors

ACCOUNT_ID = "<your-account-id>"
REGION = "cn-hangzhou"
BUCKET = "<your-vector-bucket>"


def create_vector_client():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = REGION
    cfg.account_id = ACCOUNT_ID
    return oss_vectors.Client(cfg)


client = create_vector_client()

# 向租户 A 的索引写入数据
result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
    bucket=BUCKET,
    index_name="tenantcompanya",
    vectors=[
        {
            "key": "faq_001",
            "data": {"float32": [0.1] * 1024},  # 向量维度需与索引一致
            "metadata": {"tenant": "company_a", "category": "faq"}
        }
    ]
))
print(f"租户 A 写入完成，status_code={result.status_code}")

# 向租户 B 的索引写入数据
result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
    bucket=BUCKET,
    index_name="tenantcompanyb",
    vectors=[
        {
            "key": "manual_001",
            "data": {"float32": [0.2] * 1024},  # 向量维度需与索引一致
            "metadata": {"tenant": "company_b", "category": "manual"}
        }
    ]
))
print(f"租户 B 写入完成，status_code={result.status_code}")

运行后输出：

租户 A 写入完成，status_code=200
租户 B 写入完成，status_code=200

并发检索多个索引并合并结果

将大表拆分为多张小表后，通过并发检索多个索引再合并排序，降低总响应耗时。

from concurrent.futures import ThreadPoolExecutor, as_completed

import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors

ACCOUNT_ID = "<your-account-id>"
REGION = "cn-hangzhou"
BUCKET = "<your-vector-bucket>"


def create_vector_client():
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = REGION
    cfg.account_id = ACCOUNT_ID
    return oss_vectors.Client(cfg)


def search_index(client, index_name, query_vector, top_k=10):
    """检索单个索引"""
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=BUCKET,
        index_name=index_name,
        query_vector=query_vector,
        return_metadata=True,
        return_distance=True,
        top_k=top_k,
    ))
    return {
        "index": index_name,
        "status_code": result.status_code,

        "vectors": result.vectors or [ ],

    }


def parallel_search(index_names, query_vector, top_k=10):
    """并发检索多个索引并合并结果"""
    client = create_vector_client()

    all_vectors = [ ]


    with ThreadPoolExecutor(max_workers=len(index_names)) as executor:
        futures = {
            executor.submit(search_index, client, idx, query_vector, top_k): idx
            for idx in index_names
        }
        for future in as_completed(futures):
            result = future.result()
            print(f"索引 {result['index']} 返回 {len(result['vectors'])} 条结果")
            all_vectors.extend(result["vectors"])

    # 按 distance 升序排序（距离越小越相似），取全局 TopK
    all_vectors.sort(key=lambda v: v.get("distance", float("inf")))
    return all_vectors[:top_k]


# 并发检索 3 个分表索引
indices = ["tenantcompanya", "tenantcompanyb", "tenantcompanyc"]
query_vec = {"float32": [0.1] * 1024}  # 向量维度需与索引一致

results = parallel_search(indices, query_vec, top_k=5)
print(f"\n合并后全局 Top5：")
for v in results:
    print(f"  key={v.get('key')}, distance={v.get('distance')}, metadata={v.get('metadata')}")

运行后输出：

索引 tenantcompanya 返回 1 条结果
索引 tenantcompanyb 返回 1 条结果
索引 tenantcompanyc 返回 0 条结果

合并后全局 Top5：
  key=faq_001, distance=0.0, metadata={'tenant': 'company_a', 'category': 'faq'}
  key=manual_001, distance=0.19999998807907104, metadata={'tenant': 'company_b', 'category': 'manual'}

说明：并发检索多个索引后，在客户端按 distance 排序合并结果。如需更高精度，可引入 Rerank 模型进行二次精排。

Go SDK

开始前请安装 alibabacloud-oss-go-sdk-v2 SDK：

go get github.com/aliyun/alibabacloud-oss-go-sdk-v2

确保已配置环境变量 OSS_ACCESS_KEY_ID 和 OSS_ACCESS_KEY_SECRET。

创建多租户索引

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/vectors"
)

const (
	region     = "cn-hangzhou"
	bucketName = "<your-vector-bucket>"
	accountId  = "<your-account-id>"
)

func main() {
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region).
		WithAccountId(accountId)

	client := vectors.NewVectorsClient(cfg)

	// 批量为租户创建索引

	tenantIDs := [ ]string{"companya", "companyb", "companyc"}


	for _, tenantID := range tenantIDs {
		indexName := fmt.Sprintf("tenant%s", tenantID)
		result, err := client.PutVectorIndex(context.TODO(), &vectors.PutVectorIndexRequest{
			Bucket:         oss.Ptr(bucketName),
			IndexName:      oss.Ptr(indexName),
			Dimension:      oss.Ptr(1024),
			DataType:       oss.Ptr("float32"),
			DistanceMetric: oss.Ptr("cosine"),
		})
		if err != nil {
			log.Printf("索引 %s 创建失败: %v", indexName, err)
			continue
		}
		fmt.Printf("索引 %s 创建完成，status_code=%d\n", indexName, result.StatusCode)
	}
}

运行后输出：

索引 tenantcompanya 创建完成，status_code=200
索引 tenantcompanyb 创建完成，status_code=200
索引 tenantcompanyc 创建完成，status_code=200

并发检索多个索引并合并结果

package main

import (
	"context"
	"fmt"
	"log"
	"sort"
	"sync"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/vectors"
)

const (
	region     = "cn-hangzhou"
	bucketName = "<your-vector-bucket>"
	accountId  = "<your-account-id>"
	dimension  = 1024
)


func makeVector(val float32, dim int) [ ]float32 {


	v := make([ ]float32, dim)

	for i := range v {
		v[i] = val
	}
	return v
}

func main() {
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region).
		WithAccountId(accountId)

	client := vectors.NewVectorsClient(cfg)


	indices := [ ]string{"tenantcompanya", "tenantcompanyb", "tenantcompanyc"}

	queryVector := map[string]any{"float32": makeVector(0.1, dimension)}

	var mu sync.Mutex

	var allVectors [ ]map[string]any

	var wg sync.WaitGroup

	for _, indexName := range indices {
		wg.Add(1)
		go func(idx string) {
			defer wg.Done()

			result, err := client.QueryVectors(context.TODO(), &vectors.QueryVectorsRequest{
				Bucket:         oss.Ptr(bucketName),
				IndexName:      oss.Ptr(idx),
				QueryVector:    queryVector,
				ReturnMetadata: oss.Ptr(true),
				ReturnDistance:  oss.Ptr(true),
				TopK:           oss.Ptr(10),
			})
			if err != nil {
				log.Printf("索引 %s 检索失败: %v", idx, err)
				return
			}
			fmt.Printf("索引 %s 返回 %d 条结果\n", idx, len(result.Vectors))

			mu.Lock()
			allVectors = append(allVectors, result.Vectors...)
			mu.Unlock()
		}(indexName)
	}

	wg.Wait()

	// 按 distance 升序排序，取全局 Top5
	sort.Slice(allVectors, func(i, j int) bool {
		di, _ := allVectors[i]["distance"].(float64)
		dj, _ := allVectors[j]["distance"].(float64)
		return di < dj
	})

	topK := 5
	if len(allVectors) < topK {
		topK = len(allVectors)
	}

	fmt.Printf("\n合并后全局 Top%d：\n", topK)
	for _, v := range allVectors[:topK] {
		fmt.Printf("  key=%v, distance=%v, metadata=%v\n", v["key"], v["distance"], v["metadata"])
	}
}

运行后输出：

索引 tenantcompanya 返回 1 条结果
索引 tenantcompanyc 返回 0 条结果
索引 tenantcompanyb 返回 1 条结果

合并后全局 Top2：
  key=faq_001, distance=0, metadata=map[category:faq tenant:company_a]
  key=manual_001, distance=0.19999998807907104, metadata=map[category:manual tenant:company_b]

最佳实践

索引命名规范：以租户 ID 或业务维度作为索引名称的后缀（如 tenant{tenantid}）。索引名仅支持小写字母和数字，不支持下划线和连字符。
租户数较多时：直接利用索引名称进行逻辑隔离。OSS 向量索引的创建是秒级的，管理开销极低。
追求极低延迟时：当单索引超过千万级数据量，按业务逻辑（如时间、类别）进行水平拆分，通过并发检索多个索引再合并结果。
结果重排（Rerank）：多索引表的检索结果合并后，可以根据距离相似度进行简单重排，也可引入 Rerank 模型进行二次排序。
索引清理：删除某个租户或业务的数据只需调用 DeleteVectorIndex 删除对应索引，无需逐条过滤删除。