高并发场景下，如何让你的向量语义检索快人一步？

当需要同时检索多条查询如批量问答、RAG 多路召回、多用户并发搜索逐条串行执行会导致整体耗时随查询数线性增长。通过并发执行多条检索请求可以将总耗时从 N × 单次延迟 降低到接近 1 × 单次延迟显著提升吞吐量。本文介绍两种并发方式CLI 并发和 SDK 并发适用于以下场景批量语义搜索一次性提交多条查询文本快速获取全部检索结果。RAG 多路召回为同一用户请求同时发起多条不同角度的检索降低端到端延迟。多模态批量检索同时检索文本、图片、视频等不同模态的向量数据。选择并发方式方式适用场景特点CLI 并发运维脚本、一次性批量检索、无需编写代码的快速验证输入文本自动 Embedding无需管理向量维度Shell 脚本即可实现SDK 并发业务服务集成、需要精细控制过滤条件、结果后处理、高性能后端直接调用 API可设置过滤条件复用客户端连接支持 Python 和 Go如何选择如果你有一批查询文本希望快速拿到结果而不想写代码 → 使用 CLI 并发。如果你在开发业务服务需要将向量检索嵌入到应用逻辑中 → 使用 SDK 并发。说明CLI 方式内置了 Embedding 模型调用输入文本即可检索。SDK 方式需要传入已生成的向量适合已有 Embedding 流程的场景。通过 CLI 并发检索CLI 并发通过启动多个oss-vectors-embed query进程实现并行检索。以下提供三种实现方式按复杂度递增排列。开始前请确保满足以下条件已安装 OSS Vectors Embed CLI。安装方式请参见使用OSS Vectors Embed CLI工具写入和检索向量数据。已配置环境变量OSS_ACCESS_KEY_ID、OSS_ACCESS_KEY_SECRET和DASHSCOPE_API_KEY。已创建向量 Bucket 和向量索引且索引维度与所用 Embedding 模型输出维度一致。将以下示例中的占位符替换为实际值占位符说明your-account-id阿里云账号 IDyour-vector-bucket向量 Bucket 名称your-index向量索引名称xargs 快速并发如果不需要复杂的流程控制xargs -P是最简单的 CLI 并发方式一行命令即可完成。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-P 5表示最多 5 个进程并行执行。检索结果直接输出到终端适合快速验证。如需保存结果可将输出重定向到文件。Shell 后台并发通过将多个查询命令放入后台并行执行wait等待全部完成。适合查询数量较少10 条以内的场景。#!/bin/bash ACCOUNT_IDyour-account-id REGIONcn-hangzhou BUCKETyour-vector-bucket INDEXyour-index MODELtext-embedding-v4 queries( 如何配置生命周期规则 对象存储有哪些存储类型 如何设置跨区域复制 ) mkdir -p ./query-results # 并发启动所有查询每条结果写入独立文件 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 21 done wait echo 全部查询完成结果保存在 ./query-results/运行后输出全部查询完成结果保存在 ./query-results/每个结果文件包含 JSON 格式的检索结果可通过cat ./query-results/result_0.json | python3 -m json.tool查看。控制并发数的 Shell 脚本当查询数量较多时数十条以上需要限制同时运行的进程数避免超出 API 配额。以下脚本从文件逐行读取查询文本控制最多 5 个进程同时执行。#!/bin/bash ACCOUNT_IDyour-account-id REGIONcn-hangzhou BUCKETyour-vector-bucket INDEXyour-index MODELtext-embedding-v4 MAX_CONCURRENT5 QUERY_FILE./queries.txt # 每行一条查询文本 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 21 } idx0 while IFS read -r query_text; do run_query $idx $query_text idx$((idx 1)) # 达到并发上限时等待一个任务完成再继续 if (( $(jobs -rp | wc -l) MAX_CONCURRENT )); then wait -n fi done $QUERY_FILE wait echo 全部 $idx 条查询完成运行前准备queries.txt文件每行一条查询文本如何配置生命周期规则 对象存储有哪些存储类型 如何设置跨区域复制 Bucket 标签的使用限制 如何启用版本控制运行后输出全部 5 条查询完成Python 封装 CLI 并发如果需要对 CLI 返回的结果进行后处理如解析 JSON、汇总统计可以用 Pythonasyncio封装 CLI 调用。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): 异步执行单条 CLI query 命令 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, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, ) stdout, stderr await proc.communicate() if proc.returncode 0: result json.loads(stdout.decode()) print(f查询 {query_id} 完成返回 {len(result.get(results, [ ]))} 条结果) return {query_id: query_id, query_text: query_text, result: result} else: print(f查询 {query_id} 失败: {stderr.decode()}) return {query_id: query_id, query_text: query_text, error: stderr.decode()} async def batch_query(queries: list[str]): 批量并发执行多条查询 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(parentsTrue, exist_okTrue) output_path.write_text(json.dumps(results, ensure_asciiFalse, indent2)) print(f汇总结果已保存到 {output_path}) return results if __name__ __main__: queries [ 如何配置生命周期规则, 对象存储有哪些存储类型, 如何设置跨区域复制, Bucket 标签的使用限制, 如何启用版本控制, ] asyncio.run(batch_query(queries))运行后输出查询 0 完成返回 10 条结果 查询 1 完成返回 10 条结果 查询 2 完成返回 10 条结果 查询 3 完成返回 10 条结果 查询 4 完成返回 10 条结果 汇总结果已保存到 query-results/batch_results.json通过 SDK 并发检索SDK 并发直接调用query_vectorsAPI无需启动外部进程。适合需要精细控制检索参数如设置过滤条件或将检索集成到业务服务中的场景。说明SDK 方式需要传入已生成的查询向量如 float32 数组而非原始文本。如果你的场景是从文本出发检索建议先通过 Embedding 模型生成向量或直接使用上文的 CLI 并发方式。Python SDK 并发检索使用alibabacloud-oss-v2Python SDK 通过线程池并发调用query_vectors接口。开始前请安装 SDKpip install alibabacloud-oss-v2确保已配置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET并已创建向量 Bucket 和索引。基本示例以下示例并发检索 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_filterNone): 执行单条向量检索 request oss_vectors.models.QueryVectorsRequest( bucketBUCKET, index_nameINDEX, query_vectorquery_vector, filterquery_filter, return_distanceTrue, return_metadataTrue, top_k10, ) 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_workersMAX_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(keylambda x: x[query_id]) output_path Path(./query-results/sdk_batch_results.json) output_path.parent.mkdir(parentsTrue, exist_okTrue) output_path.write_text(json.dumps(results, ensure_asciiFalse, indent2)) 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排序保存。带过滤条件的并发检索实际业务中不同查询可能需要搭配不同的过滤条件。以下示例为每条查询指定独立的filterimport 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_filterNone): 执行单条向量检索 request oss_vectors.models.QueryVectorsRequest( bucketBUCKET, index_nameINDEX, query_vectorquery_vector, filterquery_filter, return_distanceTrue, return_metadataTrue, top_k10, ) 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_workersMAX_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(keylambda x: x[query_id]) output_path Path(./query-results/sdk_batch_results.json) output_path.parent.mkdir(parentsTrue, exist_okTrue) output_path.write_text(json.dumps(results, ensure_asciiFalse, indent2)) 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_workersMAX_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, keylambda 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 条结果Go SDK 并发检索使用alibabacloud-oss-go-sdk-v2Go SDK 通过 goroutine 并发调用QueryVectors接口。开始前请安装 SDKgo get github.com/aliyun/alibabacloud-oss-go-sdk-v2确保已配置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET并已创建向量 Bucket 和索引。基本示例以下示例使用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 全部查询完成并发性能调优调优项建议说明并发数3~5query并发建议不超过 5与put命令的最大并发一致避免触发限流top_k按需设置返回结果越多单次请求延迟越高。仅返回业务所需的数量错误重试间隔 1~2 秒并发请求可能触发限流HTTP 429建议捕获错误后等待重试CLI 结果输出重定向到文件多个进程同时输出到终端会导致内容交错建议将每条结果写入独立文件SDK 客户端复用复用同一实例避免为每条查询创建新的Client重复创建会增加连接建立和认证开销相关文档使用OSS Vectors Embed CLI工具写入和检索向量数据OSS向量Bucket最佳实践快速构建多模态图片语义检索