Build an image search application

Overview

Background

Image search technology is an important feature in many modern applications. For example, you can use it to find the source of a captivating landscape painting you saw online or to find products similar to a specific piece of clothing. AnalyticDB for PostgreSQL provides image search technology that lets you search for images by using text. You enter relevant keywords, and the system provides numerous image results. You can also search for images by uploading an image. The system can then quickly find similar images or related information.

Definition

Image vector search is a method that AnalyticDB for PostgreSQL uses to search and retrieve images based on their content, such as colors, shapes, and textures. The core principle is to convert images into a mathematical representation that a computer can process: a vector (a set of numbers).

How it works

1. Feature extraction: The system extracts features that represent the content of an image. These features are then processed to form a multi-dimensional vector. These vectors must effectively and accurately reflect the features of the original image.

2. Vector storage: After feature extraction and vectorization, the system stores all image vectors in a vector-capable database and builds an index for fast retrieval.

3. Image and text search: When a user submits a query image or text, the system extracts features and vectorizes the query. It then searches the vector database for the most similar image vectors by using similarity metrics, such as Euclidean distance or cosine similarity.

4. Ranking and display: The system sorts the results by their similarity scores and displays the most relevant images to the user.

The technology stack for image search is complex and not straightforward to implement. Therefore, AnalyticDB for PostgreSQL integrates various image vector algorithms and high-performance vector search functions. This provides efficient image indexing and retrieval capabilities, which allows you to quickly build image search applications.

Prerequisites

• Your AnalyticDB for PostgreSQL instance meets the following requirements:

  • You have enabled vector search engine optimization.

  • You have created a database account.

  • You have requested a public endpoint.

• You have added the IP address of the client to the instance whitelist.

• Python 3.7 or later is installed.

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

  Important

  The version of alibabacloud-gpdb20160503 must be 3.5.1 or later.

• You have configured the AccessKey ID and AccessKey secret of a RAM user as environment variables.

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

Preparations

1. Initialize the vector database: Before you use the AnalyticDB for PostgreSQL vector database, you must initialize the vector database and the full-text search features.

2. Create a namespace: A namespace is required before you can create a vector index. You can create a new namespace or use an existing one.

3. Create a collection: Create a collection in the namespace. You can create a new collection or use an existing one based on your data type and purpose.

   Note

   You can select an embedding model in the CreateDocumentCollection step.

Upload images

Upload a single image

Upload a local image

Upload a local image and import it into the vector database. The following code provides an example:

# -*- 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:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>",
            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()

For the OpenAPI reference, see UploadDocumentAsync. This operation is asynchronous. If the call succeeds, it returns the job_id field. You can use this ID to query the job status. For more information, see Check upload progress. The following table describes the parameters.

Upload a remote image

Upload a remote image and import it into the vector database. The following code provides an example:

# -*- 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:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>",
            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()    

For the OpenAPI reference, see UploadDocumentAsync. This operation is asynchronous. If the call succeeds, it returns the job_id field. You can use this ID to query the job status. For more information, see Check upload progress. The following table describes the parameters.

Upload multiple images

The following example shows how to call the OpenAPI to upload a local compressed archive and import all images from the archive into the vector database:

# -*- 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:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>",
            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()

For the OpenAPI reference, see UploadDocumentAsync. This operation is asynchronous. If the call succeeds, it returns the job_id field. You can use this ID to query the job status. For more information, see Check upload progress. The following table describes the parameters.

Check upload progress

Image uploads are asynchronous. You must query the job status to check if the upload is complete.

Call the OpenAPI to query the image upload progress. The following code provides an example:

# -*- 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:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>",
            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()

Call the GetUploadDocumentJob operation to query the progress of an image upload. For the OpenAPI reference, see GetUploadDocumentJob. The upload task is complete when job.status='Success'. The following table describes the parameters.

Search for images

Search by text

The following code provides an example of searching 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: On Linux servers, the show() function may require an image viewer to be installed.
        img.show()
        print(cap)
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>",
            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.

When query_content is set to "dog", the test results are as follows. The query results vary based on the images in your collection.

Search by image

The following code provides an example of searching by image. The input is a local image.

# -*- 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: On Linux servers, the show() function may require an image viewer to be installed.
        img.show()
        print(cap)
class Sample:
    def __init__(self):
        pass
    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> gpdb20160503Client:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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(
            region_id="<your-instance-region-id>",
            dbinstance_id="<your-instance-id>",
            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.

Using a bicycle image as input produces the following results. The query results vary based on the images in your collection.

References

Text-to-image search with Streamlit

Streamlit

Streamlit is a Python framework for machine learning and data visualization. It can convert data scripts into web applications with just a few lines of code. The framework is written purely in Python and requires no front-end experience.

• Official documentation: streamlit and st.audio.

• Quick start: Streamlit tutorial.

• To install Streamlit, run the following command:

  pip install streamlit

Implementation example

The following code provides a simple demonstration of the text-to-image search feature by using 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:
        """
        Initialize the Client with an AccessKey pair.
        @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
        )
        # For more information about the endpoint, see 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-id>',
            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('Text-to-Image Search Demo')
text_query = st.chat_input("Enter a search term")
if text_query is None:
    st.text("Search term: ")
else:
    st.text("Search term: %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 images in your collection.

On the Text-to-Image Search Demo page, entering the search term skiing returns two relevant images: one of a person skiing down a slope and another of a person with ski poles on snowy ground. This result confirms that the text-to-image search feature retrieves images based on text input.