本文介绍基于云原生数据仓库 AnalyticDB PostgreSQL 版的图搜API,实现向量化检索图片的整体流程。
概述
背景
在数字化时代,图像搜索技术已经成为生活中不可或缺的一部分,假设您在网上看到了一幅迷人的风景画但不知道其出处,或者您想找到与某款服装相似的产品,推荐使用云原生数据仓库 AnalyticDB PostgreSQL 版的图片搜索技术,通过文本搜索图片,您只需输入相关的关键词,系统便会提供海量的图片结果供您参考。还可以通过图片搜索图片,只需要上传一张图片,系统就能快速匹配出相似的图片或相关信息,这极大地方便了我们的生活。
定义
图片向量化搜索是云原生数据仓库 AnalyticDB PostgreSQL 版根据图片内容(如颜色、形状、纹理等特征)来进行搜索和检索的方法。其核心原理是将图片转化为可以被计算机处理的数学表示形式,即向量(一组数字)。
实现原理
特征提取:首先需要从图片中提取出能够代表其内容的特征,这些特征经过处理后可以表示为一个多维的向量,这些向量必须能够有效并准确地反映原始图片的特征。
向量存储:对所有图片进行特征提取和向量化后,将其存储在支持向量化能力的数据库中,建立索引,以便快速检索。
图片、文本检索:当用户提交一个查询图片或者文本时,会进行特征提取和向量化,然后使用相似性度量方法(欧几里得距离、余弦相似性)在向量库中查找最相似的图片特征向量。
排序与显示:根据计算得到的相似性分数,将结果进行排序,并将最相关的图片展示给用户。
图片搜索整体技术栈复杂,实现起来并不简单。因此,云原生数据仓库 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>"
准备工作
创建向量索引:收集并准备要建立索引的向量数据,确保数据已经清洗和预处理,以便于索引。具体操作,请参见创建向量索引。
创建命名空间:在创建向量索引之前,需要创建一个命名空间。根据需要,可以创建一个新的命名空间或者使用现有的命名空间。具体操作,请参见创建Namespace。
创建集合:在命名空间中创建集合。根据数据的类型和用途,创建一个新的集合或者使用现有的集合。具体操作,请参见创建向量数据集。
图片上传
单张图片上传
本地图片上传
上传本地图片,导入向量库。具体代码如下:
# -*- 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配置为“狗”时,测试结果如下(查询结果与实际上传图片集有关):
图片检索
图片检索代码示例如下(输入图片为本地图片):
# -*- 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 | 待检索图片的本地地址,需要填写绝对路径。 |
输入一张自行车图片,查询结果如下(查询结果与实际上传图片集有关):
相关参考
通过Streamlit实现多模检索
Streamlit简介
Streamlit 是一个用于机器学习、数据可视化的Python框架,它能用简短的几行代码将数据脚本转换为 Web应用程序,该框架是用纯Python编写的,不需要前端经验。
快速入门: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)
参数说明:
参数 | 描述 |
your-instance-region-id | 云原生数据仓库 AnalyticDB PostgreSQL 版实例所属地域ID。 |
your-instance-name | 云原生数据仓库 AnalyticDB PostgreSQL 版实例ID。 |
your-namespace-name | 准备工作中的命名空间名称。 |
your-collection-name | 准备工作中的数据集名称。 |
your-namespace-password | 准备工作中的命名空间密码。 |
测试结果
查询结果与实际上传的文本有关: