一、Rag理论概述

由近期 RAGFlow 的火爆看 RAG 的现状与未来

二、Ragflow解析参数说明

三、♥ RagFlow源码解析

核心代码流程梳理

参考：深度解读RAGFlow的深度文档理解DeepDoc

DeepDoc的模型应该是基于paddleOCR的模型去微调训练的，开源出来的模型是onnx格式的

1、OCR识别

主要代码在ocr.py里——"E:\ragflow-main\deepdoc\vision\ocr.py"
TextRecognizer 做文字识别，TextDetector 做文本框检测，OCR整合检测和识别功能，对外提供调用

2、版面分析

版面分析主要在recognizer.py和layout_recognizer.py里：
"E:\ragflow-main\deepdoc\vision\recognizer.py"和"E:\ragflow-main\deepdoc\vision\layout_recognizer.py"
LayoutRecognizer 继承Recognizer的类，用于对文档图像进行板式分析，识别不同类型的区域，例如表格、标题、段落等。这里用的模型应该还是基于paddleocr里的版面分析模型去优化的。
Recognizer的__call__ 方法，传入图像列表和置信度阈值

3、parser功能

OCR和版面分析，都是为parser服务的，parser负责解析文档，并拆分为chunk.

框架提供了PdfParser、PlainParser、DocxParser、ExcelParser、PptParser 5种解析器。另外针对resume，提供了专门的简历解析功能

3.1 PdfParser

我们挑选重点的==PdfParser ==也就是HuParser来分析。

3.1.1 首先，初始化

def __init__(self):  
    self.ocr = OCR()  
    if hasattr(self, "model_speciess"):  
        self.layouter = LayoutRecognizer("layout." + self.model_speciess)  
    else:  
        self.layouter = LayoutRecognizer("layout")  
    self.tbl_det = TableStructureRecognizer()  
    self.updown_cnt_mdl = xgb.Booster()

对表格结构的识别需要OCR、LayoutRecognizer，以及TableStructureRecognizer互相配合，一般都是模型搭配大量的工程trick，靠一些规则来解决一些边界情况。文档解析也是这样，需要多个模型配合，结合一些规则来做，这些规则通常是经验的集合，大白话就是各种case跑出来，遇到问题就加新的规则

***************PdfParser核心的__call__

def __call__(self, fnm, need_image=True, zoomin=3, return_html=False):  
    # 转图片，处理文本，ocr识别  
    self.__images__(fnm, zoomin)  
    # 版面分析  
    self._layouts_rec(zoomin)  
    # table box 处理  
    self._table_transformer_job(zoomin)  
    # 合并文本块  
    self._text_merge()  
    self._concat_downward()  
    # 过滤分页信息  
    self._filter_forpages()  
    # 表格和图表抽取  
    tbls = self._extract_table_figure(  
        need_image, zoomin, return_html, False)  
    # 抽取的文本（去掉表格）， 表格  
    return self.__filterout_scraps(deepcopy(self.boxes), zoomin), tbls

3.1.2 pdf转图片

先读PDF文件

def __images__(self, fnm, zoomin=3, page_from=0,  
               page_to=299, callback=None):  
    self.lefted_chars = []  
    self.mean_height = []  
    self.mean_width = []  
    self.boxes = []  
    self.garbages = {}  
    self.page_cum_height = [0]  
    self.page_layout = []  
    self.page_from = page_from  
    try:  
        self.pdf = pdfplumber.open(fnm) if isinstance(  
            fnm, str) else pdfplumber.open(BytesIO(fnm))  
        self.page_images = [p.to_image(resolution=72 * zoomin).annotated for i, p in  
                            enumerate(self.pdf.pages[page_from:page_to])]  
        self.page_chars = [[c for c in page.chars if self._has_color(c)] for page in  
                           self.pdf.pages[page_from:page_to]]  
        self.total_page = len(self.pdf.pages)  
    except Exception as e:  
        self.pdf = fitz.open(fnm) if isinstance(  
            fnm, str) else fitz.open(  
            stream=fnm, filetype="pdf")  
        self.page_images = []  
        self.page_chars = []  
        mat = fitz.Matrix(zoomin, zoomin)  
        self.total_page = len(self.pdf)  
        for i, page in enumerate(self.pdf):  
            if i < page_from:  
                continue  
            if i >= page_to:  
                break  
            pix = page.get_pixmap(matrix=mat)  
            img = Image.frombytes("RGB", [pix.width, pix.height],  
                                  pix.samples)  
            self.page_images.append(img)  
            self.page_chars.append([])

然后读PDF目录结构
使用了PyPDF2库来读取pdf的目录信息

self.outlines = []  
try:  
    self.pdf = pdf2_read(fnm if isinstance(fnm, str) else BytesIO(fnm))  
    outlines = self.pdf.outline  
  
    def dfs(arr, depth):  
        for a in arr:  
            if isinstance(a, dict):  
                self.outlines.append((a["/Title"], depth))  
                continue  
            dfs(a, depth + 1)  
    dfs(outlines, 0)  
except Exception as e:  
    logging.warning(f"Outlines exception: {e}")  
if not self.outlines:  
    logging.warning(f"Miss outlines")

然后是英文文档检测，大概就是利用正则匹配

logging.info("Images converted.")  
self.is_english = [re.search(r"[a-zA-Z0-9,/¸;:'\[\]\(\)!@#$%^&*\"?<>._-]{30,}", "".join(  
    random.choices([c["text"] for c in self.page_chars[i]], k=min(100, len(self.page_chars[i]))))) for i in  
    range(len(self.page_chars))]  
if sum([1 if e else 0 for e in self.is_english]) > len(  
        self.page_images) / 2:  
    self.is_english = True  
else:  
    self.is_english = False

接着做分页处理
callback方法会更新文档解析进度，在文档页面可以查看实时进度

for i, img in enumerate(self.page_images):  
    chars = self.page_chars[i] if not self.is_english else []  
    # 计算字符的平均宽度、高度  
    self.mean_height.append(  
        np.median(sorted([c["height"] for c in chars])) if chars else 0  
    )  
    self.mean_width.append(  
        np.median(sorted([c["width"] for c in chars])) if chars else 8  
    )  
    self.page_cum_height.append(img.size[1] / zoomin)  
    j = 0  
    while j + 1 < len(chars):  
        # 对满足条件的添加空格（只包含数字、字母、逗号、句号、冒号、分号、感叹号和百分号， 两个字符宽度小于width的一半  
        if chars[j]["text"] and chars[j + 1]["text"] \  
                and re.match(r"[0-9a-zA-Z,.:;!%]+", chars[j]["text"] + chars[j + 1]["text"]) \  
                and chars[j + 1]["x0"] - chars[j]["x1"] >= min(chars[j + 1]["width"],  
                                                               chars[j]["width"]) / 2:  
            chars[j]["text"] += " "  
        j += 1  
    # if i > 0:  
    #     if not chars:    #         self.page_cum_height.append(img.size[1] / zoomin)    #     else:    #         self.page_cum_height.append(    #             np.max([c["bottom"] for c in chars]))    # OCR 识别  
    self.__ocr(i + 1, img, chars, zoomin)  
    if callback:  
        callback(prog=(i + 1) * 0.6 / len(self.page_images), msg="")

__ocr 处理
主要做的是detect，检测文本框，然后根据经验规则来对文本块做处理

def __ocr(self, pagenum, img, chars, ZM=3):  
    # 检测文本框  
    bxs = self.ocr.detect(np.array(img))  
    if not bxs:  
        self.boxes.append([])  
        return  
    bxs = [(line[0], line[1][0]) for line in bxs]  
    # 按照Y轴坐标排序  
    bxs = Recognizer.sort_Y_firstly(  
        [{"x0": b[0][0] / ZM, "x1": b[1][0] / ZM,  
          "top": b[0][1] / ZM, "text": "", "txt": t,  
          "bottom": b[-1][1] / ZM,  
          "page_number": pagenum} for b, t in bxs if b[0][0] <= b[1][0] and b[0][1] <= b[-1][1]],  
        self.mean_height[-1] / 3  
    )  
  
    # merge chars in the same rect  
    for c in Recognizer.sort_X_firstly(  
            chars, self.mean_width[pagenum - 1] // 4):  
        ii = Recognizer.find_overlapped(c, bxs)  
        if ii is None:  
            self.lefted_chars.append(c)  
            continue  
        ch = c["bottom"] - c["top"]  
        bh = bxs[ii]["bottom"] - bxs[ii]["top"]  
        if abs(ch - bh) / max(ch, bh) >= 0.7 and c["text"] != ' ':  
            self.lefted_chars.append(c)  
            continue  
        if c["text"] == " " and bxs[ii]["text"]:  
            if re.match(r"[0-9a-zA-Z,.?;:!%%]", bxs[ii]["text"][-1]):  
                bxs[ii]["text"] += " "  
        else:  
            bxs[ii]["text"] += c["text"]  
  
    for b in bxs:  
        if not b["text"]:  
            left, right, top, bott = b["x0"] * ZM, b["x1"] * \  
                ZM, b["top"] * ZM, b["bottom"] * ZM  
            b["text"] = self.ocr.recognize(np.array(img),  
                                           np.array([[left, top], [right, top], [right, bott], [left, bott]],  
                                                    dtype=np.float32))  
        del b["txt"]  
    bxs = [b for b in bxs if b["text"]]  
    if self.mean_height[-1] == 0:  
        self.mean_height[-1] = np.median([b["bottom"] - b["top"]  
                                          for b in bxs])  
    self.boxes.append(bxs)

·······························································分割线········································
实际的文件解析通过接口 /v1/document/run 进行触发的，实际的处理是在 api/db/services/task_service.py 中的 queue_tasks() 中完成的，此方法会根据文件创建一个或多个异步任务，方便异步执行。

提示词

想要通过源码了解 RAG 服务推荐优先阅读 QAnything

参考来源

对ragflow-main/deepdoc的源码剖析

RAGFlow嵌入自定义文件解析代码

向量化搜索，如果不希望引入第三方的向量数据库，那么开源的 Faiss 就是一个不错的选择：向量数据库 Faiss的git

来自工业界的知识库 RAG 服务(二)，RagFlow 源码全流程深度解析 重要点梳理

文件解析通过接口 /v1/document/run 进行触发的，实际的处理是在 api/db/services/task_service.py 中的 queue_tasks() 中完成的，此方法会根据文件创建一个或多个异步任务，方便异步执行

文件的解析是根据内容拆分为多个任务，通过 Redis 消息队列进行暂存，之后就可以离线异步处理。

直接查看对应的消息队列的消费模块，对应在 rag/svr/task_executor.py 中的 main() 方法中

在 RAGFlow 中的文件中包含了不少了**数据的清理**操作，比如在 deepdoc/vision/layout_recognizer.py 中的就包含着文档中无用内容的判断 def __is_garbage(b)

文件检索的支持 (包含混合检索)

RAGFlow 的检索目前实现的是混合检索，实现的是文本检索 + 向量检索，混合检索完全依赖 ElasticSearch 实现

文件检索的支持可以查看实际的对话处理流程，对话的 API 为 /v1/conversation/completion，实际对话的处理是在 api/db/services/dialog_service.py 中的 chat() 方法中完成

深入跟踪对话处理流程，可以看到文件的检索是在 rag/nlp/search.py 中的 search() 方法中完成。

==检索结果的重排==在 rag/nlp/search.py 中的 rerank() 中完成的，重排是基于文本匹配得分 + 向量匹配得分混合进行排序，默认文本匹配的权重为 0.3, 向量匹配的权重为 0.7;获取混合相似分之后，基于混合的相似分进行过滤和重排，默认混合得分低于 0.2 的会被过滤掉

在进行上面的检索和重排阶段中，只是进行了必要的过滤，没有限制匹配文档的数量。

实际内容可能会超过大模型的输入 token 数量，因此在调用大模型前会调用 api/db/services/dialog_service.py 文件中 message_fit_in() 根据大模型可用的 token 数量进行过滤。

将检索的内容，历史聊天记录以及问题构造为 prompt，即可作为大模型的输入了

任务1：对解析后的文档的某些字段挖空

文件解析核心函数：build()方法

"E:\ragflow-main\rag\svr\task_executor.py"

这是RAGflow核心的文件解析方法。在这个函数中，根据parser_id选择合适的解析器组，并执行文档的解析和切片

解析器

rag/app/naive.py

以默认的naive类型为例，深入对应的chunk()实现，其对应的实现在rag/app/naive.py中。此方法中包含了目前主持的docx, pdf, xlsx, md等格式的解析

文本挖空

def __ocr：“E:\ragflow-main\deepdoc\parser\pdf_parser.py”
在__ocr方法中，会对文本框进行检测和处理。你可以在这里添加逻辑来识别特定的字段，并进行挖空处理

表格内容提取

__extract_table_content “E:\ragflow-main\deepdoc\parser\docx_parser.py”
__compose_table_content “E:\ragflow-main\deepdoc\parser\docx_parser.py”
如果文档中包含表格，__extract_table_content和__compose_table_content函数会处理表格内容。你可以在这里添加逻辑来识别表格中的特定字段，并进行挖空

修改源码

需要在相应的解析器中添加逻辑来识别特定的字段，并将其替换为占位符（挖空）。这可能涉及到正则表达式的使用，以及对文档结构的理解