Vector 基于多索引表架构的大规模向量检索

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

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

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

多索引架构的优势

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

通过 CLI 按租户导入数据

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

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

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

1. 已配置环境变量 OSS_ACCESS_KEY_ID、OSS_ACCESS_KEY_SECRET 和 DASHSCOPE_API_KEY。
2. 已创建向量 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-k5--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 ossimport alibabacloud_oss_v2.vectors as oss_vectorsACCOUNT_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_providercfg.region = REGIONcfg.account_id = ACCOUNT_IDreturn 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 ossimport alibabacloud_oss_v2.vectors as oss_vectorsACCOUNT_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_providercfg.region = REGIONcfg.account_id = ACCOUNT_IDreturn 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_completedimport alibabacloud_oss_v2 as ossimport alibabacloud_oss_v2.vectors as oss_vectorsACCOUNT_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_providercfg.region = REGIONcfg.account_id = ACCOUNT_IDreturn 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): idxfor 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 升序排序（距离越小越相似），取全局 TopKall_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 mainimport ("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>")funcmain() {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=%dn", indexName, result.StatusCode)}}

运行后输出：

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

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

package mainimport ("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)funcmakeVector(val float32, dim int) [ ]float32 {v := make([ ]float32, dim)for i := range v {v[i] = val}return v}funcmain() {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.Mutexvar allVectors [ ]map[string]anyvar wg sync.WaitGroupfor _, indexName := range indices {wg.Add(1)gofunc(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 升序排序，取全局 Top5sort.Slice(allVectors, func(i, j int)bool {di, _ := allVectors[i]["distance"].(float64)dj, _ := allVectors[j]["distance"].(float64)return di < dj})topK := 5iflen(allVectors) < topK {topK = len(allVectors)}fmt.Printf("n合并后全局 Top%d：n", topK)for _, v := range allVectors[:topK] {fmt.Printf("key=%v, distance=%v, metadata=%vn", 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]

最佳实践

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