基于Tablestore的多模态图片检索系统

1. 安装 SDK

1. 确保已安装 Python 3.12 及以上版本。

2. 执行以下命令安装 Tablestore Python SDK 和阿里云百炼 SDK。

   pip install tablestore
   pip install dashscope
   pip install Pillow

2. 配置环境变量

将访问凭证配置为环境变量，确保代码安全性与跨环境可移植性。

配置前请先获取百炼平台的API Key、AccessKey，前往表格存储控制台创建实例并获取实例名称和访问地址。

export DASHSCOPE_API_KEY=<百炼平台的API KEY>
export tablestore_end_point=<Tablestore实例访问地址>
export tablestore_instance_name=<Tablestore实例名称>
export tablestore_access_key_id=<AccessKey ID>
export tablestore_access_key_secret=<AccessKey Secret>

3. 准备图片数据

支持使用自定义图片数据或教程提供的演示数据集。

git clone https://github.com/aliyun/alibabacloud-tablestore-ai-demo.git

也可直接下载演示项目文件：alibabacloud-tablestore-ai-demo-main

1. 创建数据表

# -*- coding: utf-8 -*-
"""
创建 Tablestore 数据表
"""

import os
import tablestore


def get_client():
    """创建 Tablestore 客户端"""
    endpoint = os.getenv("tablestore_end_point")
    instance_name = os.getenv("tablestore_instance_name")
    access_key_id = os.getenv("tablestore_access_key_id")
    access_key_secret = os.getenv("tablestore_access_key_secret")

    client = tablestore.OTSClient(
        endpoint,
        access_key_id,
        access_key_secret,
        instance_name,
        retry_policy=tablestore.WriteRetryPolicy(),
    )
    return client


def create_table(client, table_name):
    """创建数据表"""
    try:
        # 主键定义
        schema_of_primary_key = [("image_id", "STRING")]
        table_meta = tablestore.TableMeta(table_name, schema_of_primary_key)
        table_options = tablestore.TableOptions()
        reserved_throughput = tablestore.ReservedThroughput(tablestore.CapacityUnit(0, 0))
        client.create_table(table_meta, table_options, reserved_throughput)
        print(f"表 '{table_name}' 创建成功")
    except tablestore.OTSClientError as e:
        print(f"创建表失败 (客户端错误): {e}")
    except tablestore.OTSServiceError as e:
        if "already exist" in str(e).lower() or "OTSObjectAlreadyExist" in str(e):
            print(f"表 '{table_name}' 已存在，跳过创建")
        else:
            print(f"创建表失败 (服务端错误): {e}")


def main():
    # 配置参数
    table_name = "multi_modal_retrieval"

    print("=" * 50)
    print("创建 Tablestore 数据表")
    print("=" * 50)

    # 创建客户端
    client = get_client()
    print("Tablestore 客户端创建成功")

    # 创建表
    print("\n创建数据表...")
    create_table(client, table_name)

    print("\n" + "=" * 50)
    print("数据表创建完成!")
    print("=" * 50)


if __name__ == "__main__":
    main()

2. 创建多元索引

向量数据在 Tablestore 数据表中以字符串格式存储。要启用向量检索功能，必须创建多元索引并配置向量字段类型，以支持高维向量的相似度计算和快速检索。

# -*- coding: utf-8 -*-
"""
创建 Tablestore 多元索引
"""

import os
import tablestore


def get_client():
    """创建 Tablestore 客户端"""
    endpoint = os.getenv("tablestore_end_point")
    instance_name = os.getenv("tablestore_instance_name")
    access_key_id = os.getenv("tablestore_access_key_id")
    access_key_secret = os.getenv("tablestore_access_key_secret")

    client = tablestore.OTSClient(
        endpoint,
        access_key_id,
        access_key_secret,
        instance_name,
        retry_policy=tablestore.WriteRetryPolicy(),
    )
    return client


def create_search_index(client, table_name, index_name, dimension=1024):
    """创建多元索引"""
    try:
        # 索引字段定义
        metadata_mappings = [
            tablestore.FieldSchema(
                "image_id",
                tablestore.FieldType.KEYWORD,
                index=True,
                enable_sort_and_agg=True,
            ),
            tablestore.FieldSchema(
                "city",
                tablestore.FieldType.KEYWORD,
                index=True,
                enable_sort_and_agg=True,
            ),
            tablestore.FieldSchema(
                "height",
                tablestore.FieldType.LONG,
                index=True,
                enable_sort_and_agg=True,
            ),
            tablestore.FieldSchema(
                "width",
                tablestore.FieldType.LONG,
                index=True,
                enable_sort_and_agg=True,
            ),
            tablestore.FieldSchema(
                "vector",
                tablestore.FieldType.VECTOR,
                vector_options=tablestore.VectorOptions(
                    data_type=tablestore.VectorDataType.VD_FLOAT_32,
                    dimension=dimension,
                    metric_type=tablestore.VectorMetricType.VM_COSINE,
                ),
            ),
        ]

        index_meta = tablestore.SearchIndexMeta(metadata_mappings)
        client.create_search_index(table_name, index_name, index_meta)
        print(f"多元索引 '{index_name}' 创建成功")
    except tablestore.OTSClientError as e:
        print(f"创建索引失败 (客户端错误): {e}")
    except tablestore.OTSServiceError as e:
        if "already exist" in str(e).lower() or "OTSObjectAlreadyExist" in str(e):
            print(f"多元索引 '{index_name}' 已存在，跳过创建")
        else:
            print(f"创建索引失败 (服务端错误): {e}")


def main():
    # 配置参数
    table_name = "multi_modal_retrieval"
    index_name = "index"
    dimension = 1024

    print("=" * 50)
    print("创建 Tablestore 多元索引")
    print("=" * 50)

    # 创建客户端
    client = get_client()
    print("Tablestore 客户端创建成功")

    # 创建索引
    print("\n创建多元索引...")
    create_search_index(client, table_name, index_name, dimension)

    print("\n" + "=" * 50)
    print("多元索引创建完成!")
    print("=" * 50)


if __name__ == "__main__":
    main()

自然语言检索

# -*- coding: utf-8 -*-
"""
语义检索示例
包含多种查询场景：
1. 仅使用查询文本进行语义检索
2. 使用查询文本 + 过滤条件（城市、高度、宽度）
"""

import os

import dashscope
import tablestore
from dashscope import MultiModalEmbeddingItemText


def get_client():
    """创建 Tablestore 客户端"""
    endpoint = os.getenv("tablestore_end_point")
    instance_name = os.getenv("tablestore_instance_name")
    access_key_id = os.getenv("tablestore_access_key_id")
    access_key_secret = os.getenv("tablestore_access_key_secret")

    client = tablestore.OTSClient(
        endpoint,
        access_key_id,
        access_key_secret,
        instance_name,
        retry_policy=tablestore.WriteRetryPolicy(),
    )
    return client


def text_to_embedding(text: str) -> list[float]:
    """将文本转换为向量"""
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemText(text=text, factor=1.0)]
    )
    if resp.status_code == 200:
        return resp.output["embeddings"][0]["embedding"]
    else:
        raise Exception(f"文本向量化失败: {resp.code} - {resp.message}")


def search_by_text_only(client, table_name, index_name, query_text: str, top_k: int = 10):
    """
    场景1: 仅使用查询文本进行语义检索
    """
    print(f"\n{'='*60}")
    print(f"场景1: 仅使用查询文本检索")
    print(f"查询文本: '{query_text}'")
    print(f"返回数量: {top_k}")
    print("="*60)

    # 文本向量化
    query_vector = text_to_embedding(query_text)

    # 构建向量查询
    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
    )

    # 按分数排序
    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(
        query,
        limit=top_k,
        get_total_count=False,
        sort=sort
    )

    # 执行搜索
    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def search_with_city_filter(client, table_name, index_name, query_text: str, city: str, top_k: int = 10):
    """
    场景2: 使用查询文本 + 城市过滤条件
    """
    print(f"\n{'='*60}")
    print(f"场景2: 查询文本 + 城市过滤")
    print(f"查询文本: '{query_text}'")
    print(f"城市过滤: {city}")
    print(f"返回数量: {top_k}")
    print("="*60)

    query_vector = text_to_embedding(query_text)

    # 构建带城市过滤的向量查询
    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
        filter=tablestore.TermQuery(field_name='city', column_value=city)
    )

    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(query, limit=top_k, get_total_count=False, sort=sort)

    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def search_with_size_filter(client, table_name, index_name, query_text: str,
                             height_range: tuple = None, width_range: tuple = None, top_k: int = 10):
    """
    场景3: 使用查询文本 + 尺寸过滤条件（高度、宽度）
    """
    print(f"\n{'='*60}")
    print(f"场景3: 查询文本 + 尺寸过滤")
    print(f"查询文本: '{query_text}'")
    print(f"高度范围: {height_range}")
    print(f"宽度范围: {width_range}")
    print(f"返回数量: {top_k}")
    print("="*60)

    query_vector = text_to_embedding(query_text)

    # 构建过滤条件
    must_queries = []
    if height_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='height',
            range_from=height_range[0],
            range_to=height_range[1],
            include_lower=True,
            include_upper=True
        ))
    if width_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='width',
            range_from=width_range[0],
            range_to=width_range[1],
            include_lower=True,
            include_upper=True
        ))

    vector_filter = tablestore.BoolQuery(must_queries=must_queries) if must_queries else None

    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
        filter=vector_filter
    )

    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(query, limit=top_k, get_total_count=False, sort=sort)

    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def search_with_combined_filters(client, table_name, index_name, query_text: str,
                                  cities: list = None, height_range: tuple = None,
                                  width_range: tuple = None, top_k: int = 10):
    """
    场景4: 使用查询文本 + 组合过滤条件（城市列表、高度、宽度）
    """
    print(f"\n{'='*60}")
    print(f"场景4: 查询文本 + 组合过滤条件")
    print(f"查询文本: '{query_text}'")
    print(f"城市列表: {cities}")
    print(f"高度范围: {height_range}")
    print(f"宽度范围: {width_range}")
    print(f"返回数量: {top_k}")
    print("="*60)

    query_vector = text_to_embedding(query_text)

    # 构建组合过滤条件
    must_queries = []

    if cities and len(cities) > 0:
        must_queries.append(tablestore.TermsQuery(field_name='city', column_values=cities))

    if height_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='height',
            range_from=height_range[0],
            range_to=height_range[1],
            include_lower=True,
            include_upper=True
        ))

    if width_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='width',
            range_from=width_range[0],
            range_to=width_range[1],
            include_lower=True,
            include_upper=True
        ))

    vector_filter = tablestore.BoolQuery(must_queries=must_queries) if must_queries else None

    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
        filter=vector_filter
    )

    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(query, limit=top_k, get_total_count=False, sort=sort)

    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def parse_search_hit(hit):
    """解析搜索结果"""
    row_item = {}
    primary_key = hit.row[0]
    row_item["image_id"] = primary_key[0][1]
    attribute_columns = hit.row[1]
    for col in attribute_columns:
        key = col[0]
        val = col[1]
        row_item[key] = val
    return row_item


def main():
    # 配置参数
    table_name = "multi_modal_retrieval"
    index_name = "index"

    print("=" * 60)
    print("Tablestore 多模态语义检索示例")
    print("=" * 60)

    # 创建客户端
    client = get_client()
    print("Tablestore 客户端创建成功")

    # 场景1: 仅使用自然语言描述进行语义检索
    # 使用完整的自然语言句子，而不是简单的关键词
    search_by_text_only(
        client, table_name, index_name,
        "一只毛茸茸的小狗在草地上奔跑",
        top_k=5
    )

    # 场景2: 自然语言描述 + 城市过滤
    search_with_city_filter(
        client, table_name, index_name,
        "湖边有一棵柳树，远处是连绵的山脉",
        city="hangzhou",
        top_k=5
    )

    # 场景3: 自然语言描述 + 尺寸过滤
    # 查找高分辨率的横向图片
    search_with_size_filter(
        client, table_name, index_name,
        "夜晚灯火通明的现代化城市天际线",
        height_range=(500, 1024),
        width_range=(800, 1024),
        top_k=5
    )

    # 场景4: 自然语言描述 + 组合过滤条件
    search_with_combined_filters(
        client, table_name, index_name,
        "远处是白雪覆盖的山峰，阳光洒在雪地上闪闪发光",
        cities=["hangzhou", "shanghai", "beijing"],
        height_range=(0, 1024),
        width_range=(0, 1024),
        top_k=5
    )

    print("\n" + "=" * 60)
    print("所有检索场景演示完成!")
    print("=" * 60)


if __name__ == "__main__":
    main()

以图搜图

# -*- coding: utf-8 -*-
"""
以图搜图示例
使用本地图片进行向量化，然后在 Tablestore 中检索相似图片
"""

import base64
import os
from pathlib import Path

import dashscope
import tablestore


def get_client():
    """创建 Tablestore 客户端"""
    endpoint = os.getenv("tablestore_end_point")
    instance_name = os.getenv("tablestore_instance_name")
    access_key_id = os.getenv("tablestore_access_key_id")
    access_key_secret = os.getenv("tablestore_access_key_secret")

    client = tablestore.OTSClient(
        endpoint,
        access_key_id,
        access_key_secret,
        instance_name,
        retry_policy=tablestore.WriteRetryPolicy(),
    )
    return client


def image_to_embedding(image_path: str) -> list[float]:
    """
    将本地图片转换为向量
    """
    # 读取图片并转换为 base64
    with open(image_path, "rb") as f:
        image_data = f.read()
    base64_image = base64.b64encode(image_data).decode("utf-8")

    # 根据文件后缀确定 MIME 类型
    suffix = Path(image_path).suffix.lower()
    if suffix in [".jpg", ".jpeg"]:
        mime_type = "image/jpeg"
    elif suffix == ".png":
        mime_type = "image/png"
    elif suffix == ".gif":
        mime_type = "image/gif"
    elif suffix == ".webp":
        mime_type = "image/webp"
    else:
        mime_type = "image/jpeg"  # 默认使用 jpeg

    # 构造 data URI
    data_uri = f"data:{mime_type};base64,{base64_image}"

    # 调用多模态向量化 API
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[{"image": data_uri, "factor": 1.0}]
    )

    if resp.status_code == 200:
        return resp.output["embeddings"][0]["embedding"]
    else:
        raise Exception(f"图片向量化失败: {resp.code} - {resp.message}")


def search_by_image(client, table_name, index_name, image_path: str, top_k: int = 10):
    """
    以图搜图: 使用本地图片进行语义检索
    """
    print(f"\n{'='*60}")
    print(f"以图搜图")
    print(f"查询图片: {image_path}")
    print(f"返回数量: {top_k}")
    print("="*60)

    # 图片向量化
    print("正在对查询图片进行向量化...")
    query_vector = image_to_embedding(image_path)
    print(f"向量化完成，维度: {len(query_vector)}")

    # 构建向量查询
    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
    )

    # 按分数排序
    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(
        query,
        limit=top_k,
        get_total_count=False,
        sort=sort
    )

    # 执行搜索
    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def search_by_image_with_filter(client, table_name, index_name, image_path: str,
                                 cities: list = None, height_range: tuple = None,
                                 width_range: tuple = None, top_k: int = 10):
    """
    以图搜图 + 过滤条件: 使用本地图片进行语义检索，同时应用过滤条件
    """
    print(f"\n{'='*60}")
    print(f"以图搜图 + 过滤条件")
    print(f"查询图片: {image_path}")
    print(f"城市列表: {cities}")
    print(f"高度范围: {height_range}")
    print(f"宽度范围: {width_range}")
    print(f"返回数量: {top_k}")
    print("="*60)

    # 图片向量化
    print("正在对查询图片进行向量化...")
    query_vector = image_to_embedding(image_path)
    print(f"向量化完成，维度: {len(query_vector)}")

    # 构建过滤条件
    must_queries = []

    if cities and len(cities) > 0:
        must_queries.append(tablestore.TermsQuery(field_name='city', column_values=cities))

    if height_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='height',
            range_from=height_range[0],
            range_to=height_range[1],
            include_lower=True,
            include_upper=True
        ))

    if width_range:
        must_queries.append(tablestore.RangeQuery(
            field_name='width',
            range_from=width_range[0],
            range_to=width_range[1],
            include_lower=True,
            include_upper=True
        ))

    vector_filter = tablestore.BoolQuery(must_queries=must_queries) if must_queries else None

    # 构建向量查询
    query = tablestore.KnnVectorQuery(
        field_name='vector',
        top_k=top_k,
        float32_query_vector=query_vector,
        filter=vector_filter
    )

    # 按分数排序
    sort = tablestore.Sort(sorters=[tablestore.ScoreSort(sort_order=tablestore.SortOrder.DESC)])
    search_query = tablestore.SearchQuery(query, limit=top_k, get_total_count=False, sort=sort)

    # 执行搜索
    search_response = client.search(
        table_name=table_name,
        index_name=index_name,
        search_query=search_query,
        columns_to_get=tablestore.ColumnsToGet(
            column_names=["image_id", "city", "height", "width"],
            return_type=tablestore.ColumnReturnType.SPECIFIED
        )
    )

    print(f"\nRequest ID: {search_response.request_id}")
    print(f"\n检索结果:")
    print("-" * 60)

    for idx, hit in enumerate(search_response.search_hits):
        row_item = parse_search_hit(hit)
        print(f"{idx + 1}. 得分: {hit.score:.4f} | {row_item}")

    return search_response.search_hits


def parse_search_hit(hit):
    """解析搜索结果"""
    row_item = {}
    primary_key = hit.row[0]
    row_item["image_id"] = primary_key[0][1]
    attribute_columns = hit.row[1]
    for col in attribute_columns:
        key = col[0]
        val = col[1]
        row_item[key] = val
    return row_item


def main():
    # 配置参数
    table_name = "multi_modal_retrieval"
    index_name = "index"

    print("=" * 60)
    print("Tablestore 以图搜图示例")
    print("=" * 60)

    # 创建客户端
    client = get_client()
    print("Tablestore 客户端创建成功")

    # 获取项目根目录
    current_dir = Path(__file__).parent
    data_dir = current_dir / "data" / "photograph"

    # 获取一张示例图片作为查询图片
    sample_images = list(data_dir.glob("*.jpg"))
    if not sample_images:
        print("错误: 未找到示例图片，请确保 data/photograph 目录下有 jpg 图片")
        return

    # 使用第一张图片作为查询示例
    query_image_path = str(sample_images[0])
    print(f"\n使用示例图片: {query_image_path}")

    # 场景1: 仅使用图片进行以图搜图
    search_by_image(client, table_name, index_name, query_image_path, top_k=5)

    # 场景2: 以图搜图 + 过滤条件
    # 只搜索特定城市的相似图片
    search_by_image_with_filter(
        client, table_name, index_name,
        query_image_path,
        cities=["hangzhou", "shanghai"],
        top_k=5
    )

    # 场景3: 以图搜图 + 尺寸过滤
    # 只搜索横向的相似图片（宽度大于高度）
    search_by_image_with_filter(
        client, table_name, index_name,
        query_image_path,
        width_range=(800, 1024),
        top_k=5
    )

    print("\n" + "=" * 60)
    print("以图搜图演示完成!")
    print("=" * 60)


if __name__ == "__main__":
    main()