通义千问

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

OceanBase 数据库提供了向量类型存储、向量索引、embedding 向量搜索的能力。可以利用通义千问的 API 接口，将向量化后的数据存储在 OceanBase 数据库中，然后使用 OceanBase 的向量搜索能力查询相关数据。

前提条件

• 您已完成部署 OceanBase 数据库 V4.3.3 及以上版本并且创建了 MySQL 模式租户。创建租户 后，再参考下述步骤操作。

• 您的环境中已存在可以使用的 MySQL 租户和 MySQL 数据库和账号，并已对数据库账号授予读写权限。

• 安装 python 3.9 及以上版本 和相应 pip。

• 安装 poetry、pyobvector 和 DashScope SDK。

  pip install poetry
  pip install pyobvector
  pip install dashscope
  

• 准备 通义千问 API 密钥。

• 确保您已经在租户中设置了 ob_vector_memory_limit_percentage 配置项，以启用向量搜索功能。V4.3.5 BP3 之前的版本推荐设置值为 30，从 V4.3.5 BP3 版本开始推荐保持默认值 0。如需更精确设置此配置项，请参考 ob_vector_memory_limit_percentage 计算此值。

步骤一：获取 OceanBase 数据库连接串

联系 OceanBase 数据库部署人员或者管理员获取相应的数据库连接串，例如：

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 模式下的权限分类。

• $user_name：提供租户的连接账户。ODP 连接的常用格式：用户名@租户名#集群名 或者 集群名:租户名:用户名；直连方式格式：用户名@租户名。

• $password：提供账户密码。

更多连接串的信息，请参见 通过 OBClient 连接 OceanBase 租户。

步骤二：配置通义千问 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 密钥。

步骤三：存储向量数据到 OceanBase 数据库

1. 准备测试数据 下载预先计算好向量化数据的 CSV 文件，这个 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)
   

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

   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。

   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. 输入问题，输出相关答案。

   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.