RAG-Anything 深度解析：从多模态文档解析到知识图谱构建，全链路实战指南
当你的知识库里塞满了 PDF 论文、Excel 报表、架构图、数学公式，传统 RAG 只能看到文字——剩下的全是黑洞。RAG-Anything 要做的，就是让 AI 真正"看懂"每一页文档里的每一种内容。
一、为什么需要多模态 RAG？
1.1 传统 RAG 的致命盲区
先说结论：纯文本 RAG 在真实业务场景里，至少丢掉 40% 的信息。
这不是我瞎说。随便翻开一篇学术论文，图表占多少篇幅？一张架构图的信息密度顶得上三页文字描述。一份财报里，关键的财务数据都在表格里，文字部分只是定性的总结。传统 RAG 系统的处理方式是：把 PDF 用 OCR 或者文本提取工具转成纯文本，然后切片、向量化、存进向量库。
问题来了：
- 图片变成了一堆乱码或者直接丢失——一张系统架构图，OCR 可能提取出几个标签文字，但架构的全貌、模块间的调用关系全没了
- 表格被打散成换行分隔的文本——行列关系彻底丢失，"Q1 营收 320 万"变成了散落在不同 chunk 里的"Q1"、"营收"、"320"
- 公式变成一坨 LaTeX 碎片——$\sum_{i=1}^{n} f(x_i)$ 被切成 $\sum、i=1、n、f(x_i)$，语义全断
- 跨模态引用断裂——正文说"如上图所示"，但"上图"在另一个 chunk 里，检索时根本对不上
这就是多模态 RAG 要解决的核心问题：让知识库不只是存文字，而是存理解。
1.2 多模态 RAG 的技术演进
多模态 RAG 不是突然冒出来的，它经历了几个阶段：
阶段一：暴力拼接（2023 年中之前）
最朴素的做法——图片用 CLIP/BLIP 生成描述文本，表格转 Markdown，公式转 LaTeX 字符串，然后和正文一起扔进向量库。问题是模态间的语义关联全丢了，检索精度反而不如纯文本 RAG。
阶段二：独立索引 + 跨模态检索（2023 底 - 2024 中）
给每种模态建独立的索引：文本走文本 embedding，图片走视觉 embedding，表格走结构化 embedding。查询时做跨模态对齐。代表工作如 ColPali、UniRAG。这个方案检索精度提高了，但架构复杂，部署成本高，而且模态间的深层语义关联依然缺失。
阶段三：知识图谱统一表示（2024 底 - 至今）
把所有模态的内容统一转化为知识图谱中的实体和关系。文本中的概念、图片中的对象、表格中的数据点、公式中的变量——都是图谱中的节点，它们之间的语义关系是边。检索变成了图上的推理过程，天然支持跨模态关联。RAG-Anything 就是这个阶段的代表作。
二、RAG-Anything 架构全景
2.1 项目概览
RAG-Anything 是香港大学数据智能实验室（HKUDS）开源的全模态 RAG 框架，基于他们之前的工作 LightRAG 构建。GitHub 上已经超过 17,000 Star，是目前多模态 RAG 领域最受关注的项目之一。
核心定位一句话：一个框架内，端到端地处理所有模态的文档内容，统一构建知识图谱，统一检索生成。
项目地址：https://github.com/HKUDS/RAG-Anything
基础框架：LightRAG（EMNLP 2025）
文档解析：MinerU / Docling / PaddleOCR
语言：Python
许可证：MIT
2.2 整体架构
RAG-Anything 的架构可以用五层来理解：
┌──────────────────────────────────────────────┐
│ Query & Generation │ ← 用户查询 + LLM 生成
├──────────────────────────────────────────────┤
│ Knowledge Graph Layer │ ← 统一知识图谱
├──────────────────────────────────────────────┤
│ Multi-Modal Analysis Engine │ ← 四大分析器
├──────────────────────────────────────────────┤
│ Content Classification Router │ ← 模态路由分发
├──────────────────────────────────────────────┤
│ Document Parsing Layer │ ← MinerU / Docling
└──────────────────────────────────────────────┘
从下往上走，数据流是这样的：
- 文档输入 → PDF/Office/图片等原始文档
- 文档解析层 → 用 MinerU 或 Docling 做高保真结构抽取，保留文档的层级关系和空间位置
- 内容分类路由 → 每个内容块自动判断类型（图片/表格/公式/纯文字），分发到对应处理管线
- 多模态分析引擎 → 四个专用分析器并行处理，把每种模态的内容转化为统一的语义表示
- 知识图谱层 → 所有模态的语义表示统一构建知识图谱
- 查询与生成 → 基于 LightRAG 的图检索 + LLM 生成
2.3 关键设计决策
为什么选知识图谱而不是向量数据库作为统一表示？这是 RAG-Anything 最核心的设计决策。原因有三：
第一，跨模态关联的天然表达。 向量空间里，一张图和一个表格的 embedding 在不同子空间，要跨模态检索必须做对齐。但在知识图谱里，"2024年Q1营收"这个实体可以同时关联到文本中的描述、表格中的数值、图表中的数据点——不需要额外的对齐操作。
第二，推理能力的质变。 向量检索本质上是相似度匹配，"近邻"就是答案。但很多问题需要多跳推理："哪个季度的营收增速最快？"——你需要先找到各季度营收数据（可能在表格里），计算增速，然后比较。知识图谱上的图遍历天然支持这种多跳查询。
第三，可解释性。 向量检索返回"最相似的 5 个 chunk"，但你不知道为什么相似。知识图谱的检索路径是可追溯的：从这个实体出发，经过哪些关系，到达哪些结论——全链路白盒。
三、文档解析层：MinerU 深度解析
3.1 为什么不是 PyMuPDF 或 pdfplumber？
做 PDF 解析的库一大堆，PyMuPDF、pdfplumber、Camelot、Tabula……为什么 RAG-Anything 选了 MinerU？
核心区别：传统工具做的是文本提取，MinerU 做的是文档理解。
PyMuPDF 的逻辑是：PDF 里有文本流，我把文本流按坐标顺序读出来。这对于排版简单的文档（比如纯文本书籍）够用，但对于学术论文、技术报告这种复杂排版的文档，它会遇到：
- 多栏布局：左右两栏的文字被混在一起
- 图文混排：图片区域被当作文本流的一部分，提取出乱码
- 表格：单元格被按行读取，列关系丢失
- 页眉页脚：和正文混在一起
- 脚注和引用：散落在各处
MinerU 的做法完全不同。它使用了版面分析模型（Layout Analysis Model），先识别文档的物理结构（标题、段落、图片区域、表格区域、公式区域），然后对每个区域做针对性的内容提取。
3.2 MinerU 的处理流水线
# MinerU 的核心处理流程（简化版）
from magic_pdf.pipe.UNIPipe import UNIPipe
def parse_document_with_mineru(pdf_path: str) -> dict:
"""
使用 MinerU 解析 PDF 文档，返回结构化内容
"""
# 1. 初始化解析管道
pipe = UNIPipe(pdf_path)
# 2. 版面分析 - 识别文档的物理结构
# 输出：每个区域的类型（title/text/image/table/equation）
# 和位置信息（bounding box）
layout_result = pipe.analyze_layout()
# 3. 按区域类型分别处理
results = {
"text_blocks": [], # 文本块
"image_blocks": [], # 图片块
"table_blocks": [], # 表格块
"equation_blocks": [], # 公式块
}
for block in layout_result.blocks:
if block.type == "text":
# 文本块：保留段落层级和阅读顺序
text_content = pipe.extract_text(block)
results["text_blocks"].append({
"content": text_content,
"bbox": block.bbox,
"page": block.page_num,
"level": block.heading_level, # 标题层级
})
elif block.type == "image":
# 图片块：裁剪图片 + OCR（如有内嵌文字）
image_data = pipe.crop_image(block)
ocr_text = pipe.ocr_image(block) if block.has_text else ""
results["image_blocks"].append({
"image": image_data,
"ocr_text": ocr_text,
"caption": block.caption, # 自动关联图注
"bbox": block.bbox,
"page": block.page_num,
})
elif block.type == "table":
# 表格块：结构化提取，保留行列关系
table_html = pipe.extract_table(block)
results["table_blocks"].append({
"html": table_html,
"caption": block.caption,
"bbox": block.bbox,
"page": block.page_num,
})
elif block.type == "equation":
# 公式块：转 LaTeX
latex = pipe.extract_equation(block)
results["equation_blocks"].append({
"latex": latex,
"bbox": block.bbox,
"page": block.page_num,
})
return results
关键点在于 analyze_layout() 这一步——这是 MinerU 的核心竞争力。它使用了一个基于 DocLayout-YOLO 的版面分析模型，能够识别多达 10 种文档区域类型，准确率在学术论文场景下超过 95%。
3.3 Docling 作为备选解析器
RAG-Anything 同时支持 Docling 作为备选解析器。Docling 是 IBM 开源的文档处理库，它的优势在于：
- 对 Office 格式（.docx, .pptx, .xlsx）的支持更好
- 内置了 AI4Science 的图表理解模型
- 输出格式是 DoclingDocument，一种标准化的文档表示格式
# Docling 解析示例
from docling.document_converter import DocumentConverter
def parse_with_docling(file_path: str) -> dict:
converter = DocumentConverter()
result = converter.convert(file_path)
# Docling 的输出是结构化的 DoclingDocument
doc = result.document
# 遍历文档中的所有元素
for item in doc.iterate_items():
if hasattr(item, 'tables'):
for table in item.tables:
# 表格以 HTML 或 CSV 格式导出
table_html = table.export_to_html()
elif hasattr(item, 'pictures'):
for pic in item.pictures:
# 图片区域和关联的 caption
pass
return result
RAG-Anything 的文档解析层做了抽象，上层代码不需要关心底层用的是 MinerU 还是 Docling：
# RAG-Anything 的文档解析抽象层
class DocumentParser(ABC):
@abstractmethod
def parse(self, file_path: str) -> List[ContentBlock]:
"""解析文档，返回内容块列表"""
pass
class MinerUParser(DocumentParser):
def parse(self, file_path: str) -> List[ContentBlock]:
# MinerU 实现
pass
class DoclingParser(DocumentParser):
def parse(self, file_path: str) -> List[ContentBlock]:
# Docling 实现
pass
# 工厂函数
def create_parser(engine: str = "mineru") -> DocumentParser:
if engine == "mineru":
return MinerUParser()
elif engine == "docling":
return DoclingParser()
else:
raise ValueError(f"Unsupported parser: {engine}")
四、内容分类路由：模态分发引擎
4.1 自动模态识别
解析层输出的 ContentBlock 已经带了类型信息，但 RAG-Anything 并不直接信任解析器的类型判断。它有一个二次分类路由模块，用一个小型的多模态分类模型对每个内容块做二次确认：
ContentBlock 输入
│
├── 类型标签 = "text" → 文本置信度检查
│ ├── 置信度 > 0.9 → 直接走文本管线
│ └── 置信度 ≤ 0.9 → 重新分类
│
├── 类型标签 = "image" → 视觉内容检查
│ ├── 包含文字区域 → 图片 + OCR 双管线
│ └── 纯视觉内容 → 图片单管线
│
├── 类型标签 = "table" → 表格结构验证
│ ├── 行列结构完整 → 表格管线
│ └── 行列残缺 → 回退到图片管线
│
└── 类型标签 = "equation" → 公式验证
├── LaTeX 解析成功 → 公式管线
└── LaTeX 解析失败 → 回退到图片管线
为什么要做二次确认？因为版面分析模型在边界情况下会犯错。比如一个带边框的代码块可能被误识别为表格，一个嵌在文本中的小型公式可能被忽略。二次路由把这些错误捞回来。
4.2 并行管线架构
四种模态的处理管线是并行执行的。这一点至关重要——一篇 50 页的 PDF 可能有 100 个文本块、30 个图片、15 个表格、20 个公式。如果串行处理，光是图片分析就要跑几十分钟。
import asyncio
from concurrent.futures import ThreadPoolExecutor
class MultiModalRouter:
"""多模态内容路由器 - 并行分发到各处理管线"""
def __init__(self, max_workers: int = 4):
self.text_analyzer = TextAnalyzer()
self.image_analyzer = VisualContentAnalyzer()
self.table_analyzer = TableAnalyzer()
self.equation_analyzer = EquationAnalyzer()
self.executor = ThreadPoolExecutor(max_workers=max_workers)
async def route_and_process(
self,
content_blocks: List[ContentBlock]
) -> List[ProcessedBlock]:
"""路由并并行处理所有内容块"""
# 按类型分组
groups = {
"text": [],
"image": [],
"table": [],
"equation": [],
}
for block in content_blocks:
confirmed_type = self._confirm_type(block)
groups[confirmed_type].append(block)
# 并行处理各管线
loop = asyncio.get_event_loop()
tasks = []
if groups["text"]:
tasks.append(
loop.run_in_executor(
self.executor,
self.text_analyzer.batch_process,
groups["text"]
)
)
if groups["image"]:
tasks.append(
loop.run_in_executor(
self.executor,
self.image_analyzer.batch_process,
groups["image"]
)
)
if groups["table"]:
tasks.append(
loop.run_in_executor(
self.executor,
self.table_analyzer.batch_process,
groups["table"]
)
)
if groups["equation"]:
tasks.append(
loop.run_in_executor(
self.executor,
self.equation_analyzer.batch_process,
groups["equation"]
)
)
# 等待所有管线完成
results = await asyncio.gather(*tasks)
# 合并结果，按原文档顺序排列
all_processed = []
for batch in results:
all_processed.extend(batch)
# 恢复原始页面顺序
all_processed.sort(key=lambda x: (x.page, x.bbox.y1))
return all_processed
五、四大分析器：从模态到语义
这是 RAG-Anything 的核心——四个专用分析器，各自负责一种模态的深度语义提取。
5.1 Visual Content Analyzer（视觉内容分析器）
目标：把图片转化为 LLM 可理解的语义描述 + 结构化对象列表。
不要以为这只是调一下 GPT-4V 的 API 让它"描述一下这张图"。RAG-Anything 的视觉分析器做的是三层提取：
class VisualContentAnalyzer:
"""视觉内容分析器 - 三层语义提取"""
def analyze(self, image_block: ContentBlock) -> VisualAnalysis:
# 第一层：全局语义描述
# 用 VLM 生成图片的整体描述
global_description = self.vlm.describe_image(
image=image_block.image,
prompt=(
"Describe this image in detail. "
"Focus on: 1) What type of diagram/image this is "
"2) Key components and their relationships "
"3) Any data trends or patterns visible"
)
)
# 第二层：对象级提取
# 检测图中的关键对象，提取为结构化实体
objects = self.vlm.extract_objects(
image=image_block.image,
prompt=(
"List all named entities, components, or modules "
"visible in this image. For each, provide: "
"name, type, position (if applicable), "
"and connections to other objects."
)
)
# 第三层：图注关联
# 把图片描述和正文中的引用关联起来
# "如上图所示" → 关联到具体图片
caption_context = self._find_caption_reference(image_block)
return VisualAnalysis(
global_description=global_description,
objects=objects,
caption_context=caption_context,
source_page=image_block.page,
source_bbox=image_block.bbox,
)
三层提取的意义：
- 全局描述用于关键词检索——用户搜"系统架构图"，能命中
- 对象级提取用于实体检索——用户搜"Redis 模块"，能找到架构图中 Redis 的位置
- 图注关联用于上下文增强——检索到图片时，自动带上周围的相关文本
5.2 Table Analyzer（表格分析器）
目标：把表格转化为结构化数据 + 语义摘要，支持精确查询。
表格是最难处理的一种模态。传统的做法要么把表格转成 Markdown/HTML 文本然后切块（丢失结构），要么把表格存进 SQL 数据库（丢失语义）。RAG-Anything 的方案是双重表示：
class TableAnalyzer:
"""表格分析器 - 双重语义表示"""
def analyze(self, table_block: ContentBlock) -> TableAnalysis:
# 1. 结构化数据提取
# 将 HTML 表格转为结构化格式
structured_data = self._html_to_structured(table_block.html)
# 2. 语义摘要生成
# 用 LLM 理解表格内容，生成自然语言摘要
summary = self.llm.generate(
prompt=(
f"Analyze this table and provide:\n"
f"1. What this table is about (one sentence)\n"
f"2. Key data points and trends\n"
f"3. Notable relationships between columns\n\n"
f"Table data:\n{table_block.html}"
)
)
# 3. 单元格级语义标注
# 对关键单元格生成可检索的语义标签
cell_annotations = self._annotate_cells(structured_data)
# 4. 列级统计信息
# 自动计算数值列的统计量
column_stats = self._compute_column_stats(structured_data)
return TableAnalysis(
structured_data=structured_data,
summary=summary,
cell_annotations=cell_annotations,
column_stats=column_stats,
caption=table_block.caption,
)
def _annotate_cells(self, data: dict) -> List[CellAnnotation]:
"""对关键单元格生成语义标注"""
annotations = []
for row_idx, row in enumerate(data["rows"]):
for col_idx, cell in enumerate(row):
if cell.value is not None:
annotation = CellAnnotation(
row_header=data["row_headers"][row_idx]
if row_idx < len(data["row_headers"]) else None,
col_header=data["col_headers"][col_idx],
value=cell.value,
semantic_label=self._infer_semantic_label(
cell, data["col_headers"][col_idx]
),
)
annotations.append(annotation)
return annotations
def _compute_column_stats(self, data: dict) -> dict:
"""计算数值列的统计信息"""
stats = {}
for col_idx, header in enumerate(data["col_headers"]):
values = [
row[col_idx].numeric_value
for row in data["rows"]
if row[col_idx].is_numeric
]
if values:
stats[header] = {
"min": min(values),
"max": max(values),
"mean": sum(values) / len(values),
"count": len(values),
}
return stats
双重表示的好处：用户问"Q1 营收是多少"时，可以通过语义摘要定位到相关表格，再通过结构化数据精确提取数值——而不是在一段文本里用正则去捞。
5.3 Equation Analyzer（公式分析器）
目标：把数学公式转化为可解释的语义结构 + 变量说明。
公式的处理比图片和表格更微妙。LaTeX 本身已经是结构化的了，但 LaTeX 是排版语言，不是语义语言。$\frac{dy}{dx}$ 在 LaTeX 里就是 \frac{dy}{dx}，但你不知道 $y$ 是什么，$x$ 是什么，这个导数在上下文中的物理意义是什么。
class EquationAnalyzer:
"""公式分析器 - 从 LaTeX 到语义结构"""
def analyze(self, equation_block: ContentBlock) -> EquationAnalysis:
latex = equation_block.latex
# 1. LaTeX 语法验证和修复
clean_latex = self._validate_and_fix_latex(latex)
# 2. 变量提取
# 从公式和周围文本中提取变量定义
variables = self._extract_variables(clean_latex, equation_block.context)
# 3. 语义解释
# 用 LLM 生成公式的自然语言解释
interpretation = self.llm.generate(
prompt=(
f"Explain this mathematical formula in plain language:\n"
f"LaTeX: {clean_latex}\n"
f"Context from paper: {equation_block.context[:500]}\n"
f"Variables identified: {variables}\n\n"
f"Provide: 1) What this formula computes "
f"2) What each variable represents "
f"3) The physical/practical meaning of this formula"
)
)
# 4. 公式分类
category = self._classify_formula(clean_latex)
return EquationAnalysis(
latex=clean_latex,
variables=variables,
interpretation=interpretation,
category=category, # e.g., "loss_function", "activation", "probability"
source_page=equation_block.page,
)
def _extract_variables(self, latex: str, context: str) -> List[Variable]:
"""从公式和上下文中提取变量定义"""
variables = []
# 用 LLM 从上下文中提取变量定义
var_defs = self.llm.generate(
prompt=(
f"Extract variable definitions from this text:\n"
f"{context[:1000]}\n\n"
f"The formula is: {latex}\n"
f"List each variable with its definition."
)
)
# 解析 LLM 输出为结构化的 Variable 对象
for line in var_defs.strip().split("\n"):
if ":" in line or "=" in line:
var = self._parse_variable_line(line)
if var:
variables.append(var)
return variables
def _classify_formula(self, latex: str) -> str:
"""公式类型分类"""
# 基于模式匹配的快速分类
if r"\sum" in latex or r"\prod" in latex:
return "aggregation"
elif r"\int" in latex:
return "integral"
elif r"\frac{\partial" in latex or r"\nabla" in latex:
return "derivative"
elif r"\mathbb{E}" in latex or r"\mathbb{P}" in latex:
return "probability"
elif r"\min" in latex or r"\max" in latex or r"\arg" in latex:
return "optimization"
elif r"\mathcal{L}" in latex or "loss" in latex.lower():
return "loss_function"
else:
return "general"
5.4 Text Analyzer（文本分析器）
文本分析器看起来最简单，但实际上 RAG-Anything 在这一层做了比传统 RAG 更深的工作。它不是简单地切块 + embedding，而是做了实体-关系联合提取：
class TextAnalyzer:
"""文本分析器 - 实体关系联合提取"""
def analyze(self, text_block: ContentBlock) -> TextAnalysis:
# 1. 实体提取（Named Entity Recognition + 开放域实体）
entities = self._extract_entities(text_block.content)
# 2. 关系提取
relations = self._extract_relations(text_block.content, entities)
# 3. 段落级语义摘要
summary = self._generate_summary(text_block.content)
# 4. 与其他模态的交叉引用检测
cross_refs = self._detect_cross_references(text_block)
return TextAnalysis(
content=text_block.content,
entities=entities,
relations=relations,
summary=summary,
cross_references=cross_refs,
)
def _detect_cross_references(
self, text_block: ContentBlock
) -> List[CrossReference]:
"""检测文本中对其他模态内容的引用"""
patterns = [
r"如图\s*(\d+)", # "如图1"
r"如表\s*(\d+)", # "如表1"
r"公式\s*\((\d+)\)", # "公式(1)"
r"见\s*图\s*(\d+)", # "见图1"
r"Figure\s*(\d+)", # "Figure 1"
r"Table\s*(\d+)", # "Table 1"
r"Eq\.\s*\((\d+)\)", # "Eq. (1)"
]
refs = []
for pattern in patterns:
matches = re.finditer(pattern, text_block.content)
for match in matches:
ref_type, ref_id = self._parse_reference(match.group())
refs.append(CrossReference(
ref_type=ref_type, # "figure" / "table" / "equation"
ref_id=ref_id,
context=match.group(),
position=match.start(),
))
return refs
交叉引用检测是 RAG-Anything 的一个亮点。它让文本中"如图3所示"这样的引用能够自动关联到具体的图片实体，在知识图谱中建立跨模态的边。
六、知识图谱构建：LightRAG 的核心
6.1 从语义到图谱
四个分析器输出的语义信息，最终都要汇入统一的知识图谱。这个构建过程基于 LightRAG 的双级索引结构：
实体级索引（Entity-Level Index）：每个实体（概念、对象、变量、数据点）是图中的一个节点，带有类型标签和属性。
关系级索引（Relation-Level Index）：实体间的关系是图中的边，带有语义标签和权重。
class KnowledgeGraphBuilder:
"""知识图谱构建器 - 统一所有模态的语义"""
def build(
self,
processed_blocks: List[ProcessedBlock]
) -> KnowledgeGraph:
kg = KnowledgeGraph()
for block in processed_blocks:
if isinstance(block, VisualAnalysis):
# 图片：对象作为实体，空间关系作为边
for obj in block.objects:
kg.add_entity(
id=f"img_obj_{block.source_page}_{obj.name}",
name=obj.name,
type="visual_object",
properties={
"source_type": "image",
"page": block.source_page,
"description": obj.description,
}
)
# 对象间的关系
for conn in obj.connections:
kg.add_relation(
source=f"img_obj_{block.source_page}_{conn.from_obj}",
target=f"img_obj_{block.source_page}_{conn.to_obj}",
relation=conn.relation_type,
weight=conn.confidence,
)
elif isinstance(block, TableAnalysis):
# 表格：列头和关键数据点作为实体
for annotation in block.cell_annotations:
entity_id = (
f"tbl_{block.caption}_"
f"{annotation.row_header}_{annotation.col_header}"
)
kg.add_entity(
id=entity_id,
name=f"{annotation.col_header}: {annotation.value}",
type="data_point",
properties={
"source_type": "table",
"table_caption": block.caption,
"row": annotation.row_header,
"column": annotation.col_header,
"value": annotation.value,
}
)
elif isinstance(block, EquationAnalysis):
# 公式：变量和公式本身作为实体
kg.add_entity(
id=f"eq_{block.source_page}",
name=f"Equation: {block.latex[:50]}...",
type="equation",
properties={
"latex": block.latex,
"interpretation": block.interpretation,
"category": block.category,
}
)
for var in block.variables:
kg.add_entity(
id=f"var_{block.source_page}_{var.symbol}",
name=var.symbol,
type="variable",
properties={"definition": var.definition},
)
kg.add_relation(
source=f"eq_{block.source_page}",
target=f"var_{block.source_page}_{var.symbol}",
relation="contains_variable",
)
elif isinstance(block, TextAnalysis):
# 文本：实体和关系直接入图
for entity in block.entities:
kg.add_entity(
id=f"txt_ent_{entity.name}",
name=entity.name,
type=entity.type,
properties={
"source_type": "text",
"definition": entity.definition,
}
)
for rel in block.relations:
kg.add_relation(
source=rel.subject,
target=rel.object,
relation=rel.predicate,
weight=rel.confidence,
)
# 跨模态引用 → 建立跨模态边
for cross_ref in block.cross_references:
target_id = self._resolve_cross_reference(cross_ref)
if target_id:
kg.add_relation(
source=f"txt_ent_{block.entities[0].name}",
target=target_id,
relation=f"references_{cross_ref.ref_type}",
)
return kg
6.2 LightRAG 的双级检索
有了知识图谱，检索怎么做？LightRAG 提出了双级检索策略：
低级检索（Low-Level Retrieval）：针对具体实体和关系的精确查询。比如"Redis 在架构中的位置是什么？"——直接在图上找 Redis 节点，遍历它的关系。
高级检索（High-Level Retrieval）：针对总结性和跨领域的问题。比如"这个系统的整体设计思路是什么？"——需要聚合多个实体和关系，生成高层摘要。
class DualLevelRetriever:
"""双级检索器"""
def retrieve(
self,
query: str,
kg: KnowledgeGraph,
mode: str = "hybrid" # "low" / "high" / "hybrid"
) -> RetrievalResult:
# 1. 查询理解
query_entities = self._extract_query_entities(query)
query_type = self._classify_query(query)
if query_type == "specific" or mode == "low":
# 低级检索：精确实体匹配 + 子图提取
subgraph = self._exact_entity_search(query_entities, kg)
context = self._subgraph_to_context(subgraph)
elif query_type == "summarization" or mode == "high":
# 高级检索：社区检测 + 社区摘要
communities = self._detect_communities(kg)
relevant_communities = self._rank_communities(
communities, query_entities
)
context = self._communities_to_context(relevant_communities)
else: # hybrid
# 混合检索：两者都做，结果融合
low_result = self.retrieve(query, kg, mode="low")
high_result = self.retrieve(query, kg, mode="high")
context = self._merge_contexts(
low_result.context, high_result.context
)
return RetrievalResult(
context=context,
query_type=query_type,
retrieval_mode=mode,
)
七、端到端实战：构建一个论文知识库
说了这么多架构，来点实际的。下面我们用 RAG-Anything 从零构建一个论文知识库。
7.1 环境搭建
# 克隆项目
git clone https://github.com/HKUDS/RAG-Anything.git
cd RAG-Anything
# 安装依赖（建议用 conda 管理环境）
conda create -n rag-anything python=3.11
conda activate rag-anything
# 安装核心依赖
pip install -r requirements.txt
# 安装 MinerU（文档解析引擎）
pip install magic-pdf[full]
# 如果 GPU 可用，安装 CUDA 版本的 PyTorch
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121
7.2 基础配置
# config.py
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class RAGAnythingConfig:
# LLM 配置
llm_model: str = "gpt-4o" # 或 "deepseek-chat"
llm_api_key: str = "your-api-key"
llm_base_url: Optional[str] = None # 自定义 API 地址
# Embedding 配置
embedding_model: str = "text-embedding-3-small"
embedding_dim: int = 1536
# 文档解析配置
parser_engine: str = "mineru" # "mineru" / "docling"
ocr_enabled: bool = True
# 知识图谱配置
kg_storage: str = "json" # "json" / "neo4j" / "networkx"
kg_chunk_size: int = 1200 # 文本块大小（token）
kg_chunk_overlap: int = 100 # 块重叠大小
# 检索配置
retrieval_mode: str = "hybrid" # "low" / "high" / "hybrid"
max_retrieved_entities: int = 20
# 并行配置
max_workers: int = 4 # 并行管线数
# 存储路径
workspace: str = "./workspace"
7.3 构建论文知识库
import os
from rag_anything import RAGAnythingPipeline, RAGAnythingConfig
# 初始化配置
config = RAGAnythingConfig(
llm_model="gpt-4o",
llm_api_key=os.environ["OPENAI_API_KEY"],
parser_engine="mineru",
kg_storage="json",
retrieval_mode="hybrid",
workspace="./paper_kg",
)
# 初始化管道
pipeline = RAGAnythingPipeline(config)
# 导入论文
paper_dir = "./papers" # 放论文 PDF 的目录
for filename in os.listdir(paper_dir):
if filename.endswith(".pdf"):
filepath = os.path.join(paper_dir, filename)
print(f"Processing: {filename}")
# 一键导入：解析 → 分析 → 入图
result = pipeline.ingest(filepath)
print(f" Entities: {result.entity_count}")
print(f" Relations: {result.relation_count}")
print(f" Processing time: {result.time_seconds:.1f}s")
# 查看知识图谱统计
stats = pipeline.get_kg_stats()
print(f"\nKnowledge Graph Stats:")
print(f" Total entities: {stats.total_entities}")
print(f" Total relations: {stats.total_relations}")
print(f" Entity types: {stats.entity_type_distribution}")
print(f" Modalities covered: {stats.modality_coverage}")
7.4 多模态查询实战
# 查询 1：精确查询 - 找架构图中的 Redis 模块
answer = pipeline.query(
"在系统架构图中，Redis 模块的位置和作用是什么？"
)
print(answer.response)
# 输出示例：
# 根据论文第3页的架构图，Redis 作为缓存层位于 API Gateway 和
# 后端服务之间，主要用于：(1) 请求限流 (2) 会话存储 (3) 热点数据缓存。
# 架构中的连接关系为：API Gateway → Redis → Backend Services。
# 查询 2：数据查询 - 从表格中提取精确数值
answer = pipeline.query(
"实验中各个模型在 MMLU 基准上的分数是多少？"
)
print(answer.response)
# 输出示例：
# 根据论文表2的数据，各模型在MMLU上的表现为：
# - GPT-4o: 88.7%
# - Claude-3.5: 86.2%
# - 提出方法: 91.3%
# 提出方法在所有子任务上均优于基线模型。
# 查询 3：跨模态推理
answer = pipeline.query(
"论文中的损失函数公式和训练曲线之间有什么关系？"
)
print(answer.response)
# 输出示例：
# 论文公式(3)定义的损失函数包含三个项：
# L_total = α·L_ce + β·L_kd + γ·L_reg
# 其中 α=0.5, β=0.3, γ=0.2（见第4页参数表）。
# 从图5的训练曲线可以看出，L_kd 项的加入使得
# 收敛速度提升了约40%，而 L_reg 项有效抑制了
# 过拟合现象（验证集曲线更加平滑）。
7.5 知识图谱可视化
# 导出知识图谱用于可视化
kg = pipeline.get_knowledge_graph()
# 方法 1：导出为 GraphML（可用 Gephi 打开）
kg.export_graphml("paper_kg.graphml")
# 方法 2：用 NetworkX + Matplotlib 快速可视化
import networkx as nx
import matplotlib.pyplot as plt
G = kg.to_networkx()
# 按模态类型着色
color_map = {
"visual_object": "#FF6B6B", # 红色 - 来自图片
"data_point": "#4ECDC4", # 青色 - 来自表格
"equation": "#FFE66D", # 黄色 - 来自公式
"concept": "#95E1D3", # 绿色 - 来自文本
"variable": "#F38181", # 粉色 - 公式变量
}
node_colors = [
color_map.get(
G.nodes[n].get("type", "concept"), "#DDDDDD"
)
for n in G.nodes()
]
plt.figure(figsize=(16, 12))
pos = nx.spring_layout(G, k=2, iterations=50)
nx.draw(
G, pos,
node_color=node_colors,
node_size=80,
with_labels=False,
edge_color="#CCCCCC",
alpha=0.7,
)
plt.title("RAG-Anything Knowledge Graph")
plt.savefig("kg_visualization.png", dpi=150, bbox_inches="tight")
八、性能优化：从概念验证到生产级
8.1 文档解析加速
MinerU 的版面分析模型默认跑在 CPU 上，一篇 50 页的 PDF 解析需要 2-3 分钟。两个优化手段：
# 优化 1：启用 GPU 加速
# MinerU 支持用 ONNX Runtime 的 GPU 后端
import os
os.environ["MINERU_DEVICE"] = "cuda" # 使用 GPU
# 优化 2：批量并行解析
# 多个文档并行处理
from concurrent.futures import ProcessPoolExecutor
def batch_ingest(pipeline, file_paths, max_workers=4):
with ProcessPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(pipeline.ingest, fp): fp
for fp in file_paths
}
results = []
for future in futures:
result = future.result()
results.append(result)
return results
8.2 知识图谱存储优化
JSON 存储只适合小规模实验。当文档量超过 100 篇，实体数超过 10 万时，需要切换到图数据库：
# Neo4j 存储后端
config = RAGAnythingConfig(
kg_storage="neo4j",
neo4j_uri="bolt://localhost:7687",
neo4j_user="neo4j",
neo4j_password="your-password",
)
# 创建索引加速查询
# 在 Neo4j 中为常用查询模式创建索引
CREATE INDEX entity_name IF NOT EXISTS
FOR (e:Entity) ON (e.name);
CREATE INDEX entity_type IF NOT EXISTS
FOR (e:Entity) ON (e.type);
CREATE INDEX relation_type IF NOT EXISTS
FOR ()-[r:RELATES_TO]-() ON (r.type);
8.3 LLM 调用成本控制
多模态分析器每个都要调 LLM，一篇文档可能产生几十次 API 调用。控制成本的关键是缓存 + 批处理：
# 缓存策略：相同/相似内容的分析结果缓存
from functools import lru_cache
import hashlib
class CachedAnalyzer:
def __init__(self, analyzer, cache_dir="./cache"):
self.analyzer = analyzer
self.cache_dir = cache_dir
def analyze(self, block: ContentBlock) -> AnalysisResult:
# 生成缓存 key
cache_key = self._compute_cache_key(block)
cache_path = os.path.join(self.cache_dir, f"{cache_key}.json")
# 检查缓存
if os.path.exists(cache_path):
return AnalysisResult.from_json(
open(cache_path).read()
)
# 缓存未命中，执行分析
result = self.analyzer.analyze(block)
# 写入缓存
with open(cache_path, "w") as f:
f.write(result.to_json())
return result
def _compute_cache_key(self, block: ContentBlock) -> str:
content_hash = hashlib.md5(
block.content.encode()
).hexdigest()[:12]
return f"{block.type}_{content_hash}"
# 批处理：合并多次 LLM 调用为一次
class BatchedTableAnalyzer:
def batch_process(self, tables: List[ContentBlock]) -> List[TableAnalysis]:
# 把多个小表格的分析合并到一次 LLM 调用中
combined_prompt = self._build_batch_prompt(tables)
batch_result = self.llm.generate(prompt=combined_prompt)
return self._parse_batch_result(batch_result, tables)
def _build_batch_prompt(self, tables):
prompt = "Analyze the following tables:\n\n"
for i, table in enumerate(tables):
prompt += f"### Table {i+1}\n{table.html}\n\n"
prompt += (
"For each table, provide:\n"
"1. Summary (one sentence)\n"
"2. Key data points\n"
"3. Notable trends\n"
)
return prompt
8.4 检索精度调优
多模态检索的精度取决于两个因素：实体提取的质量和图谱检索的策略。
# 策略 1：混合检索（向量 + 图遍历）
class HybridRetriever:
def retrieve(self, query: str, kg: KnowledgeGraph):
# 向量检索找候选实体
vector_candidates = self.vector_search(query, top_k=20)
# 图遍历扩展关联实体
expanded = set()
for entity_id in vector_candidates:
# 1-hop 扩展
neighbors = kg.get_neighbors(entity_id, max_hops=2)
expanded.update(neighbors)
# 重排序
all_candidates = vector_candidates + list(expanded)
reranked = self.rerank(query, all_candidates, kg)
return reranked[:10]
# 策略 2：查询改写
# 把用户的自然语言查询改写为图查询友好的形式
def rewrite_query(query: str, llm) -> str:
rewritten = llm.generate(
prompt=(
f"Rewrite this query to be more specific for "
f"knowledge graph retrieval:\n"
f"Original: {query}\n"
f"Focus on extracting: entities, relations, "
f"and attributes to search for."
)
)
return rewritten
九、与竞品对比：RAG-Anything vs. 其他多模态 RAG 方案
| 维度 | RAG-Anything | ColPali/ColQwen | LlamaIndex Multi-Modal | Unstructured.io |
|---|---|---|---|---|
| 核心思路 | 知识图谱统一表示 | 视觉 Token 级检索 | 编排框架 + 插件 | 文档预处理平台 |
| 文档解析 | MinerU/Docling | PDF→Image→ViT | 多种解析器可选 | 自研分区模型 |
| 图表理解 | VLM 深度分析 | 视觉 Embedding | 依赖外部模型 | 结构化提取 |
| 表格处理 | 双重表示 | 视觉 Token | Markdown 转换 | HTML/CSV 导出 |
| 跨模态关联 | 知识图谱边 | 共享向量空间 | 元数据关联 | 元数据关联 |
| 检索方式 | 图遍历 + 向量 | 视觉相似度 | 向量 + 关键词 | 向量 + 关键词 |
| 多跳推理 | ✅ 原生支持 | ❌ | ❌ | ❌ |
| 可解释性 | ✅ 检索路径可追溯 | ❌ 黑盒 | 部分 | 部分 |
| 部署复杂度 | 中 | 低 | 中 | 高 |
| 适用场景 | 学术/技术文档 | 快速文档 QA | 通用 RAG 应用 | 企业文档 ETL |
核心差异：RAG-Anything 是目前唯一一个在知识图谱层面统一所有模态的 RAG 框架。其他方案要么把多模态当作"文本的补充"（LlamaIndex），要么把多模态当作"视觉检索"（ColPali），只有 RAG-Anything 把文字、图片、表格、公式放在同一个语义空间里做推理。
十、踩坑实录与最佳实践
10.1 文档解析的常见坑
坑 1：扫描版 PDF 的 OCR 质量
扫描版 PDF（尤其是老论文）的 OCR 质量是个大问题。MinerU 内置了 PaddleOCR，但识别率在低分辨率扫描件上不够理想。解决方案：
# 提升 OCR 质量的配置
ocr_config = {
"dpi": 300, # 提高渲染 DPI
"use_gpu": True, # GPU 加速
"det_model": "ch_PP-OCRv4_det", # 最新检测模型
"rec_model": "ch_PP-OCRv4_rec", # 最新识别模型
"enable_mkldnn": True, # MKL-DNN 加速
}
坑 2：超大表格的切割
有些财务报表一张表跨了 2-3 页，MinerU 会把它们识别为多个独立的表格。需要做表格合并：
def merge_cross_page_tables(table_blocks: List[ContentBlock]) -> List[ContentBlock]:
"""合并跨页的连续表格"""
merged = []
i = 0
while i < len(table_blocks):
current = table_blocks[i]
# 检查下一页是否有同名的续表
j = i + 1
while j < len(table_blocks):
next_table = table_blocks[j]
if (next_table.page == current.page + 1 and
self._is_continuation(current, next_table)):
# 合并表格内容
current = self._merge_two_tables(current, next_table)
j += 1
else:
break
merged.append(current)
i = j
return merged
def _is_continuation(self, t1: ContentBlock, t2: ContentBlock) -> bool:
"""判断两个表格是否是续表"""
# 启发式规则：
# 1. 列数相同
# 2. t2 的列头是 t1 列头的子集或相同
# 3. t2 可能有"续表"标记
if t1.column_count != t2.column_count:
return False
if "续" in t2.caption or "continued" in t2.caption.lower():
return True
return set(t2.column_headers).issubset(set(t1.column_headers))
10.2 知识图谱质量控制
问题：幻觉实体
LLM 在做实体-关系提取时会产生幻觉实体——不存在的概念、错误的关系。控制手段：
class EntityValidator:
"""实体验证器 - 过滤幻觉实体"""
def __init__(self, min_confidence: float = 0.7):
self.min_confidence = min_confidence
def validate(self, entities: List[Entity], source_text: str) -> List[Entity]:
valid = []
for entity in entities:
# 规则 1：实体名必须在原文中出现
if entity.name not in source_text:
continue
# 规则 2：置信度阈值
if entity.confidence < self.min_confidence:
continue
# 规则 3：实体名不能是停用词
if entity.name.lower() in self.stop_words:
continue
# 规则 4：实体名长度合理
if len(entity.name) < 2 or len(entity.name) > 50:
continue
valid.append(entity)
return valid
10.3 生产部署建议
推荐配置（中等规模，1000 篇文档级别）：
计算资源：
- 文档解析：1x GPU (T4/A10) + 4 CPU cores + 16GB RAM
- LLM 推理：使用 API（GPT-4o / DeepSeek）降低本地算力需求
- Neo4j：4 CPU cores + 8GB RAM + SSD
存储：
- 原始文档：约 1GB / 1000 篇论文
- 解析缓存：约 2-3GB / 1000 篇
- 知识图谱（Neo4j）：约 500MB / 1000 篇
- 向量索引：约 200MB / 1000 篇
网络：
- LLM API 调用：约 100-500 次 / 篇文档（首次导入）
- 后续增量导入：仅新增部分调用
时间：
- 首次导入：约 30-60s / 篇（含 LLM 调用）
- 增量导入：约 10-20s / 篇
- 查询响应：1-3s / 次
十一、未来展望
RAG-Anything 代表了 RAG 技术的一个重要方向：从文本检索到全模态理解。但它还不是终点，几个明显的演进方向：
1. 实时多模态 RAG
当前的 RAG-Anything 是离线构建知识图谱、在线查询。未来的方向是实时流式处理——新文档入库后，增量更新图谱，不需要全量重建。
2. Agent 驱动的主动检索
现在的 RAG 是被动响应查询。结合 AI Agent，可以实现主动检索：Agent 在执行任务时自动判断需要什么信息，主动从知识图谱中获取，而不是等用户来问。
3. 多知识图谱联邦
一个组织的知识可能分散在多个知识图谱中（不同部门、不同系统）。联邦检索——跨图谱的实体对齐和分布式查询——是一个重要的工程问题。
4. 多模态生成
当前 RAG-Anything 的生成阶段还是文本输出。未来可以基于检索到的多模态上下文，生成多模态的回答：用图表回答数据问题，用架构图回答设计问题。
总结
RAG-Anything 解决了一个真实且迫切的问题：让 RAG 系统不再对图表公式视而不见。 它的核心创新不是某一个技术点，而是把文档解析、模态路由、深度分析和知识图谱构建串成了一条完整的管线，并且在每个环节都做了工程上的优化。
如果你正在构建一个需要处理论文、技术文档、报告的知识库系统，RAG-Anything 是目前最值得投入的开源方案。它不完美——LLM 调用成本高、图谱质量控制难、对扫描文档的 OCR 依赖强——但它在"全模态统一理解"这条路上走出了最扎实的一步。
技术选型的建议：先用 JSON 存储快速验证概念，确认效果好后再切 Neo4j。先跑 10-20 篇文档看图谱质量，调好实体提取和关系提取的参数后，再扩大规模。别一上来就导入几千篇文档——图谱质量的调试成本比你想的高得多。
本文基于 RAG-Anything GitHub 仓库（HKUDS/RAG-Anything）及 LightRAG 论文（EMNLP 2025）撰写。项目持续迭代中，部分 API 可能有变化，请以最新文档为准。