--- title: "通义千问 - OceanBase 数据库 V4.3.5 | OceanBase 文档中心" description: 通义千问 通义千问 是由阿里云研发的大语言模型，用于理解和分析用户输入，可以在阿里云 模型体验中心 使用通义千问模型的 API 服务。 OceanBase 数据库提供了向量类型存储、向量索引、embedding 向量搜索的能力。可以利用通义千问的 API 接口，将向量化后的数据存储在 OceanBase 数据库中，然… --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.3.5 LTS # 通义千问 更新时间：2026-04-09 14:12:04 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.3.5/zh-CN/680.ecological-integration/1100.AI/200.tongyi-qianwen.md) [通义千问](https://tongyi.aliyun.com/) 是由阿里云研发的大语言模型，用于理解和分析用户输入，可以在阿里云 [模型体验中心](https://bailian.console.aliyun.com/?spm=a2c4g.11186623.0.0.41f9253anP7UgA#/efm/model_experience_center/text) 使用通义千问模型的 API 服务。 OceanBase 数据库提供了向量类型存储、向量索引、embedding 向量搜索的能力。可以利用通义千问的 API 接口，将向量化后的数据存储在 OceanBase 数据库中，然后使用 OceanBase 的向量搜索能力查询相关数据。 ## 前提条件 - 您已完成部署 OceanBase 数据库 V4.3.3 及以上版本并且创建了 MySQL 模式租户。[创建租户](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002013128) 后，再参考下述步骤操作。 - 您的环境中已存在可以使用的 MySQL 租户和 MySQL 数据库和账号，并已对数据库账号授予读写权限。 - 安装 [python 3.9 及以上版本](https://www.python.org/downloads/) 和相应 [pip](https://pip.pypa.io/en/stable/installation/)。 - 安装 [poetry](https://python-poetry.org/docs/)、[pyobvector](https://github.com/oceanbase/pyobvector) 和 DashScope SDK。 ```shell pip install poetry pip install pyobvector pip install dashscope ``` - 准备 [通义千问 API 密钥](https://help.aliyun.com/zh/model-studio/developer-reference/get-api-key)。 - 确保您已经在租户中设置了 `ob_vector_memory_limit_percentage` 配置项，以启用向量搜索功能。V4.3.5 BP3 之前的版本推荐设置值为 `30`，从 V4.3.5 BP3 版本开始推荐保持默认值 `0`。如需更精确设置此配置项，请参考 [ob_vector_memory_limit_percentage](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002015585) 计算此值。 ## 步骤一：获取 OceanBase 数据库连接串 联系 OceanBase 数据库部署人员或者管理员获取相应的数据库连接串，例如： ```sql obclient -h$host -P$port -u$user_name -p$password -D$database_name ``` **参数说明：** - `$host`：提供 OceanBase 数据库连接 IP。OceanBase 数据库代理（OceanBase Database Proxy，ODP）连接方式使用的是一个 ODP 地址；直连方式使用的是 OBServer 节点的 IP 地址。 - `$port`：提供 OceanBase 数据库连接端口。ODP 连接的方式默认是 `2883`，在部署 ODP 时可自定义；直连方式默认是 `2881`，在部署 OceanBase 数据库时可自定义。 - `$database_name`：需要访问的数据库名称。 #### 注意 连接租户的用户需要拥有该数据库的 `CREATE`、`INSERT`、`DROP` 和 `SELECT` 权限。更多有关用户权限的信息，请参见 [MySQL 模式下的权限分类](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002016100)。 - `$user_name`：提供租户的连接账户。ODP 连接的常用格式：`用户名@租户名#集群名` 或者 `集群名:租户名:用户名`；直连方式格式：`用户名@租户名`。 - `$password`：提供账户密码。 更多连接串的信息，请参见 [通过 OBClient 连接 OceanBase 租户](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000002013251)。 ## 步骤二：配置通义千问 API key 环境变量 对于基于 Unix 的系统（如 Ubuntu 或 MacOS），你可以在终端中运行以下命令： ```shell export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY" ``` 对于 Windows，你可以在命令提示符中使用以下命令： ```shell set DASHSCOPE_API_KEY=YOUR_DASHSCOPE_API_KEY ``` 请确保将 `YOUR_DASHSCOPE_API_KEY` 替换为你的实际通义千问 API 密钥。 ## 步骤三：存储向量数据到 OceanBase 数据库 1. 准备测试数据 下载预先计算好向量化数据的 [CSV](https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20240827/srxyhu/fine_food_reviews.csv) 文件，这个 CSV 文件中包含 1000 条美食评论数据集，最后一列是向量化之后的值，所以不需要再计算向量。 也可以使用下面的代码对 embedding 列（即向量列）重新计算，生成新的 CSV 文件。 ```shell 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) ``` 2. 运行下面的脚本，将测试数据插入 OceanBase 数据库，脚本所在的目录需要和测试数据所在的目录相同。 ```shell 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]}") ``` ## 步骤四：查询 OceanBase 数据库数据 1. 保存以下 python 脚本，命名为 `query.py`。 ```shell 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) ``` 2. 输入问题，输出相关答案。 ```shell python3 query.py 'pet food' ``` 预期结果如下： ```shell 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. ``` 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线