AnalyticDB:Use the API operations of AnalyticDB for PostgreSQL to search for images

Overview

Background information

In the digital era, image search technology has become an indispensable part of our daily lives. For example, if you want to look for the source of a landscape painting or find a specific piece of clothing online, we recommend that you use the image search technology of AnalyticDB for PostgreSQL to search for images by text. After you enter a relevant keyword, the system can return a large number of images for your reference. You can also search for images by image. After you upload an image, the system can quickly match and return similar images or related information.

Definition of vector-based image search

AnalyticDB for PostgreSQL uses the vector-based image search method to search for images based on the image features, such as color, shape, and texture. The working principle of this method is to transform images into a set of vector data that can be processed by a computer.

How vector-based image search works

1. Extracts features from images that can represent the content of the images and transforms the features into multi-dimensional vectors that can effectively and accurately reflect the image features.

2. Stores the vectors in a database that supports vectorization and creates indexes for fast search.

3. Extracts features from the image or text that you submitted, generates feature vectors, and then uses the Euclidean distance or cosine similarity measurement method to search for the most similar image feature vectors in AnalyticDB for PostgreSQL.

4. Sorts the results based on the calculated similarity scores and displays the most relevant images.

Implementing image search is difficult because the overall technology stack for image search is complex. AnalyticDB for PostgreSQL integrates a variety of image vectorization algorithms and vector search features to provide efficient image indexing and search capabilities. This helps you quickly build image search applications.

Prerequisites

• An AnalyticDB for PostgreSQL instance is created and meets all the following requirements:

  • Vector search engine optimization is enabled for the AnalyticDB for PostgreSQL instance. For more information, see Enable or disable vector search engine optimization.

  • A database account is created for the AnalyticDB for PostgreSQL instance. For more information, see Create a database account.

  • A public endpoint is obtained for the AnalyticDB for PostgreSQL instance. For more information, see Manage public endpoints.

• The IP address of your client is added to a whitelist of the AnalyticDB for PostgreSQL instance. For more information, see Configure an IP address whitelist.

• Python 3.7 or later is installed. Make sure that the Python version meets the requirements.

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

  Important

  alibabacloud-gpdb20160503 requires Python 3.5.1 or later.

• The AccessKey ID and AccessKey secret of your Resource Access Management (RAM) user are configured in environment variables. For more information, see Create an AccessKey pair.

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

Prerequisites

1. Prepare vector data that is cleansed and preprocessed and create a vector index for the data. For more information, see Create a vector index.

2. Before you create a vector index, create a namespace based on your business requirements. You can also use an existing namespace. For more information, see Create a namespace.

3. Create a document collection in the namespace based on data types and purposes. You can also use an existing document collection. For more information, see CreateDocumentCollection.

   Note

   When you call the CreateDocumentCollection operation, you can use the EmbeddingModel parameter to specify a vectorization algorithm.

Upload images

Upload a single image

Upload an on-premises image

The following sample code provides an example on how to upload an on-premises image to the AnalyticDB for PostgreSQL instance:

# -*- 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:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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()

The UploadDocumentAsync operation is an asynchronous upload operation. After you call the operation, a job ID is returned. You can use the job ID to query the upload progress. For more information, see the "Query the upload progress" section of this topic. The following table describes the parameters.

Upload a remote image

The following sample code provides an example on how to upload a remote image to the AnalyticDB for PostgreSQL instance:

# -*- 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:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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()    

The UploadDocumentAsync operation is an asynchronous upload operation. After you call the operation, a job ID is returned. You can use the job ID to query the upload progress. For more information, see the "Query the upload progress" section of this topic. The following table describes the parameters.

Upload multiple images at the same time

The following sample code provides an example on how to upload all images in an on-premises compressed package to the AnalyticDB for PostgreSQL instance:

# -*- 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:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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()

The UploadDocumentAsync operation is an asynchronous upload operation. After you call the operation, a job ID is returned. You can use the job ID to query the upload progress. For more information, see the "Query the upload progress" section of this topic. The following table describes the parameters.

Query the upload progress

The UploadDocumentAsync operation is used to upload a single image or multiple images, which is an asynchronous operation. You can call a specific API operation to query the upload progress.

The following sample code provides an example on how to query the upload progress:

# -*- 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:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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()

In this example, the GetUploadDocumentJob operation is used to query the upload progress. For more information, see GetUploadDocumentJob. If Success is returned for the job.status parameter, the upload job is complete. The following table describes the parameters.

Search for images

Search for images by text

The following sample code provides an example on how to search for images by text:

# -*- 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:
        # Note: An image browser component may be required on the Linux server to allow the SHOW() function to take effect.
        img.show()
        print(cap)

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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 = "Dog"
    show_image_text(Sample.query(query_content))

The following table describes the parameters.

If you set the query_content parameter to Dog, the following sample results are returned. The query results vary based on the images that you uploaded:

Search for images by image

The following sample code provides an example on how to search for images by image. In this example, an on-premises image is used.

# -*- 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:
        # Note: An image browser component may be required on the Linux server to allow the SHOW() function to take effect.
        img.show()
        print(cap)

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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))

The following table describes the parameters.

If you enter a bicycle image, the following sample results are returned. The query results vary based on the images that you uploaded.

References

Use Streamlit to implement multimodal search

Introduction to Streamlit

Streamlit is a Python framework for machine learning and data visualization. Streamlit is written in Python and can use simple code to convert data scripts into web applications.

• Official documentation: Streamlit and st.audio.

• Quick start: Streamlit tutorial.

• Installation command:

  pip install streamlit

Use Streamlit to search for images by text

The following sample code provides an example on how to use Streamlit to search for images by text:

# -*- 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:
        """
        Use the AccessKey ID and AccessKey secret to initialize the 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
        )
        # Specify an endpoint. For more information, view the endpoints at 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 for searching for images by text')
text_query = st.chat_input("Enter a keyword")
if text_query is None:
    st.text("Keyword: ")
else:
    st.text("Keyword: %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)

The following table describes the parameters.

Test results

The query results vary based on the text that you entered.