各位同仁，各位对LLM与信息检索技术抱有热情的开发者们，大家好！
今天，我们齐聚一堂，共同探讨一个前沿且极具挑战性的话题：多模态检索增强生成（Multimodal RAG）。我们不仅要理解它的核心理念，更要深入实践，尤其关注如何在LangChain框架下，高效地索引并检索图像、图表乃至视频片段，从而极大地拓宽我们LLM应用的信息获取能力。
传统的RAG模型，其核心在于从文本语料库中检索相关文本片段，作为上下文输入给大型语言模型（LLM），以提升其回答的准确性、时效性和减少幻觉。然而，现实世界的信息远不止文本。图像、图表、视频承载着海量的非结构化信息，这些信息对于理解复杂概念、提供视觉证据或解释动态过程至关重要。如何让我们的LLM也能“看到”并“理解”这些非文本数据，正是多模态RAG所要解决的核心问题。
1. 多模态RAG的基石：超越文本的理解
多模态RAG的根本在于将非文本信息转化为LLM能够处理的形式，并使其可检索。这通常涉及几个关键步骤：
- 特征提取与表示（Representation）：将图像、图表、视频等原始数据转化为某种向量表示（嵌入），或者将其内容转化为结构化或描述性的文本。
- 索引（Indexing）：将这些表示好的信息（无论是向量还是文本）存储在高效可检索的数据库中，通常是向量数据库。
- 检索（Retrieval）：根据用户的查询（通常是文本），从索引中找出最相关的多模态信息。
- 生成（Generation）：将检索到的多模态信息（可能经过进一步处理）与原始查询一同输入给LLM，生成最终的答案。
LangChain作为一个强大的LLM应用开发框架，为我们提供了构建多模态RAG所需的各种组件和抽象。
2. LangChain中的多模态RAG核心组件
在LangChain中实现多模态RAG，我们将主要用到以下核心组件：
- Document Loaders: 用于从各种源加载数据。我们需要自定义或利用现有加载器来处理图像、视频文件。
- Document Transformers / Text Splitters: 虽然原始的文本分割器不直接适用于图像视频，但我们会用它们处理从多模态数据中提取出的文本描述。更重要的是，我们需要多模态处理器来生成这些描述或嵌入。
- Embeddings: 将文本或多模态数据转化为向量。这是实现语义搜索的关键。
- VectorStores: 存储和检索向量嵌入的数据库。Chroma、Pinecone、Weaviate等都是常见选择。
- Retrievers: 从VectorStore中检索相关
Document的组件。 - Chains: 将上述组件连接起来，形成一个完整的RAG工作流。
我们将聚焦于如何将图像、图表和视频片段转化为LangChain的Document对象，并为其生成可检索的嵌入。
3. 图像与图表的索引与检索
图像和图表是信息密集型的数据。它们不仅包含视觉元素，还可能包含重要的文本信息（如标签、标题、数据点）。处理这类数据，我们有几种策略：
3.1 策略一：基于描述的检索 (Caption-based Retrieval)
这是最直接且广泛应用的方法。我们使用一个视觉-语言模型（VLM）为每张图片生成一个详细的文本描述（Caption）。然后，我们将这些描述作为文本，生成嵌入并存储在向量数据库中。
优点：实现简单，对LLM友好，因为LLM本身就是处理文本的。
缺点：描述的质量直接影响检索效果；丢失了图像本身的视觉信息，仅依赖于VLM的解释。
3.2 策略二：基于OCR的文本提取与检索 (OCR-based Retrieval)
特别是对于图表或包含大量文字的图像，光学字符识别（OCR）能够提取图像中的可见文本。这些文本可以直接作为描述的一部分，或单独进行索引。
优点：直接捕获图像中的显式文本信息，对于图表尤其有效。
缺点：无法理解图像的整体语义或非文本视觉信息；OCR准确性受图像质量影响。
3.3 策略三：多模态嵌入检索 (Multimodal Embedding Retrieval)
这是最先进的方法。我们使用一个多模态模型（如CLIP、OpenAI的text-embedding-3-small或gpt-4-vision-preview）直接为图像本身生成一个嵌入向量。查询时，用户的文本查询也会被同一个模型转化为向量，然后在向量空间中进行相似性匹配。
优点：保留了图像的视觉语义信息，能够理解图像的“内容”而不仅仅是描述。
缺点：需要更复杂的模型；查询时也需要将查询文本嵌入到相同的多模态空间。
在实际应用中，通常会结合使用这些策略，例如：为图像生成描述（策略一），同时进行OCR（策略二），并将这些文本信息结合起来，甚至与多模态嵌入（策略三）一起存储，以实现更鲁棒的检索。
3.3.1 实践：图像与图表的索引
让我们通过一个具体的例子，展示如何在LangChain中结合策略一和策略二来索引图像和图表。我们将使用一个简单的VLM（例如借助transformers库模拟）来生成描述，并使用Pytesseract进行OCR。
环境准备
首先，确保安装必要的库：
pip install langchain langchain-openai pypdf pillow transformers opencv-python pytesseract faiss-cpu
# Pytesseract需要额外的Tesseract OCR引擎安装，请参考其官方文档
# 例如在Debian/Ubuntu: sudo apt install tesseract-ocr
# 在macOS: brew install tesseract模拟VLM和OCR功能
由于直接在代码中运行大型VLM会很慢且需要大量资源，我们将创建一个模拟函数，或者使用OpenAI的gpt-4-vision-preview或HuggingFace transformers库中的轻量级VLM。这里为了演示方便，先用一个简单的函数模拟，然后展示如何集成真正的VLM。
import os
from PIL import Image
import pytesseract
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from typing import List, Dict, Any
import base64
import io
# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# 1. 模拟或集成VLM进行图像描述
def describe_image_mock(image_path: str) -> str:
"""
模拟一个图像描述函数。
在真实场景中，这里会调用一个VLM API或本地模型，如BLIP-2, LLaVA, 或GPT-4V。
"""
image_name = os.path.basename(image_path)
if "chart" in image_name.lower():
return f"这是一张关于销售数据的图表，标题可能是'年度销售概览'。图中可能有柱状图或折线图展示不同产品或年份的销售额。文件名: {image_name}"
elif "product" in image_name.lower():
return f"这是一张产品照片，可能展示了一个电子设备或家居用品。图中可能包含产品特写和背景环境。文件名: {image_name}"
else:
return f"这是一张普通的图片，内容未知。文件名: {image_name}"
def describe_image_with_gpt4v(image_path: str, client) -> str:
"""
使用OpenAI GPT-4 Vision API描述图像。
需要OpenAI客户端。
"""
with open(image_path, "rb") as image_file:
base64_image = base64.b64encode(image_file.read()).decode("utf-8")
try:
response = client.chat.completions.create(
model="gpt-4-vision-preview",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "详细描述这张图片的内容，包括任何可见的文本和关键视觉元素。"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high" # 或 "low"
},
},
],
}
],
max_tokens=500,
)
return response.choices[0].message.content
except Exception as e:
print(f"调用GPT-4V失败: {e}")
return describe_image_mock(image_path) # 失败时回退到mock
# 2. OCR功能
def extract_text_from_image(image_path: str) -> str:
"""
使用Pytesseract从图像中提取文本。
"""
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image, lang='chi_sim+eng') # 支持中文和英文
return text.strip()
except Exception as e:
print(f"OCR失败: {e}")
return ""
# 3. 创建LangChain Document
def create_image_document(image_path: str, client=None) -> Document:
"""
为图像创建LangChain Document。
结合了描述和OCR文本。
"""
description = ""
if client:
description = describe_image_with_gpt4v(image_path, client)
else:
description = describe_image_mock(image_path)
ocr_text = extract_text_from_image(image_path)
# 结合描述和OCR文本作为Document内容
page_content = f"图片描述: {description}nn图片中提取的文本:n{ocr_text}"
# 存储原始路径和类型作为元数据
metadata = {
"source": image_path,
"type": "image",
"description": description, # 也可以将描述作为单独元数据
"ocr_text": ocr_text # 也可以将OCR文本作为单独元数据
}
return Document(page_content=page_content, metadata=metadata)
# 4. 索引图像
def index_images(image_paths: List[str], embeddings: OpenAIEmbeddings, client=None) -> FAISS:
"""
索引给定的图像路径列表。
"""
documents = []
for img_path in image_paths:
print(f"处理图像: {img_path}")
doc = create_image_document(img_path, client)
documents.append(doc)
print(f"共生成 {len(documents)} 个图像文档。")
vectorstore = FAISS.from_documents(documents, embeddings)
print("图像索引完成。")
return vectorstore
# 示例使用
if __name__ == "__main__":
# 模拟创建一些图像文件
# 实际应用中，这些是真实存在的图像文件
os.makedirs("images", exist_ok=True)
with open("images/sales_chart.png", "w") as f: f.write("dummy chart content")
with open("images/product_overview.jpg", "w") as f: f.write("dummy product content")
with open("images/text_heavy_doc.png", "w") as f: f.write("dummy text content")
image_files = [
"images/sales_chart.png",
"images/product_overview.jpg",
"images/text_heavy_doc.png"
]
# 初始化OpenAI客户端 (如果使用GPT-4V)
# from openai import OpenAI
# openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
openai_client = None # 暂时不使用GPT-4V，用mock函数代替
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
image_vectorstore = index_images(image_files, embeddings, client=openai_client)
# 5. 检索图像
print("n--- 图像检索示例 ---")
query_image = "年度销售数据报告"
retrieved_docs_image = image_vectorstore.similarity_search(query_image, k=2)
print(f"n查询: '{query_image}'")
for i, doc in enumerate(retrieved_docs_image):
print(f"--- 检索结果 {i+1} ---")
print(f"来源: {doc.metadata['source']}")
print(f"类型: {doc.metadata['type']}")
print(f"内容预览: {doc.page_content[:200]}...")
# print(f"完整内容:n{doc.page_content}")
print("-" * 20)
query_product = "展示一个电子产品"
retrieved_docs_product = image_vectorstore.similarity_search(query_product, k=1)
print(f"n查询: '{query_product}'")
for i, doc in enumerate(retrieved_docs_product):
print(f"--- 检索结果 {i+1} ---")
print(f"来源: {doc.metadata['source']}")
print(f"类型: {doc.metadata['type']}")
print(f"内容预览: {doc.page_content[:200]}...")
print("-" * 20)代码解析：
describe_image_mock/describe_image_with_gpt4v: 这些函数负责生成图像的文本描述。实际项目中，你会集成更强大的VLM，例如使用transformers库加载BLIP-2模型，或者调用云服务（如GPT-4V、Google Gemini Vision）。extract_text_from_image: 利用Pytesseract库从图像中提取所有可识别的文本。这对于图表中的标题、轴标签、数据标注等尤其重要。create_image_document: 这是核心函数，它将图像文件路径作为输入，调用描述和OCR函数，然后将这些文本信息组合成page_content。同时，它将图像的元数据（如原始路径、类型、原始描述、OCR文本）存储在metadata字典中。一个Document对象代表了LangChain中可索引的最小信息单元。index_images: 遍历所有图像路径，为每个图像创建Document，然后使用OpenAIEmbeddings（或任何其他文本嵌入模型）将Document的page_content转换为向量，并存储到FAISS向量数据库中。- 检索: 通过
image_vectorstore.similarity_search(query, k)方法，我们可以根据文本查询找到最相关的图像Document。
3.3.2 进阶：多模态嵌入的直接应用
如果使用像CLIP或OpenAI的text-embedding-3-small（虽然它主要为文本优化，但一些多模态模型也能将其图像嵌入到同一空间）这样的模型，可以直接将图像本身转化为向量，与文本查询的向量在同一空间进行比较。LangChain目前主要通过文本嵌入来集成，因此直接的多模态图像嵌入需要更底层的集成，例如：
- 图像编码器: 使用CLIP模型将图像编码为向量。
- 文本编码器: 使用CLIP模型的文本编码器将查询文本编码为向量。
- 向量匹配: 在向量数据库中进行相似性搜索。
LangChain multi-vector-retriever是一个很好的模式，可以同时存储原始图片（或指向图片的URL）和其文本描述的嵌入。
# 示例：多向量检索模式的简化演示 (概念性)
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
import uuid
# 假设我们有一个图像编码器和文本编码器，它们将图像和文本映射到同一向量空间
class MultimodalImageEncoder:
def encode_image(self, image_path: str) -> List[float]:
# 实际这里会调用CLIP等模型进行图像编码
# 为了演示，我们返回一个基于图像内容的伪向量
import hashlib
with open(image_path, 'rb') as f:
image_bytes = f.read()
hash_val = int(hashlib.sha256(image_bytes).hexdigest(), 16)
# 伪造一个高维向量
return [(hash_val % 1000) / 1000.0 + i * 0.001 for i in range(1536)] # OpenAI embedding size
# 假设我们用OpenAIEmbeddings做文本和潜在的跨模态嵌入
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
image_encoder = MultimodalImageEncoder() # 实例化我们的图像编码器
# 创建一个存储原始图片（或指向图片的URL）的存储
# 这是一个简单的内存存储，实际中可以是S3、Blob Storage等
store = InMemoryStore()
# 创建一个向量数据库来存储文本描述的嵌入
vectorstore = Chroma(collection_name="image_descriptions", embedding_function=embeddings)
# MultiVectorRetriever
retriever = MultiVectorRetriever(
vectorstore=vectorstore,
docstore=store,
id_key="doc_id"
)
# 准备图像数据
image_files = [
"images/sales_chart.png",
"images/product_overview.jpg",
"images/text_heavy_doc.png"
]
# 索引过程
for img_path in image_files:
doc_id = str(uuid.uuid4())
# 获取图像描述和OCR文本
description = describe_image_mock(img_path)
ocr_text = extract_text_from_image(img_path)
# 创建一个用于检索的文本Document
# 这里的page_content是可检索的文本描述
text_doc = Document(
page_content=f"图片描述: {description}nn图片中提取的文本:n{ocr_text}",
metadata={"source": img_path, "type": "image", "doc_id": doc_id}
)
# 存储原始图像路径（或图像二进制数据）
# 这里的store可以存储原始图像数据，或者仅仅是图片的URL
store.mset([(doc_id, img_path)]) # 存储原始图像路径
# 将文本描述添加到向量数据库
retriever.vectorstore.add_documents([text_doc])
print("n--- 多向量图像检索示例 ---")
query = "告诉我关于销售业绩的图表"
retrieved_docs = retriever.invoke(query)
print(f"n查询: '{query}'")
for i, doc in enumerate(retrieved_docs):
print(f"--- 检索结果 {i+1} ---")
print(f"检索到的文档内容预览: {doc.page_content[:200]}...")
print(f"对应的原始图像路径: {doc.metadata['source']}") # 实际上，这里可以从store中获取原始图像
print("-" * 20)
# 如果要获取原始图像，可以这样做：
# retrieved_image_paths = [store.mget([doc.metadata['doc_id']])[0] for doc in retrieved_docs]
# print(f"原始图像文件: {retrieved_image_paths}")在这个多向量检索的例子中，我们为每个图像生成了一个唯一ID。然后：
- 我们将图像的文本描述（包含VLM描述和OCR文本）作为可检索的
Document，并将其嵌入存储在vectorstore中。 - 我们将原始的图像路径（或者实际的图像二进制数据）与这个唯一ID关联，存储在
docstore中。 - 当执行检索时，
retriever会根据查询从vectorstore中找到相关的文本描述，然后使用这些描述中的doc_id去docstore中查找对应的原始图像信息。这样，LLM就能接收到文本描述作为上下文，同时我们也能知道是哪张图片提供了这些信息。
4. 视频片段的索引与检索
视频是时间序列的图像和音频数据。处理视频比处理单张图片复杂得多。我们需要将其分解成更小的、可管理的单元，并提取多模态信息。
4.1 核心策略：视频分段与多模态信息提取
- 视频分段 (Segmentation)：将长视频切割成有意义的短片段。这可以是固定时间间隔（如每30秒一个片段），也可以是基于内容变化（如场景切换、主题变化）的动态分段。
- 关键帧提取 (Keyframe Extraction)：从每个视频片段中提取代表性的关键帧。这些关键帧可以被视为独立的图像进行处理（如图像描述、OCR）。
- 音频转录 (Audio Transcription)：使用语音转文本（STT）服务将每个视频片段的音频内容转换为文本。这对于包含演讲、对话的视频至关重要。
- 多模态摘要 (Multimodal Summarization)：对每个视频片段，结合其关键帧描述和音频转录，生成一个综合性的文本摘要。
4.2 LangChain中的实现
我们将为每个视频片段创建一个Document。这个Document的page_content将包含该片段的转录文本、关键帧描述和片段摘要。元数据将包含视频ID、片段的开始/结束时间戳、原始视频路径等。
环境准备
pip install moviepy pydub openai-whisper opencv-python ffmpeg-python
# ffmpeg需要额外安装，请参考其官方文档
# 例如在Debian/Ubuntu: sudo apt install ffmpeg
# 在macOS: brew install ffmpeg模拟视频处理功能
与图像类似，视频处理（关键帧提取、STT）也可能耗时。这里我们将使用一些库来模拟或简化这些步骤。
import moviepy.editor as mp
import speech_recognition as sr
import cv2
import numpy as np
import math
from pydub import AudioSegment
from pydub.silence import split_on_silence
import whisper # 用于本地STT，如果需要
# 1. 视频分段与关键帧提取
def extract_keyframes_and_segments(video_path: str, segment_duration_sec: int = 30, kf_per_segment: int = 1) -> List[Dict[str, Any]]:
"""
将视频分段，并从每个片段中提取关键帧。
返回一个包含片段信息和关键帧路径的列表。
"""
video = mp.VideoFileClip(video_path)
total_duration = video.duration
segments_info = []
segment_id_counter = 0
for start_time in range(0, math.ceil(total_duration), segment_duration_sec):
end_time = min(start_time + segment_duration_sec, total_duration)
if start_time >= end_time:
break
segment_id = f"{os.path.basename(video_path).split('.')[0]}_segment_{segment_id_counter}"
# 提取关键帧 (这里简化为从片段中间取一帧)
frame_time = start_time + (end_time - start_time) / 2
# 使用OpenCV从视频中精确提取帧
cap = cv2.VideoCapture(video_path)
cap.set(cv2.CAP_PROP_POS_MSEC, frame_time * 1000) # 定位到帧的时间 (毫秒)
ret, frame = cap.read()
cap.release()
keyframe_path = None
if ret:
kf_dir = os.path.join("video_keyframes", segment_id)
os.makedirs(kf_dir, exist_ok=True)
keyframe_path = os.path.join(kf_dir, f"keyframe_{int(frame_time)}.jpg")
cv2.imwrite(keyframe_path, frame)
segments_info.append({
"segment_id": segment_id,
"video_path": video_path,
"start_time": start_time,
"end_time": end_time,
"keyframe_path": keyframe_path,
"audio_path": None # 稍后填充
})
segment_id_counter += 1
video.close()
return segments_info
# 2. 音频转录 (Speech-to-Text)
def transcribe_audio_whisper(audio_path: str) -> str:
"""
使用OpenAI Whisper模型进行本地音频转录。
需要先下载Whisper模型。
"""
try:
model = whisper.load_model("base") # 或 "small", "medium"
result = model.transcribe(audio_path, fp16=False)
return result["text"].strip()
except Exception as e:
print(f"Whisper转录失败: {e}")
return ""
def transcribe_audio_segment(video_path: str, start_time: int, end_time: int, segment_id: str) -> str:
"""
从视频片段中提取音频并转录。
"""
audio_segment_path = os.path.join("video_audio_segments", f"{segment_id}.wav")
os.makedirs(os.path.dirname(audio_segment_path), exist_ok=True)
try:
# 使用moviepy提取音频片段
video = mp.VideoFileClip(video_path)
audio_clip = video.audio.subclip(start_time, end_time)
audio_clip.write_audiofile(audio_segment_path, logger=None)
audio_clip.close()
video.close()
# 转录音频片段
transcript = transcribe_audio_whisper(audio_segment_path)
return transcript
except Exception as e:
print(f"提取或转录音频片段失败 ({segment_id}): {e}")
return ""
# 3. 创建LangChain Document for Video Segments
def create_video_segment_document(segment_info: Dict[str, Any], client=None) -> Document:
"""
为视频片段创建LangChain Document。
结合了关键帧描述和音频转录。
"""
segment_id = segment_info["segment_id"]
video_path = segment_info["video_path"]
start_time = segment_info["start_time"]
end_time = segment_info["end_time"]
keyframe_path = segment_info["keyframe_path"]
# 描述关键帧 (同图像处理)
keyframe_description = ""
if keyframe_path:
keyframe_description = describe_image_with_gpt4v(keyframe_path, client) if client else describe_image_mock(keyframe_path)
# 转录音频
transcript = transcribe_audio_segment(video_path, start_time, end_time, segment_id)
# 结合所有信息
page_content = (
f"视频片段 ID: {segment_id}n"
f"时间范围: {start_time:.1f}s - {end_time:.1f}sn"
f"关键帧描述: {keyframe_description}n"
f"音频转录: {transcript}"
)
metadata = {
"source_video": video_path,
"segment_id": segment_id,
"type": "video_segment",
"start_time_sec": start_time,
"end_time_sec": end_time,
"keyframe_path": keyframe_path,
"keyframe_description": keyframe_description,
"transcript": transcript
}
return Document(page_content=page_content, metadata=metadata)
# 4. 索引视频
def index_videos(video_paths: List[str], embeddings: OpenAIEmbeddings, client=None) -> FAISS:
"""
索引给定的视频路径列表。
"""
all_segment_docs = []
for vid_path in video_paths:
print(f"处理视频: {vid_path}")
segments_info = extract_keyframes_and_segments(vid_path, segment_duration_sec=60) # 每60秒一个片段
for seg_info in segments_info:
print(f" - 处理片段: {seg_info['segment_id']} ({seg_info['start_time']:.1f}s - {seg_info['end_time']:.1f}s)")
doc = create_video_segment_document(seg_info, client)
all_segment_docs.append(doc)
print(f"共生成 {len(all_segment_docs)} 个视频片段文档。")
if not all_segment_docs:
print("没有生成任何视频片段文档，跳过向量存储创建。")
return None
vectorstore = FAISS.from_documents(all_segment_docs, embeddings)
print("视频片段索引完成。")
return vectorstore
# 示例使用
if __name__ == "__main__":
# 模拟创建一个短视频文件 (需要安装ffmpeg)
# 这个命令会创建一个10秒的空白视频
# import subprocess
# try:
# subprocess.run(["ffmpeg", "-y", "-f", "lavfi", "-i", "color=c=black:s=640x480:d=10", "-vf", "drawtext=text='Demo Video':fontcolor=white:fontsize=24:x=(w-text_w)/2:y=(h-text_h)/2", "videos/demo_video.mp4"], check=True)
# print("模拟视频创建成功: videos/demo_video.mp4")
# except Exception as e:
# print(f"无法创建模拟视频，请确保ffmpeg已安装并可执行。错误: {e}")
# print("将跳过视频索引部分。")
# video_files = []
# else:
# video_files = ["videos/demo_video.mp4"]
# 假设我们有真实的视频文件
os.makedirs("videos", exist_ok=True)
# 创建一个空的mp4文件以模拟
with open("videos/meeting_summary.mp4", "w") as f: f.write("dummy video content")
with open("videos/product_review.mp4", "w") as f: f.write("dummy video content")
video_files = ["videos/meeting_summary.mp4", "videos/product_review.mp4"]
os.makedirs("video_keyframes", exist_ok=True)
os.makedirs("video_audio_segments", exist_ok=True)
# 初始化OpenAI客户端 (如果使用GPT-4V)
# from openai import OpenAI
# openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
openai_client = None # 暂时不使用GPT-4V，用mock函数代替
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
video_vectorstore = index_videos(video_files, embeddings, client=openai_client)
if video_vectorstore:
# 5. 检索视频片段
print("n--- 视频片段检索示例 ---")
query_video = "会议中讨论了哪些关于产品发布的内容？"
retrieved_docs_video = video_vectorstore.similarity_search(query_video, k=2)
print(f"n查询: '{query_video}'")
for i, doc in enumerate(retrieved_docs_video):
print(f"--- 检索结果 {i+1} ---")
print(f"来源视频: {doc.metadata['source_video']}")
print(f"片段ID: {doc.metadata['segment_id']}")
print(f"时间: {doc.metadata['start_time_sec']:.1f}s - {doc.metadata['end_time_sec']:.1f}s")
print(f"内容预览: {doc.page_content[:300]}...")
print("-" * 20)代码解析：
extract_keyframes_and_segments: 这个函数负责将视频文件分割成指定时长的片段，并从每个片段中提取一个关键帧（这里为了简化，我们取了片段的中间帧）。实际应用中，关键帧提取可以使用更复杂的算法，例如基于帧间差异或场景检测。它返回每个片段的元信息，包括开始/结束时间、关键帧路径。transcribe_audio_whisper/transcribe_audio_segment:transcribe_audio_segment负责从视频片段中提取音频，并将其保存为临时文件。然后，transcribe_audio_whisper使用OpenAI的Whisper模型（或任何其他STT服务如Google Cloud Speech-to-Text、AssemblyAI）将音频转录为文本。create_video_segment_document: 这是视频处理的核心。它结合了：- 通过图像描述VLM对关键帧的描述。
- 通过STT获得的音频转录文本。
- 这些信息被组合成
page_content，作为可检索的文本。 - 元数据包含视频来源、片段ID、时间戳、关键帧路径等，这些对于定位和播放原始视频片段非常重要。
index_videos: 遍历所有视频路径，对每个视频进行分段和信息提取，为每个片段创建Document，然后使用OpenAIEmbeddings进行嵌入，并存储到FAISS向量数据库。- 检索: 通过
video_vectorstore.similarity_search(query, k)，用户可以根据文本查询找到最相关的视频片段Document，然后利用metadata中的时间戳信息，可以直接跳转到视频的相应位置。
5. 整合检索与生成：构建多模态RAG Chain
现在我们已经可以索引和检索不同模态的数据了。下一步是将这些检索结果输入给LLM，以完成多模态RAG的最终目标：生成高质量的、包含多模态信息的答案。
LangChain的RAG Chain通常遵循以下模式：
query -> retriever -> retrieved_documents -> LLM + prompt -> answer
对于多模态RAG，retrieved_documents现在可能包含来自图像、图表和视频片段的Document。LLM需要能够理解这些混合上下文。
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# 假设 image_vectorstore 和 video_vectorstore 已经创建
# 我们可以创建一个统一的检索器，或者根据查询类型选择不同的检索器
# 1. 创建一个统一的检索器 (合并多个向量存储)
# 这是一个简化的合并，实际中可能需要更复杂的逻辑，例如根据查询意图选择
class CombinedRetriever:
def __init__(self, image_store: FAISS, video_store: FAISS, text_store: FAISS = None):
self.image_store = image_store
self.video_store = video_store
self.text_store = text_store # 假设还有一个文本存储
def get_relevant_documents(self, query: str, k: int = 5) -> List[Document]:
retrieved_docs = []
if self.image_store:
retrieved_docs.extend(self.image_store.similarity_search(query, k=k//2 if k > 1 else 1))
if self.video_store:
retrieved_docs.extend(self.video_store.similarity_search(query, k=k//2 if k > 1 else 1))
if self.text_store:
retrieved_docs.extend(self.text_store.similarity_search(query, k=k)) # 文本可以多一点
# 简单去重和截断
unique_docs = {doc.page_content: doc for doc in retrieved_docs}.values()
return list(unique_docs)[:k]
# 假设我们也有一个文本向量存储
# text_vectorstore = FAISS.from_documents(
# [Document(page_content="关于AI技术发展的重要报告", metadata={"source": "report.pdf", "type": "text"})],
# embeddings
# )
text_vectorstore = None # 暂时不包含文本检索
combined_retriever = CombinedRetriever(image_vectorstore, video_vectorstore, text_vectorstore)
# 2. 定义LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # gpt-4o-mini支持多模态输入
# 3. 定义Prompt模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个知识渊博的助手。请根据提供的上下文信息，简洁、准确地回答用户的问题。上下文可能包含图像、图表和视频片段的描述及转录。请特别注意提及信息的来源类型（如'根据图片显示...'，'视频片段提到...'）。"),
("user", "上下文信息: {context}nn问题: {input}")
])
# 4. 创建文档组合链 (stuff documents chain)
# 这个链负责将检索到的所有文档组合成一个长字符串，作为LLM的上下文。
document_chain = create_stuff_documents_chain(llm, prompt)
# 5. 创建检索链
retrieval_chain = create_retrieval_chain(combined_retriever, document_chain)
# 6. 执行RAG查询
print("n--- 多模态RAG Chain 示例 ---")
rag_query = "销售报告中提到了哪些关键趋势？视频中关于新产品发布有什么细节？"
response = retrieval_chain.invoke({"input": rag_query})
print(f"n用户问题: {rag_query}")
print(f"nLLM回答:n{response['answer']}")
# 我们可以进一步处理response['context']来展示原始的多模态内容
print("n--- 检索到的上下文信息 ---")
for i, doc in enumerate(response['context']):
print(f"--- 上下文 {i+1} (类型: {doc.metadata.get('type', '未知')}) ---")
if doc.metadata.get('type') == 'image':
print(f" 来源图片: {doc.metadata['source']}")
print(f" 描述: {doc.metadata['description'][:100]}...")
elif doc.metadata.get('type') == 'video_segment':
print(f" 来源视频: {doc.metadata['source_video']}")
print(f" 时间: {doc.metadata['start_time_sec']:.1f}s - {doc.metadata['end_time_sec']:.1f}s")
print(f" 关键帧描述: {doc.metadata['keyframe_description'][:100]}...")
print(f" 转录: {doc.metadata['transcript'][:100]}...")
print("-" * 20)代码解析：
CombinedRetriever: 这是一个自定义的检索器，它封装了对图像和视频向量存储的查询。在实际应用中，您可能需要更智能的逻辑来合并或排序来自不同模态的检索结果，例如：- 并行检索: 同时向所有模态的向量存储发送查询。
- 混合重排: 将所有检索结果（无论模态）放在一起，使用一个跨模态的重排器（如Reranker模型）进行排序。
- 智能路由: 根据用户查询的意图（例如，如果查询包含“图片”、“图表”，则优先检索图像；如果包含“视频”、“讲话”，则优先检索视频），动态选择检索器。
ChatOpenAI(model="gpt-4o-mini"): 我们选择了一个支持多模态输入的LLM（如gpt-4o-mini，gpt-4-vision-preview或Google Gemini Pro Vision）。虽然我们在这里主要通过文本描述将多模态信息传递给LLM，但直接支持多模态输入的LLM在未来可能会更直接地处理原始图像或视频帧。ChatPromptTemplate: 设计一个合适的Prompt至关重要。它需要明确告知LLM上下文信息可能来自不同模态，并指导LLM如何引用这些信息（例如，“根据图表显示…”）。create_stuff_documents_chain: 这是LangChain的标准组件，它将检索器返回的所有Document的page_content拼接成一个大字符串，作为context变量传递给LLM。create_retrieval_chain: 将检索器和文档组合链连接起来，形成一个完整的RAG工作流。
6. 高级考量与最佳实践
- 性能与成本：多模态数据处理是计算密集型的。图像描述、视频转录、关键帧提取都需要大量CPU/GPU资源和时间。在生产环境中，考虑使用云服务（如AWS Rekognition, Google Cloud Video Intelligence, Azure AI Vision）或优化本地模型的部署。缓存结果以避免重复计算。
- 多模态融合嵌入：虽然我们使用了文本描述进行索引，但更先进的方法是使用多模态嵌入模型，将图像、文本、甚至音频直接嵌入到同一个高维向量空间。这样，查询（无论是文本还是图像）可以直接在这个空间进行相似性搜索。
- 上下文窗口管理：当检索到大量多模态上下文时，LLM的上下文窗口可能会成为瓶颈。可以考虑：
- 摘要: 对每个检索到的
Document进行二次摘要，再传递给LLM。 - 重排: 使用Reranker模型对检索到的文档进行重新排序，将最相关的文档放在前面。
- 分块策略: 对于视频，更细粒度的分段（如基于对话或场景）可以提供更精确的上下文。
- 摘要: 对每个检索到的
- 用户体验：当LLM回答中引用了多模态信息时，提供指向原始资源的链接（如图像URL、视频特定时间戳）至关重要，让用户可以验证或深入探索。
- 数据管理：多模态数据量庞大，需要健壮的数据管道进行ETL（提取、转换、加载）和版本控制。
- 错误处理与鲁棒性：图像损坏、视频编码问题、STT识别错误等都可能发生。在生产系统中需要有完善的错误处理机制。
7. 结语
多模态RAG是LLM应用发展的一个重要方向，它使得LLM能够突破文本的限制，真正感知和理解我们所处的多彩世界。通过LangChain提供的模块化组件，我们能够相对高效地构建出处理图像、图表和视频的RAG系统。这不仅提升了LLM的知识边界，也为构建更智能、更具交互性的AI应用开辟了广阔前景。挑战依然存在，但探索永无止境，期待大家将这些技术应用于更多创新场景。
感谢大家的聆听！