使用開源模型將 Tablestore 資料轉成向量

方案概覽

ModelScope（魔搭社區）旨在打造下一代開源的模型即服務共用平台，為泛 AI 開發人員提供靈活、易用、低成本的一站式模型服務供應項目，讓模型應用更簡單。ModelScope 魔搭社區彙集行業領先的預訓練模型，減少開發人員的重複研發成本，提供更加綠色環保、開源開放的 AI 開發環境和模型服務。

使用免費開源模型將儲存在 Tablestore 中的文本資料產生向量只需4步：

1. 安裝 Python SDK：使用開源模型產生向量和Table Store功能前，您需要安裝Table Store SDK 和開源模型 SDK。

2. 選擇與下載開源模型：ModelScope 魔搭社區提供了大量的文本向量 Embedding 模型，您可以通過模型庫進行選擇和下載使用。

3. 產生向量並寫入到Table Store：使用開源模型產生向量後，將向量資料寫入到Table Store資料表中使用。

4. 結果驗證：使用Table Store的資料讀取介面或者多元索引向量檢索功能查詢向量資料。

使用說明

• 開發語言：Python

• Python版本：推薦使用 Python3.9 及以上版本。

• 測試環境：本文中樣本已經過 CentOS 7 和 macOS 平台的環境驗證。

注意事項

Tablestore 多元索引中向量類型的維度、類型、距離演算法必須與開源模型中文本轉向量模型的相應配置保持一致。例如開源模型damo/nlp_corom_sentence-embedding_chinese-tiny產生的向量類型為 256 維、Float32 類型和 euclidean 歐氏距離演算法，在 Tablestore 建立多元索引時的向量類型也必須是 256 維、Float32 類型和 euclidean 歐氏距離演算法。

前提條件

• 使用阿里雲帳號或者具有Table Store操作許可權的 RAM 使用者進行操作。

  如果要使用 RAM 使用者進行操作，您需要使用阿里雲帳號建立 RAM 使用者並授予 RAM 使用者訪問Table Store（AliyunOTSFullAccess）的許可權。具體操作，請參見為RAM使用者授權。

• 已為阿里雲帳號或者 RAM 使用者建立 AccessKey。具體操作，請參見建立AccessKey。

  警告

  阿里雲帳號 AccessKey 泄露會威脅您所有資源的安全。建議您使用 RAM 使用者 AccessKey 進行操作，可以有效降低 AccessKey 泄露的風險。

• 已擷取Table Store執行個體的名稱和服務地址。具體操作，請參見服務地址。

• 已將 AccessKey（包括 AccessKey ID 和 AccessKey Secret）、執行個體名稱和執行個體的服務地址配置到環境變數中。

1. 安裝 SDK

在命令列中執行 pip 命令安裝Table Store Python SDK 和 ModelScope 相關依賴。安裝命令如下：

# 安裝Table StorePython SDK。
pip install tablestore
# 安裝ModelScope相關依賴。
pip install torch torchvision torchaudio
pip install "modelscope[framework]" -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html
pip install --use-pep517 "modelscope[nlp]" -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html

2. 選擇與下載開源模型

modelscope download --mode damo/nlp_corom_sentence-embedding_chinese-tiny

3. 產生向量並寫入到Table Store

使用開源模型進行向量產生後，將向量寫入到Table Store資料表中。您可以直接寫入向量到Table Store中，也可以將Table Store中存量的資料產生向量再寫入到Table Store中。

以下樣本用於使用 Python SDK 建立Table Store的表和多元索引，然後使用開源模型產生維度為 256 的向量，並寫入向量資料到Table Store。

import json
import os

from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks
from tablestore import OTSClient, TableMeta, TableOptions, ReservedThroughput, CapacityUnit, FieldSchema, FieldType, VectorDataType, VectorOptions, VectorMetricType, \
    SearchIndexMeta, AnalyzerType, Row, INF_MIN, INF_MAX, Direction, OTSClientError, OTSServiceError, Condition, RowExistenceExpectation

# 選擇合適的模型並填寫模型ID
pipeline_se = pipeline(Tasks.sentence_embedding, model='damo/nlp_corom_sentence-embedding_chinese-tiny')


def text_to_vector_string(text: str) -> str:
    inputs = {'source_sentence': [text]}
    result = pipeline_se(input=inputs)
    # 返回結果轉成TableStore支援的格式：float32數組字串，例如: [1, 5.1, 4.7, 0.08]
    return json.dumps(result["text_embedding"].tolist()[0])


def create_table():
    table_meta = TableMeta(table_name, [('PK_1', 'STRING')])
    table_options = TableOptions()
    reserved_throughput = ReservedThroughput(CapacityUnit(0, 0))
    tablestore_client.create_table(table_meta, table_options, reserved_throughput)


def create_search_index():
    index_meta = SearchIndexMeta([
        # 支援文本匹配查詢
        FieldSchema('field_string', FieldType.KEYWORD, index=True, enable_sort_and_agg=True),
        # 支援數字範圍查詢
        FieldSchema('field_long', FieldType.LONG, index=True, enable_sort_and_agg=True),
        # 全文檢索索引欄位
        FieldSchema('field_text', FieldType.TEXT, index=True, analyzer=AnalyzerType.MAXWORD),
        # 向量檢索欄位，使用歐氏距離作為度量，向量長度為 256
        FieldSchema("field_vector", FieldType.VECTOR,
                    vector_options=VectorOptions(
                        data_type=VectorDataType.VD_FLOAT_32,
                        dimension=256,
                        metric_type=VectorMetricType.VM_EUCLIDEAN
                    )),
    ])
    tablestore_client.create_search_index(table_name, index_name, index_meta)


def write_data_to_table():
    for i in range(100):
        pk = [('PK_1', str(i))]
        text = "一段字串，可使用者全文檢索索引。同時該欄位產生Embedding向量，寫入到下方field_vector欄位中進行向量語義相似性查詢"
        vector = text_to_vector_string(text)
        columns = [
            ('field_string', 'str-%d' % (i % 5)),
            ('field_long', i),
            ('field_text', text),
            ('field_vector', vector),
        ]
        tablestore_client.put_row(table_name, Row(pk, columns))


def get_range_and_update_vector():
    # 設定範圍讀的起始主鍵，INF_MIN是一個特殊最小值標誌位
    inclusive_start_primary_key = [('PK_1', INF_MIN)]
    # 設定範圍讀的結束主鍵，INF_MAX是一個特殊最大值標誌位
    exclusive_end_primary_key = [('PK_1', INF_MAX)]
    total = 0
    try:
        while True:
            consumed, next_start_primary_key, row_list, next_token = tablestore_client.get_range(
                table_name,
                Direction.FORWARD,
                inclusive_start_primary_key,
                exclusive_end_primary_key,
                ["field_text", "想要返回的其它欄位"],
                5000,
                max_version=1,
            )
            for row in row_list:
                total += 1
                # 擷取讀取到的"field_text"欄位
                text_field_content = row.attribute_columns[0][1]
                # 根據"field_text"欄位的內容重建向量
                vector = text_to_vector_string(text_field_content)
                update_of_attribute_columns = {
                    'PUT': [('field_vector', vector)],
                }
                update_row = Row(row.primary_key, update_of_attribute_columns)
                condition = Condition(RowExistenceExpectation.IGNORE)
                # 更新該行資料
                tablestore_client.update_row(table_name, update_row, condition)
            if next_start_primary_key is not None:
                inclusive_start_primary_key = next_start_primary_key
            else:
                break
    # 用戶端異常，一般為參數錯誤或者網路異常。
    except OTSClientError as e:
        print('get row failed, http_status:%d, error_message:%s' % (e.get_http_status(), e.get_error_message()))
    # 服務端異常，一般為參數錯誤或者流控錯誤。
    except OTSServiceError as e:
        print('get row failed, http_status:%d, error_code:%s, error_message:%s, request_id:%s' % (e.get_http_status(), e.get_error_code(), e.get_error_message(), e.get_request_id()))
    print("一共處理資料:", total)


if __name__ == '__main__':
    # 初始化 tablestore client
    end_point = os.environ.get('end_point')
    access_id = os.environ.get('access_key_id')
    access_key_secret = os.environ.get('access_key_secret')
    instance_name = os.environ.get('instance_name')
    tablestore_client = OTSClient(end_point, access_id, access_key_secret, instance_name)

    table_name = "python_demo_table_name"
    index_name = "python_demo_index_name"

    # 建立表
    create_table()
    # 建立索引
    create_search_index()
    # 方式1：直接寫入向量資料到TableStore中
    write_data_to_table()
    # 方式2：將TableStore中存量的資料產生向量再寫入到TableStore中
    get_range_and_update_vector()

4. 結果驗證

在Table Store控制台查看寫入Table Store中的向量資料。您可以使用資料讀取介面（ GetRow、BatchGetRow 和 GetRange）或者使用多元索引的向量檢索查詢向量資料。

• 使用資料讀取介面查詢向量資料

  向量資料寫入到Table Store後，您可以直接讀取表中的向量資料。具體操作，請參見讀取資料。

• 使用向量檢索功能查詢向量資料

  建立多元索引時配置向量欄位後，您可以使用向量檢索功能查詢向量資料。具體操作，請參見向量檢索。

計費說明

使用Table Store時，資料表和多元索引的資料量會佔用儲存空間，直接讀寫表中資料和使用多元索引向量檢索功能查詢資料會消耗計算資源。其中計算資源消耗會按照讀輸送量和寫輸送量計費。