OB Cloud Vector 与 OpenAI 集成

更新时间：2026-05-14 14:28:04

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

OpenAI 是一家人工智能公司，已开发出多个大语言模型，这些模型在自然语言理解和生成方面表现出色，能够生成文本、回答问题、进行对话等，可以通过 API 对这些模型进行访问。

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

前提条件

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

安装 Poetry、Pyobvector 和 OpenAI：

python3 -m pip install poetry
python3 -m pip install pyobvector
python3 -m pip install openai

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

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

步骤二：注册 LLM 平台账号

注意

获取 OpenAI API Key 需要您跳转至第三方平台完成。此操作将遵循第三方平台的收费规则，并可能产生相应费用。请在继续前，访问其官网或查阅相关文档，确认并接受其收费标准。如不同意，请勿继续操作。

获取 OpenAI API Key：

登录 OpenAI 平台。
在右上角点击 API Keys。
点击 Create API Key。
填写相关信息，点击 Create API Key。

配置 OpenAI API Key 到环境变量：

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

export OPENAI_API_KEY='your-api-key'

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

set OPENAI_API_KEY=your-api-key

请确保将 your-api-key 替换为你的实际 OpenAI API 密钥。

步骤三：向量存储与查询

存储向量数据到 OB Cloud

准备测试数据

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

from openai import OpenAI
import pandas as pd
input_datapath = "./fine_food_reviews.csv"
client = OpenAI()
# 这里使用text-embedding-ada-002嵌入模型，可以根据需要调整
def embedding_text(text, model="text-embedding-ada-002"):
    # 创建embedding 向量的方法可参考：https://community.openai.com/t/embeddings-api-documentation-needs-to-updated/475663
    res = client.embeddings.create(input=text, model=model)
    return res.data[0].embedding
df = pd.read_csv(input_datapath, index_col=0)
# 实际生成会耗时几分钟，逐行调用 OpenAI Embedding API
df["embedding"] = df.combined.apply(embedding_text)
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 脚本，命名为 openAIQuery.py。

import os
import sys
import csv
import json
from pyobvector import *
from sqlalchemy import func
from openai import OpenAI
# 获取命令行参数
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")
openAIclient = OpenAI()
# 定义生成文本向量的函数
def generate_embeddings(text, model="text-embedding-ada-002"):
    # 创建embedding 向量的方法可参考：https://community.openai.com/t/embeddings-api-documentation-needs-to-updated/475663
    res = openAIclient.embeddings.create(input=text, model=model)
    return res.data[0].embedding

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 openAIQuery.py 'pet food'

预期结果如下：

Crack for dogs.: These thing are like crack for dogs. I am not sure of the make-up but the doggies sure love them.