Run concurrent vector semantic searches-Object Storage Service(OSS)-阿里云帮助中心

Choose a concurrency method

Method	Scenarios	Features
CLI concurrency	For operations scripting, one-time batch retrieval, and quick validation without writing code.	Automatically embeds input text; does not require managing vector dimensions; and is implementable with shell scripts.
SDK concurrency	Business service integration, fine-grained control (filter conditions, result post-processing), and high-performance backends.	Direct API calls with support for filter conditions; supports client reuse; available for Python and Go.

How to choose:

To get results for a batch of query texts quickly without writing code, use CLI concurrency.
To embed vector search into your application logic when developing a business service, use SDK concurrency.

Note: The CLI method includes a built-in embedding model, allowing you to search directly with text. The SDK method requires you to provide pre-generated vectors, making it suitable for scenarios where you already have an embedding workflow.

Run concurrent searches by using the CLI

CLI concurrency runs searches in parallel by launching multiple oss-vectors-embed query processes. The following sections provide three implementation methods, ordered by increasing complexity.

Before you begin, make sure you meet the following requirements:

Install the OSS Vectors Embed CLI. For installation instructions, see Use the OSS Vectors Embed CLI tool to write and retrieve vector data.
Configure the OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, and DASHSCOPE_API_KEY environment variables.
Create a vector bucket and a vector index, and ensure that the index dimensions match the output dimensions of the embedding model you are using.

Replace the placeholders in the following examples with your actual values:

Placeholder	Description
`<your-account-id>`	Your Alibaba Cloud account ID.
`<your-vector-bucket>`	The name of your vector bucket.
`<your-index>`	The name of your vector index.

Use xargs for quick concurrency

For simple concurrency without complex flow control, xargs -P provides single-command CLI concurrency.

cat queries.txt | xargs -P 5 -I {} \
  oss-vectors-embed \
    --account-id "<your-account-id>" \
    --vectors-region "cn-hangzhou" \
    query \
    --vector-bucket-name "<your-vector-bucket>" \
    --index-name "<your-index>" \
    --model-id text-embedding-v4 \
    --text-value "{}" \
    --top-k 10

The -P 5 flag runs up to five processes in parallel. The search results are printed directly to the terminal, which is ideal for quick validation. To save the results, you can redirect the output to a file.

Use shell background jobs for concurrency

To run multiple query commands concurrently, use the & operator for background execution and wait to block until all complete. This method works well for a small number of queries (fewer than 10).

#!/bin/bash

ACCOUNT_ID="<your-account-id>"
REGION="cn-hangzhou"
BUCKET="<your-vector-bucket>"
INDEX="<your-index>"
MODEL="text-embedding-v4"

queries=(
  "How to configure a lifecycle rule"
  "What are the storage classes for object storage"
  "How to set up cross-region replication"
)

mkdir -p ./query-results

# Start all queries concurrently and write each result to a separate file
for i in "${!queries[@]}"; do
  oss-vectors-embed \
    --account-id "$ACCOUNT_ID" \
    --vectors-region "$REGION" \
    query \
    --vector-bucket-name "$BUCKET" \
    --index-name "$INDEX" \
    --model-id "$MODEL" \
    --text-value "${queries[$i]}" \
    --top-k 10 \
    --return-metadata \
    > "./query-results/result_${i}.json" 2>&1 &
done

wait
echo "All queries completed. Results are saved in ./query-results/"

Sample output:

All queries completed. Results are saved in ./query-results/

Each result file contains the search result in JSON format. You can view it by running cat ./query-results/result_0.json | python3 -m json.tool.

Use a shell script to control concurrency

For a large number of queries (dozens or more), limit concurrent processes to avoid exceeding your API quota. The following script reads queries line by line from a file and caps execution at five concurrent processes.

#!/bin/bash

ACCOUNT_ID="<your-account-id>"
REGION="cn-hangzhou"
BUCKET="<your-vector-bucket>"
INDEX="<your-index>"
MODEL="text-embedding-v4"

MAX_CONCURRENT=5
QUERY_FILE="./queries.txt"  # One query text per line

mkdir -p ./query-results

run_query() {
  local idx=$1
  local text=$2
  oss-vectors-embed \
    --account-id "$ACCOUNT_ID" \
    --vectors-region "$REGION" \
    query \
    --vector-bucket-name "$BUCKET" \
    --index-name "$INDEX" \
    --model-id "$MODEL" \
    --text-value "$text" \
    --top-k 10 \
    > "./query-results/result_${idx}.json" 2>&1
}

idx=0
while IFS= read -r query_text; do
  run_query "$idx" "$query_text" &
  idx=$((idx + 1))

  # When the concurrency limit is reached, wait for a job to finish before continuing
  if (( $(jobs -rp | wc -l) >= MAX_CONCURRENT )); then
    wait -n
  fi
done < "$QUERY_FILE"

wait
echo "All $idx queries completed."

Before running the script, create a queries.txt file with one query per line:

How to configure a lifecycle rule
What are the storage classes for object storage
How to set up cross-region replication
Limits on bucket tags
How to enable versioning

Sample output:

All 5 queries completed.

Concurrent CLI calls with Python

To post-process CLI results (for example, parse JSON or aggregate statistics), wrap the CLI calls with Python's asyncio library.

import asyncio
import json
from pathlib import Path


ACCOUNT_ID = "<your-account-id>"
REGION = "cn-hangzhou"
BUCKET = "<your-vector-bucket>"
INDEX = "<your-index>"
MODEL = "text-embedding-v4"
MAX_CONCURRENT = 5


async def run_query(semaphore: asyncio.Semaphore, query_text: str, query_id: int):
    """Asynchronously executes a single CLI query command."""
    async with semaphore:
        cmd = [
            "oss-vectors-embed",
            "--account-id", ACCOUNT_ID,
            "--vectors-region", REGION,
            "query",
            "--vector-bucket-name", BUCKET,
            "--index-name", INDEX,
            "--model-id", MODEL,
            "--text-value", query_text,
            "--top-k", "10",
            "--return-metadata",
        ]

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()

        if proc.returncode == 0:
            result = json.loads(stdout.decode())

            print(f"Query {query_id} completed, returned {len(result.get('results', []))} results.")

            return {"query_id": query_id, "query_text": query_text, "result": result}
        else:
            print(f"Query {query_id} failed: {stderr.decode()}")
            return {"query_id": query_id, "query_text": query_text, "error": stderr.decode()}


async def batch_query(queries: list[str]):
    """Executes multiple queries concurrently in a batch."""
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [
        run_query(semaphore, text, idx)
        for idx, text in enumerate(queries)
    ]
    results = await asyncio.gather(*tasks)

    output_path = Path("./query-results/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"Aggregated results saved to {output_path}")
    return results


if __name__ == "__main__":
    queries = [
        "How to configure a lifecycle rule",
        "What are the storage classes for object storage",
        "How to set up cross-region replication",
        "Limits on bucket tags",
        "How to enable versioning",
    ]
    asyncio.run(batch_query(queries))

Sample output:

Query 0 completed, returned 10 results.
Query 1 completed, returned 10 results.
Query 2 completed, returned 10 results.
Query 3 completed, returned 10 results.
Query 4 completed, returned 10 results.
Aggregated results saved to query-results/batch_results.json

Run concurrent searches by using an SDK

SDK concurrency calls the query_vectors API directly without launching external processes. Use this method when you need fine-grained control over search parameters, such as filter conditions, or when integrating search into business services.

Note: The CLI method includes a built-in embedding model, so you can search directly with text. The SDK method requires pre-generated vectors, making it suitable for scenarios where you already have an embedding workflow.

Use the Python SDK for concurrent searches

Use the alibabacloud-oss-v2 Python SDK to call the query_vectors API concurrently with a thread pool. Before you start, install the SDK:

pip install alibabacloud-oss-v2

Make sure you have configured the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables and have created a vector bucket and a vector index.

Basic example

The following example searches for five sets of vectors concurrently, using a ThreadPoolExecutor to cap concurrency at five threads:

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():
    """Creates a vector client. Reuse this client globally to avoid connection overhead."""
    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  # For public access, set this to false or remove this line.
    return oss_vectors.Client(cfg)


def run_query(client, query_vector, query_id, query_filter=None):
    """Executes a single vector search."""
    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 {query_id} completed, 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):
    """Executes multiple vector searches concurrently in a batch."""
    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"Query {idx} failed: {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"Aggregated results saved to {output_path}")
    return results


if __name__ == "__main__":
    # Example: 5 sets of query vectors (the dimension must match the index; 128 dimensions used here as an example).
    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)

Sample output:

Query 0 completed, status code: 200
Query 2 completed, status code: 200
Query 1 completed, status code: 200
Query 4 completed, status code: 200
Query 3 completed, status code: 200
Aggregated results saved to query-results/sdk_batch_results.json

Note: Because the thread pool executes tasks concurrently, the output order may differ from the submission order. However, the final results are sorted by query_id before being saved.

Run concurrent searches with filter conditions

In practice, different queries may need different filter conditions. The following example specifies a unique filter for each query:

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():
    """Creates a vector client. Reuse this client globally to avoid connection overhead."""
    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  # For public access, set this to false or remove this line.
    return oss_vectors.Client(cfg)


def run_query(client, query_vector, query_id, query_filter=None):
    """Executes a single vector search."""
    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 {query_id} completed, 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):
    """Executes multiple vector searches concurrently in a batch."""
    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"Query {idx} failed: {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"Aggregated results saved to {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,  # No filter condition.
        },
    ]

    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"Query failed: {e}")

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

        print(f"Query {r['query_id']}: returned {len(r.get('vectors', []))} results.")

Sample output:

Query 0 completed, status code: 200
Query 1 completed, status code: 200
Query 2 completed, status code: 200
Query 0: returned 10 results.
Query 1: returned 10 results.
Query 2: returned 10 results.

Use the Go SDK for concurrent searches

Use the alibabacloud-oss-go-sdk-v2 Go SDK to call the QueryVectors API concurrently with goroutines. Before you start, install the SDK:

go get github.com/aliyun/alibabacloud-oss-go-sdk-v2

Make sure you have configured the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables and have created a vector bucket and a vector index.

Basic example

The following example uses a sync.WaitGroup and a channel-based semaphore to concurrently search for five sets of vectors:

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).
		// For public access, set this to false or remove this line.
		WithUseInternalEndpoint(true)

	client := vectors.NewVectorsClient(cfg)

	// Example: 5 query vectors. The dimension must match your index.
	queryVectors := []map[string]any{
		{"float32": []float32{0.1}}, // This vector needs to be expanded to match the index dimension.
		{"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) // Use a channel as a semaphore to control concurrency.

	for i, qv := range queryVectors {
		wg.Add(1)
		sem <- struct{}{} // Acquire a semaphore; blocks if the limit is reached.

		go func(idx int, queryVector map[string]any) {
			defer wg.Done()
			defer func() { <-sem }() // Release the semaphore.

			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("Query %d failed: %v", idx, err)
				return
			}
			fmt.Printf("Query %d completed, status code: %d\n",
				idx, result.StatusCode)
		}(i, qv)
	}

	wg.Wait()
	fmt.Println("All queries completed.")
}

Sample output:

Query 0 completed, status code: 200
Query 2 completed, status code: 200
Query 1 completed, status code: 200
Query 3 completed, status code: 200
Query 4 completed, status code: 200
All queries completed.

Run concurrent searches with filter conditions

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).
		// For public access, set this to false or remove this line.
		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, // No filter condition.
		},
	}

	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("Query %d failed: %v", idx, err)
				return
			}
			fmt.Printf("Query %d completed, status code: %d\n", idx, result.StatusCode)
		}(i, task)
	}

	wg.Wait()
	fmt.Println("All queries completed.")
}

Sample output:

Query 0 completed, status code: 200
Query 1 completed, status code: 200
Query 2 completed, status code: 200
All queries completed.

Tune concurrency performance

Parameter	Recommendation	Description
Number of concurrent requests	3 to 5	Keep `query` concurrency at five or fewer. This aligns with the maximum concurrency for `put` commands and helps avoid rate limiting.
`top_k`	Set as needed	More results per request means higher latency. Return only the number of results your application needs.
Error retry	1 to 2-second interval	Concurrent requests may trigger rate limiting (HTTP 429). We recommend catching errors and waiting before retrying.
CLI result output	Redirect to files	Multiple processes writing to the terminal simultaneously can interleave output. Write each result to a separate file to prevent this.
SDK client reuse	Reuse the same instance	Creating a new `Client` for each query adds connection setup and authentication overhead. Reuse a single instance instead.

Choose a concurrency method

Run concurrent searches by using the CLI

Use xargs for quick concurrency

Use shell background jobs for concurrency

Use a shell script to control concurrency

Concurrent CLI calls with Python

Run concurrent searches by using an SDK

Use the Python SDK for concurrent searches

Basic example

Run concurrent searches with filter conditions

Use the Go SDK for concurrent searches

Basic example

Run concurrent searches with filter conditions

Tune concurrency performance

Related topics