文档

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

更新时间:

本文介绍基于云原生数据仓库 AnalyticDB 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. 创建文档库:在命名空间中创建文档库。根据数据的类型和用途,创建一个新的文档库或者使用现有的文档库。具体操作,请参见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可以查询图片上传进度。具体调用方式,请参见上传进度查询。参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

image_file_path

本地图片文件的绝对路径。

your-file-name

图片文件名称,需要包含扩展名,目前支持的扩展名有:bmp,jpg,jpeg,png和tiff。

metadata

数据集元数据信息,dict结构。

远程图片上传

上传远程图片,导入向量库。具体代码如下:

# -*- 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可以查询图片上传进度。具体调用方式,请参见上传进度查询。参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

image_file_path

远程图片文件的URL路径。

your-file-name

图片文件名称,需要包含扩展名,目前支持的扩展名有:bmp,jpg,jpeg,png和tiff。

metadata

数据集元数据信息,dict结构。

批量图片上传

以本地文件为例,调用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()
重要
  • 当前一个压缩包最多能包含100张图片。

  • 当前支持的文件压缩协议为:tar、gz、zip。

OpenAPI接口文档,请参见UploadDocumentAsync - 异步上传文档。该接口为异步上传接口,接口调用成功会返回job_id字段,根据job_id可以查询图片上传进度。具体调用方式,请参见上传进度查询。参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

compress_file_path

本地压缩包文件的绝对路径。

your-file-name

压缩包文件名,需要包含扩展名,目前支持的扩展名:tar,gz和zip。

metadata

数据集元数据信息,dict结构。

上传进度查询

单张、批量图片上传都是异步接口,需要调用进度查询接口查看图片上传进度。

调用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'时,上传任务完成。参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

job_id

图片上传接口返回的job_id。

图片检索

根据文本检索

文本检索代码示例如下:

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

参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

当query_content配置为“狗”时,测试结果如下(查询结果与实际上传图片集有关):

image.pngimage.pngimage.png

图片检索

图片检索代码示例如下(输入图片为本地图片):

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

参数说明如下:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

image_file_path

待检索图片的本地地址,需要填写绝对路径。

输入一张自行车图片,查询结果如下(查询结果与实际上传图片集有关):

image.pngimage.png

image.png

相关参考

通过Streamlit实现多模检索

Streamlit简介

Streamlit 是一个用于机器学习、数据可视化的Python框架,它能用简短的几行代码将数据脚本转换为 Web应用程序,该框架是用纯Python编写的,不需要前端经验。

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)

参数说明:

参数

描述

your-instance-region-id

云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。

your-instance-name

云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。

your-namespace-name

准备工作中的命名空间名称。

your-collection-name

准备工作中的数据集名称。

your-namespace-password

准备工作中的命名空间密码。

测试结果

查询结果与实际上传的文本有关:

image.png

  • 本页导读 (1)