名称: tos-vectors
描述: 使用 TOS Vectors 服务管理向量存储与相似性搜索。适用于处理嵌入向量、语义搜索、RAG 系统、推荐引擎,或当用户提及向量数据库、相似性搜索或 TOS Vectors 操作时。
用于通过 TOS Vectors 服务(一个为 AI 应用优化的云原生向量数据库)管理向量存储、索引和相似性搜索的综合技能。
import os
import tos
# 从环境变量获取凭证
ak = os.getenv('TOS_ACCESS_KEY')
sk = os.getenv('TOS_SECRET_KEY')
account_id = os.getenv('TOS_ACCOUNT_ID')
# 配置端点与区域
endpoint = 'https://tosvectors-cn-beijing.volces.com'
region = 'cn-beijing'
# 创建客户端
client = tos.VectorClient(ak, sk, endpoint, region)
# 1. 创建向量桶(类似数据库)
client.create_vector_bucket('my-vectors')
# 2. 创建向量索引(类似数据表)
client.create_index(
account_id=account_id,
vector_bucket_name='my-vectors',
index_name='embeddings-768d',
data_type=tos.DataType.DataTypeFloat32,
dimension=768,
distance_metric=tos.DistanceMetricType.DistanceMetricCosine
)
# 3. 插入向量
vectors = [
tos.models2.Vector(
key='doc-1',
data=tos.models2.VectorData(float32=[0.1] * 768),
metadata={'title': '文档 1', 'category': 'tech'}
)
]
client.put_vectors(
vector_bucket_name='my-vectors',
account_id=account_id,
index_name='embeddings-768d',
vectors=vectors
)
# 4. 搜索相似向量
query_vector = tos.models2.VectorData(float32=[0.1] * 768)
results = client.query_vectors(
vector_bucket_name='my-vectors',
account_id=account_id,
index_name='embeddings-768d',
query_vector=query_vector,
top_k=5,
return_distance=True,
return_metadata=True
)
创建桶
client.create_vector_bucket(bucket_name)
列出桶
result = client.list_vector_buckets(max_results=100)
for bucket in result.vector_buckets:
print(bucket.vector_bucket_name)
删除桶(必须为空)
client.delete_vector_bucket(bucket_name, account_id)
创建索引
client.create_index(
account_id=account_id,
vector_bucket_name=bucket_name,
index_name='my-index',
data_type=tos.DataType.DataTypeFloat32,
dimension=128,
distance_metric=tos.DistanceMetricType.DistanceMetricCosine
)
列出索引
result = client.list_indexes(bucket_name, account_id)
for index in result.indexes:
print(f"{index.index_name}: {index.dimension}d")
插入向量(单次最多 500 条)
vectors = []
for i in range(100):
vector = tos.models2.Vector(
key=f'vec-{i}',
data=tos.models2.VectorData(float32=[...]),
metadata={'category': 'example'}
)
vectors.append(vector)
client.put_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
vectors=vectors
)
查询相似向量(KNN 搜索)
results = client.query_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
query_vector=query_vector,
top_k=10,
filter={"$and": [{"category": "tech"}]}, # 可选的元数据过滤
return_distance=True,
return_metadata=True
)
for vec in results.vectors:
print(f"Key: {vec.key}, Distance: {vec.distance}")
按键获取向量
result = client.get_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
keys=['vec-1', 'vec-2'],
return_data=True,
return_metadata=True
)
删除向量
client.delete_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
keys=['vec-1', 'vec-2']
)
构建文档语义搜索系统:
# 索引文档
for doc in documents:
embedding = get_embedding(doc.text) # 你的嵌入模型
vector = tos.models2.Vector(
key=doc.id,
data=tos.models2.VectorData(float32=embedding),
metadata={'title': doc.title, 'content': doc.text[:500]}
)
vectors.append(vector)
client.put_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
vectors=vectors
)
# 搜索
query_embedding = get_embedding(user_query)
results = client.query_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name=index_name,
query_vector=tos.models2.VectorData(float32=query_embedding),
top_k=5,
return_metadata=True
)
为大语言模型提示检索相关上下文:
# 检索相关文档
question_embedding = get_embedding(user_question)
search_results = client.query_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name='knowledge-base',
query_vector=tos.models2.VectorData(float32=question_embedding),
top_k=3,
return_metadata=True
)
# 构建上下文
context = "\n\n".join([
v.metadata.get('content', '') for v in search_results.vectors
])
# 使用 LLM 生成答案
prompt = f"上下文:\n{context}\n\n问题: {user_question}"
基于用户偏好查找相似物品:
# 使用元数据过滤进行查询
results = client.query_vectors(
vector_bucket_name=bucket_name,
account_id=account_id,
index_name='products',
query_vector=user_preference_vector,
top_k=10,
filter={"$and": [{"category": "electronics"}, {"price_range": "mid"}]},
return_metadata=True
)
try:
result = client.create_vector_bucket(bucket_name)
except tos.exceptions.TosClientError as e:
print(f'客户端错误: {e.message}')
except tos.exceptions.TosServerError as e:
print(f'服务端错误: {e.code}, 请求 ID: {e.request_id}')
详细 API 参考,请参阅 REFERENCE.md
完整工作流示例,请参阅 WORKFLOWS.md
示例脚本,请查看 scripts/ 目录