Object Storage Service:Pengambilan vektor skala besar dengan arsitektur multi-indeks

Buat indeks multi-tenant

Beri nama indeks dengan ID tenant sebagai sufiks untuk membuat indeks vektor khusus secara batch.

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()

# Buat indeks untuk tenant secara batch
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 {index_name} created, status_code={result.status_code}")

Output:

Index tenantcompanya created, status_code=200
Index tenantcompanyb created, status_code=200
Index tenantcompanyc created, status_code=200

Tulis data berdasarkan tenant

Tulis data untuk tenant berbeda ke indeks masing-masing.

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()

# Tulis data ke indeks Tenant A
result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
    bucket=BUCKET,
    index_name="tenantcompanya",
    vectors=[
        {
            "key": "faq_001",
            "data": {"float32": [0.1] * 1024},  # Dimensi vektor harus sesuai dengan indeks.
            "metadata": {"tenant": "company_a", "category": "faq"}
        }
    ]
))
print(f"Tenant A write completed, status_code={result.status_code}")

# Tulis data ke indeks Tenant B
result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
    bucket=BUCKET,
    index_name="tenantcompanyb",
    vectors=[
        {
            "key": "manual_001",
            "data": {"float32": [0.2] * 1024},  # Dimensi vektor harus sesuai dengan indeks.
            "metadata": {"tenant": "company_b", "category": "manual"}
        }
    ]
))
print(f"Tenant B write completed, status_code={result.status_code}")

Output:

Tenant A write completed, status_code=200
Tenant B write completed, status_code=200

Pencarian paralel dan penggabungan

Setelah mempartisi satu indeks besar menjadi beberapa indeks kecil, Anda dapat mengurangi waktu respons keseluruhan dengan mencari indeks secara paralel lalu menggabungkan dan mengurutkan hasilnya.

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):
    """Cari satu indeks."""
    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):
    """Cari beberapa indeks secara paralel dan gabungkan hasilnya."""
    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"Index {result['index']} returned {len(result['vectors'])} results")
            all_vectors.extend(result["vectors"])

    # Urutkan berdasarkan jarak secara ascending (jarak lebih kecil berarti kemiripan lebih tinggi) dan ambil hasil top-K global.
    all_vectors.sort(key=lambda v: v.get("distance", float("inf")))
    return all_vectors[:top_k]


# Cari tiga indeks terpartisi secara paralel.
indices = ["tenantcompanya", "tenantcompanyb", "tenantcompanyc"]
query_vec = {"float32": [0.1] * 1024}  # Dimensi vektor harus sesuai dengan indeks.

results = parallel_search(indices, query_vec, top_k=5)
print(f"\nGlobal Top-5 after merging:")
for v in results:
    print(f"  key={v.get('key')}, distance={v.get('distance')}, metadata={v.get('metadata')}")

Output:

Index tenantcompanya returned 1 result
Index tenantcompanyb returned 1 result
Index tenantcompanyc returned 0 results

Global Top-5 after merging:
  key=faq_001, distance=0.0, metadata={'tenant': 'company_a', 'category': 'faq'}
  key=manual_001, distance=0.19999998807907104, metadata={'tenant': 'company_b', 'category': 'manual'}

Catatan: Setelah pencarian paralel di beberapa indeks, hasilnya digabungkan dan diurutkan di sisi client berdasarkan jarak. Untuk presisi lebih tinggi, Anda dapat menggunakan model Rerank untuk penyusunan ulang sekunder.

Buat indeks multi-tenant

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)

	// Buat indeks untuk tenant secara batch.
	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("Index %s creation failed: %v", indexName, err)
			continue
		}
		fmt.Printf("Index %s created, status_code=%d\n", indexName, result.StatusCode)
	}
}

Output:

Index tenantcompanya created, status_code=200
Index tenantcompanyb created, status_code=200
Index tenantcompanyc created, status_code=200

Pencarian paralel dan penggabungan

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("Index %s query failed: %v", idx, err)
				return
			}
			fmt.Printf("Index %s returned %d results\n", idx, len(result.Vectors))

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

	wg.Wait()

	// Urutkan berdasarkan jarak secara ascending dan ambil hasil top-K global.
	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("\nGlobal Top-%d after merging:\n", topK)
	for _, v := range allVectors[:topK] {
		fmt.Printf("  key=%v, distance=%v, metadata=%v\n", v["key"], v["distance"], v["metadata"])
	}
}

Output:

Index tenantcompanya returned 1 result
Index tenantcompanyc returned 0 results
Index tenantcompanyb returned 1 result

Global Top-2 after merging:
  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]