使用亚马逊Titan进行语义图片搜索

Implement semantic image search with Amazon Titan and Supabase Vector in Python.

Amazon Bedrock 是一款全托管服务，提供来自 AI21 Labs、Anthropic、Cohere、Meta、Mistral AI、Stability AI 和 Amazon 等领先 AI 公司的高性能基础模型（FM）选择。每个模型都可以通过一个通用 API 访问，这个 API 提供了一整套功能，帮助你在考虑安全、隐私和负责任 AI 的前提下构建生成式 AI 应用。

Amazon Titan 是一系列用于文本和图片生成、摘要、分类、开放式问答、信息提取以及文本或图片搜索的基础模型(FM)。

在本指南中，我们将看看如何在 Python 中使用 Amazon Titan 多模态模型和 vecs client 开始使用 Amazon Bedrock 和 Supabase Vector。

🌐 In this guide we'll look at how we can get started with Amazon Bedrock and Supabase Vector in Python using the Amazon Titan multimodal model and the vecs client.

你可以在 GitHub 上找到完整的 Python Poetry 项目应用代码。

🌐 You can find the full application code as a Python Poetry project on GitHub.

用 Poetry 创建一个新的 Python 项目 #

🌐 Create a new Python project with Poetry

Poetry 为 Python 提供了打包和依赖管理。如果你还没有安装，可以通过 pip 安装 poetry：

1
pip install poetry

然后初始化一个新项目：

🌐 Then initialize a new project:

1
poetry new aws_bedrock_image_search

用 pgvector 启动一个 Postgres 数据库 #

🌐 Spin up a Postgres database with pgvector

如果你还没有，可以去 database.new 创建一个新项目。每个 Supabase 项目都自带一个完整的 Postgres 数据库，并且预先配置了 pgvector 插件。

🌐 If you haven't already, head over to database.new and create a new project. Every Supabase project comes with a full Postgres database and the pgvector extension preconfigured.

在创建项目时，记得把你的数据库密码记下来，因为下一步构建 DB_URL 时你会用到它。

🌐 When creating your project, make sure to note down your database password as you will need it to construct the DB_URL in the next step.

你可以在你的项目仪表板上找到数据库连接字符串，点击连接。使用看起来像这样的会话池连接字符串：

🌐 You can find your database connection string on your project dashboard, click Connect. Use the Session pooler connection string which looks like this:

1
postgresql://postgres.[PROJECT-REF]:[YOUR-PASSWORD]@aws-0-[REGION].pooler.supabase.com:5432/postgres

安装依赖 #

🌐 Install the dependencies

我们需要在项目中添加以下依赖：

🌐 We will need to add the following dependencies to our project:

vecs：Supabase 向量 Python 客户端。
boto3：适用于 Python 的 AWS SDK。
matplotlib：用于显示我们的图片结果。

1
poetry add vecs boto3 matplotlib

导入必要的依赖 #

🌐 Import the necessary dependencies

在你的主 Python 脚本顶部，导入依赖，并将上面提到的 DB URL 存储在一个变量中：

🌐 At the top of your main python script, import the dependencies and store your DB URL from above in a variable:

1
import sys
2
import boto3
3
import vecs
4
import json
5
import base64
6
from matplotlib import pyplot as plt
7
from matplotlib import image as mpimg
8
from typing import Optional
9
10
DB_CONNECTION = "postgresql://postgres.[PROJECT-REF]:[YOUR-PASSWORD]@aws-0-[REGION].pooler.supabase.com:5432/postgres"

接下来，获取你的 AWS 账户凭证，然后实例化 boto3 客户端：

🌐 Next, get the credentials to your AWS account and instantiate the boto3 client:

1
bedrock_client = boto3.client(
2
    'bedrock-runtime',
3
    region_name='us-west-2',
4
    # Credentials from your AWS account
5
    aws_access_key_id='<replace_your_own_credentials>',
6
    aws_secret_access_key='<replace_your_own_credentials>',
7
    aws_session_token='<replace_your_own_credentials>',
8
)

为你的图片创建嵌入 #

🌐 Create embeddings for your images

在你的项目根目录下，创建一个名为 images 的新文件夹并添加一些图片。你可以使用 GitHub 上示例项目中的图片，或者在 Unsplash 上找到免费的图片。

🌐 In the root of your project, create a new folder called images and add some images. You can use the images from the example project on GitHub or you can find license free images on Unsplash.

要向 Amazon Bedrock API 发送图片，我们需要将它们编码为 base64 字符串。创建以下辅助方法：

🌐 To send images to the Amazon Bedrock API we need to need to encode them as base64 strings. Create the following helper methods:

1
def readFileAsBase64(file_path):
2
    """Encode image as base64 string."""
3
    try:
4
        with open(file_path, "rb") as image_file:
5
            input_image = base64.b64encode(image_file.read()).decode("utf8")
6
        return input_image
7
    except:
8
        print("bad file name")
9
        sys.exit(0)
10
11
12
def construct_bedrock_image_body(base64_string):
13
    """Construct the request body.
14
15
    https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-embed-mm.html
16
    """
17
    return json.dumps(
18
        {
19
            "inputImage": base64_string,
20
            "embeddingConfig": {"outputEmbeddingLength": 1024},
21
        }
22
    )
23
24
25
def get_embedding_from_titan_multimodal(body):
26
    """Invoke the Amazon Titan Model via API request."""
27
    response = bedrock_client.invoke_model(
28
        body=body,
29
        modelId="amazon.titan-embed-image-v1",
30
        accept="application/json",
31
        contentType="application/json",
32
    )
33
34
    response_body = json.loads(response.get("body").read())
35
    print(response_body)
36
    return response_body["embedding"]
37
38
39
def encode_image(file_path):
40
    """Generate embedding for the image at file_path."""
41
    base64_string = readFileAsBase64(file_path)
42
    body = construct_bedrock_image_body(base64_string)
43
    emb = get_embedding_from_titan_multimodal(body)
44
    return emb

接下来，创建一个 seed 方法，它将创建一个新的 Supabase 向量集合，为你的图片生成嵌入，并将这些嵌入插入或更新到你的数据库中：

🌐 Next, create a seed method, which will create a new Supabase Vector Collection, generate embeddings for your images, and upsert the embeddings into your database:

1
def seed():
2
    # create vector store client
3
    vx = vecs.create_client(DB_CONNECTION)
4
5
    # get or create a collection of vectors with 1024 dimensions
6
    images = vx.get_or_create_collection(name="image_vectors", dimension=1024)
7
8
    # Generate image embeddings with Amazon Titan Model
9
    img_emb1 = encode_image('./images/one.jpg')
10
    img_emb2 = encode_image('./images/two.jpg')
11
    img_emb3 = encode_image('./images/three.jpg')
12
    img_emb4 = encode_image('./images/four.jpg')
13
14
    # add records to the *images* collection
15
    images.upsert(
16
        records=[
17
            (
18
                "one.jpg",       # the vector's identifier
19
                img_emb1,        # the vector. list or np.array
20
                {"type": "jpg"}  # associated  metadata
21
            ), (
22
                "two.jpg",
23
                img_emb2,
24
                {"type": "jpg"}
25
            ), (
26
                "three.jpg",
27
                img_emb3,
28
                {"type": "jpg"}
29
            ), (
30
                "four.jpg",
31
                img_emb4,
32
                {"type": "jpg"}
33
            )
34
        ]
35
    )
36
    print("Inserted images")
37
38
    # index the collection for fast search performance
39
    images.create_index()
40
    print("Created index")

把这个方法作为脚本添加到你的 pyproject.toml 文件里：

🌐 Add this method as a script in your pyproject.toml file:

1
[tool.poetry.scripts]
2
seed = "image_search.main:seed"
3
search = "image_search.main:search"

在使用 poetry shell 激活虚拟环境后，你现在可以通过 poetry run seed 运行你的种子脚本。你可以在 Supabase 仪表板中查看生成的嵌入，方法是访问表格编辑器，选择 vecs 模式，然后选择 image_vectors 表。

🌐 After activating the virtual environment with poetry shell you can now run your seed script via poetry run seed. You can inspect the generated embeddings in your Supabase Dashboard by visiting the Table Editor, selecting the vecs schema, and the image_vectors table.

根据文字查询进行图片搜索 #

🌐 Perform an image search from a text query

我们可以使用 Supabase Vector 来查询我们的嵌入。我们可以用一张图片作为搜索输入，或者从字符串输入生成嵌入：

🌐 We can use Supabase Vector to query our embeddings. We can either use an image as the search input or generate an embedding from a string input:

1
def search(query_term: Optional[str] = None):
2
    if query_term is None:
3
        query_term = sys.argv[1]
4
5
    # create vector store client
6
    vx = vecs.create_client(DB_CONNECTION)
7
    images = vx.get_or_create_collection(name="image_vectors", dimension=1024)
8
9
    # Encode text query
10
    text_emb = get_embedding_from_titan_multimodal(json.dumps(
11
        {
12
            "inputText": query_term,
13
            "embeddingConfig": {"outputEmbeddingLength": 1024},
14
        }
15
    ))
16
17
    # query the collection filtering metadata for "type" = "jpg"
18
    results = images.query(
19
        data=text_emb,                      # required
20
        limit=1,                            # number of records to return
21
        filters={"type": {"$eq": "jpg"}},   # metadata filters
22
    )
23
    result = results[0]
24
    print(result)
25
    plt.title(result)
26
    image = mpimg.imread('./images/' + result)
27
    plt.imshow(image)
28
    plt.show()

通过将查询限制为一个结果，我们可以向用户显示最相关的图片。最后，我们使用 matplotlib 向用户展示图片结果。

🌐 By limiting the query to one result, we can show the most relevant image to the user. Finally we use matplotlib to show the image result to the user.

去试试看吧，运行 poetry run search，你就会看到一张“红砖墙前的自行车”的图片。

🌐 Go ahead and test it out by running poetry run search and you will be presented with an image of a "bike in front of a red brick wall".

结论 #

🌐 Conclusion

只需几行 Python 代码，你就可以使用 Amazon Titan 多模态模型和 Supabase 向量实现图片搜索以及反向图片搜索。

🌐 With a couple of lines of Python you are able to implement image search as well as reverse image search using the Amazon Titan multimodal model and Supabase Vector.