Tablestore：寫入向量資料

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.model.search.vector.VectorUtils;

import java.util.Random;
import java.util.UUID;


// 產生隨機向量的輔助方法。
private static float[] generateRandomFloats(int length, Random random) {
    float[] result = new float[length];
    for (int i = 0; i < length; i++) {
        result[i] = random.nextFloat();
    }
    return result;
}

// 批量寫入資料。
private static void batchWriteRow(SyncClient tableStoreClient) throws Exception {
    
    Random random = new Random();
    // 寫入 1 千行資料，每 100 行一個批次。
    for (int i = 0; i < 10; i++) {
        BatchWriteRowRequest batchWriteRowRequest = new BatchWriteRowRequest();
        for (int j = 0; j < 100; j++) {
            // 使用者的業務資料。
            String text = "一段字串，可使用者全文檢索索引。同時該欄位產生 Embedding 向量，寫入到下方 field_vector 欄位中進行向量語義相似性查詢";
            // 轉化好的向量，需使用者進行轉換。
            float[] vector = generateRandomFloats(1024,random);
            RowPutChange rowPutChange = new RowPutChange("TABLE_NAME");
            // 設定主鍵。
            rowPutChange.setPrimaryKey(PrimaryKeyBuilder.createPrimaryKeyBuilder().addPrimaryKeyColumn("PK_1", PrimaryKeyValue.fromString(UUID.randomUUID().toString())).build());
            // 設定屬性列。
            rowPutChange.addColumn("field_string", ColumnValue.fromLong(i));
            rowPutChange.addColumn("field_long", ColumnValue.fromLong(i * 100 + j));
            rowPutChange.addColumn("field_text", ColumnValue.fromString(text));
            // 通過binary格式寫入向量資料
            rowPutChange.addColumn("field_vector", ColumnValue.fromBinary(VectorUtils.toBytes(vector)));

            batchWriteRowRequest.addRowChange(rowPutChange);
        }
        BatchWriteRowResponse batchWriteRowResponse = tableStoreClient.batchWriteRow(batchWriteRowRequest);
        System.out.println("批量寫入是否全部成功：" + batchWriteRowResponse.isAllSucceed());
        if (!batchWriteRowResponse.isAllSucceed()) {
            for (BatchWriteRowResponse.RowResult rowResult : batchWriteRowResponse.getFailedRows()) {
                System.out.println("失敗的行：" + batchWriteRowRequest.getRowChange(rowResult.getTableName(), rowResult.getIndex()).getPrimaryKey());
                System.out.println("失敗原因：" + rowResult.getError());
            }
        }
    }
}

import time
import tablestore.utils
from tablestore import *

def batch_write_vector(rows_count):
    print('Begin prepare data: %d' % rows_count)
    batch_write_row_reqs = BatchWriteRowRequest()
    put_row_items = []
    for i in range(rows_count):
        pk = [('PK1', i)]

        cols = [('field_string', 'key%03d' % i),
                ('field_long', i),
                ('field_text', '一些文本'),
                ('field_vector', tablestore.utils.VectorUtils.floats_to_bytes([0.1, 0.2, 0.3, 0.4]))]

        put_row_item = PutRowItem(Row(pk,cols),Condition(RowExistenceExpectation.IGNORE))
        put_row_items.append(put_row_item)
    batch_write_row_reqs.add(TableInBatchWriteRowItem(table_name, put_row_items))
    client.batch_write_row(batch_write_row_reqs)

    print('End prepare data.')
    print('Wait for data sync to search index.')
    time.sleep(60)

public class VectorUtils {
    private static final ByteOrder order = ByteOrder.LITTLE_ENDIAN;

    /**
     * 將float[]轉換為二進位
     * @param vector 需要轉換的向量
     * @return byte 二進位化後的資料
     */
    public static byte[] toBytes(float[] vector) {
        if (vector == null || vector.length == 0) {
            throw new ClientException("vector is null or empty");
        }
        ByteBuffer buffer = ByteBuffer.allocate(vector.length * 4);
        buffer.order(order);
        for (float value : vector) {
            buffer.putFloat(value);
        }
        return buffer.array();
    }

    /**
     * 將二進位化後的資料轉換回float[]
     * @param bytes 二進位化後的資料
     * @return Float 原向量
     */
    public static float[] toFloats(byte[] bytes) {
        int length = bytes.length / 4;
        if (bytes.length % 4 != 0 || length == 0) {
            throw new ClientException("bytes length is not multiple of 4(SIZE_OF_FLOAT32) or length is 0");
        }
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        buffer.order(order);
        float[] vector = new float[length];
        buffer.asFloatBuffer().get(vector);
        return vector;
    }
}

// Float32ToBytes 將[]float32 轉換為位元組
func Float32ToBytes(vector []float32) ([]byte, error) {
	if len(vector) == 0 {
		return nil, errors.New("vector is null or empty")
	}
	data := make([]byte, 4*len(vector))

	for i, v := range vector {
		binary.LittleEndian.PutUint32(data[i*4:(i+1)*4], math.Float32bits(v))
	}
	return data, nil
}

// ToFloat32 將位元組轉回[]float32
func ToFloat32(data []byte) ([]float32, error) {
	if data == nil {
		return nil, errors.New("bytes is null")
	}
	if len(data)%4 != 0 || len(data) == 0 {
		return nil, errors.New("bytes length is not multiple of 4(SIZE_OF_FLOAT32) or length is 0")
	}
	floats := make([]float32, len(data)/4)
	buf := bytes.NewReader(data)

	for i := range floats {
		if err := binary.Read(buf, binary.LittleEndian, &floats[i]); err != nil {
			return nil, err
		}
	}

	return floats, nil
}

class VectorUtils:
    # 將floats轉換為bytearray
    @staticmethod
    def floats_to_bytes(floats):
        if not isinstance(floats, (list, tuple)) or not all(isinstance(f, float) for f in floats):
            raise TypeError("Input must be a list/tuple of floats")
        if len(floats) == 0:
            raise ValueError("vector is empty")
        return bytearray(struct.pack('<' + 'f' * len(floats), *floats))
      
    # 將bytearray轉換回floats
    @staticmethod
    def bytes_to_floats(byte_data):
        if not isinstance(byte_data, bytearray):
            raise TypeError("Input must be a bytearray object")
        num_floats = len(byte_data) // 4
        if len(byte_data) % 4 != 0 or num_floats == 0:
            raise ValueError("bytes length is not multiple of 4(SIZE_OF_FLOAT32) or length is 0")
        floats = struct.unpack('<' + 'f' * num_floats, byte_data)
        return list(floats)