文档解析与文本切分利器¶
学习目标¶
- 掌握系统中用于解析不同文档格式（PDF, DOCX, PPTX, Images）的核心工具。
- 理解光学字符识别（OCR）工具
RapidOCR如何集成并应用于文档解析流程。 - 了解系统中提供的两种文本切分工具：基于规则的递归切分器和基于模型的语义切分器。
- 熟悉
edu_document_loaders和edu_text_spliter目录下各脚本的具体实现和功能。
文档解析工具 (edu_document_loaders/)¶
为了从各种常见的 IT 教育文档格式中提取信息，系统实现了一系列专门的加载器（Loaders）。这些加载器不仅能提取文档中的原生文本，还能利用 OCR 技术识别并提取图片中嵌入的文字。它们都继承自 Langchain 的 BaseLoader，并实现了 lazy_load 方法来按需生成 Document 对象。
OCR 引擎核心 (edu_ocr.py)¶
该脚本提供了一个标准化的函数 get_ocr() 来初始化和获取 OCR 识别引擎实例。这是所有需要图片文字识别功能的加载器的基础。
- 功能: 初始化
RapidOCR实例。
- 特点：
- 引擎选择: 优先尝试
rapidocr_paddle(利用 PaddlePaddle 推理，推荐 GPU 环境)，若失败则回退到rapidocr_onnxruntime(利用 ONNX Runtime 推理，适合 CPU 环境或需要跨平台部署的场景)。 - 参数控制: 允许通过
use_cuda参数控制是否启用 GPU 加速（如果使用 PaddlePaddle 引擎）。
- 引擎选择: 优先尝试
# edu_document_loaders/edu_ocr.py 源码
from typing import TYPE_CHECKING
'''
paddleocr：解析图片中的文字，也可以进行表格识别
rapidocr_paddle 和 rapidocr_onnxruntime 两种导入方式
主要区别在于它们所使用的推理引擎和硬件支持
选择哪种方式最合适取决于你的硬件环境和性能需求。
当你有 GPU 且追求速度时：使用 rapidocr_paddle。PaddlePaddle 原生支持在 GPU 上推理 PaddleOCR 模型，速度更快。
当只有 CPU 且需要高效推理时：使用 rapidocr_onnxruntime。它在 CPU 上进行了优化，资源占用较低.
'''
def get_ocr(use_cuda: bool = True) -> "RapidOCR":
try:
from rapidocr_paddle import RapidOCR
'''
det_use_cuda=True：启用检测模型的GPU加速。cls_use_cuda=True：启用分类模型的GPU加速。rec_use_cuda=True：启用识别模型的GPU加速。
'''
ocr = RapidOCR(det_use_cuda=use_cuda, cls_use_cuda=use_cuda, rec_use_cuda=use_cuda)
except ImportError:
#
from rapidocr_onnxruntime import RapidOCR
ocr = RapidOCR()
return ocr
PDF 文档加载器 (edu_pdfloader.py)¶
OCRPDFLoader 类专门用于处理 PDF 文件。
- 功能: 解析 PDF，提取文本和图片中的文字。
- 依赖:
PyMuPDF(fitz),Pillow,numpy,opencv-python,tqdm以及edu_ocr.py。
- 核心逻辑:
- 使用
fitz.open()打开 PDF。 - 逐页 (
page) 处理。 - 使用
page.get_text()提取原生文本。 - 使用
page.get_image_info(xrefs=True)获取页面上的图片信息。 - OCR 应用: 对获取到的图片，检查其尺寸是否超过预设阈值
PDF_OCR_THRESHOLD（默认为页面宽高的 60%）。仅对大于阈值的图片执行 OCR。 - 处理页面旋转 (
page.rotation)，确保 OCR 时图像方向正确。 - 调用
get_ocr()获取的 OCR 实例识别图片文字。 - 合并原生文本和 OCR 结果。
- 使用
tqdm显示处理进度。
- 使用
# edu_document_loaders/edu_pdfloader.py 源码
import cv2
import fitz # pyMuPDF里面的fitz包，不要与pip install fitz混淆
import numpy as np
from PIL import Image
from tqdm import tqdm
from typing import Iterator
from edu_ocr import get_ocr
from langchain_core.documents import Document
from langchain_core.document_loaders import BaseLoader
from langchain.text_splitter import CharacterTextSplitter
# PDF OCR 控制：只对宽高超过页面一定比例（图片宽/页面宽，图片高/页面高）的图片进行 OCR。
# 这样可以避免 PDF 中一些小图片的干扰，提高非扫描版 PDF 处理速度
PDF_OCR_THRESHOLD = (0.6, 0.6)
class OCRPDFLoader(BaseLoader):
"""An example document loader that reads a file line by line."""
def __init__(self, file_path: str) -> None:
"""Initialize the loader with a file path.
Args:
file_path: The path to the file to load.
"""
self.file_path = file_path
def lazy_load(self) -> Iterator[Document]:
# <-- Does not take any arguments
"""A lazy loader that reads a file line by line.
When you're implementing lazy load methods, you should use a generator
to yield documents one by one.
"""
line = self.pdf2text()
yield Document(page_content=line, metadata={"source": self.file_path})
def pdf2text(self):
ocr = get_ocr()
# 打开pdf文件
doc = fitz.open(self.file_path)
## 获取页数
# print(f'len(doc)-->{len(doc)}')
resp = ""
b_unit = tqdm(total=doc.page_count, desc="OCRPDFLoader context page index: 0")
for i, page in enumerate(doc):
b_unit.set_description("OCRPDFLoader context page index: {}".format(i))
b_unit.refresh()
# 提取文本：默认使用 "text" 模式提取文本。
text = page.get_text("")
resp += text + "\n"
# print(f'resp-->{resp}')
# 获取图片：获得所有显示的图像的元信息列表。
# 它适用于所有文档类型，不仅限于 PDF。
img_list = page.get_image_info(xrefs=True)
# print(f'img_list--》{img_list}')
# print(f'img_list--》{len(img_list)}')
for img in img_list:
# xref一种编号，指向该图像对象在PDF文件中的位置，程序可以通过这个编号快速定位和提取图像数据。
if xref := img.get("xref"):
# 图像在页面上的位置和尺寸。
bbox = img["bbox"]
# 检查图片尺寸是否超过设定的阈值
# if ((bbox[2] - bbox[0]) / (page.rect.width) < PDF_OCR_THRESHOLD[0]
# or (bbox[3] - bbox[1]) / (page.rect.height) < PDF_OCR_THRESHOLD[1]):
# continue
pix = fitz.Pixmap(doc, xref)
# print(f'page.rotation-->{page.rotation}')
if int(page.rotation) != 0: # 如果Page有旋转角度，则旋转图片
img_array = np.frombuffer(pix.samples, dtype=np.uint8).reshape(pix.height, pix.width, -1)
tmp_img = Image.fromarray(img_array)
ori_img = cv2.cvtColor(np.array(tmp_img), cv2.COLOR_RGB2BGR)
rot_img = self.rotate_img(img=ori_img, angle=360 - page.rotation)
img_array = cv2.cvtColor(rot_img, cv2.COLOR_RGB2BGR)
else:
img_array = np.frombuffer(pix.samples, dtype=np.uint8).reshape(pix.height, pix.width, -1)
# result：包含了图像中检测到的所有文本框的位置、文本内容和置信度信息。
# _：它是一个包含了时间数据的列表，可以用于优化模型运行速度。
result, _ = ocr(img_array)
if result:
ocr_result = [line[1] for line in result]
resp += "\n".join(ocr_result)
# 更新进度
b_unit.update(1)
return resp
if __name__ == '__main__':
pdf_loader = OCRPDFLoader(file_path="./data/Python机器学习基础教程.pdf")
doc = pdf_loader.load()
print(type(doc))
print(doc)
# text_spliter = CharacterTextSplitter(chunk_size=300, chunk_overlap=20)
# result = text_spliter.split_documents(doc)
# print(len(result))
# print(result[0])
(注意：上述代码中 self.rotate_img 方法未在提供的代码段中定义，实际使用时需要确保该方法存在或移除相关调用)
Word 文档加载器 (edu_docloader.py)¶
OCRDOCLoader 类用于处理 .docx 文件。
- 功能: 解析 DOCX 文件，提取段落、表格文本，并对嵌入的图片进行 OCR。
- 依赖:
python-docx,Pillow,numpy,tqdm以及edu_ocr.py。
- 核心逻辑:
- 使用
docx.Document()打开 DOCX 文件。 - 定义
iter_block_items辅助函数，用于统一遍历文档中的段落 (Paragraph) 和表格 (Table) 块。 - 遍历所有块：
- 如果是段落，提取
block.text。同时，使用 XPath (.//pic:pic,.//a:blip/@r:embed) 查找并提取段落内嵌入的图片。对提取的图片执行 OCR。 - 如果是表格，遍历所有单元格 (
cell)，提取单元格内段落的文本。
- 如果是段落，提取
- 合并所有提取的文本和 OCR 结果。
- 使用
tqdm显示处理进度。
- 使用
# edu_document_loaders/edu_docloader.py 源码
from typing import Iterator
from .edu_ocr import get_ocr
# 导入必要的模块
from tqdm import tqdm
from docx.table import _Cell, Table # 用于处理表格
from docx.oxml.table import CT_Tbl # 用于处理表格XML结构
from docx.oxml.text.paragraph import CT_P # 用于处理段落XML结构
from docx.text.paragraph import Paragraph # 用于处理段落内容
from docx import Document as Docu1
from docx.document import Document as Docu2
from docx import ImagePart # 用于处理Word文档和图片
from PIL import Image # 用于处理图片
from io import BytesIO # 用于将字节流转换为图片
import numpy as np # 用于处理数组
from langchain_core.documents import Document
from langchain_core.document_loaders import BaseLoader
class OCRDOCLoader(BaseLoader):
"""An example document loader that reads a file line by line."""
def __init__(self, filepath: str) -> None:
"""Initialize the loader with a file path.
Args:
filepath_path: The path to the filepath to load.
"""
self.filepath = filepath
def lazy_load(self) -> Iterator[Document]:
# <-- Does not take any arguments
"""A lazy loader that reads a file line by line.
When you're implementing lazy load methods, you should use a generator
to yield documents one by one.
"""
line = self.doc2text(self.filepath)
yield Document(page_content=line, metadata={"source": self.filepath})
def doc2text(self, filepath):
# 创建OCR识别对象
ocr = get_ocr()
# print(f'ocr--》{ocr}') # 输出OCR对象信息
# 读取Word文档
doc = Docu1(filepath)
# print(f'doc-->{doc}') # 输出读取到的文档信息
# 定义一个空字符串用于存储最终的文本内容
resp = ""
# 定义一个迭代器，用于遍历文档中的块（段落、表格等）
def iter_block_items(parent):
# 判断parent对象类型，如果是Document类型，则获取其元素
if isinstance(parent, Docu2):
parent_elm = parent.element.body
# 如果是表格单元格类型，获取单元格的XML元素
elif isinstance(parent, _Cell):
parent_elm = parent._tc
else:
raise ValueError("OCRDOCLoader parse fail") # 如果都不是，则抛出错误
# print(f'parent_elm--》{parent_elm}')
# print('*'*80)
# 遍历parent_elm中的所有子元素
for child in parent_elm.iterchildren():
# print(f'child--》{child}')
if isinstance(child, CT_P): # 如果是段落类型
yield Paragraph(child, parent) # 返回段落
elif isinstance(child, CT_Tbl): # 如果是表格类型
yield Table(child, parent) # 返回表格
# print(f'doc.paragraphs-->{doc.paragraphs}')
# print(f'doc.tables-->{doc.tables}')
# 创建进度条，表示文档处理的进度
b_unit = tqdm(total=len(doc.paragraphs) + len(doc.tables),
desc="OCRDOCLoader block index: 0")
# 遍历文档中的所有块（段落和表格）
for i, block in enumerate(iter_block_items(doc)):
# 更新进度条描述
b_unit.set_description("OCRDOCLoader block index: {}".format(i))
b_unit.refresh() # 刷新进度条
# 如果块是段落类型
if isinstance(block, Paragraph):
resp += block.text.strip() + "\n" # 将段落文本加入到返回字符串中
# 获取段落中的所有图片
images = block._element.xpath('.//pic:pic')
for image in images:
# 遍历图片，获取图片ID
for img_id in image.xpath('.//a:blip/@r:embed'):
part = doc.part.related_parts[img_id] # 根据图片ID获取图片对象
if isinstance(part, ImagePart): # 如果该部分是图片
# BytesIO 是 Python 内置的 io 模块中的一个类，用于在内存中读写二进制数据
# part._blob 通常表示从某个文档（如 DOCX 文件）中提取的二进制内容。
image = Image.open(BytesIO(part._blob)) # 打开图片
result, _ = ocr(np.array(image)) # 使用OCR识别图片中的文字
if result: # 如果识别结果不为空
ocr_result = [line[1] for line in result] # 提取识别出的文字
resp += "\n".join(ocr_result) # 将识别结果加入返回文本中
# 如果块是表格类型
elif isinstance(block, Table):
# 遍历表格中的所有行和单元格
for row in block.rows:
for cell in row.cells:
for paragraph in cell.paragraphs:
resp += paragraph.text.strip() + "\n" # 将单元格内的段落文本加入返回文本中
# 更新进度条
b_unit.update(1)
# 返回提取的文本内容
return resp
if __name__ == '__main__':
docx_loader = OCRDOCLoader(filepath='./data/b.docx')
doc = docx_loader.load()
print(doc)
PowerPoint 文档加载器 (edu_pptloader.py)¶
OCRPPTLoader 类用于处理 .ppt 和 .pptx 文件。
- 功能: 解析 PPT/PPTX 文件，提取形状（文本框、表格）、图片中的文本。
- 依赖:
python-pptx,Pillow,numpy,tqdm以及edu_ocr.py。
-
核心逻辑:
-
使用
pptx.Presentation()打开演示文稿。 -
逐张幻灯片 (
slide) 处理。 -
顺序处理: 将幻灯片上的形状 (
shape) 按视觉顺序（top,left坐标）排序。 -
定义 extract_text 递归函数处理单个形状：
- 提取文本框 (
shape.has_text_frame) 的文本。 - 提取表格 (
shape.has_table) 内所有单元格的文本。 - 如果形状是图片 (
shape.shape_type == 13)，提取图片数据 (shape.image.blob)，执行 OCR。 - 如果形状是组合 (
shape.shape_type == 6)，递归调用extract_text处理其包含的子形状。
- 提取文本框 (
-
遍历排序后的形状，调用
extract_text。 -
合并所有提取的文本和 OCR 结果。
-
使用
tqdm显示处理进度。
-
# edu_document_loaders/edu_pptloader.py 源码
from typing import Iterator
from edu_ocr import get_ocr
from langchain_core.documents import Document
from langchain_core.document_loaders import BaseLoader
from pptx import Presentation
from PIL import Image
import numpy as np
from io import BytesIO
from tqdm import tqdm
class OCRPPTLoader(BaseLoader):
"""An example document loader that reads a file line by line."""
def __init__(self, filepath: str) -> None:
"""Initialize the loader with a file path.
Args:
filepath: The path to the ppt to load.
"""
self.filepath = filepath
def lazy_load(self) -> Iterator[Document]:
# <-- Does not take any arguments
"""A lazy loader that reads a file line by line.
When you're implementing lazy load methods, you should use a generator
to yield documents one by one.
"""
line = self.ppt2text(self.filepath)
yield Document(page_content=line, metadata={"source": self.filepath})
def ppt2text(self, filepath):
# 打开指定路径的 PowerPoint 文件
prs = Presentation(filepath)
print(f'prs-->{prs}')
# 获取 OCR 功能的实例
ocr = get_ocr()
# 初始化一个空字符串，用于存储提取的文本内容
resp = ""
def extract_text(shape):
# nonlocal指明resp非全局非局部，而是外部嵌套函数中的变量，
# 允许内部函数访问和修改外部函数中定义的变量resp
nonlocal resp
# 检查形状是否有文本框
if shape.has_text_frame:
# 将文本框中的文本添加到resp中，并去掉前后空格
resp += shape.text.strip() + "\n"
# 检查形状是否为表格
if shape.has_table:
# 遍历表格的每一行
for row in shape.table.rows:
# 遍历每一行中的每个单元格
for cell in row.cells:
# 遍历单元格中的每个段落
for paragraph in cell.text_frame.paragraphs:
# 将单元格中的文本添加到resp中，并去掉前后空格
resp += paragraph.text.strip() + "\n"
# 检查形状是否为图片（shape_type == 13）
if shape.shape_type == 13: # 13 表示图片
# 使用 BytesIO 打开图片数据并转换为图像对象
image = Image.open(BytesIO(shape.image.blob))
# 使用 OCR 处理图像并获取结果
result, _ = ocr(np.array(image))
if result: # 如果 OCR 有结果
# 提取 OCR 结果中的文本行
ocr_result = [line[1] for line in result]
# 将 OCR 提取的文本添加到resp中，以换行分隔
resp += "\n".join(ocr_result)
# 检查形状是否为组合形状（shape_type == 6）
elif shape.shape_type == 6: # 6 表示组合
# 遍历组合形状中的每个子形状，递归调用extract_text函数
for child_shape in shape.shapes:
extract_text(child_shape)
# 创建一个进度条，用于显示幻灯片处理进度，初始总数为幻灯片数量
b_unit = tqdm(total=len(prs.slides), desc="OCRPPTLoader slide index: 1")
# 遍历所有幻灯片
for slide_number, slide in enumerate(prs.slides, start=1):
# 更新进度条描述，显示当前处理的幻灯片索引
b_unit.set_description("OCRPPTLoader slide index: {}".format(slide_number))
b_unit.refresh() # 刷新进度条显示
# 按照从上到下、从左到右的顺序对形状进行排序遍历
sorted_shapes = sorted(slide.shapes, key=lambda x: (x.top, x.left))
for shape in sorted_shapes:
extract_text(shape) # 调用extract_text函数提取当前形状的文本内容
b_unit.update(1) # 更新进度条，表示处理了一张幻灯片
return resp # 返回提取到的所有文本内容
if __name__ == '__main__':
img_loader = OCRPPTLoader(filepath='./data/01.pptx')
doc = img_loader.load()
print(doc)
图像文件加载器 (edu_imgloader.py)¶
OCRIMGLoader 类用于直接处理图像文件（如 .png, .jpg）。
- 功能: 对单个图像文件执行 OCR。
- 依赖:
Pillow,numpy以及edu_ocr.py。
- 核心逻辑:
- 接收图像文件路径
img_path。 - 调用
get_ocr()获取 OCR 实例。 - 直接对图像文件执行 OCR。
- 将 OCR 结果（所有识别出的文本行）合并成一个字符串。
- 接收图像文件路径
# edu_document_loaders/edu_imgloader.py 源码
from typing import Iterator
from edu_ocr import get_ocr
from langchain_core.documents import Document
from langchain_core.document_loaders import BaseLoader
class OCRIMGLoader(BaseLoader):
"""An example document loader that reads a file line by line."""
def __init__(self, img_path: str) -> None:
"""Initialize the loader with a file path.
Args:
img_path: The path to the img to load.
"""
self.img_path = img_path
def lazy_load(self) -> Iterator[Document]:
# <-- Does not take any arguments
"""A lazy loader that reads a file line by line.
When you're implementing lazy load methods, you should use a generator
to yield documents one by one.
"""
line = self.img2text()
yield Document(page_content=line, metadata={"source": self.img_path})
def img2text(self):
resp = ""
ocr = get_ocr()
result, _ = ocr(self.img_path)
if result:
ocr_result = [line[1] for line in result]
resp += "\n".join(ocr_result)
return resp
if __name__ == '__main__':
img_loader = OCRIMGLoader(img_path='./data/test_img.png')
doc = img_loader.load()
print(doc)
文本切分工具 (edu_text_spliter/)¶
将解析得到的长文本切分成适合向量化和检索的小块是 RAG 流程中的关键一步。本系统提供了两种文本切分工具。
中文递归文本切分器 (edu_chinese_recursive_text_splitter.py)¶
ChineseRecursiveTextSplitter 类是针对中文文本特点定制的切分器。
- 功能: 将长文本按照预设的中文分隔符递归地切分成指定大小的块。
- 继承:
langchain.text_splitter.RecursiveCharacterTextSplitter。
- 核心定制:
_separators: 定义了用于切分的、按优先级排列的分隔符列表，包括常见的中文标点和换行符，如["\n\n", "\n", "。|！|？", "\.\s|\!\s|\?\s", "；|;\s", "，|,\s"]。这有助于在切分时尽量保持句子的完整性。- 支持通过正则表达式定义分隔符 (
is_separator_regex=True)。 - 通过
chunk_size和chunk_overlap控制切分块的大小和重叠。
# edu_text_spliter/edu_chinese_recursive_text_splitter.py 源码
import re
from typing import List, Optional, Any
from langchain.text_splitter import RecursiveCharacterTextSplitter
import logging
logger = logging.getLogger(__name__)
def _split_text_with_regex_from_end(
text: str, separator: str, keep_separator: bool
) -> List[str]:
# Now that we have the separator, split the text
if separator:
if keep_separator:
# The parentheses in the pattern keep the delimiters in the result.
_splits = re.split(f"({separator})", text)
splits = ["".join(i) for i in zip(_splits[0::2], _splits[1::2])]
if len(_splits) % 2 == 1:
splits += _splits[-1:]
# splits = [_splits[0]] + splits
else:
splits = re.split(separator, text)
else:
splits = list(text)
return [s for s in splits if s != ""]
class ChineseRecursiveTextSplitter(RecursiveCharacterTextSplitter):
def __init__(
self,
separators: Optional[List[str]] = None,
keep_separator: bool = True,
is_separator_regex: bool = True,
**kwargs: Any,
) -> None:
"""Create a new TextSplitter."""
super().__init__(keep_separator=keep_separator, **kwargs)
self._separators = separators or [
"\n\n",
"\n",
"。|！|？",
"\.\s|\!\s|\?\s",
"；|;\s",
"，|,\s"
]
self._is_separator_regex = is_separator_regex
def _split_text(self, text: str, separators: List[str]) -> List[str]:
"""Split incoming text and return chunks."""
final_chunks = []
# Get appropriate separator to use
separator = separators[-1]
new_separators = []
for i, _s in enumerate(separators):
_separator = _s if self._is_separator_regex else re.escape(_s)
if _s == "":
separator = _s
break
if re.search(_separator, text):
separator = _s
new_separators = separators[i + 1:]
break
_separator = separator if self._is_separator_regex else re.escape(separator)
splits = _split_text_with_regex_from_end(text, _separator, self._keep_separator)
# Now go merging things, recursively splitting longer texts.
_good_splits = []
_separator = "" if self._keep_separator else separator
for s in splits:
if self._length_function(s) < self._chunk_size:
_good_splits.append(s)
else:
if _good_splits:
merged_text = self._merge_splits(_good_splits, _separator)
final_chunks.extend(merged_text)
_good_splits = []
if not new_separators:
final_chunks.append(s)
else:
other_info = self._split_text(s, new_separators)
final_chunks.extend(other_info)
if _good_splits:
merged_text = self._merge_splits(_good_splits, _separator)
final_chunks.extend(merged_text)
return [re.sub(r"\n{2,}", "\n", chunk.strip()) for chunk in final_chunks if chunk.strip()!=""]
if __name__ == "__main__":
text_splitter = ChineseRecursiveTextSplitter(
keep_separator=True,
is_separator_regex=True,
chunk_size=150,
chunk_overlap=10
)
ls = [
"""中国对外贸易形势报告（75页）。前 10 个月，一般贸易进出口 19.5 万亿元，增长 25.1%， 比整体进出口增速高出 2.9 个百分点，占进出口总额的 61.7%，较去年同期提升 1.6 个百分点。其中，一般贸易出口 10.6 万亿元，增长 25.3%，占出口总额的 60.9%，提升 1.5 个百分点；进口8.9万亿元，增长24.9%，占进口总额的62.7%， 提升 1.8 个百分点。加工贸易进出口 6.8 万亿元，增长 11.8%， 占进出口总额的 21.5%，减少 2.0 个百分点。其中，出口增 长 10.4%，占出口总额的 24.3%，减少 2.6 个百分点；进口增 长 14.2%，占进口总额的 18.0%，减少 1.2 个百分点。此外， 以保税物流方式进出口 3.96 万亿元，增长 27.9%。其中，出 口 1.47 万亿元，增长 38.9%；进口 2.49 万亿元，增长 22.2%。前三季度，中国服务贸易继续保持快速增长态势。服务 进出口总额 37834.3 亿元，增长 11.6%；其中服务出口 17820.9 亿元，增长 27.3%；进口 20013.4 亿元，增长 0.5%，进口增 速实现了疫情以来的首次转正。服务出口增幅大于进口 26.8 个百分点，带动服务贸易逆差下降 62.9%至 2192.5 亿元。服 务贸易结构持续优化，知识密集型服务进出口 16917.7 亿元， 增长 13.3%，占服务进出口总额的比重达到 44.7%，提升 0.7 个百分点。 二、中国对外贸易发展环境分析和展望 全球疫情起伏反复，经济复苏分化加剧，大宗商品价格 上涨、能源紧缺、运力紧张及发达经济体政策调整外溢等风 险交织叠加。同时也要看到，我国经济长期向好的趋势没有 改变，外贸企业韧性和活力不断增强，新业态新模式加快发 展，创新转型步伐提速。产业链供应链面临挑战。美欧等加快出台制造业回迁计 划，加速产业链供应链本土布局，跨国公司调整产业链供应 链，全球双链面临新一轮重构，区域化、近岸化、本土化、 短链化趋势凸显。疫苗供应不足，制造业“缺芯”、物流受限、 运价高企，全球产业链供应链面临压力。 全球通胀持续高位运行。能源价格上涨加大主要经济体 的通胀压力，增加全球经济复苏的不确定性。世界银行今年 10 月发布《大宗商品市场展望》指出，能源价格在 2021 年 大涨逾 80%，并且仍将在 2022 年小幅上涨。IMF 指出，全 球通胀上行风险加剧，通胀前景存在巨大不确定性。""",
]
# text = """"""
for inum, text in enumerate(ls):
print(inum)
chunks = text_splitter.split_text(text)
for chunk in chunks:
print(chunk)
基于模型的语义切分器 (edu_model_text_spliter.py)¶
AliTextSplitter 类提供了另一种基于 AI 模型的文本切分方法。
- 功能: 利用预训练的文档语义分割模型对文本进行切分。
- 继承:
langchain.text_splitter.CharacterTextSplitter。
- 核心逻辑:
- 初始化时指定是否处理 PDF 文本（包含特定的换行符和空格处理逻辑）。
- 调用
modelscope.pipeline加载指定的文档分割模型（代码中为MODEL_PATH['segment_model']['ali_model']，需要配置configs.py或直接指定模型路径/名称，如 'damo/nlp_bert_document-segmentation_chinese-base'）。模型运行在 CPU 上。 - 将输入文本传递给模型 pipeline 进行处理。
- 模型返回按语义分割好的文本段落，脚本将其整理成列表返回。
- 优势: 理论上能更好地根据内容的语义关联性进行切分，而不是仅仅依赖标点符号。
- 劣势: 需要额外加载一个模型，增加了计算开销和依赖。
# edu_text_spliter/edu_model_text_spliter.py 源码
from langchain.text_splitter import CharacterTextSplitter
import re
from typing import List
from modelscope.pipelines import pipeline
# from configs import MODEL_PATH # Assume MODEL_PATH is defined elsewhere or replaced
# Placeholder for MODEL_PATH if configs.py is not available
MODEL_PATH = {
'segment_model': {
# Replace with the actual model name or path from ModelScope
'ali_model': 'damo/nlp_bert_document-segmentation_chinese-base'
}
}
class AliTextSplitter(CharacterTextSplitter):
def __init__(self, pdf: bool = False, **kwargs):
super().__init__(**kwargs)
self.pdf = pdf
# Initialize the pipeline here or ensure it's initialized before split_text is called
# Consider adding error handling for model loading
try:
self.pipeline = pipeline(
task="document-segmentation",
model=MODEL_PATH['segment_model']['ali_model'],
device="cpu" # Specify CPU device
)
except Exception as e:
print(f"Error initializing ModelScope pipeline: {e}")
self.pipeline = None
def split_text(self, text: str) -> List[str]:
if not self.pipeline:
print("ModelScope pipeline not initialized. Returning empty list.")
return []
# Preprocessing specific to PDF text if needed
if self.pdf:
text = re.sub(r"\n{3,}", r"\n", text)
# Replace multiple spaces with a single space
text = re.sub('\s+', " ", text)
# Consider removing single newlines carefully, might merge unrelated lines
# text = text.replace("\n", " ") # This might be too aggressive
text = re.sub("\n\n", "\n", text) # Keep paragraph breaks
try:
result = self.pipeline(documents=text)
# The default output format might be a single string with "\n\t" separators
sent_list = [segment.strip() for segment in result["text"].split("\n\t") if segment.strip()]
return sent_list
except Exception as e:
print(f"Error during ModelScope document segmentation: {e}")
# Fallback behavior: maybe split by paragraph or return the original text in a list
return text.split('\n\n') # Simple fallback
if __name__ == '__main__':
# Example usage requires modelscope and relevant model downloaded
# pip install "modelscope[nlp]" tensorflow torch -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html
sample_text = """移动端语音唤醒模型，检测关键词为“小云小云”。
模型主体为4层FSMN结构，使用CTC训练准则，参数量750K，适用于移动端设备运行。
模型输入为Fbank特征，输出为基于char建模的中文全集token预测，测试工具根据每一帧的预测数据进行后处理得到输入音频的实时检测结果。
模型训练采用“basetrain + finetune”的模式，basetrain过程使用大量内部移动端数据，在此基础上，使用1万条设备端录制安静场景“小云小云”数据进行微调，得到最终面向业务的模型。
后续用户可在basetrain模型基础上，使用其他关键词数据进行微调，得到新的语音唤醒模型，但暂时未开放模型finetune功能。"""
# Assuming MODEL_PATH is correctly configured and modelscope is installed
model_split = AliTextSplitter()
result = model_split.split_text(text=sample_text)
print(result)
# Expected output (example, actual output depends on the model):
# ['移动端语音唤醒模型，检测关键词为“小云小云”。', '模型主体为4层FSMN结构，使用CTC训练准则，参数量750K，适用于移动端设备运行。', '模型输入为Fbank特征，输出为基于char建模的中文全集token预测，测试工具根据每一帧的预测数据进行后处理得到输入音频的实时检测结果。', '模型训练采用“basetrain + finetune”的模式，basetrain过程使用大量内部移动端数据，在此基础上，使用1万条设备端录制安静场景“小云小云”数据进行微调，得到最终面向业务的模型。', '后续用户可在basetrain模型基础上，使用其他关键词数据进行微调，得到新的语音唤醒模型，但暂时未开放模型finetune功能。']
本章总结¶
本章我们详细介绍了 EduRAG 系统中用于处理原始文档和切分文本的核心工具。在文档解析方面，我们学习了 edu_document_loaders 目录下的各个加载器如何利用 PyMuPDF, python-docx, python-pptx 等库结合 RapidOCR（通过 edu_ocr.py 提供）来处理 PDF、DOCX、PPTX 及图像文件，有效提取文本和图片中的文字。在文本切分方面，我们探讨了 edu_text_spliter 目录提供的两种工具：ChineseRecursiveTextSplitter（针对中文优化的、基于规则的递归切分器）和 AliTextSplitter（利用 AI 模型进行语义切分）。理解这些底层工具的功能和实现是掌握 RAG 系统数据处理流程的关键。