通义千问

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

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

前提条件

您已完成部署 seekdb。
您的环境中已存在可以使用的 MySQL 数据库和账号，并已对数据库账号授予读写权限。
安装 python 3.9 及以上版本和相应 pip。

安装 poetry、pyobvector 和 DashScope SDK。

pip install poetry
pip install pyobvector
pip install dashscope

准备通义千问 API 密钥。

步骤一：获取 seekdb 连接串

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

obclient -h$host -P$port -u$user_name -p$password -D$database_name

参数说明：

$host：提供 seekdb 连接 IP 地址。
$port：提供 seekdb 连接端口，默认是 2881。
$database_name：需要访问的数据库名称。

提示
连接的用户需要拥有该数据库的 CREATE、INSERT、DROP 和 SELECT 权限。
$user_name：提供数据库连接账户。
$password：提供账户密码。

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

步骤三：存储向量数据到 seekdb

准备测试数据下载预先计算好向量化数据的 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)

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

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]}")

步骤四：查询 seekdb 数据

保存以下 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.

前提条件​

步骤一：获取 seekdb 连接串​

步骤二：配置通义千问 API key 环境变量​

步骤三：存储向量数据到 seekdb​

步骤四：查询 seekdb 数据​

Contents

前提条件

步骤一：获取 seekdb 连接串

步骤二：配置通义千问 API key 环境变量

步骤三：存储向量数据到 seekdb

步骤四：查询 seekdb 数据