构建多模态图片语义检索系统

使用OSS的向量Bucket存储能力,结合大模型服务平台百炼的多模态 Embedding 模型,可以为海量图片构建智能语义检索系统,实现基于自然语言描述的文搜图能力,适用于电商商品搜索、智能相册、媒体资产管理等场景。

方案概览

搭建多模态图片语义检索系统,包含以下步骤:

  1. 准备图片数据:准备待检索的图片数据集,上传至Bucket进行存储。

  2. 向量化处理图片:使用百炼多模态Embedding模型将图片转换为高维向量。

  3. 创建向量索引并写入数据:创建向量Bucket 并建立向量索引,将图片向量数据写入索引。

  4. 执行语义检索:将文本查询转换为向量,在向量索引中执行相似度搜索,并可通过元数据进行过滤。

环境准备

获取访问凭证

安装SDK

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

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

    pip install alibabacloud-oss-v2
    pip install dashscope

配置环境变量

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

# 百炼 API Key
export DASHSCOPE_API_KEY=<您的百炼API-KEY>

# OSS 访问凭证
export oss_test_access_key_id=<您的AccessKey ID>
export oss_test_access_key_secret=<您的AccessKey Secret>
export oss_test_region=<您的Region,如cn-hangzhou>
export oss_test_account_id=<您的阿里云账号ID>

1. 准备图片数据

将本地图片上传到Bucket。百炼的Embedding模型需要通过URL访问这些图片进行向量化处理。以下代码演示了如何批量上传本地文件夹中的图片到指定的Bucket。

 -*- coding: utf-8 -*-
"""
示例:使用文件上传管理器上传图片

本示例展示如何使用 OSS SDK 的文件上传管理器进行更高效的文件上传。
适用于大文件或需要断点续传的场景。
"""

import os
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2.models import PutObjectRequest


def create_oss_client():
    """创建 OSS Client"""
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    return oss.Client(cfg)


def upload_with_uploader(client, bucket_name: str, local_path: str, oss_key: str):
    """
    使用上传管理器上传文件
    
    Args:
        client: OSS Client
        bucket_name: OSS Bucket 名称
        local_path: 本地文件路径
        oss_key: OSS 对象键
    """
    # 创建上传管理器
    uploader = client.uploader()
    
    # 执行上传
    result = uploader.upload_file(
        filepath=local_path,
        request=PutObjectRequest(
            bucket=bucket_name,
            key=oss_key
        )
    )
    return result


def main():
    client = create_oss_client()
    
    bucket_name = "your-bucket-name"
    # 注意:文末的 Github仓库里 data/photograph/ 目录已包含测试图片,可直接使用
    # 也可以修改 local_image_path 变量,指向自己的图片目录
    local_image_path = "data/photograph/"
    oss_prefix = "photograph/"
    
    image_files = os.listdir(local_image_path)
    print(f"待上传图片数量: {len(image_files)}")
    
    for i, image_name in enumerate(image_files, 1):
        local_path = os.path.join(local_image_path, image_name)
        oss_key = f"{oss_prefix}{image_name}"
        
        try:
            result = upload_with_uploader(client, bucket_name, local_path, oss_key)
            print(f"[{i}/{len(image_files)}] 上传成功: {image_name}, status: {result.status_code}")
        except Exception as e:
            print(f"[{i}/{len(image_files)}] 上传失败 {image_name}: {e}")
    
    print(f"\n上传完成!")


if __name__ == "__main__":
    main()

2. 向量化处理图片

使用百炼多模态 Embedding 模型将图片转换为 1024 维向量。

import dashscope
from dashscope import MultiModalEmbeddingItemImage


def embedding_image(image_url: str) -> list[float]:
    """
    将图片转换为向量
    
    Args:
        image_url: 图片的 URL 地址(支持 OSS 地址或公网 URL)
    
    Returns:
        1024 维的向量列表
    """
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemImage(image=image_url, factor=1.0)]
    )
    return resp.output["embeddings"][0]["embedding"]


def main():
    # 示例图片 URL(需替换为实际可访问的图片地址,如为图片为私有权限,需要填入带签名的临时 URL)
    image_url = "http://your-bucket-name.oss-cn-hangzhou.aliyuncs.com/photograph/Zsd0YhBa8LM.jpg"
    
    print(f"正在对图片进行向量化: {image_url}")
    
    # 调用 Embedding API
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemImage(image=image_url, factor=1.0)]
    )
    
    # 打印完整响应
    print("\n完整响应:")
    print(resp)
    
    # 获取向量
    embedding = resp.output["embeddings"][0]["embedding"]
    print(f"\n向量维度: {len(embedding)}")
    print(f"向量前10个元素: {embedding[:10]}")


if __name__ == "__main__":
    main()

3. 创建向量索引并写入数据

3.1 创建向量Bucket

创建一个向量Bucket,作为存储所有向量数据和索引的容器。

# -*- coding: utf-8 -*-
"""
示例:创建 VectorBucket

本示例展示如何创建一个 OSS VectorBucket。

前提条件:
1. 已安装 alibabacloud-oss-v2: pip install alibabacloud-oss-v2
2. 已设置环境变量(参考初始化 Client 示例)
"""

import os
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors


def main():
    # 从环境变量获取凭证
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # VectorBucket 名称
    vector_bucket_name = "my-test-2"
    
    print(f"正在创建 VectorBucket: {vector_bucket_name}")
    
    try:
        # 创建 VectorBucket
        result = client.put_vector_bucket(oss_vectors.models.PutVectorBucketRequest(
            bucket=vector_bucket_name,
        ))
        print(f"创建成功!")
        print(f"  status code: {result.status_code}")
        print(f"  request id: {result.request_id}")
    except Exception as e:
        print(f"创建失败: {e}")
        print("提示: 如果 Bucket 已存在,会返回错误")


if __name__ == "__main__":
    main()

3.2 创建向量索引

Bucket创建成功后,需要在其内部创建一个向量索引。索引定义了向量的结构(如维度)和检索方式(如距离度量),是存储和查询向量数据的基础。

# -*- coding: utf-8 -*-
"""
示例:创建向量索引(Index)

本示例展示如何在 VectorBucket 中创建向量索引。

前提条件:
1. 已安装 alibabacloud-oss-v2: pip install alibabacloud-oss-v2
2. 已设置环境变量
3. 已创建 VectorBucket
"""

import os
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors


def main():
    # 从环境变量获取凭证
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # 配置参数
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    dimension = 1024  # 百炼多模态模型输出向量维度
    
    print(f"正在创建向量索引:")
    print(f"  Bucket: {vector_bucket_name}")
    print(f"  Index: {vector_index_name}")
    print(f"  维度: {dimension}")
    
    # 创建向量索引
    result = client.put_vector_index(oss_vectors.models.PutVectorIndexRequest(
        bucket=vector_bucket_name,
        index_name=vector_index_name,
        dimension=dimension,
        data_type='float32',           # 向量数据类型
        distance_metric='cosine',       # 距离度量方式:余弦相似度
        metadata={
            "nonFilterableMetadataKeys": ["key1", "key2"]  # 不参与过滤的元数据字段
        }
    ))   
    print(f"\n创建成功!")
    print(f"  status code: {result.status_code}")
    print(f"  request id: {result.request_id}")

if __name__ == "__main__":
    main()

参数说明

参数

说明

dimension

向量维度,需与 Embedding 模型输出维度一致

data_type

向量数据类型,支持 float32

distance_metric

距离度量方式,支持 cosine、euclidean

metadata

元数据配置,配置非过滤元数据字段,用于存储不参与搜索过滤的附加信息,作为向量数据的描述信息

3.3 写入向量数据

索引准备就绪后,下一步是将向量数据上传到指定的向量索引,以便后续进行检索。

# -*- coding: utf-8 -*-
"""
示例:批量写入图片向量数据

本示例展示如何批量写入已经向量化的图片数据到向量索引中。

前提条件:
1. 已安装 alibabacloud-oss-v2: pip install alibabacloud-oss-v2
2. 已设置环境变量
3. 已创建向量索引
4. 已准备好图片向量数据文件(data/data.json)
"""

import os
import json
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors


def main():
    # 从环境变量获取凭证
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # 配置参数
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    
    # 加载预处理好的图片向量数据
    # 注意:文末的 Github仓库里 data/目录已包含测试文件,可直接使用
    # 也可以修改 data_file 变量,指向自己的图片目录
    data_file = "./data/data.json"
    print(f"正在加载图片向量数据: {data_file}")
    
    image_data_array = []
    with open(data_file, "r") as f:
        image_data_array = json.load(f)
    
    print(f"共加载 {len(image_data_array)} 条图片向量数据")
    
    # 打印数据样例
    if len(image_data_array) > 0:
        sample = image_data_array[0]
        print(f"\n数据样例:")
        print(f"  key: {sample.get('key', 'N/A')}")
        if 'metadata' in sample:
            print(f"  metadata: {sample['metadata']}")
        if 'data' in sample and 'float32' in sample['data']:
            print(f"  向量维度: {len(sample['data']['float32'])}")
    
    # 批量写入,每批 500 条
    batch_size = 500
    vectors = []
    total_written = 0
    
    print(f"\n开始批量写入 (batch_size={batch_size})...")
    
    for idx in range(len(image_data_array)):
        vectors.append(image_data_array[idx])
        
        if len(vectors) == batch_size:
            result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
                bucket=vector_bucket_name,
                index_name=vector_index_name,
                vectors=vectors,
            ))
            total_written += len(vectors)
            print(f"  已写入 {total_written}/{len(image_data_array)} 条, "
                  f"status code: {result.status_code}")
            vectors = []
    
    # 写入剩余数据
    if len(vectors) > 0:
        result = client.put_vectors(oss_vectors.models.PutVectorsRequest(
            bucket=vector_bucket_name,
            index_name=vector_index_name,
            vectors=vectors,
        ))
        total_written += len(vectors)
        print(f"  已写入 {total_written}/{len(image_data_array)} 条, "
              f"status code: {result.status_code}")
    
    print(f"\n写入完成! 共写入 {total_written} 条向量数据")


if __name__ == "__main__":
    main()

4. 执行语义检索

使用自然语言文本作为查询条件,在向量索引中检索最相似的图片。

4.1 基础检索

将查询文本(例如“狗狗”)向量化后,在索引中查找最接近的Top-K个图片向量。

# -*- coding: utf-8 -*-
"""
示例:向量检索查询(Query Vector)

本示例展示如何使用向量进行相似度检索,支持元数据过滤。

前提条件:
1. 已安装 alibabacloud-oss-v2 和 dashscope
2. 已设置环境变量
3. 已设置百炼 API Key: export DASHSCOPE_API_KEY=<您的 API Key>
4. 已写入向量数据
"""

import os
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors
import dashscope
from dashscope import MultiModalEmbeddingItemText


def embedding(text: str) -> list[float]:
    """
    文本向量化用于将查询文本转换为向量,实现文搜图能力
    
    Args:
        text: 待转换的文本
    
    Returns:
        1024 维的向量列表
    """
    return dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemText(text=text, factor=1.0)]
    ).output["embeddings"][0]["embedding"]


def main():
    # 从环境变量获取凭证
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    # 初始化 Client
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    client = oss_vectors.Client(cfg)
    
    # 配置参数
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    
    # 查询文本
    query_text = "狗狗"
    
    print(f"正在进行向量检索:")
    print(f"  Bucket: {vector_bucket_name}")
    print(f"  Index: {vector_index_name}")
    print(f"  查询文本: {query_text}")
    
    # 将查询文本转换为向量
    print(f"\n正在将查询文本转换为向量...")
    query_vector = embedding(query_text)
    print(f"  向量维度: {len(query_vector)}")
    
    # 执行向量检索
    print(f"\n执行向量检索 ...")
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=vector_bucket_name,
        index_name=vector_index_name,
        query_vector={
            "float32":query_vector
        },
        top_k=5,                    # 返回最相似的 5 个结果
        return_distance=True,       # 返回相似度距离
        return_metadata=True,       # 返回元数据
    ))
    
    print(f"\n检索结果 (共 {len(result.vectors)} 条):")
    for i, vector in enumerate(result.vectors, 1):
        print(f"\n  [{i}] key: {vector.get('key', 'N/A')}")
        if 'distance' in vector:
            print(f"      距离: {vector['distance']:.6f}")
        if 'metadata' in vector:
            print(f"      元数据: {vector['metadata']}")
    
    # 不带过滤条件的检索
    print(f"\n" + "=" * 50)
    print(f"执行向量检索 (无过滤条件)...")
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=vector_bucket_name,
        index_name=vector_index_name,
        query_vector={
            "float32": query_vector
        },
        top_k=5,
        return_distance=True,
        return_metadata=True,
    ))
    
    print(f"\n检索结果 (共 {len(result.vectors)} 条):")
    for i, vector in enumerate(result.vectors, 1):
        print(f"  [{i}] key: {vector.get('key', 'N/A')}, "
              f"距离: {vector.get('distance', 'N/A')}")


if __name__ == "__main__":
    main()

4.2 结合过滤条件检索

在进行向量相似度检索的同时,可以根据图片的元数据(如cityheight)进行精确过滤,以缩小检索范围。向量检索支持使用$in$and$or等操作符对元数据进行过滤。

# -*- coding: utf-8 -*-
"""
示例:高级向量检索查询(Advanced Query)

本示例展示向量检索的高级用法,包括复杂过滤条件和多个查询示例。

前提条件:
1. 已安装 alibabacloud-oss-v2 和 dashscope
2. 已设置环境变量
3. 已设置百炼 API Key: export DASHSCOPE_API_KEY=<您的 API Key>
4. 已写入向量数据
"""

import os
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.vectors as oss_vectors
import dashscope
from dashscope import MultiModalEmbeddingItemText


def embedding(text: str) -> list[float]:
    """文本向量化用于将查询文本转换为向量,实现文搜图能力"""
    return dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[MultiModalEmbeddingItemText(text=text, factor=1.0)]
    ).output["embeddings"][0]["embedding"]


def create_client():
    """创建 OSS Vector Client"""
    access_key_id = os.environ.get('oss_test_access_key_id')
    access_key_secret = os.environ.get('oss_test_access_key_secret')
    region = os.environ.get('oss_test_region')
    account_id = os.environ.get('oss_test_account_id')
    
    cfg = oss.config.load_default()
    cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id, access_key_secret
    )
    cfg.region = region
    cfg.account_id = account_id
    return oss_vectors.Client(cfg)


def query_with_filter(client, bucket, index, query_text, filter_body, top_k=5):
    """执行带过滤条件的向量检索"""
    result = client.query_vectors(oss_vectors.models.QueryVectorsRequest(
        bucket=bucket,
        index_name=index,
        query_vector={"float32": embedding(query_text)},
        filter=filter_body,
        top_k=top_k,
        return_distance=True,
        return_metadata=True,
    ))
    return result.vectors


def main():
    client = create_client()
    
    vector_bucket_name = "my-test-2"
    vector_index_name = "test1"
    
    print("=" * 60)
    print("高级向量检索示例")
    print("=" * 60)
    
    # 示例 1: $in 操作符 - 匹配多个城市
    print("\n【示例 1】使用 $in 操作符 - 查询杭州或上海的图片")
    print("-" * 40)
    filter_in = {
        "city": {"$in": ["hangzhou", "shanghai"]}
    }
    results = query_with_filter(client, vector_bucket_name, vector_index_name, 
                                "城市风景", filter_in)
    print(f"查询: '城市风景', 过滤: city in ['hangzhou', 'shanghai']")
    print(f"结果数量: {len(results)}")
    for v in results[:3]:
        print(f"  - {v.get('key')}: {v.get('metadata', {}).get('city', 'N/A')}")
    
    # 示例 2: $and 操作符 - 组合多个条件
    print("\n【示例 2】使用 $and 操作符 - 组合多个过滤条件")
    print("-" * 40)
    filter_and = {
        "$and": [
            {"city": {"$in": ["hangzhou", "shanghai"]}},
            {"height": {"$in": ["1024"]}}
        ]
    }
    results = query_with_filter(client, vector_bucket_name, vector_index_name,
                                "高楼大厦", filter_and)
    print(f"查询: '高楼大厦', 过滤: city in [hangzhou, shanghai] AND height=1024")
    print(f"结果数量: {len(results)}")
    for v in results[:3]:
        meta = v.get('metadata', {})
        print(f"  - {v.get('key')}: city={meta.get('city')}, height={meta.get('height')}")
    
    # 示例 3: 不同查询文本的比较
    print("\n【示例 3】不同查询文本的语义检索效果")
    print("-" * 40)
    query_texts = ["狗狗", "海边日落", "城市夜景", "美食"]
    
    for qt in query_texts:
        results = query_with_filter(client, vector_bucket_name, vector_index_name,
                                    qt, None, top_k=3)
        print(f"\n查询: '{qt}'")
        for i, v in enumerate(results, 1):
            print(f"  [{i}] {v.get('key')}, 距离: {v.get('distance', 0):.4f}")


if __name__ == "__main__":
    main()

过滤条件说明

操作符

说明

示例

$in

包含在列表中

{"city": {"$in": ["hangzhou", "beijing"]}}

$and

逻辑与

{"$and": [条件1, 条件2]}

$or

逻辑或

{"$or": [条件1, 条件2]}

复杂过滤示例

{
    "$and": [
        {"city": {"$in": ["hangzhou", "shanghai"]}},
        {
            "$or": [
                {"height": "1024"},
                {"height": "768"}
            ]
        }
    ]
}

4.3 构建可视化检索界面

为直观地展示检索效果,可使用Gradio构建一个简单的Web界面,提供一个包含文本输入、条件过滤和图片结果展示的交互式检索界面。

  1. 安装Web UI 框架

    pip install gradio==5.44.1
  2. 将以下代码保存为 gradio_app.py

    # -*- coding: utf-8 -*-
    
    import json
    import logging
    import os
    
    import alibabacloud_oss_v2 as oss
    import alibabacloud_oss_v2.vectors as oss_vectors
    import dashscope
    import gradio as gr
    from PIL import Image
    from dashscope import MultiModalEmbeddingItemText
    
    logging.basicConfig(level=logging.INFO)
    
    logger = logging.getLogger(__name__)
    
    
    class Util:
        access_key_id = os.environ.get('oss_test_access_key_id')
        access_key_secret = os.environ.get('oss_test_access_key_secret')
        region = os.environ.get('oss_test_region')
        account_id = os.environ.get('oss_test_account_id')
    
        cfg = oss.config.load_default()
        cfg.credentials_provider = oss.credentials.StaticCredentialsProvider(access_key_id, access_key_secret)
        cfg.region = region
        cfg.account_id = account_id
        client = oss_vectors.Client(cfg)
    
        vector_bucket_name = "my-test-2"
        vector_index_name = "test1"
        dimension = 1024
    
        @staticmethod
        def embedding(text) -> list[float]:
            return dashscope.MultiModalEmbedding.call(
                model="multimodal-embedding-v1",
                input=[MultiModalEmbeddingItemText(text=text, factor=1.0)]
            ).output["embeddings"][0]["embedding"]
    
        @staticmethod
        def query_text(text: str, top_k: int = 5, city: list[str] = None, height: list[str] = None, return_meta: bool = True, return_distance: bool = True) -> list[tuple[Image.Image, str]]:
            logger.info(f"search text:{text}, top_k:{top_k}, city:{city}, height:{height}")
    
            sub_filter = []
            if city is not None and len(city) > 0:
                sub_filter.append({"city": {"$in": city}})
            if height is not None and len(height) > 0:
                sub_filter.append({"height": {"$in": height}})
            if len(sub_filter) > 0:
                filter_body = {"$and": sub_filter}
            else:
                filter_body = None
    
            result = Util.client.query_vectors(oss_vectors.models.QueryVectorsRequest(
                bucket=Util.vector_bucket_name,
                index_name=Util.vector_index_name,
                query_vector={
                    "float32": Util.embedding(text)
                },
                filter=filter_body,
                top_k=top_k,
                return_distance=return_distance,
                return_metadata=return_meta,
            ))
    
            gallery_data = []
            current_dir = os.path.dirname(os.path.abspath(__file__))
            # 前端展示依赖本地图片文件,请确保已按仓库结构准备图片资源:
            # - 默认使用文末仓库内 data/photograph/ 目录下的图片,前端会读取并展示其中的文件;
            # - 如需使用自己的图片,可将图片放入其他目录,并修改下方 path 指向该目录。
            for vector in result.vectors:
                file_path = os.path.join(current_dir, "data/photograph/", vector["key"])
                img = Image.open(file_path)
                gallery_data.append((img, json.dumps(vector)))
            ret = gallery_data
            logger.info(f"search text:{text}, top_k:{top_k}, request_id:{result.request_id}, ret:{ret}")
            return ret
    
        @staticmethod
        def on_gallery_box_select(evt: gr.SelectData):
            result = ""
            img_data = evt.value["caption"]
            img_data = json.loads(img_data)
            for key in img_data:
                img_data_item = img_data[key]
                if type(img_data_item) is str:
                    img_data_item = img_data_item.replace("\n", "\\n").replace("\t", "\\t").replace("\r", "\\r")
                if type(img_data_item) is dict:
                    for sub_key in img_data_item:
                        img_data_item[sub_key] = img_data_item[sub_key].replace("\n", "\\n").replace("\t", "\\t").replace("\r", "\\r")
                        result += f' - **{sub_key}**: &nbsp; {img_data_item[sub_key]}\r\n'
                    continue
                result += f' - **{key}**: &nbsp; {img_data_item}\r\n'
            return result
    
    
    with gr.Blocks(title="OSS Demo") as demo:
        with gr.Tab("OSS QueryVector 图片示例") as search_tab:
            with gr.Row():
                query_text_box = gr.Textbox(label='query_text', interactive=True, value="狗狗")
                top_k_box = gr.Slider(minimum=1, maximum=30, value=10, step=1, label='top_k', interactive=True)
                with gr.Column():
                    return_meta_box = gr.Checkbox(label='return_meta', interactive=True, value=True)
                    return_distance_box = gr.Checkbox(label='return_distance', interactive=True, value=True)
            with gr.Row():
                city_box = gr.Dropdown(label='city', multiselect=True, choices=["hangzhou", "shanghai", "beijing", "shenzhen", "guangzhou"])
                height_box = gr.Dropdown(label='height', multiselect=True, choices=["1024", "683", "768", "576"])
            with gr.Row():
                query_button = gr.Button(value="query", variant='primary')
            with gr.Row():
                with gr.Column(scale=8):
                    gallery_box = gr.Gallery(columns=5, show_label=False, preview=False, allow_preview=False, visible=True, show_download_button=False)
                with gr.Column(scale=2):
                    with gr.Row(variant="panel"):
                        md_box = gr.Markdown(visible=True, elem_classes="image_detail")
                gallery_box.select(Util.on_gallery_box_select, [], [md_box])
            query_button.click(
                Util.query_text,
                inputs=[
                    query_text_box,
                    top_k_box,
                    city_box,
                    height_box,
                    return_meta_box,
                    return_distance_box
                ],
                outputs=[
                    gallery_box,
                ],
                concurrency_limit=1,
            )
    
    if __name__ == "__main__":
        demo.launch(server_name="0.0.0.0", server_port=7860)
  3. 启动界面

    python gradio_app.py

    启动成功后,访问 http://localhost:7860 即可使用检索界面。检索示例:输入"狗狗" → 返回包含狗的图片。2025-11-26_19-32-56 (1)

    功能

    说明

    query_text

    输入自然语言描述,如"狗狗"、"山峰"等

    top_k

    设置返回结果数量(1-30)

    city

    按城市过滤(支持多选)

    height

    按图片高度过滤(支持多选)

    return_meta

    是否返回元数据信息

    return_distance

    是否返回相似度距离

相关文档

本教程的完整项目工程请参见https://github.com/aliyun/alibabacloud-oss-vector-index-demo