幫助中心
首頁 Object Storage Service 用户指南 OSS Vectors OSS Vectors 最佳實務 基於多索引表架構的大規模向量檢索

基於多索引表架構的大規模向量檢索

更新時間：
Copy as MD
產品

隨著 RAG 和語義搜尋需求的增長，向量檢索系統常面臨兩類挑戰：

  • 多租戶隔離：SaaS 服務商為大量企業客戶提供知識庫，或企業內部各部門擁有獨立知識庫，要求資料嚴格隔離。

  • 超大規模資料：單索引資料量達到千萬甚至億級時，檢索延遲顯著增加，難以滿足即時性要求。

OSS Vectors 支援同一帳號在同一地區建立大量向量索引（Index）。通過多索引架構，可以按租戶或業務維度拆分資料，兼顧隔離性和檢索效能。

image.png

多索引架構的優勢

  • 資料隔離：不同租戶或業務的資料存放在獨立索引中，從底層避免跨租戶資料泄露。

  • 檢索提速：將大表拆分為多張小表，縮小單次檢索範圍；配合并發檢索多個索引再合并結果，可有效降低總響應耗時。

  • 營運靈活：不同索引可獨立配置維度、模型、相似性演算法。刪除某個租戶的資料只需刪除對應索引，無需逐條過濾刪除。

通過 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 刪除對應索引，無需逐條過濾刪除。

上一篇：構建向量與原始檔案的映射關係下一篇：OSS Tables