1. RAGFlow PDF解析器 #

1.1 什么是RAGFlowPdfParser #

RAGFlowPdfParser是RAGFlow提供的PDF解析器基础类，它封装了完整的PDF解析流程：

1. OCR识别：识别PDF中的文字
2. 布局识别：识别文档的布局结构
3. 表格识别：识别表格的结构和内容
4. 文本合并：将OCR识别的文字按布局合并成段落
5. 内容提取：提取表格和图片信息

1.2 简单示例 #

下面是一个完整的示例，演示如何使用RAGFlowPdfParser解析PDF：

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入RAGFlow的PDF解析器
from deepdoc.parser.pdf_parser import RAGFlowPdfParser

# 定义PDF文件路径（请替换为你的PDF文件路径）
pdf_file = './data/zhidu_employee.pdf'

# 创建PDF解析器实例
# 初始化时会加载OCR、布局识别、表格识别等模型（首次运行可能需要一些时间）
print("正在初始化PDF解析器...")
parser = RAGFlowPdfParser()
print("PDF解析器初始化完成！")

# 解析PDF文件
# need_image=True: 提取图片
# zoomin=3: PDF转图片的放大倍数，越大越清晰但处理越慢
# return_html=False: 表格不转换为HTML格式
print(f"正在解析PDF文件: {pdf_file}")
text_content, tables_and_images = parser(
    pdf_file, 
    need_image=True, 
    zoomin=3, 
    return_html=False
)

# 处理文字内容
# text_content是一个字符串，用\n\n分隔不同页面，用\n分隔不同段落
# 每个段落包含文字内容和位置信息，格式为：文字@@页码\tx0\tx1\ty0\ty1##
print("\n========== 文字内容 ==========")
pages = text_content.split('\n\n')  # 按页面分割
for page_num, page in enumerate(pages, 1):
    print(f"\n 第 {page_num} 页 ")
    paragraphs = page.split('\n')  # 按段落分割
    for para in paragraphs:
        # 移除位置信息标签，只显示文字
        # 位置信息格式：文字@@页码\tx0\tx1\ty0\ty1##
        clean_text = parser.remove_tag(para)
        if clean_text.strip():  # 只显示非空段落
            print(clean_text)

# 处理表格和图片
# tables_and_images是一个列表，每个元素是(图片对象, 文字描述)的元组
print("\n========== 表格和图片 ==========")
for idx, (img, txt) in enumerate(tables_and_images, 1):
    print(f"\n 表格/图片 {idx} ")
    print(f"描述: {txt}")

    # 如果有图片对象，可以保存
    if img:
        output_path = f"./output/table_image_{idx}.jpg"
        os.makedirs("./output", exist_ok=True)
        img.save(output_path)
        print(f"图片已保存到: {output_path}")

print("\n解析完成！")

代码说明：

1. 导入模块：导入RAGFlowPdfParser类
2. 初始化解析器：创建解析器实例，会自动加载所需模型
3. 解析PDF：调用解析器，传入PDF文件路径和参数
4. 处理文字：文字内容用特殊格式存储，包含位置信息
5. 处理表格图片：表格和图片作为独立元素返回

返回结果格式说明：

• 文字信息：用\n\n分隔页面，用\n分隔段落

  • 格式：文字内容@@页码\tx0\tx1\ty0\ty1##
  • 例如：教职工考勤管理制度@@1\t242.0\t387.0\t80.7\t95.7##
  • 表示：文字为"教职工考勤管理制度"，在第1页，位置坐标为(x0=242.0, x1=387.0, y0=80.7, y1=95.7)

• 表格和图片信息：列表格式，每个元素是(图片对象, 文字描述)的元组

2.3 解析流程 #

RAGFlowPdfParser的解析过程包含以下步骤：

PDF文件
  ↓
__images__: OCR识别（将PDF转为图片，识别文字）
  ↓
_layouts_rec: 布局识别（识别标题、正文、表格等区域）
  ↓
_table_transformer_job: 表格识别（识别表格结构）
  ↓
_text_merge: 文本合并（将OCR识别的文字按布局合并）
  ↓
_concat_downward: 向下合并（合并垂直方向的文本块）
  ↓
_filter_forpages: 页面过滤（过滤目录、致谢等页面）
  ↓
_extract_table_figure: 提取表格和图片
  ↓
__filterout_scraps: 过滤碎片（过滤掉过短的文本）
  ↓
返回结果

重要说明：

• _filter_forpages和__filterout_scraps这两个过滤步骤是可选的
• 如果文档必须包含目录或致谢，可以去掉_filter_forpages
• 过滤规则可以根据实际需求调整

2. RAGFlow应用解析器 #

2.1 什么是应用解析器 #

RAGFlow在基础解析器之上，提供了针对不同场景的应用解析器（app），它们位于rag/app目录下：

• naive.py：通用PDF解析器
• one.py：单页文档解析器（整个文档作为一个chunk）
• laws.py：法律文档解析器（使用法律专用布局模型）
• paper.py：论文解析器（处理双栏布局）
• manual.py：手册解析器
• book.py：书籍解析器

这些应用解析器继承自RAGFlowPdfParser，重写了__call__方法，并提供了chunk方法用于文档分块。

2.2 通用解析器：naive.py #

2.2.1 什么是naive解析器 #

naive解析器是RAGFlow提供的通用PDF解析器，适用于大多数常见的PDF文档。

特点：

• 适用于单栏布局的PDF文档
• 文本和表格分开返回
• 支持自定义分块大小和分隔符

局限性：

• 多栏布局可能出现顺序混乱
• 表格和文本分开，可能缺失上下文

2.2.2 使用naive解析器 #

下面是一个完整的使用示例：

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入naive解析器
from rag.app.naive import Pdf, chunk

# 定义一个空的回调函数（用于显示进度，这里简化处理）
def dummy(prog=None, msg=""):
    # prog: 进度值（0-1）
    # msg: 进度消息
    if msg:
        print(f"进度: {msg}")

# 定义PDF文件路径
pdf_file = './data/zhidu_employee.pdf'

# 配置解析参数
parser_config = {
    "chunk_token_num": 256,      # 每个chunk的token数量（大约对应256个中文字符）
    "delimiter": "\n!?。；！？",  # 分块的分隔符（在这些符号处可以分割）
    "layout_recognize": True      # 是否进行布局识别
}

# 使用chunk方法解析并分块
# chunk方法会：
# 1. 调用Pdf类解析PDF
# 2. 按照配置的参数进行分块
# 3. 返回LangChain Document格式的结果
print(f"正在解析PDF文件: {pdf_file}")
naive_chunks = chunk(
    pdf_file, 
    callback=dummy, 
    parser_config=parser_config
)

# 处理分块结果
# 每个chunk是一个字典，包含：
# - content_with_weight: 内容文本
# - metadata: 元数据（来源、位置等）
print(f"\n共生成 {len(naive_chunks)} 个chunk")
print("\n========== 分块内容 ==========")
for idx, data in enumerate(naive_chunks, 1):
    print(f"\n Chunk {idx} ")
    print(data['content_with_weight'])
    print(f"元数据: {data.get('metadata', {})}")

# 可以将chunks转换为LangChain Document并保存到向量数据库
from langchain_core.documents import Document

documents = []
for chunk_data in naive_chunks:
    doc = Document(
        page_content=chunk_data['content_with_weight'],
        metadata=chunk_data.get('metadata', {})
    )
    documents.append(doc)

print(f"\n已转换为 {len(documents)} 个LangChain Document")
print("可以保存到向量数据库用于RAG检索")

代码说明：

1. 导入模块：导入naive解析器的Pdf类和chunk函数
2. 定义回调函数：用于显示解析进度（可选）
3. 配置参数：
   • chunk_token_num：每个chunk的大小
   • delimiter：分块分隔符
4. 调用chunk方法：解析PDF并自动分块
5. 处理结果：每个chunk包含内容和元数据

参数说明：

• chunk_token_num：控制每个chunk的大小，值越大chunk越大
• delimiter：分块时优先在这些符号处分割，保持语义完整性

2.3 单页解析器：one.py #

2.3.1 什么是one解析器 #

one解析器将整个PDF文档作为一个chunk处理，适用于：

• 单栏布局的文档
• 内容较少的文档
• 需要保持文档完整性的场景

特点：

• 整个文档作为一个chunk
• 表格和图片按文档中的位置顺序输出
• 保持文档的完整性

局限性：

• 多栏布局可能出现顺序混乱
• 不适合内容很长的文档（chunk太大）

2.3.2 使用one解析器 #

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入one解析器
from rag.app.one import Pdf, chunk

# 定义回调函数
def dummy(prog=None, msg=""):
    if msg:
        print(f"进度: {msg}")

# 定义PDF文件路径
pdf_file = './files/红楼梦.pdf'

# 使用chunk方法解析
# one解析器不需要额外的配置参数
print(f"正在解析PDF文件: {pdf_file}")
chunks = chunk(pdf_file, callback=dummy)

# 处理结果
# one解析器通常只返回一个chunk（整个文档）
print(f"\n共生成 {len(chunks)} 个chunk")
for idx, data in enumerate(chunks, 1):
    print(f"\n Chunk {idx} ")
    print(data['content_with_weight'])

2.4 法律文档解析器：laws.py #

2.4.1 什么是laws解析器 #

laws解析器专门用于解析法律文档，特点：

• 使用法律专用布局识别模型（layout.laws.onnx）
• 识别文档的层级关系（编、章、节、条等）
• 在分块时保留层级信息

适用场景：

• 法律法规文档
• 合同文档
• 有明确层级结构的文档

2.4.2 使用laws解析器 #

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入laws解析器
from rag.app.laws import Pdf, chunk

# 定义回调函数
def dummy(prog=None, msg=""):
    if msg:
        print(f"进度: {msg}")

# 定义PDF文件路径（法律文档）
pdf_file = './data/laws.pdf'

# 使用chunk方法解析
print(f"正在解析法律文档: {pdf_file}")
chunks = chunk(pdf_file, callback=dummy)

# 处理结果
# laws解析器会在chunk中保留层级信息
print(f"\n共生成 {len(chunks)} 个chunk")
for idx, data in enumerate(chunks, 1):
    print(f"\n Chunk {idx} ")
    print(data['content_with_weight'])
    print(f"元数据: {data.get('metadata', {})}")

层级识别说明：

laws解析器会识别以下层级模式：

• 第X编/部分
• 第X章
• 第X节
• 第X条
• 数字编号（1.1, 1.2.3等）

在分块时，会为每个chunk添加上一级层级的说明，保持层级关系。

2.5 论文解析器：paper.py #

2.5.1 什么是paper解析器 #

paper解析器专门用于解析学术论文，特点：

• 处理双栏布局（论文常见格式）
• 按正确顺序输出：先左栏，再右栏
• 使用论文专用布局模型（layout.paper.onnx）

适用场景：

• 学术论文PDF
• 期刊文章
• 双栏布局的文档

局限性：

• 对于单栏和双栏混合的论文，可能输出混乱
• 某些特殊字体可能导致OCR识别乱码

2.5.2 使用paper解析器 #

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入paper解析器
from rag.app.paper import Pdf, chunk

# 定义回调函数
def dummy(prog=None, msg=""):
    if msg:
        print(f"进度: {msg}")

# 定义PDF文件路径（论文）
pdf_file = './data/paper3.pdf'

# 使用chunk方法解析
print(f"正在解析论文: {pdf_file}")
chunks = chunk(pdf_file, callback=dummy)

# 处理结果
print(f"\n共生成 {len(chunks)} 个chunk")
for idx, data in enumerate(chunks, 1):
    print(f"\n Chunk {idx} ")
    print(data['content_with_weight'])

2.6 应用解析器总结 #

解析器	适用场景	特点	局限性
naive	通用单栏PDF	灵活配置，支持自定义分块	多栏布局顺序混乱
one	短文档	保持文档完整性	不适合长文档
laws	法律文档	识别层级关系	不处理表格
paper	学术论文	处理双栏布局	混合布局可能混乱

选择建议：

• 普通文档：使用naive
• 短文档：使用one
• 法律文档：使用laws
• 学术论文：使用paper

3. 自定义PDF解析器 #

3.1 为什么需要自定义解析器 #

虽然RAGFlow提供了多种应用解析器，但在实际应用中，你可能需要：

1. 保持顺序性：确保文本、表格、图片按文档中的顺序输出
2. 添加上下文：为表格和图片添加前后文本作为上下文
3. 图片查询：将图片转换为文本描述，支持通过自然语言查询图片
4. 表格总结：对表格进行总结，支持分析型问题

3.2 自定义解析器的实现思路 #

我们的自定义解析器将实现以下功能：

1. 位置排序：根据文本、表格、图片在文档中的位置进行排序
2. 表格上下文：为表格添加前一段文本作为上下文
3. 图片描述：使用视觉大模型将图片转换为文本描述
4. 图片上下文：为图片添加前后文本作为上下文
5. 表格总结：使用LLM对表格进行总结

3.3 完整实现示例 #

下面是一个完整的自定义PDF解析器实现：

# 导入必要的库
import os
import sys
import re
import uuid

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入RAGFlow的基础解析器
from deepdoc.parser import PdfParser

# 导入LangChain相关
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 导入工具函数（需要自己实现或从utils导入）
from model import RagLLM
from utils import table_to_summary, image_to_txt

# 导入RAGFlow的NLP工具
from rag.nlp import bullets_category, title_frequency


class MyPdf(PdfParser):
    """
    自定义PDF解析器

    功能：
    1. 按位置排序文本、表格、图片
    2. 为表格和图片添加上下文
    3. 支持图片描述和表格总结
    """

    def __init__(self):
        # 设置使用手册布局模型（可以根据需要修改）
        self.model_speciess = "manual"
        # 调用父类初始化
        super().__init__()

    def __call__(self, filename, binary=None, from_page=0,
                 to_page=100000, zoomin=3, callback=None):
        """
        解析PDF文件

        参数:
            filename: PDF文件路径
            binary: PDF二进制数据（可选）
            from_page: 起始页码（从0开始）
            to_page: 结束页码
            zoomin: PDF转图片的放大倍数
            callback: 进度回调函数

        返回:
            sorted_sections: 排序后的内容列表
        """

        # 步骤1: OCR识别
        callback(msg="OCR is running...")
        self.__images__(
            filename if not binary else binary,
            zoomin,
            from_page,
            to_page,
            callback
        )
        callback(msg="OCR finished")

        # 步骤2: 布局识别
        from timeit import default_timer as timer
        start = timer()
        self._layouts_rec(zoomin, drop=True)  # drop=True表示过滤页头页脚
        callback(0.63, "Layout analysis finished.")
        print("layouts:", timer() - start)

        # 步骤3: 表格识别
        self._table_transformer_job(zoomin)
        callback(0.65, "Table analysis finished.")

        # 步骤4: 文本合并
        self._text_merge()
        callback(0.67, "Text merging finished")

        # 步骤5: 提取表格和图片
        # 参数说明：
        # - True: 需要图片
        # - zoomin: 放大倍数
        # - True: 返回HTML格式
        # - True: 返回位置信息
        tbls = self._extract_table_figure(True, zoomin, True, True, return_html_and_text=True)

        # 步骤6: 向下合并文本
        self._concat_downward()

        # 步骤7: 清理文本中的多余空白
        for b in self.boxes:
            # 移除连续的空格、制表符、全角空格等，替换为单个空格
            b["text"] = re.sub(r"([\t 　]|\u3000){2,}", " ", b["text"].strip())

        # 步骤8: 构建sections列表
        # 每个section包含：(文本, 布局编号, 位置信息)
        sections = [(b["text"], b.get("layout_no", ""), self.get_position(b, zoomin))
                    for i, b in enumerate(self.boxes)]

        # 步骤9: 分析文档结构（用于更好的排序）
        # 确定文档中的项目符号类型
        bull = bullets_category([txt for txt, _, _ in sections])

        # 分析标题级别
        # most_level: 最常见的标题级别
        # levels: 每个section的级别
        most_level, levels = title_frequency(
            bull, [(txt, l) for txt, l, poss in sections])

        # 确保每个section都有对应的级别
        assert len(sections) == len(levels)

        # 步骤10: 为每个section分配章节ID
        sec_ids = []  # 存储章节ID
        sid = 0       # 当前章节ID

        for i, lvl in enumerate(levels):
            # 当遇到新的标题级别时，增加章节ID
            if lvl <= most_level and i > 0 and lvl != levels[i - 1]:
                sid += 1
            sec_ids.append(sid)

        # 步骤11: 重新构建sections，添加章节ID
        # 格式：(文本, 章节ID, 位置信息, None)
        sections = [(txt, sec_ids[i], poss, None) 
                   for i, (txt, _, poss) in enumerate(sections)]

        # 步骤12: 添加表格和图片到sections
        for (img, rows), poss in tbls:
            if not rows:
                continue
            # 表格格式：(表格内容, -1(表示表格), 位置信息, 图片对象)
            sections.append((rows, -1,
                            [(p[0] + 1 - from_page, p[1], p[2], p[3], p[4]) 
                             for p in poss], img))

        # 步骤13: 根据位置排序
        # 排序规则：
        # 1. 首先按页码排序
        # 2. 然后按垂直位置（从上到下）
        # 3. 最后按水平位置（从左到右）
        sorted_sections = sorted(sections, key=lambda x: (
            x[-2][0][0],  # 页码
            x[-2][0][3],  # 垂直位置（y坐标）
            x[-2][0][1]   # 水平位置（x坐标）
        ))

        return sorted_sections


def pre_chunk(filename, binary=None, from_page=0,
              to_page=100000, zoomin=3, callback=None):
    """
    预处理：提取文本、表格、图片，并添加上下文

    返回:
        txt_chunks: 文本chunks列表
        table_chunks: 表格chunks列表（已添加上下文）
        image_chunks: 图片chunks列表（已添加上下文）
    """

    # 创建解析器并解析PDF
    parser = MyPdf()
    sections = parser(filename, binary, from_page, to_page, zoomin, callback)

    # 步骤1: 识别每个section的类型
    section_types = []
    for section in sections:
        txt, _, _, img = section

        # 处理表格内容（可能是字典或列表格式）
        if isinstance(txt, dict):
            txt = txt.get('html', '')
        elif isinstance(txt, list):
            txt = txt[0] if txt else ''

        # 判断类型
        content_type = 'text'
        if img is not None:
            if 'table' in str(txt).lower():
                content_type = 'table'
            else:
                content_type = 'image'
        section_types.append(content_type)

    # 步骤2: 分别处理文本、表格、图片
    txt_chunks = []
    table_chunks = []
    image_chunks = []

    for idx, section in enumerate(sections):
        txt, _, _, img = section
        content_type = section_types[idx]

        # 处理表格
        if content_type == 'table':
            # 向前查找最近的文本作为上下文
            pre_text_idx = idx - 1
            while pre_text_idx >= 0 and section_types[pre_text_idx] != 'text':
                pre_text_idx -= 1

            # 获取前一段文本
            if pre_text_idx >= 0:
                pre_text = sections[pre_text_idx][0]
                # 合并表格和上下文
                combined_text = f"{pre_text} \n {txt}"
                table_chunks.append(combined_text)

        # 处理图片
        elif content_type == 'image':
            # 向前查找最近的文本
            pre_text_idx = idx - 1
            while pre_text_idx >= 0 and section_types[pre_text_idx] != 'text':
                pre_text_idx -= 1

            # 向后查找最近的文本
            post_text_idx = idx + 1
            while post_text_idx < len(sections) and section_types[post_text_idx] != 'text':
                post_text_idx += 1

            # 获取前后文本
            pre_text = sections[pre_text_idx][0] if pre_text_idx >= 0 else ""
            post_text = sections[post_text_idx][0] if post_text_idx < len(sections) else ""

            # 保存图片信息（文本描述、图片对象、上下文）
            image_chunks.append((pre_text, txt, img, post_text))

        # 处理文本
        else:  # content_type == 'text'
            # 如果文本很短且已经作为表格或图片的上下文，则跳过
            if len(str(txt)) < 20:
                # 检查是否已经作为表格的上下文
                is_table_context = any(str(txt) in str(table_txt) for table_txt in table_chunks)
                # 检查是否已经作为图片的上下文
                is_image_context = any(
                    str(txt) in [str(post_text), str(pre_text)] 
                    for pre_text, _, _, post_text in image_chunks
                )

                if is_table_context or is_image_context:
                    continue

            txt_chunks.append(str(txt))

    return txt_chunks, table_chunks, image_chunks


def chunk(filename, binary=None, from_page=0,
          to_page=100000, zoomin=3, callback=None, llm=None,
          chunk_size=512, chunk_overlap=30):
    """
    最终分块函数：生成LangChain Document列表

    参数:
        filename: PDF文件路径
        binary: PDF二进制数据
        from_page: 起始页码
        to_page: 结束页码
        zoomin: 放大倍数
        callback: 进度回调
        llm: 语言模型实例（用于表格总结和图片描述）
        chunk_size: 文本chunk大小
        chunk_overlap: chunk重叠大小

    返回:
        docs: LangChain Document列表
    """

    # 步骤1: 预处理，获取文本、表格、图片
    txt_chunks, table_chunks, image_chunks = pre_chunk(
        filename, binary, from_page, to_page, zoomin, callback
    )

    # 如果没有提供LLM，创建一个
    if llm is None:
        llm = RagLLM()

    docs = []

    # 步骤2: 处理表格（总结）
    print("正在处理表格...")
    for table_chunk in table_chunks:
        # 使用LLM对表格进行总结
        table_summary = table_to_summary(table_chunk, llm)

        # 创建Document，metadata中保存原始表格内容
        doc = Document(
            page_content=table_summary,  # 总结作为主要内容
            metadata={
                "source": "table",
                "content": table_chunk  # 原始表格内容保存在metadata中
            }
        )
        docs.append(doc)

    # 步骤3: 处理图片（描述）
    print("正在处理图片...")
    img_path = "./doc_images"
    os.makedirs(img_path, exist_ok=True)

    for image_chunk in image_chunks:
        pre_text, txt, img, post_text = image_chunk

        # 构建图片描述提示词
        prompt = f"""
请根据图片的相关上下文和信息，用中文详细描述一下图中的内容，比如时间，地点，人物，事情，功能等。
如果有数据请提取出数据，并做分析，最后对内容进行总结陈述，不要超过512个字符。

上下文:
{pre_text}

{post_text}

图表信息：{txt}
"""

        # 使用视觉大模型描述图片
        image_desc = image_to_txt(img, prompt, 512)

        # 保存图片到本地
        img_file_name = f"{img_path}/image_{uuid.uuid4()}.png"
        img.save(img_file_name)

        # 创建Document，metadata中保存图片路径
        doc = Document(
            page_content=image_desc[0] if isinstance(image_desc, list) else image_desc,
            metadata={
                "source": "image",
                "content": img_file_name  # 图片路径保存在metadata中
            }
        )
        docs.append(doc)

    # 步骤4: 处理文本（分块）
    print("正在处理文本...")
    all_texts = "\n".join(txt_chunks)

    # 使用递归字符分割器进行分块
    r_spliter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=[
            "\n\n",      # 双换行（段落分隔）
            "\n",        # 单换行
            ".",         # 英文句号
            "\uff0e",    # 全角句号
            "\u3002",    # 中文句号
            ",",         # 英文逗号
            "\uff0c",    # 全角逗号
            "\u3001",    # 中文顿号
        ]
    )

    # 分割文本
    chunks = r_spliter.split_text(all_texts)

    # 创建Document
    for chunk_text in chunks:
        doc = Document(
            page_content=chunk_text,
            metadata={
                "source": "text",
                "content": ""
            }
        )
        docs.append(doc)

    print(f"处理完成！共生成 {len(docs)} 个Document")
    return docs


# 使用示例
if __name__ == "__main__":
    # 定义回调函数
    def dummy(prog=None, msg=""):
        if msg:
            print(f"进度: {msg}")

    # 定义PDF文件路径
    pdf_file = './files/红楼梦.pdf'

    # 解析并分块
    documents = chunk(
        pdf_file,
        callback=dummy,
        chunk_size=512,
        chunk_overlap=30
    )

    # 查看结果
    print(f"\n共生成 {len(documents)} 个Document")
    for idx, doc in enumerate(documents[:5], 1):  # 只显示前5个
        print(f"\n Document {idx} ")
        print(f"类型: {doc.metadata['source']}")
        print(f"内容: {doc.page_content[:200]}...")  # 只显示前200个字符

3.4 视觉大模型的使用 #

3.4.1 什么是视觉大模型 #

视觉大模型（Vision Language Model, VLM）可以理解图片内容并用自然语言描述。我们使用MiniCPM-V 2.6模型。

MiniCPM-V 2.6特点：

• 仅8B参数，性能接近GPT-4V
• 支持图片理解
• 可以部署在本地（需要约6GB显存）

3.4.2 安装和使用Ollama #

# 安装Ollama（如果还没有）
# Windows: 下载安装包从 https://ollama.ai
# Linux/Mac: curl -fsSL https://ollama.ai/install.sh | sh

# 拉取MiniCPM-V模型
ollama pull minicpm-v

3.4.3 使用视觉大模型描述图片 #

# 导入必要的库
from openai import OpenAI
import base64
import os

def image_to_base64(image_path):
    """
    将图片转换为base64编码

    参数:
        image_path: 图片文件路径

    返回:
        dict: 包含base64编码的字典
    """
    # 确定图片的MIME类型
    ext = os.path.splitext(image_path)[-1].lower()
    if ext == '.png':
        mime_type = 'image/png'
    elif ext in ['.jpg', '.jpeg']:
        mime_type = 'image/jpeg'
    elif ext == '.gif':
        mime_type = 'image/gif'
    else:
        raise ValueError(f"不支持的图片格式: {ext}")

    # 读取图片文件并转换为base64
    with open(image_path, "rb") as image_file:
        encoded_string = base64.b64encode(image_file.read()).decode('utf-8')

    # 构造base64字符串
    base64_image = f"data:{mime_type};base64,{encoded_string}"

    return {"image_url": base64_image}


def describe_image(image_path, prompt="请详细描述这张图片的内容"):
    """
    使用视觉大模型描述图片

    参数:
        image_path: 图片路径
        prompt: 描述提示词

    返回:
        str: 图片描述文本
    """
    # 创建OpenAI客户端（连接到本地Ollama）
    client = OpenAI(
        base_url='http://localhost:11434/v1/',  # Ollama的API地址
        api_key='ollama',  # Ollama不需要真实的API key
    )

    # 将图片转换为base64
    result = image_to_base64(image_path)

    # 调用API描述图片
    response = client.chat.completions.create(
        model="minicpm-v",  # 使用的模型名称
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": result['image_url'],
                    },
                ],
            }
        ],
        max_tokens=3000,  # 最大生成token数
    )

    # 返回描述文本
    return response.choices[0].message.content


# 使用示例
if __name__ == "__main__":
    # 描述图片
    image_path = './test_image.jpg'
    description = describe_image(
        image_path, 
        "请详细描述这张图片，包括其中的文字、图表、数据等信息"
    )
    print("图片描述：")
    print(description)

3.5 表格总结功能 #

3.5.1 为什么需要表格总结 #

表格数据在RAG系统中面临两个挑战：

1. 分析型问题：用户问"这个表格说明了什么？"需要总结
2. 细节型问题：用户问"2022年的收入是多少？"需要精确检索

表格总结可以：

• 提供表格的概括性描述，支持分析型问题
• 保留原始表格内容，支持细节型问题

3.5.2 表格总结实现 #

# 导入必要的库
from model import RagLLM

def table_to_summary(tables_text, llm):
    """
    对表格文本信息进行总结

    参数:
        tables_text: 表格文本信息，可能包含表格的标题和内容
        llm: 语言模型实例

    返回:
        str: 表格内容的总结
    """
    # 构建提示词
    prompt = f"""
请对以下表格信息进行全面而简洁的总结。提取表格中的关键信息，包括主题、重要数据点和潜在的见解。
确保总结涵盖表格的主要内容，并保持客观准确。

表格信息:
{tables_text}

请提供总结:
"""

    # 调用语言模型生成总结
    summary = llm(prompt, temperature=0.3, max_tokens=512)

    return summary


# 使用示例
if __name__ == "__main__":
    # 创建LLM实例
    llm = RagLLM()

    # 表格文本（包含上下文）
    table_text = """
    营业收入统计表

    项目               | 2022年        | 2023年
    ||
    营业收入   |  200万                      |   300万
    营业成本   |  150万                      |   220万
    净利润     |  50万                       |   80万
    """

    # 生成总结
    summary = table_to_summary(table_text, llm)
    print("表格总结：")
    print(summary)

4. 完整使用示例 #

下面是一个完整的示例，演示如何使用自定义解析器：

# 导入必要的库
import os
import sys

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.abspath('.'))

# 导入自定义解析器（假设已经保存为my_pdf_parser.py）
from my_pdf_parser import chunk
from model import RagLLM

# 定义回调函数
def progress_callback(prog=None, msg=""):
    """进度回调函数"""
    if msg:
        print(f"[进度] {msg}")
    elif prog is not None:
        print(f"[进度] {prog*100:.1f}%")

# 定义PDF文件路径
pdf_file = './files/红楼梦.pdf'

# 创建LLM实例（用于表格总结和图片描述）
llm = RagLLM()

# 解析PDF并生成Document列表
print("开始解析PDF...")
documents = chunk(
    pdf_file,
    callback=progress_callback,
    llm=llm,
    chunk_size=512,
    chunk_overlap=30
)

# 查看结果统计
text_count = sum(1 for doc in documents if doc.metadata['source'] == 'text')
table_count = sum(1 for doc in documents if doc.metadata['source'] == 'table')
image_count = sum(1 for doc in documents if doc.metadata['source'] == 'image')

print(f"\n解析完成！")
print(f"文本chunks: {text_count} 个")
print(f"表格chunks: {table_count} 个")
print(f"图片chunks: {image_count} 个")
print(f"总计: {len(documents)} 个Document")

# 查看示例结果
print("\n========== 示例结果 ==========")
for idx, doc in enumerate(documents[:3], 1):
    print(f"\n Document {idx} ({doc.metadata['source']}) ")
    content_preview = doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content
    print(f"内容: {content_preview}")
    print(f"元数据: {doc.metadata}")

# 可以保存到向量数据库
# from langchain_chroma import Chroma
# from model import RagEmbedding
# 
# embedding_model = RagEmbedding()
# vector_db = Chroma.from_documents(
#     documents,
#     embedding_model.get_embedding_fun(),
#     collection_name="my_pdf_collection"
# )

5. 总结与最佳实践 #

5.1 解析器选择建议 #

场景	推荐解析器	说明
普通单栏PDF	naive	灵活配置，适合大多数场景
短文档	one	保持完整性
法律文档	laws	识别层级关系
学术论文	paper	处理双栏布局
需要顺序和上下文	自定义解析器	完全控制解析流程

5.2 常见问题处理 #

问题1：多栏布局顺序混乱

• 解决方案：使用自定义解析器，根据位置信息排序

问题2：表格缺失上下文

• 解决方案：在自定义解析器中为表格添加前后文本

问题3：图片无法查询

• 解决方案：使用视觉大模型将图片转换为文本描述

问题4：表格理解困难

• 解决方案：对表格进行总结，同时保留原始内容

5.3 性能优化建议 #

1. 调整zoomin参数：值越大精度越高但速度越慢，建议3-5
2. 使用GPU：如果有GPU，使用GPU版本的onnxruntime
3. 批量处理：对于大量文档，考虑并行处理
4. 缓存结果：解析结果可以缓存，避免重复解析