概述

背景

在數字化時代，圖像搜索技術已經成為生活中不可或缺的一部分，假設您在網上看到了一幅迷人的風景畫但不知道其出處，或者您想找到與某款服裝相似的產品，推薦使用云原生數據倉庫 AnalyticDB PostgreSQL 版的圖片搜索技術，通過文本搜索圖片，您只需輸入相關的關鍵詞，系統便會提供海量的圖片結果供您參考。還可以通過圖片搜索圖片，只需要上傳一張圖片，系統就能快速匹配出相似的圖片或相關信息，這極大地方便了我們的生活。

定義

圖片向量化搜索是云原生數據倉庫 AnalyticDB PostgreSQL 版根據圖片內容（如顏色、形狀、紋理等特征）來進行搜索和檢索的方法。其核心原理是將圖片轉化為可以被計算機處理的數學表示形式，即向量（一組數字）。

實現原理

1. 特征提取：首先需要從圖片中提取出能夠代表其內容的特征，這些特征經過處理后可以表示為一個多維的向量，這些向量必須能夠有效并準確地反映原始圖片的特征。

2. 向量存儲：對所有圖片進行特征提取和向量化后，將其存儲在支持向量化能力的數據庫中，建立索引，以便快速檢索。

3. 圖片、文本檢索：當用戶提交一個查詢圖片或者文本時，會進行特征提取和向量化，然后使用相似性度量方法（歐幾里得距離、余弦相似性）在向量庫中查找最相似的圖片特征向量。

4. 排序與顯示：根據計算得到的相似性分數，將結果進行排序，并將最相關的圖片展示給用戶。

圖片搜索整體技術棧復雜，實現起來并不簡單。因此，云原生數據倉庫 AnalyticDB PostgreSQL 版集合了多種圖片向量算法及高效向量檢索功能，提供高效的圖片索引及檢索能力，方便客戶快速構建圖搜應用。

前提條件

• 云原生數據倉庫 AnalyticDB PostgreSQL 版實例需同時滿足以下條件：

  • 實例已開啟向量引擎優化。具體操作，請參見開啟或關閉向量檢索引擎優化。

  • 實例已創建數據庫賬號。具體操作，請參見創建數據庫賬號。

  • 實例已申請外網地址。具體操作，請參見管理外網地址。

• 已將客戶端的IP地址添加至實例的白名單。具體操作，請參見設置白名單。

• 已安裝Python3.7及以上版本環境，請確保相應的Python版本。

  pip install alibabacloud-gpdb20160503
  pip install alibabacloud-tea-OpenAPI
  pip install alibabacloud-tea-util
  pip install alibabacloud-OpenAPI-util

  重要

  alibabacloud-gpdb20160503版本需要3.5.1及以上版本。

• 已將RAM用戶的AccessKey ID和AccessKey Secret配置到環境變量，請參見創建AccessKey。

  export ALIBABA_CLOUD_ACCESS_KEY_ID = "<YOUR_ALIBABA_CLOUD_ACCESS_KEY_ID>"
  export ALIBABA_CLOUD_ACCESS_KEY_SECRED = "<YOUR_ALIBABA_CLOUD_ACCESS_KEY_SECRET>"
  

準備工作

1. 創建向量索引：收集并準備要建立索引的向量數據，確保數據已經清洗和預處理，以便于索引。具體操作，請參見創建向量索引。

2. 創建命名空間：在創建向量索引之前，需要創建一個命名空間。根據需要，可以創建一個新的命名空間或者使用現有的命名空間。具體操作，請參見創建Namespace。

3. 創建文檔庫：在命名空間中創建文檔庫。根據數據的類型和用途，創建一個新的文檔庫或者使用現有的文檔庫。具體操作，請參見CreateDocumentCollection - 創建文檔庫。

   說明

   在CreateDocumentCollection步驟下可以選擇embedding模型。

圖片上傳

單張圖片上傳

本地圖片上傳

上傳本地圖片，導入向量庫。具體代碼如下：

# -*- coding: utf-8 -*-
import os
import sys

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def main() -> None:
        meta_data = {metadata}
        f = open("<image_file_path>", "rb")

        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        upload_document_async_request = gpdb_20160503_models.UploadDocumentAsyncAdvanceRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            file_name="<your-file-name>",
            file_url_object=f,
            dry_run=False,
            metadata=meta_data,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.upload_document_async_advance(upload_document_async_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))
        except Exception as error:
            print(error)

if __name__ == '__main__':
    Sample.main()

OpenAPI接口文檔，請參見UploadDocumentAsync- 異步上傳文檔。該接口為異步上傳接口，接口調用成功會返回job_id字段，根據job_id可以查詢圖片上傳進度。具體調用方式，請參見上傳進度查詢。參數說明如下：

遠程圖片上傳

上傳遠程圖片，導入向量庫。具體代碼如下：

# -*- coding: utf-8 -*-
import os
import sys

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def main() -> None:
        file_url = "<image_file_url>"
        meta_data = {metadata}

        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        upload_document_async_request = gpdb_20160503_models.UploadDocumentAsyncRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            file_name="<your-file-name>",
            file_url=file_url,
            dry_run=False,
            metadata=meta_data,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.upload_document_async_with_options(upload_document_async_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))
        except Exception as error:
            print(error)

if __name__ == '__main__':
    Sample.main()    

OpenAPI接口文檔，請參見UploadDocumentAsync - 異步上傳文檔。該接口為異步上傳接口，接口調用成功會返回job_id字段，根據job_id可以查詢圖片上傳進度。具體調用方式，請參見上傳進度查詢。參數說明如下：

批量圖片上傳

以本地文件為例，調用OpenAPI上傳本地壓縮包，導入壓縮包里的所有圖片到向量庫中。具體代碼如下：

# -*- coding: utf-8 -*-
import os
import sys

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def main() -> None:
        meta_data = {metadata}
        f = open("<compress_file_path>", "rb")

        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        upload_document_async_request = gpdb_20160503_models.UploadDocumentAsyncAdvanceRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            file_name="<your-file-name>",
            file_url_object=f,
            dry_run=False,
            metadata=meta_data,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.upload_document_async_advance(upload_document_async_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))
        except Exception as error:
            print(error)

if __name__ == '__main__':
    Sample.main()

OpenAPI接口文檔，請參見UploadDocumentAsync - 異步上傳文檔。該接口為異步上傳接口，接口調用成功會返回job_id字段，根據job_id可以查詢圖片上傳進度。具體調用方式，請參見上傳進度查詢。參數說明如下：

上傳進度查詢

單張、批量圖片上傳都是異步接口，需要調用進度查詢接口查看圖片上傳進度。

調用OpenAPI 查詢上傳圖片進度。具體代碼如下：

# -*- coding: utf-8 -*-
import os
import sys

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def main() -> None:
        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        get_upload_document_request = gpdb_20160503_models.GetUploadDocumentJobRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            job_id="<job_id>",
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.get_upload_document_job_with_options(get_upload_document_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))
        except Exception as error:
            print(error)

if __name__ == '__main__':
    Sample.main()

調用GetUploadDocumentJob接口查詢上傳圖片的進度。OpenAPI接口文檔，請參見GetUploadDocumentJob - 獲取上傳文檔任務。當job.status='Success'時，上傳任務完成。參數說明如下：

圖片檢索

根據文本檢索

文本檢索代碼示例如下：

# -*- coding: utf-8 -*-
import os
import sys

from urllib.request import urlopen
from PIL import Image

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

def show_image_text(image_text_list):
    for img, cap in image_text_list:
        # 注意：show() 函數在 Linux 服務器上可能需要安裝必要的圖像瀏覽器組件才生效
        img.show()
        print(cap)

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def query(content: str) -> []:
        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        query_content_request = gpdb_20160503_models.QueryContentRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            content=content,
            top_k=3,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.query_content_with_options(query_content_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))

            if response.status_code != 200:
                raise Exception(f"query_content failed, result: {response.body}")

            image_list = []
            for match_item in response.body.matches.match_list:
                url = match_item.file_url
                caption = match_item.metadata.get("caption")
                print("url: %s, caption: %s" % (url, caption))

                img = Image.open(urlopen(url))
                image_list.append((img, caption))
            return image_list
        except Exception as error:
            print(error)

if __name__ == '__main__':
    query_content = "狗"
    show_image_text(Sample.query(query_content))

參數說明如下：

當query_content配置為“狗”時，測試結果如下（查詢結果與實際上傳圖片集有關）：

圖片檢索

圖片檢索代碼示例如下（輸入圖片為本地圖片）：

# -*- coding: utf-8 -*-
import os
import sys

from urllib.request import urlopen
from PIL import Image

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

def show_image_text(image_text_list):
    for img, cap in image_text_list:
        # 注意：show() 函數在 Linux 服務器上可能需要安裝必要的圖像瀏覽器組件才生效
        img.show()
        print(cap)

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def query(file_path: str) -> []:
        client = Sample.create_client(os.environ["<ALIBABA_CLOUD_ACCESS_KEY_ID>"], os.environ["<ALIBABA_CLOUD_ACCESS_KEY_SECRET>"])
        f = open(file_path, 'rb')
        filename = os.path.basename(file_path)
        query_content_request = gpdb_20160503_models.QueryContentAdvanceRequest(
        query_content_request = gpdb_20160503_models.QueryContentRequest(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-name>",
            namespace="<your-namespace-name>",
            namespace_password="<your-namespace-password>",
            collection="<your-collection-name>",
            file_url_object=f,
            file_name=filename,
            top_k=3,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.query_content_advance(query_content_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))

            if response.status_code != 200:
                raise Exception(f"query_content failed, result: {response.body}")

            image_list = []
            for match_item in response.body.matches.match_list:
                url = match_item.file_url
                caption = match_item.metadata.get("caption")
                print("url: %s, caption: %s" % (url, caption))

                img = Image.open(urlopen(url))
                image_list.append((img, caption))
            return image_list
        except Exception as error:
            print(error)

if __name__ == '__main__':
    query_file_path = "<image_file_path>"
    show_image_text(Sample.query(query_file_path))

參數說明如下：

輸入一張自行車圖片，查詢結果如下（查詢結果與實際上傳圖片集有關）：

相關參考

通過Streamlit實現多模檢索

Streamlit簡介

Streamlit 是一個用于機器學習、數據可視化的Python框架，它能用簡短的幾行代碼將數據腳本轉換為 Web應用程序，該框架是用純Python編寫的，不需要前端經驗。

• 官方文檔：streamlit、st.audio。

• 快速入門：streamlit教程。

• 安裝方式如下：

  pip install streamlit

Streamlit實現文本搜圖

使用Streamlit進行簡單的文本搜圖功能演示，具體代碼如下：

# -*- coding: utf-8 -*-
import os
import streamlit as st

from alibabacloud_gpdb20160503.client import Client as gpdb20160503Client
from alibabacloud_tea_OpenAPI import models as open_api_models
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
from alibabacloud_tea_util import models as util_models

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        使用AK&SK初始化賬號Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # Endpoint 請參考 https://api.aliyun.com/product/gpdb
        config.endpoint = f'gpdb.aliyuncs.com'
        return gpdb20160503Client(config)

    @staticmethod
    def query(content: str) -> []:
        client = Sample.create_client(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'])
        query_content_request = gpdb_20160503_models.QueryContentRequest(
            region_id='{your-instance-region-id}',
            dbinstance_id='{your-instance-name}',
            namespace='{your-namespace-name}',
            namespace_password='{your-namespace-password}',
            collection='{your-collection-name}',
            content=content,
            top_k=3,
        )
        runtime = util_models.RuntimeOptions()
        try:
            response = client.query_content_with_options(query_content_request, runtime)
            print("response code: %s, response body: %s\n" % (response.status_code, response.body))

            if response.status_code != 200:
                raise Exception(f"query_content failed, result: {response.body}")

            image_list = []
            for match_item in response.body.matches.match_list:
                url = match_item.file_url
                caption = match_item.metadata.get("caption")
                print("url: %s, caption: %s" % (url, caption))
                image_list.append((url, caption))
            return image_list
        except Exception as error:
            print(error)

# markdown
st.header('文本搜圖Demo')
text_query = st.chat_input("請輸入檢索詞")
if text_query is None:
    st.text("檢索詞: ")
else:
    st.text("檢索詞: %s" % text_query)

if text_query:
    image_text_list = Sample.query(text_query)
    for url, cap in image_text_list:
        st.image(url)
        st.text("Description: " + cap)

參數說明：

測試結果

查詢結果與實際上傳的文本有關：