Hcc的Blog

嵌入式 · AI · 折腾不止

0%

动手做 Rag(1) -- 文档处理系统的设计

发表于 分类于 ，

目标

  实现 PPT 文件的读取、清洗、分块、向量化。

写在前面的一些 Py 基础

抽象基类 (ABC) 和 @abstractmethod

  ABC 是一个合同模板，它自己不能直接使用，只能被子类继承。它用 @abstractmethod 标记的方法，子类必须实现，否则连对象都创建不了。目的是避免子类忘记写关键方法，把错误从运行时提前到代码启动时暴露。

1
2
3
4
5
from abc import ABC, abstractmethod
class MyBase(ABC):
    @abstractmethod
    def do_something(self):
        ...   # 只定义接口，不写具体内容

@classmethod 类方法与普通方法的区别

  实例方法(self)：绑定到实例，必须先创建对象才能调用，处理的是某一只具体的狗的数据。
  类方法(cls)：绑定到类本身，直接用 类名.方法() 调用，处理的是整个类共有的信息。

kwargs

  **kwargs 是 Python 里的万能收纳盒，专门存放多传的带名字的参数。其全称为 keyword arguments，拆开看：** 表示打包，kwargs 是变量名。举个例子：

1
2
3
4
def make_coffee(**kwargs):
    print(kwargs)

make_coffee(糖="少糖", 温度="热", 杯型="大")

  所有多余的参数都会被收进一个字典，函数不需要提前知道你有多少种参数。

  *args 负责收没名字的参数，打包成元组。
  而 **kwargs 负责收 有名字 的参数，打包成字典。

any 和 Any

  any 是 Python 的内置函数，用来判断一个可迭代的对象（列表、元组）有没有至少一个为真的元素：

1
2
print(any([False, False, True]))   # True（有 True）
print(any([0, 0, 0]))             # False（全是 0，视为假）

  Any 从 typing 模块导入，表示任意类型，表示这里可以接受任何类型，不要检查他：

1
2
3
from typing import Any

def __init__(self, file_path: str, **kwargs: Any):

@dataclass ———— 自动生成基础代码

  正常写一个类，需要写一大堆初始化属性：

1
2
3
4
class Document:
    def __init__(self, page_content, metadata):
        self.page_content = page_content
        self.metadata = metadata

  使用 @dataclass 装饰器，只需要列出属性，一些方法会自动创建，省时省力不易出错。

常用正则表达式

  re 是 Python 的正则表达式模块，主要用于按照某种规则查找、匹配、替换字符串中的内容。接下来我们学习几个最常用的函数：
  ①、re.sub() ———— 替换
  这是文本清洗中最常用的，可以实现字符的替换，其格式如下：

1
2
3
4
5
re.sub(
    pattern,
    replacement,
    string
)

  ②、re.search() ———— 查找
  判断字符串里面有没有某种模式。例如，判断文件是不是 PPT：

1
2
3
4
5
6
7
8
import re
filename="report.pptx"
result = re.search(
    r"\.pptx$",
    filename
)
if result:
    print("PPT文件")

  ③、re.findall() ———— 找出所有匹配内容
  例如提取 PPT 中的所有数字：

1
2
3
4
5
numbers = re.findall(
    r"\d+",
    text
)
print(numbers)

  以下是一些正则表达式常见的符号

  字符匹配

表达式 含义 例子
. 任意字符 a,b,1
\d 数字 0-9
\w 字母数字下划线 abc_1
\s 空白 空格换行
[abc] a/b/c其中一个 apple

  数量控制

符号 含义
* 0次或多次
+ 1次或多次
? 0次或1次
{n} n次
{n,m} n到m次

strip 方法

  Python strip() 方法用于移除字符串头尾指定的字符（默认为空格）或字符序列。注意：该方法只能删除开头或是结尾的字符，不能删除中间部分的字符。

enumerate 函数

  enumerate() 函数用于将一个可遍历的数据对象(如列表、元组或字符串)组合为一个索引序列，同时列出数据和数据下标，一般用在 for 循环当中。

文档处理基类

  我们先创建 FileProcess 文件夹，新建 file_process_base.py，创建文件处理基类。后续其他的文件处理类都基于此类创建。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any

@dataclass
class Document:
    page_content: str
    metadata: dict = field(default_factory=dict)

class BaseLoader(ABC):
    """
    文档加载器抽象基类
    """
    def __init__(self, file_path: str, **kwargs: Any):
        self.file_path = file_path
        self.kwargs = kwargs
    
    @abstractmethod
    def load(self) -> list[Document]:
        """加载文档并返回 Document 列表"""


class BaseCleaner(ABC):
    """
    文档清洗器抽象基类
    """
    @abstractmethod
    def clean(self, documents: list[Document]) -> list[Document]:
        """清洗文本"""

class BaseSplitter(ABC):
    """
    文档清洗器抽象基类
    """
    @abstractmethod
    def split(self, documents: list[Document]) -> list[Document]:
        """文本分割"""

class BaseProcessor:
    """
    文档处理基类
    """
    def __init__(self, 
                 loader: BaseLoader, 
                 cleaner: BaseCleaner, 
                 splitter: BaseSplitter
                ):
        self.loader = loader
        self.cleaner = cleaner
        self.splitter = splitter

    def run(self):
        return

  创建加载、清洗、分块基类，然后再创建处理大类。这样就可以对每一种文件实行不同的策略。

PPTx_process

  创建 PPTx_process.py 文件，继承三种基类。接下来我们依次实现三种方法。

加载

  我先尝试使用 markitdown 工具将 PPT 转换为 MD 文件，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
from FileProcess.file_process_base import BaseLoader, BaseCleaner, BaseSplitter
from FileProcess.file_process_base import Document
from markitdown import MarkItDown

class PPTx_Loader(BaseLoader):
    def load(self, file_path: str) -> list[Document]:
        """加载文档并返回 Document 列表"""
        md = MarkItDown(enable_plugins=False)
        result = md.convert(file_path)
        doc = Document(page_content=result.text_content, 
                       metadata={"file": f"{file_path}"})
        return [doc]

  但是这样存在一个问题，markitdown 获取的信息是一整个文档，并没有分页。询问 GPT 后，其建议我们使用 python-pptx 来获取信息，具体代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from FileProcess.file_process_base import BaseLoader, BaseCleaner, BaseSplitter
from FileProcess.file_process_base import Document
from pptx import Presentation

class PPTx_Loader(BaseLoader):
    def load(self, file_path: str) -> list[Document]:
        """加载文档并返回 Document 列表"""
        docs = []

        # 使用 python-pptx 打开ppt文件
        #
        # Presentation会读取:
        # - 幻灯片数量
        # - 每一页的shape
        # - 文本框内容
        #
        prs = Presentation(file_path)

        # 遍历 PPT 中每一页
        # prs.slides:
        #   是一个幻灯片集合
        # enumerate:
        #   同时获取:
        #       index -> 页码索引
        #       slide -> 当前页对象
        # 例如:
        # 第1页:
        # index = 0
        # slide = 第一页对象
        # 第2页:
        # index = 1
        for index, slide in enumerate(prs.slides):
            texts = []

            # 遍历所有元素，此处只保留文本
            for shape in slide.shapes:
                if hasattr(shape, "text"):
                    texts.append(shape.text)

            # 合并所有文档
            content = "\n".join(texts)

            doc = Document(page_content=content,
                        metadata={
                            "file":file_path,
                            "slide":index+1,
                            "type":"pptx"
                        })
            docs.append(doc)
        return docs

清洗

  接下来我们清洗文档。有很多清洗的步骤是可以通用的，因此我们直接将其定义在基类内。这里我们首先实现了三种方法：分别是去除空格、去除特殊符号，去除短文本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def clean_whitespace(text: str):
    text = re.sub(r'\s+', ' ', text)
    text = text.strip()
    return text

def clean_symbol(text: str):
        text=text.replace("•","")
        text=text.replace("▪","")
        return text

def remove_short_text(text):
    if len(text)<5:
        return ""
    return text

  接下来我们实现一个更高级的用法，去除多余的重复内容。企业 PPT 经常有重复的公司信息等内容，会干扰 Embedding 效果。
  我们先统计每一行出现的次数

1
2
3
4
5
6
7
8
9
def remove_repeat_lines(self, docs: list[Document], threshold=0.5):
    from collections import Counter
    counter = Counter()
    for doc in docs:
        lines = doc.page_content.split("\n")
        for line in lines:
            line = line.strip()
            if line:
                counter[line] += 1

  接下来就是对于出现次数大于阈值的词进行删除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
total = len(docs)
# 无序不重复元素集
remove_lines=set()

for line,count in counter.items():
    if count/total > threshold:
        remove_lines.add(line)

for doc in docs:
    lines = []
    for line in doc.page_content.split("\n"):
        if line.strip not in remove_lines:
            lines.append(line)
    doc.page_content = "\n".join(lines)

return docs

  完成后，我们写对文档的清洗代码。

1
2
3
4
5
6
7
8
class PPTxCleaner(BaseCleaner):
    def clean(self, docs: list[Document]) -> list[Document]:
        for doc in docs:
            doc.page_content = self.clean_whitespace(doc.page_content)
            doc.page_content = self.clean_symbol(doc.page_content)
            doc.page_content = self.remove_short_text(doc.page_content)
        docs = self.remove_repeat_lines(docs)
        return docs

分块

  对于 PPT 分块，我们应该以页为整体，优先整理一页内的内容，如果一页内的内容过长，再通过滑动窗口分块分为不同的 chunk。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class PPTxSplitter(BaseSplitter):

    def __init__(
        self,
        chunk_size=256,
        chunk_overlap=50
    ):
        """
        参数:

        chunk_size:
            每个chunk最大字符数

        chunk_overlap:
            chunk之间重叠字符数
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap


    def _split_text(self, text: str):
        """简单滑动窗口切分"""
        chunks = []
        start = 0
        length = len(text)
        while start < length:
            end = start+self.chunk_size
            chunk = text[start:end]
            chunks.append(chunk)
            start = end - self.chunk_overlap
        return chunks

    def split(self, 
              docs: list[Document], 
              ) -> list[Chunk]:
        chunks = []

        for doc in docs:
            text = doc.page_content
            if len(text) <= self.chunk_size:
                chunks.append(
                    Chunk(
                        chunk_content=doc.page_content,
                        metadata=doc.metadata
                    )
                )

            # PPT 优先一页内容
            elif len(text) <= 4*self.chunk_size:
                chunks.append(
                    Chunk(
                        chunk_content=doc.page_content,
                        metadata=doc.metadata
                    )
                )

            # 超长文本再切割
            else:
                sub_chunks = self._split_text(text)

                for index, chunk in enumerate(sub_chunks):
                    new_chunk = Chunk(
                        chunk_content=chunk,
                        metadata={
                            **doc.metadata,
                            # 标记这是第几个chunk
                            "chunk_id":index
                        }
                    )
                    chunks.append(new_chunk)
        return chunks

  So Good！我们现在已经完成了加载、清洗、分块等一系列操作。接下来定义 Processor 基类。

1
2
3
4
5
6
7
class PPTxProcessor(BaseProcessor):
    def __init__(self):
        super().__init__(
            loader=PPTxLoader(),
            cleaner=PPTxCleaner(),
            splitter=PPTxSplitter()
        )

  然后只需要在 main.py 中定义此基类，运行 run 方法，就可以成功进行处理并输出 Chunks。

Embedding

  新建 embedding 文件夹，创建 base_embedding.py 文件，定义基类。

1
2
3
4
5
6
7
8
from abc import ABC, abstractmethod
from FileProcess.file_process_base import Chunk

class BaseEmbedding(ABC):

    @abstractmethod
    def embed(self, chunks:list[Chunk]):
        pass

  接下来新建一个文件，定义 TongyiEmbedding 类，继承基类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class TongyiEmbedding(BaseEmbedding):
    def __init__(self, model):
        self.client = OpenAI(api_key=api_key,
                             base_url=base_url,)
        self.model = model

    def embed(self, chunks:list[Chunk]) -> list[VectorChunk]:
        texts=[]
        for chunk in chunks:
            texts.append(chunk.chunk_content)
        response = self.client.embeddings.create(
            model = self.model,
            input=texts
        )

        vectors:list[VectorChunk] = []
        for item, chunk in zip(response.data, chunks):
            vectors.append(
                VectorChunk(chunk_content=chunk.chunk_content,
                            vector=item.embedding,
                            metadata=chunk.metadata)
            )
        return vectors

  So Good!那么到现在为止，我们的向量化也已经完成！那么接下来，我们就应该实现入库了！

Chroma 入库

  作为一个初代版本，我们先使用 Chroma 作为数据库，它小、轻量，适合个人 Demo。

  同样的 为了方便后续的代码发展，我们新建 VectorStore 文件夹，新建 base_store.py，写一个基类。

1
2
3
4
5
6
7
8
9
10
11
from abc import ABC, abstractmethod

class BaseVectorStore(ABC):

    @abstractmethod
    def add(self, vectors):
        pass

    @abstractmethod
    def search(self,query_vector):
        pass

  接下来我们实现 Chromadb 的类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class ChromaVectorStore(BaseVectorStore):
    def __init__(self):
        # 使用持久化本地 Chroma 数据库
        self.client = chromadb.PersistentClient(
            path="./Chroma_db"
        )

        # collection 为一张向量表
        self.collection = (
            self.client
            .get_or_create_collection(
                "company_docs"
            )
        )

  Chroma 初始化已经写好，接下来就是录入和查询函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def add(self, chunks:list[VectorChunk]):
    ids = []
    docs = []
    embeddings = []
    metadatas = []

    for index, chunk in enumerate(chunks):

        ids.append(f"chunk_{index}")
        docs.append(chunk.chunk_content)
        embeddings.append(chunk.vector)
        metadatas.append(chunk.metadata)

    self.collection.add(
        ids = ids,
        documents=docs,
        metadatas=metadatas,
        embeddings=embeddings
    )

def search(self, vector:list[float], top_k:int):
    result = self.collection.query(
        query_embeddings=[
            vector
        ],
        n_results=top_k
    )

  至此，所有的基础功能已经完成。我们应该写处理文档、模型回复的工作流了！在下一篇日志中我会开始！