RAG组件--向量数据库检索（搜索）、度量类型和条件过滤

Milvus混合检索实战

稠密向量

稠密向量通常表示为具有固定长度的浮点数数组，例如[0.2, 0.7, 0.1, 0.8, 0.3, …, 0.5]。这些向量的维数通常在数百到数千之间，例如 128、256、768 或 1024。每个维度都捕获对象的特定语义特征，使其通过相似度计算适用于各种场景。

• 定义：由固定长度的浮点数数组表示，如$[0.2,0.7,0.1,0.8,0.3,…,0.5]$
• 维度范围：通常在数百到数千之间（如128/256/768/1024维）
• 特性：每个维度捕获对象的特定语义特征，通过相似度计算适用于多种场景
• 语义表示：坐标值表示语义维度强度，点间距反映语义相似度（越近越相似）

稀疏向量

稀疏向量的特点是向量维数高且非零值较少。这种结构使其特别适合传统的信息检索应用。在大多数情况下，稀疏向量中使用的维度数对应于一种或多种语言中的不同标记。每个维度都被赋予一个值，该值指示该标记在文档中的相对重要性。这种布局对于涉及文本匹配的任务非常有利。

• 特点：维度数高但非零值少，传统信息检索中常用
• 表示方式：
  • 早期：二进制表示（0/1）
  • 现代：通过机器学习习得的实数值（如$\{2:0.2,…,9997:0.5,9999:0.7\}$）
• 适用场景：文本匹配任务，维度对应语言中的不同标记，值表示标记在文档中的相对重要性

混合检索

工作流程：

1. 通过嵌入模型（如BERT和Transformers）生成密集向量。
2. 通过BM25、BGE-M3、SPLADE等嵌入模型生成稀疏向量。（可以使用 Function生成稀疏向量）
3. 创建一个集合并定义包含密集和稀疏矢量场的集合模式。
4. 将稀疏密集向量插入到上一步刚刚创建的集合中。
5. 进行混合搜索：对密集向量进行 ANN 搜索将返回一组前 K 个最相似的结果，对稀疏向量进行文本匹配也将返回一组前 K 个结果。
6. 归一化：将两组top-K结果的得分进行归一化，将得分转换到[0,1]之间的范围。
7. 选择合适的重排序策略，对两组top-K结果进行合并、重排序，最终返回一组top-K结果。

重排序

优势：同时保留密集向量的语义相似度和稀疏向量的关键字查询优势

实战代码（说明）

# @Time：2025/9/2 09:15
# @Author：jinglv
import json
import time

from pymilvus import (
    connections,
    utility,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
    AnnSearchRequest,
    WeightedRanker
)
from pymilvus.exceptions import MilvusException
from pymilvus.model.hybrid import BGEM3EmbeddingFunction  # pip install "pymilvus[model]"

# 0. 配置 (方便修改)
DATA_PATH = "../../../data/灭神纪/战斗场景.json"
COLLECTION_NAME = "wukong_hybrid_v4"  # 使用新的集合名以避免旧数据冲突
MILVUS_URI = "http://82.157.193.65:19530"  # 使用新的数据库文件
BATCH_SIZE = 50  # 可以尝试减小批次大小，例如 10 或 20，进行测试
DEVICE = "cpu"  # 或者 "cuda" 如果有GPU并已正确配置

print("脚本开始执行...")

# 1. 加载数据
print(f"1. 正在从 {DATA_PATH} 加载数据...")
try:
    with open(DATA_PATH, 'r', encoding='utf-8') as f:
        dataset = json.load(f)
except FileNotFoundError:
    print(f"错误: 数据文件 {DATA_PATH} 未找到。请检查路径。")
    exit()
except json.JSONDecodeError:
    print(f"错误: 数据文件 {DATA_PATH} JSON 格式错误。")
    exit()

docs = []
metadata = []
for item in dataset.get('data', []):  # 使用 .get 避免 'data' 键不存在的错误
    text_parts = [item.get('title', ''), item.get('description', '')]
    if 'combat_details' in item and isinstance(item['combat_details'], dict):
        text_parts.extend(item['combat_details'].get('combat_style', []))
        text_parts.extend(item['combat_details'].get('abilities_used', []))
    if 'scene_info' in item and isinstance(item['scene_info'], dict):
        text_parts.extend([
            item['scene_info'].get('location', ''),
            item['scene_info'].get('environment', ''),
            item['scene_info'].get('time_of_day', '')
        ])
    # 过滤掉 None 和空字符串，然后连接
    docs.append(' '.join(filter(None, [str(part).strip() for part in text_parts if part])))
    metadata.append(item)

if not docs:
    print("错误: 未能从数据文件中加载任何文档。请检查文件内容和结构。")
    exit()
print(f"数据加载完成，共 {len(docs)} 条文档。")

# 2. 生成向量
print("2. 正在生成向量...")
try:
    ef = BGEM3EmbeddingFunction(use_fp16=False, device=DEVICE)
    docs_to_embed = docs
    print(f"将为 {len(docs_to_embed)} 条文档生成向量...")
    docs_embeddings = ef(docs_to_embed)
    print("向量生成完成。")
    print(f"  密集向量维度: {ef.dim['dense']}")
    if "sparse" in docs_embeddings and docs_embeddings["sparse"].shape[0] > 0:
        print(f"  稀疏向量类型 (整体): {type(docs_embeddings['sparse'])}")
        #  打印第一个稀疏向量的形状和部分内容以供检查
        first_sparse_vector_row_obj = docs_embeddings['sparse'][0]  # 这会得到一个表示单行的稀疏数组对象
        print(f"  第一个稀疏向量 (行对象类型): {type(first_sparse_vector_row_obj)}")
        print(f"  第一个稀疏向量 (行对象形状): {first_sparse_vector_row_obj.shape}")
        if hasattr(first_sparse_vector_row_obj, 'col') and hasattr(first_sparse_vector_row_obj, 'data'):
            print(f"  第一个稀疏向量 (部分列索引/col): {first_sparse_vector_row_obj.col[:5]}")
            print(f"  第一个稀疏向量 (部分数据/data): {first_sparse_vector_row_obj.data[:5]}")
        elif hasattr(first_sparse_vector_row_obj, 'indices') and hasattr(first_sparse_vector_row_obj,
                                                                         'data'):  # Fallback for other types
            print(f"  第一个稀疏向量 (部分索引/indices): {first_sparse_vector_row_obj.indices[:5]}")
            print(f"  第一个稀疏向量 (部分数据/data): {first_sparse_vector_row_obj.data[:5]}")
        else:
            print("  无法直接获取第一个稀疏向量的列索引和数据属性。")
    else:
        print("警告: 未生成稀疏向量或稀疏向量为空。")

except Exception as e:
    print(f"生成向量时发生错误: {e}")
    exit()

# 3. 连接Milvus
print(f"3. 正在连接 Milvus (URI: {MILVUS_URI})...")
try:
    connections.connect(uri=MILVUS_URI)
    print("成功连接到 Milvus。")
except MilvusException as e:
    print(f"连接 Milvus 失败: {e}")
    exit()

# 4. 创建集合
print(f"4. 正在准备集合 '{COLLECTION_NAME}'...")
fields = [
    FieldSchema(name="pk", dtype=DataType.VARCHAR, is_primary=True, auto_id=True, max_length=100),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=100),
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=128),
    FieldSchema(name="location", dtype=DataType.VARCHAR, max_length=256),
    FieldSchema(name="environment", dtype=DataType.VARCHAR, max_length=128),
    FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR),
    FieldSchema(name="dense_vector", dtype=DataType.FLOAT_VECTOR, dim=ef.dim["dense"])
]
schema = CollectionSchema(fields, description="Wukong Hybrid Search Collection v4")

try:
    if utility.has_collection(COLLECTION_NAME):
        print(f"集合 '{COLLECTION_NAME}' 已存在，正在删除...")
        utility.drop_collection(COLLECTION_NAME)
        print(f"集合 '{COLLECTION_NAME}' 删除成功。")
        time.sleep(1)

    print(f"正在创建集合 '{COLLECTION_NAME}'...")
    collection = Collection(name=COLLECTION_NAME, schema=schema, consistency_level="Strong")
    print(f"集合 '{COLLECTION_NAME}' 创建成功。")

    print("正在为 sparse_vector 创建索引 (SPARSE_INVERTED_INDEX, IP)...")
    collection.create_index("sparse_vector", {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"})
    print("sparse_vector 索引创建成功。")
    time.sleep(0.5)

    print("正在为 dense_vector 创建索引 (AUTOINDEX, IP)...")
    collection.create_index("dense_vector", {"index_type": "AUTOINDEX", "metric_type": "IP"})
    print("dense_vector 索引创建成功。")
    time.sleep(0.5)

    print(f"正在加载集合 '{COLLECTION_NAME}'...")
    collection.load()
    print(f"集合 '{COLLECTION_NAME}' 加载成功。")

except MilvusException as e:
    print(f"创建或加载集合/索引时发生 Milvus 错误: {e}")
    exit()
except Exception as e:
    print(f"创建或加载集合/索引时发生未知错误: {e}")
    exit()

# 5. 插入数据
print("5. 正在准备插入数据...")
num_docs_to_insert = len(docs_to_embed)
try:
    for i in range(0, num_docs_to_insert, BATCH_SIZE):
        end_idx = min(i + BATCH_SIZE, num_docs_to_insert)
        batch_data = []
        print(f"  正在准备批次 {i // BATCH_SIZE + 1} (索引 {i} 到 {end_idx - 1})...")

        for j in range(i, end_idx):
            item_metadata = metadata[j]

            # 关键：转换稀疏向量格式
            # 当从 csr_array 索引一行时，可能得到 coo_array 或其他稀疏格式
            sparse_row_obj = docs_embeddings["sparse"][j]
            # coo_array 使用 .col 和 .data
            if hasattr(sparse_row_obj, 'col') and hasattr(sparse_row_obj, 'data'):
                milvus_sparse_vector = {int(idx_col): float(val) for idx_col, val in
                                        zip(sparse_row_obj.col, sparse_row_obj.data)}
            # csr_array (如果直接是行 csr_array) 使用 .indices 和 .data
            elif hasattr(sparse_row_obj, 'indices') and hasattr(sparse_row_obj, 'data'):
                milvus_sparse_vector = {int(idx_col): float(val) for idx_col, val in
                                        zip(sparse_row_obj.indices, sparse_row_obj.data)}
            else:
                print(f"警告: 无法识别的稀疏行对象类型 {type(sparse_row_obj)} 在索引 {j}。跳过此条。")
                continue  # 或者引发错误

            doc_text = docs_to_embed[j]
            if len(doc_text) > 65530:
                doc_text = doc_text[:65530]

            title_text = item_metadata.get("title", "N/A")
            if len(title_text) > 500:
                title_text = title_text[:500]

            batch_data.append({
                "text": doc_text,
                "id": str(item_metadata.get("id", f"unknown_id_{j}")),
                "title": title_text,
                "category": item_metadata.get("category", "N/A"),
                "location": item_metadata.get("scene_info", {}).get("location", "N/A"),
                "environment": item_metadata.get("scene_info", {}).get("environment", "N/A"),
                "sparse_vector": milvus_sparse_vector,
                "dense_vector": docs_embeddings["dense"][j].tolist()
            })

        if not batch_data:  # 如果批次中所有稀疏向量都无法处理
            print(f"  批次 {i // BATCH_SIZE + 1} 为空，跳过插入。")
            continue

        print(f"  正在插入批次 {i // BATCH_SIZE + 1} ({len(batch_data)} 条记录)...")
        insert_result = collection.insert(batch_data)
        print(f"  批次 {i // BATCH_SIZE + 1} 插入成功, 主键: {insert_result.primary_keys[:5]}...")
        collection.flush()
        print(f"  批次 {i // BATCH_SIZE + 1} flush 完成。")
        time.sleep(0.5)

    print(f"所有数据插入完成。总共 {collection.num_entities} 条实体。")

except MilvusException as e:
    print(f"插入数据时发生 Milvus 错误: {e}")
    if 'batch_data' in locals() and batch_data:
        print("问题批次的第一条数据（部分）:")
        print(f"  Text: {batch_data[0]['text'][:100]}...")
        print(f"  ID: {batch_data[0]['id']}")
        print(f"  Title: {batch_data[0]['title']}")
    exit()
except Exception as e:
    print(f"插入数据时发生未知错误: {e}")
    if 'batch_data' in locals() and batch_data:
        print("问题批次的第一条数据（部分）:")
        print(f"  Text: {batch_data[0]['text'][:100]}...")
    exit()


# 6. 混合搜索 (示例)
def hybrid_search(query, category=None, environment=None, limit=5, weights=None):
    if weights is None:
        weights = {"sparse": 0.5, "dense": 0.5}

    print(f"\n6. 执行混合搜索: '{query}'")
    print(f"   Category: {category}, Environment: {environment}, Limit: {limit}, Weights: {weights}")

    try:
        query_embeddings = ef([query])

        conditions = []
        if category:
            conditions.append(f'category == "{category}"')
        if environment:
            conditions.append(f'environment == "{environment}"')
        expr = " && ".join(conditions) if conditions else None
        print(f"   过滤表达式: {expr}")

        search_params_dense = {"metric_type": "IP", "params": {}}
        search_params_sparse = {"metric_type": "IP", "params": {}}

        if expr:
            search_params_dense["expr"] = expr
            search_params_sparse["expr"] = expr

        dense_req = AnnSearchRequest(
            data=[query_embeddings["dense"][0].tolist()],
            anns_field="dense_vector",
            param=search_params_dense,
            limit=limit
        )

        # 转换查询稀疏向量格式
        query_sparse_row_obj = query_embeddings["sparse"][0]  # 索引返回单行稀疏对象
        if hasattr(query_sparse_row_obj, 'col') and hasattr(query_sparse_row_obj, 'data'):
            query_milvus_sparse_vector = {int(idx): float(val) for idx, val in
                                          zip(query_sparse_row_obj.col, query_sparse_row_obj.data)}
        elif hasattr(query_sparse_row_obj, 'indices') and hasattr(query_sparse_row_obj, 'data'):
            query_milvus_sparse_vector = {int(idx): float(val) for idx, val in
                                          zip(query_sparse_row_obj.indices, query_sparse_row_obj.data)}
        else:
            print(f"错误: 无法识别的查询稀疏向量类型 {type(query_sparse_row_obj)}。")
            return []

        sparse_req = AnnSearchRequest(
            data=[query_milvus_sparse_vector],
            anns_field="sparse_vector",
            param=search_params_sparse,
            limit=limit
        )

        rerank = WeightedRanker(weights["sparse"], weights["dense"])

        print("   发送混合搜索请求到 Milvus...")
        results = collection.hybrid_search(
            reqs=[sparse_req, dense_req],
            rerank=rerank,
            limit=limit,
            output_fields=["text", "id", "title", "category", "location", "environment", "pk"]
        )

        print("   搜索完成。结果:")
        if not results or not results[0]:
            print("   未找到结果。")
            return []

        processed_results = []
        for hit in results[0]:
            processed_results.append({
                "id": hit.entity.get("id"),
                "pk": hit.id,
                "title": hit.entity.get("title"),
                "text_preview": hit.entity.get("text", "")[:200] + "...",
                "category": hit.entity.get("category"),
                "location": hit.entity.get("location"),
                "environment": hit.entity.get("environment"),
                "distance": hit.distance
            })
        return processed_results

    except MilvusException as e:
        print(f"混合搜索时发生 Milvus 错误: {e}")
        return []
    except Exception as e:
        print(f"混合搜索时发生未知错误: {e}")
        return []


# 示例搜索调用
if collection.num_entities > 0:
    print("\n开始示例搜索...")
    search_results = hybrid_search("孙悟空的战斗技巧", category="神魔大战", limit=3)
    if search_results:
        for res in search_results:
            print(f"  - PK: {res['pk']}, Title: {res['title']}, Distance: {res['distance']:.4f}")
            print(f"    Category: {res['category']}, Location: {res['location']}")
            print(f"    Preview: {res['text_preview']}\n")

    search_results_filtered = hybrid_search("火焰山的战斗", environment="火山", limit=2)
    if search_results_filtered:
        for res in search_results_filtered:
            print(f"  - PK: {res['pk']}, Title: {res['title']}, Distance: {res['distance']:.4f}")
            print(f"    Category: {res['category']}, Location: {res['location']}, Environment: {res['environment']}")
            print(f"    Preview: {res['text_preview']}\n")
else:
    print("\n集合中没有实体，跳过示例搜索。")

print("\n脚本执行完毕。")

使用BGE-M3嵌入：

• 特点：多任务整合模型，可生成密集向量和稀疏向量
• 实现：BGEM3EmbeddingFunction(use_fp16=False, device=”cpu”)

混合检索逻辑

• 关键组件
  • AnnSearchRequest：处理两种向量类型的检索请求
  • 字段定义：需包含sparse_vector和dense_vector字段
• 索引创建
• 权重参数设计
  • 设计原则：权重值为相对比例而非百分比（如{“sparse”:0.7,”dense”:1.0}）
  • 调整策略：
    • 密集向量效果更好时增加dense权重
    • 稀疏向量效果更好时增加sparse权重
    • 效果相当时设置相近权重

Milvus多模态检索实战

多模态检索

• 本质理解: 多模态检索可理解为混合检索，即在原有检索基础上增加一种向量表示方式。
• 应用场景: 适用于需要同时支持文字和图片检索的场景，如博客图片搜索引擎、游戏场景搜索等。
• 核心组件:
  • 多模态编码器：如Visualized-BGE模型
  • 向量数据库：存储向量和元数据
  • 相似度搜索：基于余弦相似度等算法
• 模型特点:
  • 基于BGE文本嵌入框架，增加图像token嵌入能力
  • 保留原BGE模型的强大文本嵌入能力
  • 支持多种检索任务：多模态知识检索、组合图像检索等
• 模型版本:
  • bge-visualized-base-en-v1.5：768维，英文版
  • bge-visualized-m3：1024维，多语言版
• 安装使用:
  • 需要按官方指导逐步安装模型
  • 模型下载后需上传到服务器指定路径
  • 核心依赖包：torchvision、timm、einops等

visual_bge模型下载安装

github说明地址：https://github.com/FlagOpen/FlagEmbedding/tree/master/research/visual_bge

1. 下载工程安装依赖

   git clone https://github.com/FlagOpen/FlagEmbedding.git
   cd FlagEmbedding/research/visual_bge
   pip install -e .

1. 安装其他需要的依赖

   pip install torchvision timm einops ftfy

1. 测试是否安装成功:pip show visual-bge

   Name: visual_bge
   Version: 0.1.0
   Summary: visual_bge
   Home-page: https://github.com/FlagOpen/FlagEmbedding/tree/master/research/visual_bge
   Author: 
   Author-email: 
   License: 
   Location: /Users/jinglv/PycharmProjects/llm-rag-system/.venv/lib/python3.12/site-packages
   Editable project location: /Users/jinglv/PycharmProjects/llm-rag-system/FlagEmbedding/research/visual_bge
   Requires: einops, ftfy, timm, torchvision
   Required-by: 

导入报错问题：ModuleNotFoundError: No module named 'visual_bge'

• visualbge：是在下载的FlagEmbedding源码中，去找对应的位置，有包导入错误，或者不是python的包（目录下缺少：`_init.py`）补上对应的部分即可。

实战代码（说明）

# @Time：2025/9/2 09:45
# @Author：jinglv
"""
多模态图像检索系统：基于Visualized-BGE和Milvus实现
功能：对图像和文本进行多模态编码，并在图像数据库中检索相似内容
"""

import json
from dataclasses import dataclass
from typing import List

import cv2
import numpy as np
# ==================== 1. 初始化编码器 ====================
import torch
from PIL import Image
from pymilvus import MilvusClient
from tqdm import tqdm

from FlagEmbedding.research.visual_bge.visual_bge.modeling import Visualized_BGE


# from visual_bge.modeling import Visualized_BGE


class WukongEncoder:
    """多模态编码器：将图像和文本编码成向量"""

    def __init__(self, model_name: str, model_path: str):
        self.model = Visualized_BGE(model_name_bge=model_name, model_weight=model_path)
        self.model.eval()

    def encode_query(self, image_path: str, text: str) -> list[float]:
        """编码图像和文本的组合查询"""
        with torch.no_grad():
            query_emb = self.model.encode(image=image_path, text=text)
        return query_emb.tolist()[0]

    def encode_image(self, image_path: str) -> list[float]:
        """仅编码图像"""
        with torch.no_grad():
            query_emb = self.model.encode(image=image_path)
        return query_emb.tolist()[0]


# 初始化编码器
model_name = "BAAI/bge-m3"
model_path = "../../../visualized_models/Visualized_m3.pth"
encoder = WukongEncoder(model_name, model_path)


# ==================== 2. 数据集管理 ====================
@dataclass
class WukongImage:
    """图像元数据结构"""
    image_id: str
    file_path: str
    title: str
    category: str
    description: str
    tags: List[str]
    game_chapter: str
    location: str
    characters: List[str]
    abilities_shown: List[str]
    environment: str
    time_of_day: str


class WukongDataset:
    """图像数据集管理类"""

    def __init__(self, data_dir: str, metadata_path: str):
        self.data_dir = data_dir
        self.metadata_path = metadata_path
        self.images: List[WukongImage] = []
        self._load_metadata()

    def _load_metadata(self):
        """加载图像元数据"""
        with open(self.metadata_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            for img_data in data['images']:
                # 确保图片路径是相对于 data_dir 的
                img_data['file_path'] = f"{self.data_dir}/{img_data['file_path'].split('/')[-1]}"
                self.images.append(WukongImage(**img_data))


# 初始化数据集
dataset = WukongDataset("../../../data/多模态", "../../../data/多模态/metadata.json")

# ==================== 3. 生成图像嵌入 ====================
# 为所有图像生成嵌入向量
image_dict = {}
for image in tqdm(dataset.images, desc="生成图片嵌入"):
    try:
        image_dict[image.file_path] = encoder.encode_image(image.file_path)
    except Exception as e:
        print(f"处理图片 {image.file_path} 失败：{str(e)}")
        continue

print(f"成功编码 {len(image_dict)} 张图片")

# ==================== 4. Milvus向量库设置 ====================
# 连接/创建Milvus数据库
collection_name = "wukong_scenes"
milvus_client = MilvusClient(uri="http://82.157.193.65:19530")

# 创建向量集合
dim = len(list(image_dict.values())[0])
milvus_client.create_collection(
    collection_name=collection_name,
    dimension=dim,
    auto_id=True,
    enable_dynamic_field=True
)

# 插入数据到Milvus
insert_data = []
for image in dataset.images:
    if image.file_path in image_dict:
        insert_data.append({
            "image_path": image.file_path,
            "vector": image_dict[image.file_path],
            "title": image.title,
            "category": image.category,
            "description": image.description,
            "tags": ",".join(image.tags),
            "game_chapter": image.game_chapter,
            "location": image.location,
            "characters": ",".join(image.characters),
            "abilities": ",".join(image.abilities_shown),
            "environment": image.environment,
            "time_of_day": image.time_of_day
        })

result = milvus_client.insert(
    collection_name=collection_name,
    data=insert_data
)
print(f"索引构建完成，共插入 {result['insert_count']} 条记录")


# ==================== 5. 搜索功能实现 ====================
def search_similar_images(
        query_image: str,
        query_text: str,
        limit: int = 9
) -> List[dict]:
    """
    搜索相似图像
    参数:
        query_image: 查询图像路径
        query_text: 查询文本
        limit: 返回结果数量
    返回:
        检索结果列表
    """
    # 生成查询向量
    query_vec = encoder.encode_query(query_image, query_text)

    # 构建搜索参数
    search_params = {
        "metric_type": "COSINE",
        "params": {
            "nprobe": 10,
            "radius": 0.1,
            "range_filter": 0.8
        }
    }

    # 执行搜索
    results = milvus_client.search(
        collection_name=collection_name,
        data=[query_vec],
        output_fields=[
            "image_path", "title", "category", "description",
            "tags", "game_chapter", "location", "characters",
            "abilities", "environment", "time_of_day"
        ],
        limit=limit,
        search_params=search_params
    )[0]

    return results


# ==================== 6. 可视化函数 ====================
def visualize_results(query_image: str, results: List[dict], output_path: str):
    """
    可视化搜索结果
    参数:
        query_image: 查询图像路径
        results: 搜索结果列表
        output_path: 输出图像路径
    """
    # 设置图片大小和网格参数
    img_size = (300, 300)
    grid_size = (3, 3)

    # 创建画布
    canvas_height = img_size[0] * (grid_size[0] + 1)
    canvas_width = img_size[1] * (grid_size[1] + 1)
    canvas = np.full((canvas_height, canvas_width, 3), 255, dtype=np.uint8)

    # 添加查询图片
    query_img = Image.open(query_image).convert("RGB")
    query_array = np.array(query_img)
    query_resized = cv2.resize(query_array, (img_size[0] - 20, img_size[1] - 20))
    bordered_query = cv2.copyMakeBorder(
        query_resized, 10, 10, 10, 10,
        cv2.BORDER_CONSTANT,
        value=(255, 0, 0)
    )
    canvas[:img_size[0], :img_size[1]] = bordered_query

    # 添加结果图片
    for idx, result in enumerate(results[:grid_size[0] * grid_size[1]]):
        row = (idx // grid_size[1]) + 1
        col = idx % grid_size[1]

        img = Image.open(result["entity"]["image_path"]).convert("RGB")
        img_array = np.array(img)
        resized = cv2.resize(img_array, (img_size[0], img_size[1]))

        y_start = row * img_size[0]
        x_start = col * img_size[1]

        canvas[y_start:y_start + img_size[0], x_start:x_start + img_size[1]] = resized

        # 添加相似度分数
        score_text = f"Score: {result['distance']:.2f}"
        cv2.putText(
            canvas,
            score_text,
            (x_start + 10, y_start + img_size[0] - 10),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.5,
            (0, 0, 0),
            1
        )

    cv2.imwrite(output_path, canvas)


# ==================== 7. 执行查询示例 ====================
# 执行查询
query_image = "../../../data/多模态/query_image.jpg"
query_text = "寻找悟空面对建筑物战斗场景"

results = search_similar_images(
    query_image=query_image,
    query_text=query_text,
    limit=9
)

# 输出详细信息
print("\n搜索结果:")
for idx, result in enumerate(results):
    print(f"\n结果 {idx}:")
    print(f"图片：{result['entity']['image_path']}")
    print(f"标题：{result['entity']['title']}")
    print(f"描述：{result['entity']['description']}")
    print(f"相似度分数：{result['distance']:.4f}")

# 可视化结果
visualize_results(query_image, results, "search_results.jpg")

• 权重参数设计:

  • sparse权重0.7，dense权重1.0
  • 权重值不需要加起来等于1，表示相对重要性比例
  • 稠密向量重要性是稀疏向量的约1.43倍(1.0/0.7)
  • 可根据实际效果动态调整权重比例

• 数据集管理

  • 数据结构:
    • 包含图像ID、文件路径、标题、类别等基础信息
    • 详细描述字段：如战斗场景描述、角色、环境等
    • 标签列表：如”群战”、”水特效”等关键词
  • 构建方法:
    • 可使用大模型自动生成图像元数据
    • 元数据与图像文件需保持对应关系
    • 支持自定义字段扩展
• 检索功能的实现
  • 检索方式:
    • 支持纯图像检索
    • 支持图文混合检索
    • 支持带过滤条件的检索
  • 核心流程:
    • 初始化编码器并加载模型
    • 为所有图像生成嵌入向量
    • 创建Milvus向量集合
    • 执行相似度搜索
    • 可视化展示结果
  • 混合检索实现:
    • 同时编码图像和文本生成查询向量
    • 可设置过滤条件缩小搜索范围
    • 支持调整稀疏和稠密向量的权重比例

其它多模态解决方案

• Weaviate：原生支持图像、文本等多种模态的 Any-to-Any 检索 。
• Qdrant：结合 ImageBind、LlamaIndex 等模型，可实现音频、图像、文本等跨模态向量化检索。
• OpenSearch：通过神经插件（Neural Search）和 Titan 多模态嵌入模型，提供云端托管的多模态搜索能力。
• Pinecone：全托管向量库，开箱即可实现图像、音频、视频等多模态向量检索。
• Haystack：在 NLP 检索框架基础上，集成 CLIP 等模型，可扩展为多模态搜索系统。
• Vespa：支持多向量索引与复杂排名函数，擅长电商、新闻等场景下的多模态检索。