OB Cloud Vector 与通义千问集成

更新时间：2026-07-04 20:37:26

OceanBase 从 V4.3.3 开始支持向量类型存储、向量索引、embedding 向量检索的能力。可以将向量化后的数据存储在 OceanBase，供下一步的检索使用。

通义千问是由阿里云研发的大语言模型，用于理解和分析用户输入，可以在阿里云模型体验中心使用通义千问模型的 API 服务。

本文介绍了如何利用通义千问 API，将向量化后的数据存储在 OB Cloud，并使用 OB Cloud 的向量检索能力查询相关数据。

前提条件

您的环境中有可用的事务型（MySQL）集群实例。

请参考创建租户完成租户创建后，再参考下述步骤操作。

您的环境中已存在可以使用的 MySQL 兼容模式租户和 MySQL 数据库和账号，并已对数据库账号授予读写权限。如需创建，详情参见创建账号和创建数据库（仅 MySQL）。
您拥有项目管理员或实例管理员角色可对项目中的实例进行读写操作，如无权限，可联系组织管理员添加权限。
安装 Python 3.9 及以上版本和相应 pip。如果您的机器上 Python 版本较低，可以使用 Miniconda 来创建新的 Python 3.9 及以上的环境，具体可参考 Miniconda 安装指南。

安装 Poetry、Pyobvector 和 DashScope：

python3 -m pip install poetry
python3 -m pip install pyobvector
python3 -m pip install dashscope

步骤一：获取数据库连接信息

在下拉框中，根据 ID 选择您的集群实例。
进入 实例工作台 页面。
单击连接，选择 获取连接串。
在弹出框中选择 使用公共网络。
获取访问地址，选择 添加当前浏览器 IP 地址。
填写数据库相关信息，复制连接串。

步骤二：注册 LLM 平台账号

注册阿里云百炼账号，开通模型服务并获取 API 密钥。

注意

开通阿里云百炼大模型服务需要您跳转至第三方平台完成。此操作将遵循第三方平台的收费规则，并可能产生相应费用。请在继续前，访问其官网或查阅相关文档，确认并接受其收费标准。如不同意，请勿继续操作。

点击开通模型服务

确认开通模型服务

阿里云百炼

获取阿里云百炼 API Key

配置 API Key 到环境变量：

对于基于 Unix 的系统（如 Ubuntu 或 MacOS），你可以在终端中运行以下命令：

export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

对于 Windows 系统，你可以在命令提示符中运行以下命令：

set DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

请确保将 YOUR_DASHSCOPE_API_KEY 替换为你的实际通义千问 API 密钥。

步骤三：向量存储与查询

存储向量数据到 OB Cloud

准备测试数据

下载预先计算好向量化数据的 CSV 文件，该文件是阿里云的一个包含 1000 条美食评论的公开数据集，最后一列是向量化之后的值，所以不需要再计算向量。也可以使用下面的代码对 embedding 列（即向量列）重新计算，生成新的 CSV 文件。

import dashscope
import pandas as pd
input_datapath = "./fine_food_reviews.csv"
# 这里使用text_embedding_v1嵌入模型，可以根据需要调整
def generate_embeddings(text):
    rsp = dashscope.TextEmbedding.call(model=TextEmbedding.Models.text_embedding_v1, input=text)
    embeddings = [record['embedding'] for record in rsp.output['embeddings']]
    return embeddings if isinstance(text, list) else embeddings[0]
df = pd.read_csv(input_datapath, index_col=0)
# 实际生成会耗时几分钟，逐行调用通义千问 Embedding API
df["embedding"] = df.combined.apply(generate_embeddings)
output_datapath = './fine_food_reviews_self_embeddings.csv'
df.to_csv(output_datapath)

运行下面的脚本，将测试数据插入 OB Cloud，脚本所在的目录需要和测试数据所在的目录相同。

import os
import sys
import csv
import json
from pyobvector import *
from sqlalchemy import Column, Integer, String
# 使用pyobvector连接OB,用户名和密码中如果有@符号用%40代替
client = ObVecClient(uri="host:port", user="username",password="****",db_name="test")
# 事先准备的测试数据集，已进行了向量化，默认放在python脚本相同的目录下，如果是自己重新向量化的，需要替换为对应的文件
file_name = "fine_food_reviews.csv"
file_path = os.path.join("./", file_name)
# 定义列，向量化的列放在了最后一个字段
cols = [
    Column('id', Integer, primary_key=True, autoincrement=False),
    Column('product_id', String(256), nullable=True),
    Column('user_id', String(256), nullable=True),
    Column('score', Integer, nullable=True),
    Column('summary', String(2048), nullable=True),
    Column('text', String(8192), nullable=True),
    Column('combined', String(8192), nullable=True),
    Column('n_tokens', Integer, nullable=True),
    Column('embedding', VECTOR(1536))
]
# 表名
table_name = 'fine_food_reviews'
# 如果表不存在就创建表
if not client.check_table_exists(table_name):
    client.create_table(table_name,columns=cols)
    # 为向量列创建索引
    client.create_index(
        table_name=table_name,
        is_vec_index=True,
        index_name='vidx',
        column_names=['embedding'],
        vidx_params='distance=l2, type=hnsw, lib=vsag',
    )
# 打开并读取 CSV 文件
with open(file_name, mode='r', newline='', encoding='utf-8') as csvfile:
    csvreader = csv.reader(csvfile)
    # 读取标题行
    headers = next(csvreader)
    print("Headers:", headers)
    batch = [] # 存储数据，每10行插入一次到数据库
    for i, row in enumerate(csvreader):
        # CSV 文件有9个字段: id,product_id,user_id,score,summary,text,combined,n_tokens,embedding
        if not row:
            break
        food_review_line= {'id':row[0],'product_id':row[1],'user_id':row[2],'score':row[3],'summary':row[4],'text':row[5],\
        'combined':row[6],'n_tokens':row[7],'embedding':json.loads(row[8])}
        batch.append(food_review_line)
        # 每 10 行插入一次
        if (i + 1) % 10 == 0:
            client.insert(table_name,batch)
            batch = []  # 清空缓存
    # 插入剩余的行(如果有)
    if batch:
        client.insert(table_name,batch)
# 检查表中的数据，确保所有的数据已经插入
count_sql = f"select count(*) from {table_name};"
cursor = client.perform_raw_text_sql(count_sql)
result = cursor.fetchone()
print(f"导入数据总条数:{result[0]}")

查询向量数据

保存以下 python 脚本，命名为 query.py。

import os
import sys
import csv
import json
from pyobvector import *
from sqlalchemy import func
import dashscope
# 获取命令行参数
if len(sys.argv) != 2:
    print("请输入一个查询语句。")
    sys.exit()
queryStatement = sys.argv[1]
# 使用pyobvector连接OB,用户名和密码中如果有@符号用%40代替
client = ObVecClient(uri="host:port", user="usename",password="****",db_name="test")
# 定义生成文本向量的函数
def generate_embeddings(text):
    rsp = dashscope.TextEmbedding.call(model=TextEmbedding.Models.text_embedding_v1, input=text)
    embeddings = [record['embedding'] for record in rsp.output['embeddings']]
    return embeddings if isinstance(text, list) else embeddings[0]

def query_ob(query, tableName, vector_name="embedding", top_k=1):
    embedding = generate_embeddings(query)
    # 执行近似最近邻搜索
    res = client.ann_search(
        table_name=tableName,
        vec_data=embedding,
        vec_column_name=vector_name,
        distance_func=func.l2_distance,
        topk=top_k,
        output_column_names=['combined']
    )
    for row in res:
        print(str(row[0]).replace("Title: ", "").replace("; Content: ", ": "))
# 表名
table_name = 'fine_food_reviews'
query_ob(queryStatement,table_name,'embedding',1)

输入问题，输出相关答案。

python3 query.py 'pet food'

预期结果如下：

This is so good!: I purchased this after my sister sent a small bag to me in a gift box. I loved it so much I wanted to find it to buy for myself and keep it around. I always look on Amazon because you can find everything here and true enough, I found this wonderful candy. It is nice to keep in your purse for when you are out and about and get a dry throat or a tickle in the back of your throat. It is also nice to have in a candy dish at home for guests to try.