RAG 中图片解析方案全景
核心问题：图片里可能藏着关键信息（图表、流程图、截图、扫描件），如果跳过图片，RAG 的知识就是残缺的
图片在 RAG 中的挑战
RAG 处理文档时遇到图片，有几种选择：
┌─────────────────────────────────────────────────────────┐
│ 策略一：直接跳过图片 │
│ → 简单，但损失大量信息 │
│ → 图表数据、流程图逻辑全部丢失 │
│ │
│ 策略二：OCR 提取图片中的文字 │
│ → 只能提取文字，理解不了图表含义 │
│ → 适合：截图、扫描件 │
│ │
│ 策略三：多模态模型理解图片语义 │
│ → 真正"看懂"图片，生成文字描述 │
│ → 适合：图表、流程图、示意图 │
│ │
│ 策略四：以图搜图（图片向量化） │
│ → 图片转向量，支持图片作为查询输入 │
│ → 适合：图片检索场景 │
└─────────────────────────────────────────────────────────┘
常见图片类型及对应处理方式：
截图/扫描文字 ──→ OCR（PaddleOCR）
柱状图/折线图 ──→ 多模态模型（描述数据趋势）
流程图/架构图 ──→ 多模态模型（描述逻辑关系）
表格截图 ──→ OCR + 表格识别
产品图/照片 ──→ 多模态模型（描述图片内容）
公式截图 ──→ 专用公式识别（LaTeX）
整体技术方案架构
┌──────────────────────────────────────────────────────────────┐
│ RAG 图片处理全流程 │
│ │
│ PDF/文档输入 │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 图片提取 │ ← PyMuPDF / pdfplumber │
│ │ （从PDF抽出） │ │
│ └────────┬────────┘ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 图片分类 │ ← 判断图片类型 │
│ │ （是什么图） │ 纯文字截图 / 图表 / 流程图 / 照片 │
│ └────────┬────────┘ │
│ ↓ │
│ ┌──────┴──────────────────────┐ │
│ ↓ ↓ │
│ 文字类图片 语义类图片 │
│ (截图/扫描) (图表/流程图) │
│ ↓ ↓ │
│ OCR 提取 多模态模型 │
│ PaddleOCR GPT-4V / Qwen-VL │
│ ↓ ↓ │
│ └──────────┬──────────────────┘ │
│ ↓ │
│ ┌─────────────────────────┐ │
│ │ 文字描述（Caption） │ "该图表显示2024年Q1销售额..." │
│ └────────────┬────────────┘ │
│ ↓ │
│ ┌─────────────────────────┐ │
│ │ 向量化存储 │ 与正文一起存入向量数据库 │
│ └─────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
方案一：OCR（文字类图片）
PaddleOCR —— 中文首选
import fitz
import numpy as np
from PIL import Image
from paddleocr import PaddleOCR
import io
class ImageOCRParser:
def __init__(self):
self.ocr = PaddleOCR(
use_angle_cls=True, # 自动纠正倾斜文字
lang='ch', # 中英文混合
use_gpu=True,
show_log=False
)
def extract_images_from_pdf(self, pdf_path: str) -> list:
"""从 PDF 提取所有图片"""
doc = fitz.open(pdf_path)
images = []
for page_num, page in enumerate(doc):
# 方式1：提取嵌入的图片资源
for img_index, img in enumerate(page.get_images()):
xref = img[0]
base_image = doc.extract_image(xref)
images.append({
"page": page_num + 1,
"index": img_index,
"bytes": base_image["image"],
"ext": base_image["ext"],
"bbox": None # 图片在页面中的位置
})
# 方式2：整页渲染（适合扫描件）
# mat = fitz.Matrix(2.0, 2.0) # 2x 分辨率
# pix = page.get_pixmap(matrix=mat)
# images.append({"page": page_num+1, "bytes": pix.tobytes()})
doc.close()
return images
def ocr_image(self, image_bytes: bytes) -> str:
"""对图片做 OCR"""
img = Image.open(io.BytesIO(image_bytes))
img_array = np.array(img)
result = self.ocr.ocr(img_array, cls=True)
if not result or not result[0]:
return ""
# 按 Y 坐标排序，保证阅读顺序
lines = sorted(result[0], key=lambda x: x[0][0][1])
texts = []
for line in lines:
bbox, (text, confidence) = line
if confidence > 0.75:
texts.append(text)
return "\n".join(texts)
def is_text_image(self, image_bytes: bytes) -> bool:
"""
判断图片是否主要包含文字
策略：OCR 识别出的字符数 / 图片像素数
"""
img = Image.open(io.BytesIO(image_bytes))
img_array = np.array(img)
result = self.ocr.ocr(img_array, cls=True)
if not result or not result[0]:
return False
total_chars = sum(len(line[1][0]) for line in result[0])
# 字符数超过阈值，认为是文字图片
return total_chars > 20
def parse_pdf_images(self, pdf_path: str) -> list:
images = self.extract_images_from_pdf(pdf_path)
results = []
for img_info in images:
ocr_text = self.ocr_image(img_info["bytes"])
if ocr_text.strip():
results.append({
"type": "image_ocr",
"page": img_info["page"],
"content": ocr_text,
"source": "ocr"
})
return results
方案二：多模态大模型（语义理解）
核心思路：把图片喂给 Vision LLM，让它用文字描述图片内容，描述结果存入向量库参与检索
GPT-4o / GPT-4V
import base64
import httpx
from openai import OpenAI
class VisionLLMParser:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
def image_to_base64(self, image_bytes: bytes) -> str:
return base64.b64encode(image_bytes).decode("utf-8")
def describe_image(
self,
image_bytes: bytes,
context: str = "", # 图片周围的文字上下文
image_type: str = "auto" # chart/flowchart/screenshot/photo
) -> str:
"""
用多模态模型生成图片描述
context: 图片在文档中的上下文（前后文字）
"""
base64_image = self.image_to_base64(image_bytes)
# 根据图片类型调整 Prompt
prompts = {
"chart": """
请详细描述这张图表：
1. 图表类型（柱状图/折线图/饼图等）
2. 坐标轴含义和单位
3. 关键数据点和数值
4. 整体趋势或规律
5. 最重要的结论
输出纯文字描述，不要使用Markdown格式。
""",
"flowchart": """
请描述这张流程图/架构图：
1. 整体功能/目的
2. 主要组成部分
3. 关键流程步骤和逻辑关系
4. 输入输出
输出纯文字描述。
""",
"auto": """
请详细描述这张图片的内容，
重点提取其中的关键信息、数据、文字和逻辑关系。
如果包含数据图表，请提取具体数值。
如果包含文字，请完整转录。
输出纯文字，便于后续文本检索。
"""
}
prompt = prompts.get(image_type, prompts["auto"])
# 如果有上下文，补充进去
if context:
prompt += f"\n\n该图片在文档中的上下文：{context[:500]}"
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high" # high/low/auto
}
}
]
}
],
max_tokens=1000
)
return response.choices[0].message.content
def batch_describe(
self, images: list, max_workers: int = 5
) -> list:
"""批量并发处理图片"""
from concurrent.futures import ThreadPoolExecutor, as_completed
results = [None] * len(images)
def process_one(idx, img_info):
description = self.describe_image(
img_info["bytes"],
context=img_info.get("context", ""),
image_type=img_info.get("type", "auto")
)
return idx, description
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(process_one, i, img): i
for i, img in enumerate(images)
}
for future in as_completed(futures):
idx, desc = future.result()
results[idx] = desc
return results
Qwen-VL —— 开源多模态，中文最强
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.generation import GenerationConfig
import torch
from PIL import Image
import io
class QwenVLParser:
"""
Qwen-VL-Chat：阿里开源多模态模型
中文理解能力强，可本地部署，数据不出境
"""
def __init__(self, model_path: str = "Qwen/Qwen-VL-Chat"):
self.tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True
)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map="cuda",
trust_remote_code=True,
bf16=True # 节省显存
).eval()
self.model.generation_config = GenerationConfig.from_pretrained(
model_path, trust_remote_code=True
)
def describe_image(
self, image_bytes: bytes, context: str = ""
) -> str:
# 保存临时文件（Qwen-VL 需要文件路径）
import tempfile, os
with tempfile.NamedTemporaryFile(
suffix=".jpg", delete=False
) as tmp:
tmp.write(image_bytes)
tmp_path = tmp.name
try:
prompt = "请详细描述这张图片的内容，提取所有关键信息、数据和文字。"
if context:
prompt += f"图片的上下文背景：{context[:300]}"
query = self.tokenizer.from_list_format([
{"image": tmp_path},
{"text": prompt}
])
response, _ = self.model.chat(
self.tokenizer,
query=query,
history=None
)
return response
finally:
os.unlink(tmp_path) # 清理临时文件
class InternVLParser:
"""
InternVL2：开源多模态效果最强之一
支持高分辨率图片，表格/图表理解优秀
"""
def __init__(self, model_path="OpenGVLab/InternVL2-8B"):
import torchvision.transforms as T
from transformers import AutoModel
self.model = AutoModel.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="cuda",
trust_remote_code=True
).eval()
self.tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=True
)
def describe_image(self, image_bytes: bytes) -> str:
from PIL import Image
import torchvision.transforms as T
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
# 预处理
transform = T.Compose([
T.Resize((448, 448)),
T.ToTensor(),
T.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
pixel_values = transform(img).unsqueeze(0).to(
torch.bfloat16
).cuda()
question = "<image>\n请详细描述图片内容，提取所有关键信息。"
generation_config = {
"max_new_tokens": 1024,
"do_sample": False
}
response = self.model.chat(
self.tokenizer,
pixel_values,
question,
generation_config
)
return response
方案三：图表专项解析
针对柱状图、折线图、饼图等数据图表，提取精确数值
class ChartParser:
"""
图表专项解析：
目标是提取图表中的精确数据，而不只是描述
"""
def __init__(self, openai_api_key: str):
self.client = OpenAI(api_key=openai_api_key)
def extract_chart_data(self, image_bytes: bytes) -> dict:
"""
从图表中提取结构化数据
输出 JSON 格式，便于后续处理
"""
base64_img = base64.b64encode(image_bytes).decode()
prompt = """
分析这张数据图表，请以JSON格式输出：
{
"chart_type": "图表类型(bar/line/pie/scatter等)",
"title": "图表标题",
"x_axis": {"label": "X轴名称", "unit": "单位"},
"y_axis": {"label": "Y轴名称", "unit": "单位"},
"data": [
{"label": "系列名", "values": [{"x": "...", "y": ...}]}
],
"key_insights": ["关键洞察1", "关键洞察2"],
"summary": "一句话总结图表含义"
}
如果某字段无法识别，填null。只输出JSON，不要其他内容。
"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/jpeg;base64,{base64_img}",
"detail": "high"
}}
]
}],
max_tokens=2000,
response_format={"type": "json_object"}
)
import json
chart_data = json.loads(response.choices[0].message.content)
# 同时生成自然语言描述（用于向量检索）
natural_desc = self._to_natural_language(chart_data)
return {
"structured": chart_data,
"natural_language": natural_desc
}
def _to_natural_language(self, chart_data: dict) -> str:
"""将结构化图表数据转为自然语言（便于向量检索）"""
parts = []
if chart_data.get("title"):
parts.append(f"图表标题：{chart_data['title']}")
if chart_data.get("chart_type"):
parts.append(f"图表类型：{chart_data['chart_type']}")
if chart_data.get("summary"):
parts.append(f"内容摘要：{chart_data['summary']}")
if chart_data.get("key_insights"):
insights = "；".join(chart_data["key_insights"])
parts.append(f"关键洞察：{insights}")
# 提取具体数值
if chart_data.get("data"):
for series in chart_data["data"]:
if series.get("values"):
vals = series["values"]
label = series.get("label", "")
nums = [str(v.get("y", "")) for v in vals[:5]]
parts.append(f"{label}数据：{', '.join(nums)}")
return "\n".join(parts)
方案四：多模态 RAG（图文联合检索）
不把图片转成文字，而是把图片直接向量化，支持用文字查询检索到图片
from PIL import Image
import torch
import io
class MultiModalRAG:
"""
使用 CLIP 模型将图片和文字映射到同一向量空间
实现：文字查询 → 召回相关图片
"""
def __init__(self):
from transformers import CLIPModel, CLIPProcessor
# CLIP：OpenAI 出品，图文向量对齐
self.model = CLIPModel.from_pretrained(
"openai/clip-vit-large-patch14"
)
self.processor = CLIPProcessor.from_pretrained(
"openai/clip-vit-large-patch14"
)
self.model.eval()
def image_to_vector(self, image_bytes: bytes) -> list:
"""图片 → 向量"""
image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
inputs = self.processor(images=image, return_tensors="pt")
with torch.no_grad():
image_features = self.model.get_image_features(**inputs)
# 归一化
image_features = image_features / image_features.norm(
dim=-1, keepdim=True
)
return image_features[0].tolist()
def text_to_vector(self, text: str) -> list:
"""文字 → 向量（与图片向量在同一空间）"""
inputs = self.processor(
text=[text],
return_tensors="pt",
padding=True,
truncation=True
)
with torch.no_grad():
text_features = self.model.get_text_features(**inputs)
text_features = text_features / text_features.norm(
dim=-1, keepdim=True
)
return text_features[0].tolist()
def build_image_index(
self, images: list, vector_db
):
"""
构建图片向量索引
images: [{"bytes": ..., "page": ..., "description": ...}]
"""
for img_info in images:
vector = self.image_to_vector(img_info["bytes"])
# 存入向量数据库
vector_db.upsert({
"id": f"img_{img_info['page']}_{img_info['index']}",
"vector": vector,
"metadata": {
"type": "image",
"page": img_info["page"],
"description": img_info.get("description", ""),
"image_base64": base64.b64encode(
img_info["bytes"]
).decode()
}
})
def search_by_text(
self, query: str, vector_db, top_k: int = 5
) -> list:
"""用文字查询检索相关图片"""
query_vector = self.text_to_vector(query)
results = vector_db.search(query_vector, top_k=top_k)
return results
完整集成方案
import fitz
import base64
import hashlib
from pathlib import Path
from dataclasses import dataclass, field
from typing import List, Optional, Literal
@dataclass
class ImageChunk:
"""图片解析后的 Chunk"""
image_id: str
page_num: int
image_bytes: bytes
ocr_text: str # OCR 提取的文字
description: str # 多模态模型生成的描述
image_type: str # chart/flowchart/screenshot/photo
context_before: str # 图片前的文字
context_after: str # 图片后的文字
vector: Optional[list] = None # CLIP 图片向量
@property
def rag_content(self) -> str:
"""
组合成 RAG 最终使用的文字内容
综合 OCR + 描述 + 上下文
"""
parts = []
if self.context_before:
parts.append(f"[图片上文] {self.context_before[-200:]}")
parts.append(f"[图片描述] {self.description}")
if self.ocr_text:
parts.append(f"[图片文字] {self.ocr_text}")
if self.context_after:
parts.append(f"[图片下文] {self.context_after[:200:]}")
return "\n".join(parts)
class SmartImageParser:
"""
智能图片解析器：
自动判断图片类型，选择最合适的解析策略
"""
def __init__(
self,
vision_model: str = "gpt-4o", # gpt-4o / qwen-vl / internvl
openai_api_key: str = None,
use_ocr: bool = True,
use_clip: bool = False, # 是否做图片向量化
):
self.vision_model = vision_model
self.use_ocr = use_ocr
self.use_clip = use_clip
# 初始化 OCR
if use_ocr:
from paddleocr import PaddleOCR
self.ocr_engine = PaddleOCR(
use_angle_cls=True, lang='ch',
use_gpu=True, show_log=False
)
# 初始化视觉模型
if vision_model == "gpt-4o":
from openai import OpenAI
self.llm_client = OpenAI(api_key=openai_api_key)
elif vision_model == "qwen-vl":
self.llm_client = QwenVLParser()
elif vision_model == "internvl":
self.llm_client = InternVLParser()
# 初始化 CLIP
if use_clip:
self.clip = MultiModalRAG()
def parse_pdf(self, pdf_path: str) -> List[ImageChunk]:
"""解析 PDF 中所有图片"""
doc = fitz.open(pdf_path)
chunks = []
# 提取每页的文字（用于上下文）
page_texts = {
i: page.get_text()
for i, page in enumerate(doc)
}
for page_num, page in enumerate(doc):
images = page.get_images(full=True)
for img_idx, img in enumerate(images):
xref = img[0]
base_img = doc.extract_image(xref)
image_bytes = base_img["image"]
# 过滤太小的图片（可能是图标/装饰）
img_obj = Image.open(io.BytesIO(image_bytes))
w, h = img_obj.size
if w < 100 or h < 100:
continue
# 生成唯一ID
img_hash = hashlib.md5(image_bytes).hexdigest()[:8]
image_id = f"p{page_num+1}_img{img_idx}_{img_hash}"
# 获取上下文文字
page_text = page_texts.get(page_num, "")
prev_text = page_texts.get(page_num - 1, "")
next_text = page_texts.get(page_num + 1, "")
# 1. OCR 提取文字
ocr_text = ""
if self.use_ocr:
ocr_text = self._do_ocr(image_bytes)
# 2. 判断图片类型
img_type = self._classify_image(
image_bytes, ocr_text
)
# 3. 多模态理解
description = self._describe_image(
image_bytes,
context=page_text[:500],
img_type=img_type
)
# 4. CLIP 向量化（可选）
vector = None
if self.use_clip:
vector = self.clip.image_to_vector(image_bytes)
chunk = ImageChunk(
image_id=image_id,
page_num=page_num + 1,
image_bytes=image_bytes,
ocr_text=ocr_text,
description=description,
image_type=img_type,
context_before=prev_text[-300:] + page_text[:300],
context_after=page_text[-300:] + next_text[:300],
vector=vector
)
chunks.append(chunk)
doc.close()
return chunks
def _do_ocr(self, image_bytes: bytes) -> str:
"""OCR 识别"""
import numpy as np
img_array = np.array(
Image.open(io.BytesIO(image_bytes)).convert("RGB")
)
result = self.ocr_engine.ocr(img_array, cls=True)
if not result or not result[0]:
return ""
return "\n".join(
line[1][0]
for line in result[0]
if line[1][1] > 0.75
)
def _classify_image(
self, image_bytes: bytes, ocr_text: str
) -> str:
"""
图片类型分类
根据 OCR 文字密度和图片特征简单判断
"""
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
w, h = img.size
# 文字密度高 → 截图类
if len(ocr_text) > 100:
return "screenshot"
# 宽高比接近 → 可能是图表
if 0.5 < w/h < 2.0 and len(ocr_text) > 10:
return "chart"
return "photo"
def _describe_image(
self,
image_bytes: bytes,
context: str = "",
img_type: str = "auto"
) -> str:
"""调用视觉模型生成描述"""
if self.vision_model == "gpt-4o":
return self._gpt4v_describe(
image_bytes, context, img_type
)
elif self.vision_model in ("qwen-vl", "internvl"):
return self.llm_client.describe_image(
image_bytes, context
)
return ""
def _gpt4v_describe(
self, image_bytes, context, img_type
) -> str:
base64_img = base64.b64encode(image_bytes).decode()
type_prompts = {
"chart": "这是一张数据图表，请提取所有数据值、趋势和关键结论。",
"screenshot": "这是一张截图，请完整识别并转录所有文字内容。",
"photo": "请描述这张图片的主要内容和关键信息。",
"auto": "请详细描述图片内容，提取所有关键信息。"
}
prompt = type_prompts.get(img_type, type_prompts["auto"])
if context:
prompt += f"\n上下文：{context[:400]}"
resp = self.llm_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {
"url": f"data:image/jpeg;base64,{base64_img}",
"detail": "high"
}}
]}],
max_tokens=800
)
return resp.choices[0].message.content
# ─── 使用示例 ────────────────────────────────────────────────
def build_rag_with_images(pdf_path: str, vector_db):
parser = SmartImageParser(
vision_model="gpt-4o",
openai_api_key="sk-...",
use_ocr=True,
use_clip=True
)
image_chunks = parser.parse_pdf(pdf_path)
for chunk in image_chunks:
# 存入向量数据库
vector_db.add({
"id": chunk.image_id,
"content": chunk.rag_content, # 文字描述（用于文本向量）
"image_vector": chunk.vector, # 图片向量（用于图文检索）
"metadata": {
"type": "image",
"page": chunk.page_num,
"image_type": chunk.image_type,
"has_ocr": bool(chunk.ocr_text),
# 存原图 base64，生成答案时可以展示
"image_base64": base64.b64encode(
chunk.image_bytes
).decode()
}
})
print(f"共解析 {len(image_chunks)} 张图片")
方案对比
┌──────────────────┬────────┬────────┬────────┬───────────────┐
│ 方案 │ 成本 │ 效果 │ 速度 │ 适合场景 │
├──────────────────┼────────┼────────┼────────┼───────────────┤
│ 跳过图片 │ 零 │ ❌ │ 最快 │ 图片无关紧要 │
│ OCR（PaddleOCR）│ 低 │ ⭐⭐⭐ │ 快 │ 文字截图/扫描 │
│ GPT-4o Vision │ 高 │ ⭐⭐⭐⭐⭐ │ 中 │ 各类图片 │
│ Qwen-VL（本地） │ 低 │ ⭐⭐⭐⭐ │ 中 │ 中文，不出境 │
│ InternVL2（本地）│ 低 │ ⭐⭐⭐⭐⭐ │ 中 │ 复杂图表 │
│ CLIP 向量化 │ 低 │ ⭐⭐⭐ │ 快 │ 图文检索 │
└──────────────────┴────────┴────────┴────────┴───────────────┘
选型建议
图片中主要是什么？
│
├── 文字/截图/扫描
│ └──→ PaddleOCR（快、准、免费）
│
├── 数据图表（柱状/折线/饼图）
│ ├── 要精确数值 ──→ GPT-4o + JSON结构化提取
│ └── 要语义描述 ──→ InternVL2（本地）
│
├── 流程图/架构图
│ └──→ GPT-4o / Qwen-VL（理解逻辑关系）
│
├── 中文文档，数据不出境
│ └──→ Qwen-VL / InternVL2（本地部署）
│
└── 需要"以文搜图"检索
└──→ CLIP 向量化 + 向量数据库
最佳实践：生产级 RAG 系统通常组合使用 —— PaddleOCR 兜底提取文字 + 多模态模型（GPT-4o 或 InternVL2）理解语义 + 上下文拼接增强检索，三者结合才能最大限度保留图片中的信息。