多模态 RAG

传统 RAG 只处理文本，但企业文档中充满了图表、表格、PDF 扫描件和图片。多模态 RAG（Multimodal RAG）将检索增强生成扩展到图像、表格和混合文档。

多模态 RAG 架构

文档多模态解析

"""
多模态文档解析器
"""
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
class ContentType(Enum):
TEXT = "text"
TABLE = "table"
IMAGE = "image"
CODE = "code"
@dataclass
class ParsedElement:
"""解析后的文档元素"""
content_type: ContentType
text: str = ""
image_path: str = ""
table_data: list[list[str]] = field(default_factory=list)
page_number: int = 0
metadata: dict = field(default_factory=dict)
class MultimodalDocParser:
"""多模态文档解析器"""
def parse_pdf(self, pdf_path: str) -> list[ParsedElement]:
"""
解析 PDF 文件，提取文本、表格、图片
实际生产中可使用：
- unstructured: 通用文档解析
- PyMuPDF: PDF 文本和图片提取
- Camelot/Tabula: PDF 表格提取
- DocTR: OCR 文字识别
"""
elements = []
# 文本提取（示意）
elements.append(ParsedElement(
content_type=ContentType.TEXT,
text="提取的文本内容",
page_number=1,
))
# 表格提取（示意）
elements.append(ParsedElement(
content_type=ContentType.TABLE,
table_data=[["列1", "列2"], ["值1", "值2"]],
page_number=2,
))
# 图片提取（示意）
elements.append(ParsedElement(
content_type=ContentType.IMAGE,
image_path="/tmp/extracted_img_001.png",
page_number=3,
))
return elements
def table_to_text(self, table: list[list[str]]) -> str:
"""将表格转为文本描述，便于 Embedding"""
if not table:
return ""
headers = table[0]
rows = table[1:]
lines = [" | ".join(headers)]
for row in rows:
lines.append(" | ".join(row))
return "\n".join(lines)

多模态 Embedding 策略

"""
多模态 Embedding 管理器
"""
from dataclasses import dataclass
import hashlib
@dataclass
class MultimodalEmbedding:
"""多模态嵌入结果"""
element_id: str
content_type: ContentType
vector: list[float]
text_repr: str
source_page: int
class MultimodalEmbedder:
"""多模态嵌入生成器"""
def __init__(self, text_embedder, vision_llm=None):
self.text_embedder = text_embedder
self.vision_llm = vision_llm
def embed_element(self, element: ParsedElement) -> MultimodalEmbedding:
"""为文档元素生成嵌入"""
element_id = hashlib.md5(
f"{element.content_type.value}_{element.text[:50]}".encode()
).hexdigest()
if element.content_type == ContentType.TEXT:
text_repr = element.text
vector = self.text_embedder.embed(text_repr)
elif element.content_type == ContentType.TABLE:
text_repr = self._table_to_description(element.table_data)
vector = self.text_embedder.embed(text_repr)
elif element.content_type == ContentType.IMAGE:
text_repr = self._describe_image(element.image_path)
vector = self.text_embedder.embed(text_repr)
else:
text_repr = element.text
vector = self.text_embedder.embed(text_repr)
return MultimodalEmbedding(
element_id=element_id,
content_type=element.content_type,
vector=vector,
text_repr=text_repr,
source_page=element.page_number,
)
def _table_to_description(self, table_data: list[list[str]]) -> str:
"""表格转描述"""
if not table_data:
return "空表格"
headers = table_data[0]
row_count = len(table_data) - 1
return f"表格包含 {len(headers)} 列 ({', '.join(headers)})，共 {row_count} 行数据"
def _describe_image(self, image_path: str) -> str:
"""用视觉模型描述图片"""
if self.vision_llm:
return self.vision_llm.describe(image_path)
return f"图片: {Path(image_path).name}"

多模态 RAG 查询引擎

"""
多模态 RAG 查询引擎
"""
from dataclasses import dataclass
@dataclass
class MultimodalRAGConfig:
"""多模态 RAG 配置"""
text_top_k: int = 5
image_top_k: int = 2
table_top_k: int = 3
use_vision_model: bool = True
class MultimodalRAGEngine:
"""多模态 RAG 引擎"""
def __init__(self, vector_store, llm_client, config: MultimodalRAGConfig | None = None):
self.store = vector_store
self.llm = llm_client
self.config = config or MultimodalRAGConfig()
def query(self, question: str) -> dict:
"""多模态查询"""
# 检索多种类型的结果
results = self.store.search(question, top_k=10)
# 按类型分组
texts = [r for r in results if r["type"] == "text"][:self.config.text_top_k]
tables = [r for r in results if r["type"] == "table"][:self.config.table_top_k]
images = [r for r in results if r["type"] == "image"][:self.config.image_top_k]
# 构建多模态上下文
context_parts = []
if texts:
context_parts.append("## 相关文本\n" + "\n".join(t["text_repr"] for t in texts))
if tables:
context_parts.append("## 相关表格\n" + "\n".join(t["text_repr"] for t in tables))
if images:
context_parts.append("## 相关图片描述\n" + "\n".join(i["text_repr"] for i in images))
context = "\n\n".join(context_parts)
# 生成回答
answer = self.llm.generate(
f"根据以下多模态资料回答问题。\n\n{context}\n\n问题：{question}\n回答："
)
return {
"answer": answer,
"text_sources": len(texts),
"table_sources": len(tables),
"image_sources": len(images),
}

工具生态

本章小结

下一章：可观测性与监控