并发执行向量语义检索

基本示例

以下示例并发检索 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
    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
    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)

	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)

	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
全部查询完成