多模态AIAgent开发实践 邓立国、周驰岷、邓淇等 清华大学出版社【行情 报价 价格 评测】-京东

《多模态AI Agent开发实践》1~6章试读-CSDN博客

多模态数据检索是多模态智能体实现“快速获取相关数据、辅助推理决策”的核心能力，其核心是基于特征向量的相似性检索，结合多模态数据类型的差异性，采用差异化检索策略。本节将讲解多模态数据检索的核心原理、常用策略，结合前文集成的向量数据库与qwen-vl-plus大模型，实现多模态检索实操，提升检索精度与效率。

5.4.1  多模态数据检索核心原理

多模态数据检索的核心是“特征向量相似性匹配”，具体原理如下：

（1）检索触发：多模态智能体接收用户指令（如图像查询、文本提问、语音指令），通过qwen-vl-plus解析指令核心需求。

（2）特征生成：将用户输入的多模态指令转换为特征向量（如文本指令生成文本特征、图像查询生成图像特征）。

（3）相似性匹配：向量数据库计算用户输入特征向量，以及库中存储的多模态特征向量的相似度（常用余弦相似度、欧氏距离）。

（4）结果返回：返回相似度Top N的多模态数据，结合qwen-vl-plus大模型，将检索结果整理为自然语言响应，反馈给用户。

其核心优势：无须依赖传统的关键词匹配，可实现跨模态检索（如用文本查询相似图片、用图片查询相关音频），适配多模态智能体的交互需求。

5.4.2  常用多模态数据检索策略

结合多模态场景与向量数据库特性，重点讲解4种常用检索策略，均适配qwen-vl-plus大模型与指定依赖版本，通过实操案例说明用法。

1. 单一模态检索（基础策略）

单一模态检索是指“输入与输出为同一模态”，如文本查询文本、图片查询图片，是最基础、最常用的检索策略，代码示例（基于Chroma数据库）：

# 单一模态检索（文本→文本、图片→图片）

from langchain.vectorstores import Chroma

from langchain_community.embeddings import QwenEmbeddings

# 初始化数据库与嵌入模型

embeddings = QwenEmbeddings(api_key=os.getenv("QWen_API_KEY"))

chroma_db = Chroma(persist_directory="./chroma_multimodal_db", embedding_function=embeddings, collection_name="multimodal_collection")

# 1. 文本→文本检索（查询与工业巡检相关的文本）

text_query = "工业巡检管道维护方法"

text_results = chroma_db.similarity_search(text_query, k=2)

print("文本→文本检索结果：")

for result in text_results:

    print(f"- 内容：{result.page_content[:80]}...")

# 2. 图片→图片检索（用一幅图片查询相似图片）

# 加载查询图片，提取特征向量

query_img_loader = ImageLoader("industrial_query.jpg")

query_img_doc = query_img_loader.load()[0]

query_img_desc = llm.invoke(f"简洁描述图片核心内容：{query_img_doc.page_content}")

query_img_embedding = embeddings.embed_query(query_img_desc)

# 检索相似图片

img_results = chroma_db.similarity_search_by_vector(query_img_embedding, k=1, filter={"type": "image"})

print("\n图片→图片检索结果：")

print(f"- 图片路径：{img_results[0].metadata['image_path']}，描述：{img_results[0].page_content[:80]}...")

2. 跨模态检索（核心策略）

跨模态检索是指“输入与输出为不同模态”，如文本查询图片、音频查询文本，是多模态智能体的核心检索能力，代码示例（文本→图片、音频→文本）：

# 跨模态检索（文本→图片、音频→文本）

# 1. 文本→图片检索（用文本指令查询相关图片）

text_to_img_query = "工业巡检管道泄漏图像"

text_to_img_embedding = embeddings.embed_query(text_to_img_query)

img_results = chroma_db.similarity_search_by_vector(text_to_img_embedding, k=1, filter={"type": "image"})

print("文本→图片检索结果：")

print(f"- 图片路径：{img_results[0].metadata['image_path']}，描述：{img_results[0].page_content[:80]}...")

# 2. 音频→文本检索（用音频转写文本查询相关文本）

# 加载音频，转写为文本并提取特征

audio_query_loader = WhisperLoader("industrial_query_audio.mp3", model="base", language="zh")

audio_query_doc = audio_query_loader.load()[0]

audio_query_text = audio_query_doc.page_content

audio_to_text_embedding = embeddings.embed_query(audio_query_text)

text_results = chroma_db.similarity_search_by_vector(audio_to_text_embedding, k=2, filter={"type": "text"})

print("\n音频→文本检索结果：")

for result in text_results:

    print(f"- 内容：{result.page_content[:80]}...")

3. 多条件筛选检索（精准策略）

结合元数据筛选与相似性检索，实现精准检索，适用于多模态数据量大、需精准定位的场景，代码示例：

# 多条件筛选检索（筛选文本类型＋工业巡检主题）

query = "管道故障检测"

# 筛选条件：数据类型为文本，且内容与管道故障相关

filtered_results = chroma_db.similarity_search(

    query,

    k=2,

    filter={"type": "text"}  # 筛选文本类型数据

)

# 结合qwen-vl-plus进一步筛选，确保与管道故障相关

refined_results = []

for result in filtered_results:

    relevance = llm.invoke(f"判断以下文本是否与管道故障检测相关：{result.page_content}，仅回答是或否")

    if relevance == "是":

        refined_results.append(result)

print("多条件筛选检索结果：")

for result in refined_results:

    print(f"- 内容：{result.page_content[:80]}...")

4. 批量检索（高效策略）

针对大量多模态查询需求，采用批量检索方式，提升检索效率，代码示例：

# 批量检索（同时查询多个文本指令，获取对应结果）

batch_queries = ["管道泄漏检测", "设备破损识别", "巡检报告生成"]

# 批量生成查询特征向量

batch_embeddings = embeddings.embed_documents(batch_queries)

# 批量检索

batch_results = chroma_db.similarity_search_by_vector(batch_embeddings, k=1)

print("批量检索结果：")

for i, (query, results) in enumerate(zip(batch_queries, batch_results)):

    print(f"{i+1}. 查询：{query}，结果：{results[0].page_content[:80]}...")

5.4.3  多模态数据检索策略综合案例实现

【示例5.1】本示例对以上四种模态检索的综合实现。

# Multimodal_data_retrieval.py

# -*- coding: utf-8 -*-

import os

import warnings

from dotenv import load_dotenv

import whisper

from PIL import Image

# 新版稳定导入（无报错）

from langchain_chroma import Chroma

from langchain_community.embeddings import DashScopeEmbeddings

from langchain_community.llms import Tongyi

from langchain_core.documents import Document

# 忽略警告

warnings.filterwarnings("ignore")

load_dotenv()

# ===================== 1. 核心组件初始化 =====================

# 通义千问嵌入模型

embeddings = DashScopeEmbeddings(

    model="text-embedding-v2",

    dashscope_api_key=os.getenv("QWEN_API_KEY")

)

# 通义千问 VL 多模态大模型

llm = Tongyi(

    dashscope_api_key=os.getenv("QWEN_API_KEY"),

    model_name="qwen-vl-plus",

    temperature=0.0

)

# 向量数据库

chroma_db = Chroma(

    persist_directory="./chroma_multimodal_db",

    embedding_function=embeddings,

    collection_name="multimodal_collection"

)

# ===================== 2. 加载真实图片 ＋ 真实音频工具 =====================

def get_image_description(image_path: str) -> str:

    """使用Qwen-VL-Plus 真实理解图片内容"""

    prompt = f"图片描述：<img>{image_path}</img> 请用简洁文字描述这幅工业巡检图片的核心内容"

    return llm.invoke(prompt)

def audio_to_text(audio_path: str) -> str:

    """使用Whisper 真实语音转文字"""

    model = whisper.load_model("base")

    result = model.transcribe(audio_path, language="zh")

    return result["text"].strip()

# ===================== 3. 初始化多模态数据（含真实图片元数据） =====================

def init_multimodal_data():

    try:

        chroma_db.delete_collection()

    except:

        pass

    db = Chroma(

        persist_directory="./chroma_multimodal_db",

        embedding_function=embeddings,

        collection_name="multimodal_collection"

    )

    # 文本数据

    text_docs = [

        Document(

            page_content="工业巡检管道维护方法：定期检查管道接口密封性，清理管道内壁污垢，及时更换老化管道",

            metadata={"type": "text", "topic": "工业巡检"}

        ),

        Document(

            page_content="管道故障检测流程：超声波探伤、观察渗漏、定位故障、上报维修",

            metadata={"type": "text", "topic": "工业巡检"}

        ),

        Document(

            page_content="设备破损识别标准：外壳变形、零件脱落、运行异响均需停机检修",

            metadata={"type": "text", "topic": "工业巡检"}

        )

    ]

    # 真实图片数据（读取真实图片并生成描述）

    img_desc = get_image_description("industrial_leak.jpg")

    img_docs = [

        Document(

            page_content=img_desc,

            metadata={

                "type": "image",

                "image_path": "industrial_leak.jpg",

                "topic": "工业巡检"

            }

        )

    ]

    db.add_documents(text_docs + img_docs)

    print("✅ 真实多模态数据初始化完成！")

    return db

chroma_db = init_multimodal_data()

# ===================== 策略1：单一模态检索 =====================

def single_modal_search():

    print("\n" + "="*50)

    print(" 策略1：单一模态检索")

    print("="*50)

    # 文本→文本

    text_query = "工业巡检管道维护方法"

    text_results = chroma_db.similarity_search(text_query, k=2, filter={"type": "text"})

    print("🔍  文本→文本检索结果：")

    for i, res in enumerate(text_results):

        print(f"{i+1}. {res.page_content}")

    # 图片→图片（真实查询图）

    print("\n 图片→图片检索（真实图片查询）：")

    query_img_desc = get_image_description("industrial_leak.jpg")

    query_emb = embeddings.embed_query(query_img_desc)

    img_res = chroma_db.similarity_search_by_vector(query_emb, k=1, filter={"type": "image"})

    print(f"图片路径：{img_res[0].metadata['image_path']}")

    print(f"图片内容：{img_res[0].page_content}")

# ===================== 策略2：跨模态检索（真实图片 ＋ 真实音频）=====================

def cross_modal_search():

    print("\n" + "="*50)

    print(" 策略2：跨模态检索（真实文件版）")

    print("="*50)

    # 1. 文本 → 真实图片

    print(" 文本→图片（检索真实 industrial_leak.jpg）：")

    text_query = "工业管道泄漏的现场图片"

    emb = embeddings.embed_query(text_query)

    img_res = chroma_db.similarity_search_by_vector(emb, k=1, filter={"type": "image"})

    print(f"✅ 匹配到真实图片：{img_res[0].metadata['image_path']}")

    print(f"图片内容：{img_res[0].page_content}")

    # 2. 真实音频 → 文本（读取真实MP3并转写）

    print("\n 音频→文本（读取真实 industrial_query_audio.mp3）：")

    try:

        audio_text = audio_to_text("industrial_query_audio.mp3")

        print(f"音频转文字结果：{audio_text}")

        audio_emb = embeddings.embed_query(audio_text)

        text_res = chroma_db.similarity_search_by_vector(audio_emb, k=2, filter={"type": "text"})

        for i, res in enumerate(text_res):

            print(f"{i+1}. {res.page_content}")

    except Exception as e:

        print(f"音频文件未找到，使用模拟内容演示：{e}")

        audio_text = "管道故障检测流程是什么"

        audio_emb = embeddings.embed_query(audio_text)

        text_res = chroma_db.similarity_search_by_vector(audio_emb, k=2, filter={"type": "text"})

        for i, res in enumerate(text_res):

            print(f"{i+1}. {res.page_content}")

# ===================== 策略3：多条件筛选检索 =====================

def filtered_search():

    print("\n" + "="*50)

    print(" 策略3：多条件筛选检索")

    print("="*50)

    query = "管道故障检测"

    results = chroma_db.similarity_search(query, k=2, filter={"type": "text"})

    refined = []

    for res in results:

        try:

            ans = llm.invoke(f"以下内容是否与管道故障检测相关？只回答是或否：{res.page_content}").strip()

            if "是" in ans:

                refined.append(res)

        except:

            refined.append(res)

    print(" 大模型二次校验结果：")

    for i, res in enumerate(refined):

        print(f"{i+1}. {res.page_content}")

# ===================== 策略4：批量检索 =====================

def batch_search():

    print("\n" + "="*50)

    print(" 策略4：批量检索")

    print("="*50)

    queries = ["管道泄漏检测", "设备破损识别", "巡检报告生成"]

    embs = embeddings.embed_documents(queries)

    results = [chroma_db.similarity_search_by_vector(e, k=1) for e in embs]

    for i, (q, r) in enumerate(zip(queries, results)):

        print(f"{i+1}. 查询：{q}")

        print(f"   结果：{r[0].page_content}\n")

# ===================== 运行全部 =====================

if __name__ == "__main__":

    single_modal_search()

    cross_modal_search()    # ✅ 已接入真实图片 ＋ 真实音频

    filtered_search()

    batch_search()

    print("\n 4种多模态检索案例全部运行成功！")

运行输出：

✅ 真实多模态数据初始化完成！

==================================================

 策略1：单一模态检索

==================================================

 文本→文本检索结果：

1. 工业巡检管道维护方法：定期检查管道接口密封性，清理管道内壁污垢，及时更换老化管道

2. 管道故障检测流程：超声波探伤、观察渗漏、定位故障、上报维修

 图片→图片检索（真实图片查询）：

图片路径：industrial_leak.jpg

图片内容：图片显示工业现场一处管道连接处存在明显液体泄漏，泄漏物呈深色，在管壁及下方地面形成湿润痕迹和积液；周边可见锈蚀、油污及部分巡检标识，环境为典型室内厂房或设备间。

==================================================

 策略2：跨模态检索（真实文件版）

==================================================

 文本→图片（检索真实 industrial_leak.jpg）：

✅ 匹配到真实图片：industrial_leak.jpg

图片内容：图片显示工业现场一处管道连接处存在明显液体泄漏，泄漏物呈深色，在管壁及下方地面形成湿润痕迹和积液；周边可见锈蚀、油污及部分巡检标识，环境为典型室内厂房或设备间。

 音频→文本（读取真实 industrial_query_audio.mp3）：

音频转文字结果：幫我分析這張工業尋減圖像,看看管道是否有洩漏。詳細說明洩漏位置,特徵和嚴重程度。

1. 管道故障检测流程：超声波探伤、观察渗漏、定位故障、上报维修

2. 工业巡检管道维护方法：定期检查管道接口密封性，清理管道内壁污垢，及时更换老化管道

==================================================

 策略3：多条件筛选检索

==================================================

 大模型二次校验结果：

1. 管道故障检测流程：超声波探伤、观察渗漏、定位故障、上报维修

2. 工业巡检管道维护方法：定期检查管道接口密封性，清理管道内壁污垢，及时更换老化管道

==================================================

 策略4：批量检索

==================================================

1. 查询：管道泄漏检测

   结果：管道故障检测流程：超声波探伤、观察渗漏、定位故障、上报维修

2. 查询：设备破损识别

   结果：设备破损识别标准：外壳变形、零件脱落、运行异响均需停机检修

3. 查询：巡检报告生成

   结果：工业巡检管道维护方法：定期检查管道接口密封性，清理管道内壁污垢，及时更换老化管道

 4种多模态检索案例 全部运行成功！

5.4.4  多模态检索优化技巧

（1）检索精度优化：调整k值（检索数量），结合qwen-vl-plus对检索结果进行二次筛选，剔除不相关数据；优化特征提取方法，提升特征向量的准确性。

（2）检索效率优化：采用多条件筛选（filter参数），减少检索范围；生产环境使用Pinecone/Milvus分布式数据库，实现并行检索；对大量数据进行分区存储（如按数据类型分区）。

（3）跨模态适配：统一多模态特征向量维度，确保跨模态检索时相似度计算准确；对不同模态的特征向量进行归一化处理，提升检索一致性。

（4）缓存优化：将高频查询的检索结果缓存到本地（如用字典、文件缓存），减少重复检索，提升智能体响应速度。