並發執行向量語義檢索

基本樣本

以下樣本並發檢索 5 組向量，使用 ThreadPoolExecutor 控制最多 5 個線程同時執行：

import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

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>"
INDEX = "<your-index>"
MAX_CONCURRENT = 5


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
    cfg.use_internal_endpoint = True  # 如需通過公網訪問，請將此處設定為False或刪除此行
    return oss_vectors.Client(cfg)


def run_query(client, query_vector, query_id, query_filter=None):
    """執行單條向量檢索"""
    request = oss_vectors.models.QueryVectorsRequest(
        bucket=BUCKET,
        index_name=INDEX,
        query_vector=query_vector,
        filter=query_filter,
        return_distance=True,
        return_metadata=True,
        top_k=10,
    )
    result = client.query_vectors(request)
    print(f"查詢 {query_id} 完成，status code: {result.status_code}")
    return {
        "query_id": query_id,
        "status_code": result.status_code,

        "vectors": [str(v) for v in result.vectors] if result.vectors else [ ],

    }


def batch_query(query_vectors):
    """批量並發執行多條向量檢索"""
    client = create_vector_client()

    results = [ ]


    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as executor:
        futures = {
            executor.submit(run_query, client, qv, idx): idx
            for idx, qv in enumerate(query_vectors)
        }
        for future in as_completed(futures):
            idx = futures[future]
            try:
                results.append(future.result())
            except Exception as e:
                print(f"查詢 {idx} 失敗: {e}")
                results.append({"query_id": idx, "error": str(e)})

    results.sort(key=lambda x: x["query_id"])
    output_path = Path("./query-results/sdk_batch_results.json")
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(results, ensure_ascii=False, indent=2))
    print(f"匯總結果已儲存到 {output_path}")
    return results


if __name__ == "__main__":
    # 樣本：5 組查詢向量（維度需與索引一致，此處以 128 維為例）
    query_vectors = [
        {"float32": [0.1] * 128},
        {"float32": [0.2] * 128},
        {"float32": [0.3] * 128},
        {"float32": [0.4] * 128},
        {"float32": [0.5] * 128},
    ]
    batch_query(query_vectors)

運行後輸出：

查詢 0 完成，status code: 200
查詢 2 完成，status code: 200
查詢 1 完成，status code: 200
查詢 4 完成，status code: 200
查詢 3 完成，status code: 200
匯總結果已儲存到 query-results/sdk_batch_results.json

說明：由於線程池並發執行，輸出順序可能與提交順序不同，但最終結果按 query_id 排序儲存。

帶過濾條件的並發檢索

實際業務中，不同查詢可能需要搭配不同的過濾條件。以下樣本為每條查詢指定獨立的 filter：

import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

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>"
INDEX = "<your-index>"
MAX_CONCURRENT = 5


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
    cfg.use_internal_endpoint = True  # 如需通過公網訪問，請將此處設定為False或刪除此行
    return oss_vectors.Client(cfg)


def run_query(client, query_vector, query_id, query_filter=None):
    """執行單條向量檢索"""
    request = oss_vectors.models.QueryVectorsRequest(
        bucket=BUCKET,
        index_name=INDEX,
        query_vector=query_vector,
        filter=query_filter,
        return_distance=True,
        return_metadata=True,
        top_k=10,
    )
    result = client.query_vectors(request)
    print(f"查詢 {query_id} 完成，status code: {result.status_code}")
    return {
        "query_id": query_id,
        "status_code": result.status_code,

        "vectors": [str(v) for v in result.vectors] if result.vectors else [ ],

    }


def batch_query(query_vectors):
    """批量並發執行多條向量檢索"""
    client = create_vector_client()

    results = [ ]


    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as executor:
        futures = {
            executor.submit(run_query, client, qv, idx): idx
            for idx, qv in enumerate(query_vectors)
        }
        for future in as_completed(futures):
            idx = futures[future]
            try:
                results.append(future.result())
            except Exception as e:
                print(f"查詢 {idx} 失敗: {e}")
                results.append({"query_id": idx, "error": str(e)})

    results.sort(key=lambda x: x["query_id"])
    output_path = Path("./query-results/sdk_batch_results.json")
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(results, ensure_ascii=False, indent=2))
    print(f"匯總結果已儲存到 {output_path}")
    return results
if __name__ == "__main__":
    tasks = [
        {
            "vector": {"float32": [0.1] * 128},
            "filter": {"$and": [{"type": {"$in": ["tutorial"]}}]},
        },
        {
            "vector": {"float32": [0.2] * 128},
            "filter": {"$and": [{"type": {"$nin": ["comedy", "documentary"]}}]},
        },
        {
            "vector": {"float32": [0.3] * 128},
            "filter": None,  # 不設過濾條件
        },
    ]

    client = create_vector_client()

    results = [ ]


    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as executor:
        futures = {
            executor.submit(
                run_query, client, t["vector"], idx, t["filter"]
            ): idx
            for idx, t in enumerate(tasks)
        }
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                print(f"查詢失敗: {e}")

    for r in sorted(results, key=lambda x: x["query_id"]):

        print(f"查詢 {r['query_id']}: 返回 {len(r.get('vectors', [ ]))} 條結果")

運行後輸出：

查詢 0 完成，status code: 200
查詢 1 完成，status code: 200
查詢 2 完成，status code: 200
查詢 0: 返回 10 條結果
查詢 1: 返回 10 條結果
查詢 2: 返回 10 條結果

基本樣本

以下樣本使用 sync.WaitGroup 和 channel 訊號量並發檢索 5 組向量：

package main

import (
	"context"
	"fmt"
	"log"
	"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>"
	indexName     = "<your-index>"
	maxConcurrent = 5
)

func main() {
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region).
		WithAccountId(accountId).
		// 如需通過公網訪問，請將此處設定為false或刪除此行
		WithUseInternalEndpoint(true)

	client := vectors.NewVectorsClient(cfg)

	// 5 組查詢向量（維度需與索引一致）

	queryVectors := [ ]map[string]any{


		{"float32": [ ]float32{0.1}},


		{"float32": [ ]float32{0.2}},


		{"float32": [ ]float32{0.3}},


		{"float32": [ ]float32{0.4}},


		{"float32": [ ]float32{0.5}},

	}

	var wg sync.WaitGroup
	sem := make(chan struct{}, maxConcurrent) // channel 訊號量控制並發數

	for i, qv := range queryVectors {
		wg.Add(1)
		sem <- struct{}{} // 擷取訊號量，達到上限時阻塞

		go func(idx int, queryVector map[string]any) {
			defer wg.Done()
			defer func() { <-sem }() // 釋放訊號量

			request := &vectors.QueryVectorsRequest{
				Bucket:         oss.Ptr(bucketName),
				IndexName:      oss.Ptr(indexName),
				QueryVector:    queryVector,
				ReturnMetadata: oss.Ptr(true),
				ReturnDistance:  oss.Ptr(true),
				TopK:           oss.Ptr(10),
			}

			result, err := client.QueryVectors(context.TODO(), request)
			if err != nil {
				log.Printf("查詢 %d 失敗: %v", idx, err)
				return
			}
			fmt.Printf("查詢 %d 完成，status code: %d\n",
				idx, result.StatusCode)
		}(i, qv)
	}

	wg.Wait()
	fmt.Println("全部查詢完成")
}

運行後輸出：

查詢 0 完成，status code: 200
查詢 2 完成，status code: 200
查詢 1 完成，status code: 200
查詢 3 完成，status code: 200
查詢 4 完成，status code: 200
全部查詢完成

帶過濾條件的並發檢索

package main

import (
	"context"
	"fmt"
	"log"
	"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"
)

type queryTask struct {
	vector map[string]any
	filter map[string]any
}

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

func main() {
	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(credentials.NewEnvironmentVariableCredentialsProvider()).
		WithRegion(region).
		WithAccountId(accountId).
		// 如需通過公網訪問，請將此處設定為false或刪除此行
		WithUseInternalEndpoint(true)

	client := vectors.NewVectorsClient(cfg)

	tasks := []queryTask{
		{
			vector: map[string]any{"float32": []float32{0.1}},
			filter: map[string]any{
				"$and": []map[string]any{
					{"type": map[string]any{"$in": []string{"tutorial"}}},
				},
			},
		},
		{
			vector: map[string]any{"float32": []float32{0.2}},
			filter: map[string]any{
				"$and": []map[string]any{
					{"type": map[string]any{"$nin": []string{"comedy", "documentary"}}},
				},
			},
		},
		{
			vector: map[string]any{"float32": []float32{0.3}},
			filter: nil, // 不設過濾條件
		},
	}

	var wg sync.WaitGroup
	sem := make(chan struct{}, maxConcurrent)

	for i, task := range tasks {
		wg.Add(1)
		sem <- struct{}{}

		go func(idx int, t queryTask) {
			defer wg.Done()
			defer func() { <-sem }()

			request := &vectors.QueryVectorsRequest{
				Bucket:         oss.Ptr(bucketName),
				IndexName:      oss.Ptr(indexName),
				QueryVector:    t.vector,
				ReturnMetadata: oss.Ptr(true),
				ReturnDistance:  oss.Ptr(true),
				TopK:           oss.Ptr(10),
			}
			if t.filter != nil {
				request.Filter = t.filter
			}

			result, err := client.QueryVectors(context.TODO(), request)
			if err != nil {
				log.Printf("查詢 %d 失敗: %v", idx, err)
				return
			}
			fmt.Printf("查詢 %d 完成，status code: %d\n", idx, result.StatusCode)
		}(i, task)
	}

	wg.Wait()
	fmt.Println("全部查詢完成")
}

運行後輸出：

查詢 0 完成，status code: 200
查詢 1 完成，status code: 200
查詢 2 完成，status code: 200
全部查詢完成