RAG进阶实战：PaddleOCR+混合检索打造高精度知识库

type

status

date

slug

summary

category

icon

password

网址

在大型语言模型（LLM）的应用落地中，RAG（检索增强生成）是解决模型幻觉和知识时效性的关键技术。

而在RAG的诸多场景中，基于多文档高精度智能分析与问答系统，也就是知识库又必然是我们最常遇到，且企业场景最刚需的一类。

那么如何做好知识库？

本文将以开源项目Paddle-ERNIE-RAG为例，对其关键技术进行说明介绍。

项目地址：https://github.com/LiaoYFBH/Paddle-ERNIE-RAG

该系统集成了在线 OCR 解析、Milvus 混合检索（向量+关键词）以及多维度的重排序（Reranker）策略，可以提升低资源环境下的检索准确率，以实现高精度多文档分析与问答。

系统架构概览

本项目的系统主要由四个核心模块组成：

1.数据提取层：使用在线 OCR API 进行高精度的文档布局分析（Layout Parsing）。

2.存储层：利用 Milvus 向量数据库存储 Dense Embedding，同时维护倒排索引以支持关键词检索。

3.检索与问答层：实现向量检索与关键词检索的加权融合（RRF），集成 ERNIE 大模型 API 接口生成回答。

4.应用层：基于 Gradio 构建交互界面。

🔗 项目资源

• 🐙 GitHub 代码仓库：https://github.com/LiaoYFBH/Paddle-ERNIE-RAG

• 🚀 星河社区在线应用：https://aistudio.baidu.com/application/detail/107183

关键技术实现

2.1 PP-StructureV3 文档解析

针对科研论文中常见的双栏排版、公式混排及图表嵌入问题，传统的 PyPDF2 等纯文本提取工具往往力不从心（容易导致段落乱序、表格崩坏）。

为此，本项目在 backend.py 中封装了 OnlinePDFParser 类，直接集成 PP-StructureV3 在线 API 进行高精度的文档布局分析（Layout Parsing）。

该方案具备三大核心优势：

• 结构化输出：直接返回 Markdown 格式（自动识别标题层级、段落边界）。

• 图表提取：在解析文本的同时，自动提取文档中的图片并转存，为后续的“多模态问答”提供素材。

• 上下文保留：基于滑动窗口进行切分，防止关键信息在切片边界丢失。

2.1.1 核心解析逻辑

在 backend.py 中，我们构建了 API 请求，将 PDF 文件流发送至服务端，并解析返回的 layoutParsingResults，提取出清洗后的 Markdown 文本和图片资源。

backend.py (OnlinePDFParser 类核心逻辑摘要)

def predict(self, file_path):

1. 文件转 Base64

with open(file_path, "rb") as file:

file_data = base64.b64encode(file.read()).decode("ascii")

2. 构造请求 Payload

payload = {

"file": file_data,

"fileType": 1, # PDF 类型

"useChartRecognition": False, # 根据需求配置

"useDocOrientationClassify": False

}

3. 发送请求获取 Layout Parsing 结果

response = requests.post(self.api_url, json=payload, headers=headers)

res_json = response.json()

4. 提取 Markdown 文本与图片

parsingresults = resjson.get("result", {}).get("layoutParsingResults", [])

mock_outputs = []

for item in parsing_results:

md_text = item.get("markdown", {}).get("text", "")

images = item.get("markdown", {}).get("images", {})

... (后续图片下载与文本清洗逻辑)

mockoutputs.append(MockResult(mdtext, images))

return mock_outputs, "Success"

2.1.2 滑动窗口文本分块

拿到结构化的 Markdown 文本后，为了避免语义被生硬切断（例如一句话跨了两个 chunk），我们实现了一个带有 overlap（重叠区）的滑动窗口分块策略。

backend.py

def splittextintochunks(text: str, chunksize: int = 300, overlap: int = 120) -> list:

"""基于滑动窗口的文本分块，保留 overlap 长度的重叠上下文"""

if not text: return []

lines = [line.strip() for line in text.split("\n") if line.strip()]

chunks = []

current_chunk = []

current_length = 0

for line in lines:

while len(line) > chunk_size:

处理超长单行

part = line[:chunk_size]

line = line[chunk_size:]

current_chunk.append(part)

... (切分逻辑) ...

current_chunk.append(line)

current_length += len(line)

当累积长度超过阈值，生成一个 chunk

if currentlength > chunksize:

chunks.append("\n".join(current_chunk))

回退：保留最后 overlap 长度的文本作为下一个 chunk 的开头

overlaptext = currentchunk[-1][-overlap:] if current_chunk else ""

currentchunk = [overlaptext] if overlap_text else []

currentlength = len(overlaptext)

if current_chunk:

chunks.append("\n".join(current_chunk).strip())

return chunks

2.2 Milvus 向量库与混合检索策略

2.2.1 知识库命名的工程化处理

在实际部署中，Milvus 等向量数据库对集合名称（Collection Name）通常有严格的命名限制。为了解决这一问题，我们在后端代码中实现了一套透明的编解码机制。

1.编码 (Encode)：当用户创建如“物理论文”的库时，系统将其 UTF-8 字节转换为 Hex 字符串，并添加 kb_ 前缀。

2.解码 (Decode)：在前端展示时，自动将 Hex 字符串反解为原始中文。

import binascii

import re

def encodename(uiname):

"""把中文名称转为 Milvus 合法的 Hex 字符串"""

if not ui_name: return ""

如果是纯英文/数字/下划线，直接返回

if re.match(r'^[a-zA-Z][a-zA-Z0-9]*$', ui_name):

return ui_name

Hex 编码并加前缀 kb_

hexstr = binascii.hexlify(uiname.encode('utf-8')).decode('utf-8')

return f"kb{hexstr}"

def decodename(realname):

"""把 Hex 字符串转回中文"""

if realname.startswith("kb"):

try:

hexstr = realname[3:]

return binascii.unhexlify(hex_str).decode('utf-8')

except:

return real_name

2.2.2 向量化入库与元数据绑定

在 OCR 解析并将长文本切分为 Chunks 后，系统并非简单地将文本存入数据库，而是执行了“向量化 + 元数据绑定”的关键步骤。

为了支持后续的精确溯源（Citation）和多模态问答，我们在设计 Milvus Schema 时，除了存储 384 维的 Dense Vector 外，还强制绑定了 filename（文件名）、page（页码）和 chunk_id（切片 ID）等标量字段。

这一过程在 vectorstore.py 中通过 insertdocuments 方法实现，采用批量 Embedding 策略以减少网络开销：

vector_store.py

def insert_documents(self, documents):

"""批量向量化并写入 Milvus"""

if not documents: return

1. 提取纯文本列表，批量请求 Embedding 模型

texts = [doc['content'] for doc in documents]

embeddings = self.get_embeddings(texts)

2. 数据清洗：过滤掉 Embedding 失败的坏数据

validdocs, validvectors = [], []

for i, emb in enumerate(embeddings):

if emb and len(emb) == 384: # 确保向量维度正确

valid_docs.append(documents[i])

valid_vectors.append(emb)

3. 组装列式数据 (Columnar Format)

Milvus insert 接口要求各字段数据以列表形式传入

data = [

[doc['filename'] for doc in valid_docs], # Scalar: 文件名

[doc['page'] for doc in valid_docs], # Scalar: 页码 (用于溯源)

[doc['chunkid'] for doc in validdocs], # Scalar: 切片ID

[doc['content'] for doc in valid_docs], # Scalar: 原始内容 (用于关键词检索)

valid_vectors # Vector: 语义向量

]

4. 执行插入与持久化

self.collection.insert(data)

self.collection.flush()

2.2.3 混合检索策略

检索前，系统首先利用 LLM 生成的问题的双语翻译，避免中文问题询问英文文档，使得关键词不匹配，以最大化语义覆盖。随后并行执行两路检索：

1.Dense (向量检索)：捕捉语义相似度（例如“简谐振子”与“弹簧振子”的语义关联）。

2.Sparse (关键词检索)：弥补向量模型对专有名词或精确数字匹配的不足（例如精确匹配公式中的变量名）。

向量检索容易因语义泛化而召回错误概念（如“弹簧振子”与“简谐振子”），而关键词检索能确保专有名词的精确命中，从而大幅提升准确率。

然后执行：

-RRF (倒排融合)：系统内部使用倒排秩融合算法 (Reciprocal Rank Fusion) 将两路结果合并，确保多样性。

vector_store.py 中的检索逻辑摘要

def search(self, query: str, top_k: int = 10, \*\*kwargs):

'''向量检索(Dense+Keyword)+RRF 融合'''

1. 向量检索 (Dense)

dense_results = []

queryvector = self.embeddingclient.get_embedding(query) # ... (Milvus search code) ...

2. 关键词检索 (Keyword)

通过 jieba 分词后构建 like "%keyword%" 查询

keywordresults = self.keywordsearch(query, topk=top_k * 5, expr=expr)

3. RRF 融合

rank_dict = {}

def applyrrf(resultslist, k=60, weight=1.0):

for rank, item in enumerate(results_list):

docid = item.get('id') or item.get('chunkid')

if docid not in rankdict:

rankdict[docid] = {"data": item, "score": 0.0}

RRF 核心公式

rankdict[docid]["score"] += weight * (1.0 / (k + rank))

applyrrf(denseresults, weight=4.0)

applyrrf(keywordresults, weight=1.0)

4. 排序输出

sorteddocs = sorted(rankdict.values(), key=lambda x: x['score'], reverse=True)

return [item['data'] for item in sorteddocs[:topk * 2]]

2.3 综合重排序算法

检索回来的片段（Chunks）需要进一步精排。在reranker_v2.py中，设计了一套综合打分算法。

评分维度包括：

1.模糊匹配（Fuzzy Score）：使用 fuzzywuzzy 计算 Query 与 Content 的字面重合度。

2.关键词覆盖率（Keyword Coverage）：计算 Query 中的核心词在文档片段中的出现比例。

3.语义相似度：来自 Milvus 的原始向量距离。

4.长度惩罚与位置偏置：对过短的片段进行惩罚，对 Milvus 召回的排名靠前的片段给予位置奖励。

5.专有名词：

• 英文（看“大小写”特征）：使用正则 \b[A-Z][a-z]+\b|[A-Z]{2,}，专门匹配首字母大写的单词（如 "Milvus"）或全大写的缩写（如 "RAG"），因为在英文中这些通常代表专有名词。

• 中文（看“连续性”特征）：由于中文没有大小写，策略变成了“切分+长度”：使用非中文字符作为分隔符切断句子，保留所有连续出现 2 个及以上\\的汉字片段（如“简谐振子”），将其视为潜在实体。

具体的分数占比见下图：

这种基于规则与语义结合的重排序策略，在无训练数据的情况下，比纯黑盒模型更具可解释性。

reranker_v2.py

def calculatecomposite_score(self, query: str, chunk: Dict[str, Any]) -> float:

content = chunk.get('content', '')

1. 字面重合度 (FuzzyWuzzy)

fuzzyscore = fuzz.partialratio(query, content)

2. 关键词覆盖率

querykeywords = self.extract_keywords(query)

contentkeywords = self.extract_keywords(content)

keywordcoverage = (len(querykeywords & contentkeywords) / len(querykeywords)) * 100 if query_keywords else 0

3. 向量语义分 (归一化)

milvusdistance = chunk.get('semanticscore', 0)

milvussimilarity = 100 / (1 + milvusdistance * 0.1)

4. 长度惩罚 (偏好 200-600 字的段落)

content_len = len(content)

if 200 <= content_len <= 600:

length_score = 100

else:

... (惩罚逻辑)

lengthscore = 100 - min(50, abs(contentlen - 400) / 20)

加权求和

base_score = (

fuzzy_score * 0.25 +

keyword_coverage * 0.25 +

milvus_similarity * 0.35 +

length_score * 0.15

)

位置权重

position_bonus = 0

if 'milvus_rank' in chunk:

rank = chunk['milvus_rank']

position_bonus = max(0, 20 - rank)

专有名词额外加分 (Bonus)

propernounbonus = 30 if self.checkproper_nouns(query, content) else 0

return basescore + propernoun_bonus

2.4 API 速率限制与自适应保护

在调用 LLM 或 Embedding 服务时，偶尔会遇到429 Too Many Requests。本项目在ernie_client.py中实现了自适应降速机制：

遇到限流时的处理逻辑

if isratelimit:

self.adaptiveslow_down() # 永久增加请求间隔

wait_time = (2 ** attempt) + random.uniform(1.0, 3.0) # 指数退避

time.sleep(wait_time)

def adaptiveslow_down(self):

"""触发自适应降级：遇到限流时，永久增加全局请求间隔"""

self.currentdelay = min(self.currentdelay * 2.0, 1...