基于AnalyticDB for PostgreSQL图搜API构建图搜应用

概述

背景

在数字化时代，图像搜索技术已经成为生活中不可或缺的一部分，假设您在网上看到了一幅迷人的风景画但不知道其出处，或者您想找到与某款服装相似的产品，推荐使用云原生数据仓库 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. 创建集合：在命名空间中创建集合。根据数据的类型和用途，创建一个新的集合或者使用现有的集合。具体操作，请参见创建向量数据集。

图片上传

单张图片上传

本地图片上传

上传本地图片，导入向量库。具体代码如下：

# -*- 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))

参数说明如下：

输入一张自行车图片的公网URL地址，测试结果如下（查询结果与实际上传图片集有关）：

相关参考

通过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)

参数说明：

测试结果

查询结果与实际上传的图片有关：