随着 AI 技术的发展，多模态混合检索（如图文互搜）成为主流需求。用户希望既能通过上传图片找到相似图片，也能通过文字描述搜索目标图像。实现该需求的核心在于两点：

向量生成：将图片、文本等不同模态数据转化为统一维度的向量。

向量检索：高效存储和查询海量向量，快速匹配相似结果。

在本文实践中，我们可以使用方舟中提供的图文向量化模型、图像理解模型生成 Milvus 实体，使用向量数据库 Milvus 版进行向量存储与近邻检索，二者结合起来可以实现图文多模态混合搜索的完整功能。

整体架构

架构组成

离线数据写入：将图片数据集通过图生文模型，文本向量模型，图像向量模型生成 Milvus 数据实体写入向量数据库 Milvus 版。

用户在线查询：用户可以通过输入文本、上传图像等方式进行搜索，将用户输入转换为向量后在Milvus中进行搜索得到结果。

核心流程：

数据写入：图片 > 方舟图像向量化 > 生成向量 > 写入向量数据库 Milvus。

检索：

图片搜图：用户上传图片 > 图片生成向量 > 向量数据库 Milvus 版检索相似向量 > 返回关联图片。

文本搜图：用户输入文本 > 文本生成向量 > 向量数据库 Milvus 版检索相似向量 > 返回关联图片。

前提条件

已创建实例且实例处于运行中状态。具体操作，请参见创建实例。

已获取实例的访问地址。具体操作，请参见连接实例。

已在火山方舟控制台创建用于所需模型的在线推理接入点，并配置接入点限流，具体操作，请参见获取 Endpoint ID（创建自定义推理接入点）和管理推理接入点。

已安装 pymilvus。安装命令如下。

# Install specific PyMilvus version
pip install pymilvus==2.5.8

# Update PyMilvus to the newest version
pip install --upgrade pymilvus

# Verify installation success
python -m pip list | grep pymilvus

已安装火山方舟 SDK。具体操作，请参见安装与初始化。

已下载 Milvus 官方数据集并解压。命令如下。

wget https://github.com/milvus-io/pymilvus-assets/releases/download/imagedata/reverse_image_search.zip
unzip -q -o reverse_image_search.zip

数据处理

核心代码逻辑

读取单张图片，转为 base64 编码（方舟模型要求图片输入为 base64）。

调用方舟图生文接入点，生成描述文本。

调用方舟 embedding 模型接入点，生成图片和文本向量。

图片数据处理

本地图像数据需要转换成 base64 编码，或者需要提供可以访问的图片 URL。

生成文本。

使用 doubao-1.5-thinking-vision-pro 模型为图片生成描述和关键词。

from volcenginesdkarkruntime import Ark

def image_gen_text(image_path):
image_base64 = encode_image(image_path)

ark_client = Ark(
api_key="<your_api_key>",
base_url="https://ark-cn-beijing.bytedance.net/api/v3",
timeout=120,
max_retries=2,
)

response = ark_client.chat.completions.create(
# 替换为您的推理接入点 ID
model="<your_model_id>",
messages=[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
# 需要注意：传入 Base64 编码前需要增加前缀 data:image/{图片格式};base64,{Base64 编码}：
"url": f"data:image/jpeg;base64,{image_base64}"
},
},
{
"type": "text",
"text": "先用 50 字内的文字描述这张图片，然后再给出 5 个关键词",
},
],
}
],
)
return response.choices[0].message.content

生成图片向量。

使用 Doubao-embedding-vision 模型根据图片生成向量。

from volcenginesdkarkruntime import Ark

def image_embedding(image_path):
image_base64 = encode_image(image_path)

ark_client = Ark(
api_key="<your_api_key>",
base_url="https://ark-cn-beijing.bytedance.net/api/v3",
timeout=120,
max_retries=2,
)

resp = ark_client.multimodal_embeddings.create(
model="<your_model_id>",
input=[
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
)
embedding = resp.data.embedding
# 确保向量是numpy数组并展平为一维
embedding = np.array(embedding).flatten()
print(f"向量维度：{embedding.shape}")
return embedding

生成文本向量。

使用 Doubao-embedding-vision 模型根据图片描述生成文本向量。

from volcenginesdkarkruntime import Ark

def text_embedding(text):
ark_client = Ark(
api_key="<your_api_key>",
base_url="https://ark-cn-beijing.bytedance.net/api/v3",
timeout=120,
max_retries=2,
)

resp = ark_client.multimodal_embeddings.create(
model="<your_model_id>",
input=[
{
"type": "text",
"text": text
}
]
)
embedding = resp.data.embedding
# 确保向量是numpy数组并展平为一维
embedding = np.array(embedding).flatten()
print(f"向量维度：{embedding.shape}")
return embedding

数据写入

设置 Schema 和 Index。

向量数据库 Milvus版中 Collection 类似传统数据库中的“表”，需先定义集合结构（含向量字段、元数据字段）。本示例将向量的 index_type 设置为 auto_index。metric_type 为距离类型，在搜索时如果要指定距离类型需要与 index_params 保持一致，支持的距离类型可参考度量类型。

字段名称	数据类型	索引类型	说明	度量类型
id	int64	auto_index	自动 ID。	-
image_path	varchar(512)	auto_index	图片地址。	-
image_embedding	float_vector(2048)	auto_index	图片向量。	L2
image_description	varchar(1024)	auto_index	图片描述。	-
image_desc_embedding	float_vector(2048)	auto_index	描述向量。	L2

from pymilvus import MilvusClient
from pymilvus import DataType

milvus_client = MilvusClient(
uri="http://milvus-xxx.milvus.ivolces.com:19530", # Milvus 实例的访问地址
token="<user>:<password>", # Milvus 实例的用户名和密码，替换为创建实例的用户名密码
)

def create_schema():
schema = MilvusClient.create_schema(
auto_id=False,
enable_dynamic_field=True,
)

schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, auto_id=True)
schema.add_field(field_name="image_path", datatype=DataType.VARCHAR, max_length=512)
schema.add_field(field_name="image_embedding", datatype=DataType.FLOAT_VECTOR, dim=2048)
schema.add_field(field_name="image_description", datatype=DataType.VARCHAR, max_length=1024)
schema.add_field(field_name="image_desc_embedding", datatype=DataType.FLOAT_VECTOR, dim=2048)

return schema

def create_index_params():
index_params = milvus_client.prepare_index_params()

index_params.add_index(
field_name="id",
index_type="AUTOINDEX"
)
index_params.add_index(
field_name="image_embedding",
index_type="AUTOINDEX",
metric_type="L2"
)
index_params.add_index(
field_name="image_description",
index_type="AUTOINDEX"
)
index_params.add_index(
field_name="image_desc_embedding",
index_type="AUTOINDEX",
metric_type="L2"
)

return index_params

您可根据实际需要将 index_type 设置为其他类型，例如 IVF_FLAT 或 HNSW。详情请参见 IVF_FLAT 和 HNSW。代码如下。

# IVF_FLAT类型
index_params.add_index(
field_name="your_vector_field_name", # Name of the vector field to be indexed
index_type="IVF_FLAT", # Type of the index to create
index_name="vector_index", # Name of the index to create
metric_type="L2", # Metric type used to measure similarity
params={
"nlist": 64, # Number of clusters for the index
} # Index building params
)

# HNSW类型
index_params.add_index(
field_name="your_vector_field_name", # Name of the vector field to be indexed
index_type="HNSW", # Type of the index to create
index_name="vector_index", # Name of the index to create
metric_type="L2", # Metric type used to measure similarity
params={
"M": 64, # Maximum number of neighbors each node can connect to in the graph
"efConstruction": 100 # Number of candidate neighbors considered for connection during index construction
} # Index building params
)

创建 Collection。

在已创建的向量数据库 Milvus 版实例中创建 Collection，完整代码如下。具体操作，请参见创建 Collection。

from pymilvus import MilvusClient
from pymilvus import DataType

milvus_client = MilvusClient(
uri="http://milvus-xxx.milvus.ivolces.com:19530", # Milvus 实例的访问地址
token="<user>:<password>", # Milvus 实例的用户名和密码，替换为创建实例的用户名密码
)

def create_collection_if_not_exists(collection_name):
if collection_name in milvus_client.list_collections():
return

schema = MilvusClient.create_schema(
auto_id=False,
enable_dynamic_field=True,
)

schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, auto_id=True)
schema.add_field(field_name="image_path", datatype=DataType.VARCHAR, max_length=512)
schema.add_field(field_name="image_embedding", datatype=DataType.FLOAT_VECTOR, dim=2048)
schema.add_field(field_name="image_description", datatype=DataType.VARCHAR, max_length=1024)
schema.add_field(field_name="image_desc_embedding", datatype=DataType.FLOAT_VECTOR, dim=2048)

index_params = milvus_client.prepare_index_params()

index_params.add_index(
field_name="id",
index_type="AUTOINDEX"
)
index_params.add_index(
field_name="image_embedding",
index_type="AUTOINDEX",
metric_type="L2"
)
index_params.add_index(
field_name="image_description",
index_type="AUTOINDEX"
)
index_params.add_index(
field_name="image_desc_embedding",
index_type="AUTOINDEX",
metric_type="L2"
)

milvus_client.create_collection(
collection_name=collection_name,
schema=schema,
index_params=index_params,
consistency_level="Bounded",
num_shards=1
)
print(f"Created collection {collection_name}")

批量写入图片数据。

从下载的数据集中的 reverse_image_search.csv 文件逐条读取图片数据写入向量数据库 Milvus 版。代码如下。

import pandas as pd
from tqdm import tqdm
from pymilvus import MilvusClient
from pymilvus import DataType

milvus_client = MilvusClient(
uri="http://milvus-xxx.milvus.ivolces.com:19530", # Milvus 实例的访问地址
token="<user>:<password>", # Milvus 实例的用户名和密码，替换为创建实例的用户名密码
)

def batch_insert():
df = pd.read_csv("reverse_image_search.csv")
success_count = 0
failure_count = 0
collection_name = "image_search_collection_demo"
create_collection_if_not_exists(collection_name)
for image_path in tqdm(df["path"].tolist(), desc="生成图像embedding"):
print(f"开始处理{image_path}")
try:
image_vector = image_embedding(image_path)
image_desc = image_gen_text(image_path)
image_desc_vector = text_embedding(image_desc)
milvus_client.insert(
collection_name=collection_name,
data=[{"image_path": image_path, "image_embedding":image_vector, "image_description": image_desc, "image_desc_embedding": image_desc_vector}]
)
print(f"成功插入{image_path}")
success_count += 1
time.sleep(1) # 控制API调用频率
except Exception as e:
print(f"处理{image_path}失败，已跳过: {str(e)}")
failure_count += 1
print(f"成功插入 {success_count} 张图片，失败 {failure_count} 张图片")