概述

背景

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

定義

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

實(shí)現(xiàn)原理

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

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

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

4. 排序與顯示：根據(jù)計(jì)算得到的相似性分?jǐn)?shù)，將結(jié)果進(jìn)行排序，并將最相關(guān)的圖片展示給用戶。

圖片搜索整體技術(shù)棧復(fù)雜，實(shí)現(xiàn)起來并不簡(jiǎn)單。因此，云原生數(shù)據(jù)倉庫 AnalyticDB PostgreSQL 版集合了多種圖片向量算法及高效向量檢索功能，提供高效的圖片索引及檢索能力，方便客戶快速構(gòu)建圖搜應(yīng)用。

前提條件

• 云原生數(shù)據(jù)倉庫 AnalyticDB PostgreSQL 版實(shí)例需同時(shí)滿足以下條件：

  • 實(shí)例已開啟向量引擎優(yōu)化。具體操作，請(qǐng)參見開啟或關(guān)閉向量檢索引擎優(yōu)化。

  • 實(shí)例已創(chuàng)建數(shù)據(jù)庫賬號(hào)。具體操作，請(qǐng)參見創(chuàng)建數(shù)據(jù)庫賬號(hào)。

  • 實(shí)例已申請(qǐng)外網(wǎng)地址。具體操作，請(qǐng)參見管理外網(wǎng)地址。

• 已將客戶端的IP地址添加至實(shí)例的白名單。具體操作，請(qǐng)參見設(shè)置白名單。

• 已安裝Python3.7及以上版本環(huán)境，請(qǐng)確保相應(yīng)的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配置到環(huán)境變量，請(qǐng)參見創(chuàng)建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>"
  

準(zhǔn)備工作

1. 創(chuàng)建向量索引：收集并準(zhǔn)備要建立索引的向量數(shù)據(jù)，確保數(shù)據(jù)已經(jīng)清洗和預(yù)處理，以便于索引。具體操作，請(qǐng)參見創(chuàng)建向量索引。

2. 創(chuàng)建命名空間：在創(chuàng)建向量索引之前，需要?jiǎng)?chuàng)建一個(gè)命名空間。根據(jù)需要，可以創(chuàng)建一個(gè)新的命名空間或者使用現(xiàn)有的命名空間。具體操作，請(qǐng)參見創(chuàng)建Namespace。

3. 創(chuàng)建文檔庫：在命名空間中創(chuàng)建文檔庫。根據(jù)數(shù)據(jù)的類型和用途，創(chuàng)建一個(gè)新的文檔庫或者使用現(xiàn)有的文檔庫。具體操作，請(qǐng)參見CreateDocumentCollection - 創(chuàng)建文檔庫。

   說明

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

圖片上傳

單張圖片上傳

本地圖片上傳

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

# -*- 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初始化賬號(hào)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 請(qǐng)參考 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接口文檔，請(qǐng)參見UploadDocumentAsync- 異步上傳文檔。該接口為異步上傳接口，接口調(diào)用成功會(huì)返回job_id字段，根據(jù)job_id可以查詢圖片上傳進(jìn)度。具體調(diào)用方式，請(qǐng)參見上傳進(jìn)度查詢。參數(shù)說明如下：

遠(yuǎn)程圖片上傳

上傳遠(yuǎn)程圖片，導(dǎo)入向量庫。具體代碼如下：

# -*- 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初始化賬號(hào)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 請(qǐng)參考 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接口文檔，請(qǐng)參見UploadDocumentAsync - 異步上傳文檔。該接口為異步上傳接口，接口調(diào)用成功會(huì)返回job_id字段，根據(jù)job_id可以查詢圖片上傳進(jìn)度。具體調(diào)用方式，請(qǐng)參見上傳進(jìn)度查詢。參數(shù)說明如下：

批量圖片上傳

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

# -*- 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初始化賬號(hào)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 請(qǐng)參考 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接口文檔，請(qǐng)參見UploadDocumentAsync - 異步上傳文檔。該接口為異步上傳接口，接口調(diào)用成功會(huì)返回job_id字段，根據(jù)job_id可以查詢圖片上傳進(jìn)度。具體調(diào)用方式，請(qǐng)參見上傳進(jìn)度查詢。參數(shù)說明如下：

上傳進(jìn)度查詢

單張、批量圖片上傳都是異步接口，需要調(diào)用進(jìn)度查詢接口查看圖片上傳進(jìn)度。

調(diào)用OpenAPI 查詢上傳圖片進(jìn)度。具體代碼如下：

# -*- 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初始化賬號(hào)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 請(qǐng)參考 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()

調(diào)用GetUploadDocumentJob接口查詢上傳圖片的進(jìn)度。OpenAPI接口文檔，請(qǐng)參見GetUploadDocumentJob - 獲取上傳文檔任務(wù)。當(dāng)job.status='Success'時(shí)，上傳任務(wù)完成。參數(shù)說明如下：

圖片檢索

根據(jù)文本檢索

文本檢索代碼示例如下：

# -*- 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() 函數(shù)在 Linux 服務(wù)器上可能需要安裝必要的圖像瀏覽器組件才生效
        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初始化賬號(hào)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 請(qǐng)參考 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))

參數(shù)說明如下：

當(dāng)query_content配置為“狗”時(shí)，測(cè)試結(jié)果如下（查詢結(jié)果與實(shí)際上傳圖片集有關(guān)）：

圖片檢索

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

# -*- 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() 函數(shù)在 Linux 服務(wù)器上可能需要安裝必要的圖像瀏覽器組件才生效
        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初始化賬號(hào)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 請(qǐng)參考 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))

參數(shù)說明如下：

輸入一張自行車圖片，查詢結(jié)果如下（查詢結(jié)果與實(shí)際上傳圖片集有關(guān)）：

相關(guān)參考

通過Streamlit實(shí)現(xiàn)多模檢索

Streamlit簡(jiǎn)介

Streamlit 是一個(gè)用于機(jī)器學(xué)習(xí)、數(shù)據(jù)可視化的Python框架，它能用簡(jiǎn)短的幾行代碼將數(shù)據(jù)腳本轉(zhuǎn)換為 Web應(yīng)用程序，該框架是用純Python編寫的，不需要前端經(jīng)驗(yàn)。

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

• 快速入門：streamlit教程。

• 安裝方式如下：

  pip install streamlit

Streamlit實(shí)現(xiàn)文本搜圖

使用Streamlit進(jìn)行簡(jiǎn)單的文本搜圖功能演示，具體代碼如下：

# -*- 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初始化賬號(hào)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 請(qǐng)參考 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("請(qǐng)輸入檢索詞")
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)

參數(shù)說明：

測(cè)試結(jié)果

查詢結(jié)果與實(shí)際上傳的文本有關(guān)：