Object Storage Service：OSS向量Bucket最佳實務：快速構建多模態圖片語義檢索

使用阿里雲百鍊 SDK 和 OSS SDK

以阿里雲百鍊提供的多模態 Embedding 模型 multimodal-embedding-v1 為例，將原始圖片轉換為 1024 維向量，並寫入 OSS 向量 Bucket 的向量索引中。

4.1 產生向量資料

import dashscope
from dashscope import MultiModalEmbeddingItemImage


def embedding_image(image_url: str) -> list[float]:
    """
    將圖片轉換為向量
    
    Args:
        image_url: 圖片的 URL 地址（支援 OSS 地址或公網 URL）
    
    Returns:
        1024 維的向量列表
    """
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemImage(image=image_url, factor=1.0)]
    )
    return resp.output["embeddings"][0]["embedding"]


def main():
    # 樣本圖片 URL（需替換為實際可訪問的圖片地址，如為圖片為私人許可權，需要填入帶簽名的臨時 URL）
    image_url = "http://your-bucket-name.oss-cn-hangzhou.aliyuncs.com/photograph/Zsd0YhBa8LM.jpg"
    
    print(f"正在對圖片進行向量化: {image_url}")
    
    # 調用 Embedding API
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemImage(image=image_url, factor=1.0)]
    )
    
    # 列印完整響應
    print("\n完整響應:")
    print(resp)
    
    # 擷取向量
    embedding = resp.output["embeddings"][0]["embedding"]
    print(f"\n向量維度: {len(embedding)}")
    print(f"向量前10個元素: {embedding[:10]}")


if __name__ == "__main__":
    main()

4.2 寫入向量資料

# -*- coding: utf-8 -*-
"""
樣本：批量寫入圖片向量資料

本樣本展示如何批量寫入已經向量化的圖片資料到向量索引中。

前提條件：
1. 已安裝 alibabacloud-oss-v2: pip install alibabacloud-oss-v2
2. 已設定環境變數
3. 已建立向量索引
4. 已準備好圖片向量資料檔案（data/data.json）
"""

import os
import json
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors


def main():
    # 從環境變數擷取憑證
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # 配置參數
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    
    # 載入預先處理好的圖片向量資料
    # 注意：文末的 Github倉庫裡 data/目錄已包含測試檔案，可直接使用
    # 也可以修改 data_file 變數，指向自己的圖片目錄
    data_file = "./data/data.json"
    print(f"正在載入圖片向量資料: {data_file}")
    
    image_data_array = []
    with open(data_file, "r") as f:
        image_data_array = json.load(f)
    
    print(f"共載入 {len(image_data_array)} 條圖片向量資料")
    
    # 列印資料範例
    if len(image_data_array) > 0:
        sample = image_data_array[0]
        print(f"\n資料範例:")
        print(f"  key: {sample.get('key', 'N/A')}")
        if 'metadata' in sample:
            print(f"  metadata: {sample['metadata']}")
        if 'data' in sample and 'float32' in sample['data']:
            print(f"  向量維度: {len(sample['data']['float32'])}")
    
    # 批量寫入，每批 500 條
    batch_size = 500
    vectors = []
    total_written = 0
    
    print(f"\n開始批量寫入 (batch_size={batch_size})...")
    
    for idx in range(len(image_data_array)):
        vectors.append(image_data_array[idx])
        
        if len(vectors) == batch_size:
            result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
                bucket=vector_bucket_name,
                index_name=vector_index_name,
                vectors=vectors,
            ))
            total_written += len(vectors)
            print(f"  已寫入 {total_written}/{len(image_data_array)} 條, "
                  f"status code: {result.status_code}")
            vectors = []
    
    # 寫入剩餘資料
    if len(vectors) > 0:
        result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
            bucket=vector_bucket_name,
            index_name=vector_index_name,
            vectors=vectors,
        ))
        total_written += len(vectors)
        print(f"  已寫入 {total_written}/{len(image_data_array)} 條, "
              f"status code: {result.status_code}")
    
    print(f"\n寫入完成! 共寫入 {total_written} 條向量資料")


if __name__ == "__main__":
    main()

使用 OSS-Vectors-Embed-CLI

OSS-Vectors-Embed-CLI 調用阿里雲百鍊向量模型，對 OSS 中的原始檔案或本地檔案進行向量化，並將向量結果寫入 OSS 向量 Bucket；也可用於發起多模態語義檢索。工具支援批量處理、自訂配置，以及寫入標量中繼資料等能力。

oss-vectors-embed \
  --account-id <your-account-id> \                      //阿里雲 ID
  --vectors-region cn-hangzhou \                        //OSS 向量 Bucket 所在地區
  put \
  --region cn-hangzhou \                                //OSS 原始檔案所在 Bucket 的地區
  --vector-bucket-name my-vector-bucket \               //OSS 向量 Bucket 名稱
  --index-name my-index \                               //OSS 向量 Index 名稱
  --model-id multimodal-embedding-v1 \                  //使用的向量模型
  --image "oss://bucket/path/*"                         //對 OSS 該 Prefix 下的檔案批量發起向量化和向量結果的寫入
  --filename-as-key                                     //將原始檔案 Key 值自訂設定為向量 Key 值

使用阿里雲百鍊 SDK 和 OSS SDK

# -*- coding: utf-8 -*-
"""
樣本：向量檢索查詢（Query Vector）

本樣本展示如何使用向量進行相似性檢索，支援中繼資料過濾。

前提條件：
1. 已安裝 alibabacloud-oss-v2 和 dashscope
2. 已設定環境變數
3. 已設定百鍊 API Key: export DASHSCOPE_API_KEY=<您的 API Key>
4. 已寫入向量資料
"""

import os
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors
import dashscope
from dashscope import MultiModalEmbeddingItemText


def embedding(text: str) -> list[float]:
    """
    文本向量化用於將查詢文本轉換為向量，實現文搜圖能力
    
    Args:
        text: 待轉換的文本
    
    Returns:
        1024 維的向量列表
    """
    return dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemText(text=text, factor=1.0)]
    ).output["embeddings"][0]["embedding"]


def main():
    # 從環境變數擷取憑證
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # 配置參數
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    
    # 查詢文本
    query_text = "狗狗"
    
    print(f"進行中向量檢索:")
    print(f"  Bucket: {vector_bucket_name}")
    print(f"  Index: {vector_index_name}")
    print(f"  查詢文本: {query_text}")
    
    # 將查詢文本轉換為向量
    print(f"\n正在將查詢文本轉換為向量...")
    query_vector = embedding(query_text)
    print(f"  向量維度: {len(query_vector)}")
    
    # 執行向量檢索
    print(f"\n執行向量檢索 ...")
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=vector_bucket_name,
        index_name=vector_index_name,
        query_vector={
            "float32":query_vector
        },
        top_k=5,                    # 返回最相似的 5 個結果
        return_distance=True,       # 返回相似性距離
        return_metadata=True,       # 返回中繼資料
    ))
    
    print(f"\n檢索結果 (共 {len(result.vectors)} 條):")
    for i, vector in enumerate(result.vectors, 1):
        print(f"\n  [{i}] key: {vector.get('key', 'N/A')}")
        if 'distance' in vector:
            print(f"      距離: {vector['distance']:.6f}")
        if 'metadata' in vector:
            print(f"      中繼資料: {vector['metadata']}")
    
    # 不帶過濾條件的檢索
    print(f"\n" + "=" * 50)
    print(f"執行向量檢索 (無過濾條件)...")
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=vector_bucket_name,
        index_name=vector_index_name,
        query_vector={
            "float32": query_vector
        },
        top_k=5,
        return_distance=True,
        return_metadata=True,
    ))
    
    print(f"\n檢索結果 (共 {len(result.vectors)} 條):")
    for i, vector in enumerate(result.vectors, 1):
        print(f"  [{i}] key: {vector.get('key', 'N/A')}, "
              f"距離: {vector.get('distance', 'N/A')}")


if __name__ == "__main__":
    main()

使用 OSS-Vectors-Embed-CLI

oss-vectors-embed \
  --account-id <your-account-id> \
  --vectors-region cn-hangzhou \
  query \
  --vector-bucket-name my-vector-bucket \
  --index-name my-index \
  --model-id multimodal-embedding-v1 \
  --text-value "狗狗" \               // Query 檢索文本
  --top-k 100                        // 檢索返回 100 個最相似的向量結果

{
  "results": [
    {
      "Key": "myimage03.jpg",                                        //檢索返回的向量結果，返迴向量 Key-Value 和標量中繼資料
      "metadata": {
        "OSSVECTORS-EMBED-SRC-CONTENT-TYPE": "TEXT",
        "OSSVECTORS-EMBED-SRC-LOCATION":  "./images/photo.jpg",     //CLI 會自動在向量 Bucket 該向量所在行添加源資訊欄位（OSSVECTORS-EMBED-SRC-*），用於追溯向量來源。
      }
    },
    ...
  ],
  "summary": {
    "queryType": "text",
    "model": "multimodal-embedding-v1",
    "index": "my-index",
    "resultsFound": 100,
    "queryDimensions": 1024
  }
}