简介:PDF文本提取与文档结构化是构建企业知识库、法规条文检索和问答机器人的基础环节。其原理是先判断文本层与扫描件,利用版面裁剪、锚点正则还原章节与“第X条”,再把条号、主题、正文等字段化入库,配合中文分词、FTS5 trigram、BM25与向量召回完成关键词与语义的混合检索。这样既能按条号精确定位,也能按主题聚合和口语化问句召回,技术价值在于提升检索准确率、支持增量更新与版本diff。以《华为基本法》PDF为例,pdfplumber解析条文、SQLite FTS5建索引、jieba术语词典与RRF融合,正适合制度文档、培训题库和企业文化知识库等场景。
1. 从一份《华为基本法》PDF 说起:条文型文档为什么必须先结构化
拿到《华为基本法.pdf》的人,动机通常很朴素——想引用其中某一条。比如「压强原则」到底写在第几条、第十条讲的是什么、股权分配的依据落在哪一段。真去 Ctrl+F 大概率会失望:PDF 里「第 二 十 三 条」可能被排版拆成两行,「压强原则」四个字中间夹着引号或空格,双栏页面导出的文本顺序还会左右串行,搜出来的结果对不上号。条文型文档的价值从来不在「能不能读」,而在「能不能按条号定位、按主题聚合、按语义召回」。把章标题、条号、括号里的主题词(核心价值观、经营模式、中间试验)拆成独立字段之后,做内部培训题库、企业文化知识库,还是接一个问答机器人,用的都是同一份数据。这条链路可以拆成四步:还原文本层、按锚点切条文、字段化入库并校验、做关键词与语义的混合检索。
2. 用 pdfplumber 还原章节层级与「第X条」锚点
2.1 先确认 PDF 有没有可用文本层
动手之前别急着写解析逻辑,先花两分钟判断这份文件是原生导出还是扫描件。判断依据非常直接:page.chars的长度和page.images的分布。
import pdfplumber with pdfplumber.open("华为基本法.pdf") as pdf: page = pdf.pages[2] print("chars:", len(page.chars)) # 0 或个位数,基本可以判定无文本层 print("images:", len(page.images)) # 整页一张大图,是扫描件的强信号 print(repr(page.extract_text()[:200]))如果chars为 0、images是一张覆盖整页的大图,那就先走 OCR 补一层文本(ocrmypdf 做整页识别、或按页送 PaddleOCR 取行坐标),再回到下面的流程。反过来,chars有几千个,说明文字是矢量的,直接解析文本层,别浪费 OCR 的时间——OCR 对「第X条」这种编号的误识别率(把「条」认成「奈」、「十三」认成「1三」)会让后面所有正则都失效。
2.2 逐页抽行,合并被排版打断的条文
原生 PDF 的文本抽取有两个绕不开的坑:一是双栏版面,extract_text默认按 y 坐标从上往下扫,会把左右两栏的行交错拼在一起;二是字符间距,中文导出时很多字之间带零点几磅的间隙,容差设大了会把两行并一行,设小了会把一条正文拆成七八段。
import pdfplumber, re def iter_lines(path): with pdfplumber.open(path) as pdf: for pno, page in enumerate(pdf.pages, start=1): w, h = page.width, page.height # 双栏版面:按中线切两半,左栏读完再读右栏,避免左右串行 halves = [page.crop((0, 0, w / 2, h)), page.crop((w / 2, 0, w, h))] for half in halves: text = half.extract_text(x_tolerance=1.5, y_tolerance=3) or "" for line in text.splitlines(): line = re.sub(r"\s+", "", line) # 中文正文里的空格多为排版噪声 if line: yield pno, line关键参数的实际影响,可以对照下面这张表来调:
| 参数 | 常用默认 | 建议取值 | 调错时的现象 |
|---|---|---|---|
x_tolerance | 3 | 1.5~2 | 偏大:同行词被粘成乱序长串;偏小:一行被拆成多行 |
y_tolerance | 3 | 3~5 | 偏大:两条正文并成一行的末尾;偏小:同一条被切成多段 |
layout | False | 双栏时保持 False | 开启后只做等宽重排,栏间顺序反而更乱 |
| 裁剪方式 | 整页 | 按中线 crop | 不裁剪时,第二栏的行会插进第一栏条文中间 |
合并逻辑本身很简单:只要当前行不以任何锚点开头,就把它接到上一行末尾。难点全在容差和版面裁剪上。
2.3 用锚点正则切出章、节、条与主题词
《华为基本法》的行首结构相当规整,一共四类锚点:「第X章」章标题、「一、」节标题、「第X条」条文起头、以及紧跟在条号后面的全角括号主题词,例如「(核心价值观)」「(经营模式)」「(中间试验)」。中文数字要转成整数,才能做后续的连续性校验。
import re CN = "一二三四五六七八九十百零" RE_CHAPTER = re.compile(rf"^第([{CN}]+)章(.*)$") RE_SECTION = re.compile(r"^([一二三四五六七八九十]+)、(.*)$") RE_ARTICLE = re.compile(rf"^第([{CN}]+)条(.*)$") RE_TOPIC = re.compile(r"^(([^)]{1,10}))") DIGIT = {c: i for i, c in enumerate("零一二三四五六七八九")} def cn2int(s: str) -> int: """把「二十三」「十」「一百零五」这类写法转成 int""" if s == "十": return 10 total, unit = 0, 1 for ch in reversed(s): if ch == "十": unit = 10 elif ch == "百": unit = 100 else: total += DIGIT[ch] * unit return total有了转换函数,状态机就只剩十几行:维护chapter、section两个上下文变量,遇到条文锚点就新开一条记录,否则把行追加到当前条文正文。
def parse(lines): records, chapter, section = [], None, None for pno, line in lines: m = RE_CHAPTER.match(line) if m: chapter, section = f"第{cn2int(m.group(1))}章 {m.group(2)}".strip(), None continue m = RE_SECTION.match(line) if m: section = f"{m.group(1)}、{m.group(2)}" continue m = RE_ARTICLE.match(line) if m: rest = m.group(3) t = RE_TOPIC.match(rest) records.append({ "art_no": cn2int(m.group(1)), "chapter": chapter, "section": section, "topic": t.group(1) if t else None, "body": RE_TOPIC.sub("", rest, count=1), "page": pno, }) continue if records and line: records[-1]["body"] += line # 续行并入上一条 return records这段代码有两个值得说的取舍。主题词用RE_TOPIC单独摘出来存字段,而不是留在正文里——因为「核心价值观」「经营模式」这些词既是最常见的检索入口,也是后面做主题聚合时的天然标签。续行合并只认「上一条记录」,不做跨页判断;如果某条正文被分页截断,靠页码字段回原文核对即可,不值得为此引入复杂的跨页状态机。
跑完这一步,正常应该得到 28 条记录(第一章七条、第二章二十一条),章号分布是 1 到 2。数量不对就回去看第二节的容差参数,八成是合并粒度出了问题。
3. 条文入库:SQLite 表结构、中文分词与一致性校验
3.1 字段怎么设计,为什么不用一张大文本表
切分结果落地成什么形态,决定了后面能不能做局部更新和精确定位。一张「id + 全文」的表看着省事,但改一条条文就得整表重写,而且没法按章、按主题过滤。字段拆开的收益在第一次做增量更新时就会体现出来。
| 字段 | 类型 | 来源 | 用途 |
|---|---|---|---|
art_no | INTEGER | 条号中文转整数 | 主键,支持「第几条」精确查询与连续性校验 |
chapter | TEXT | 章标题锚点 | 按章筛选、做目录树 |
section | TEXT | 节标题锚点 | 章下的二级分组 |
topic | TEXT | 条号后的全角括号 | 主题聚合、高频标签统计 |
body | TEXT | 条文正文 | 检索主体、向量化的输入 |
page | INTEGER | 抽取时的页码 | 回原文核对、定位 PDF 页 |
sha1 | TEXT | 正文归一化后哈希 | 版本 diff,判断某条是否被改过 |
3.2 建表与 FTS5 全文索引
SQLite 自带 FTS5,做条文检索足够用,但中文分词必须显式指定,否则默认的unicode61会把一整串连续汉字当成一个 token——搜「股权」命中不了「股权结构要保持动态合理性」。
PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS article ( art_no INTEGER PRIMARY KEY, chapter TEXT NOT NULL, section TEXT, topic TEXT, body TEXT NOT NULL, page INTEGER, sha1 TEXT ); -- 中文场景用 trigram:按三字符滑窗建索引,子串匹配天然可用 CREATE VIRTUAL TABLE IF NOT EXISTS article_fts USING fts5(topic, body, content='article', content_rowid='art_no', tokenize='trigram');trigram分词器需要 SQLite 3.34 及以上版本,先SELECT sqlite_version();确认。它有个必须知道的边界:查询串短于 3 个字符时命中率会明显下降,两字词像「股权」「质量」「利润」这类,最好走第 4 章的 BM25 通道兜底。反过来,trigram对「终生效能费用比」这种长术语的精度很高,因为它是纯子串匹配,不受分词歧义影响。
3.3 编号连续性与脏字符校验
入库之后先跑两条校验 SQL,比人工翻 PDF 快得多。第一条找断号,用的是自连接找「下一个数字没人占」的位置:
SELECT a.art_no + 1 AS missing_no FROM article a LEFT JOIN article b ON b.art_no = a.art_no + 1 WHERE b.art_no IS NULL AND a.art_no < (SELECT MAX(art_no) FROM article);第二条找过短的正文,用来抓「正文被误判成锚点行」或「续行没并进来」的情况:
SELECT art_no, topic, length(body) AS n FROM article WHERE length(body) < 20 ORDER BY n;原文里还有一类需要统一清理的噪声:第二十八条出现「大油田、大森林、大煤矿,, 」这样的逗号残留,以及「情操,, ,也包含」这类半角标点混排。一次性替换掉比在正则里逐条处理更干净:
UPDATE article SET body = replace(replace(body, ',,', '……'), ' ', ''); UPDATE article SET body = replace(body, ' ,', ',');注意:标点替换会改变
body的哈希值,必须在计算sha1之前执行,否则每次重跑都会把所有条文标成「已修改」。
校验通过后的数据就是一个可查询的条文库,SELECT body FROM article WHERE art_no = 23;能直接拿到「压强原则」那一条的原文,页数字段还能反查出处。
4. jieba 术语词典 + BM25 与向量召回的混合检索
4.1 先补一份领域词典,再谈分词质量
通用分词器在管理类文本上表现一般:「知识资本化」会被切成「知识/资本/化」,「终生效能费用比」会被切成四五个碎片,BM25 的倒排索引跟着一起退化。低成本的做法是在分词前把文档里反复出现的术语加进自定义词典。
import jieba TERMS = ["知识资本化", "终生效能费用比", "压强原则", "中间试验", "价值分配", "按劳分配", "按资分配", "核心层", "战略联盟", "宽频带、高振幅", "窄频带、高振幅", "事业可持续成长"] for w in TERMS: jieba.add_word(w, freq=1000) # freq 500~2000 之间比较稳 STOP = set("我们 你们 他们 的 了 是 在 和 与 就 都 而 及 或 等 这 那 有 为 以 对 不 要".split()) def tokenize(text: str): return [w for w in jieba.lcut(text) if len(w) > 1 and w not in STOP and not w.isdigit()]freq这个参数容易踩坑:给得太低,词照样被切碎;给到几万,分词器会为了保住这个词去破坏周边边界,把「不迁就有功的员工」拆得很奇怪。词典里的词必须是从条文里实打实抓出来的,不要凭印象加通用词。
4.2 BM25 起底,向量补语义,用 RRF 融合
28 条条文的数据量下,rank_bm25这种纯 Python 实现完全够用,不需要上 Elasticsearch。BM25 的优势是术语命中准,短查询「股权分配的依据」能精确落到第十九条;弱点是问句换了说法就抓瞎——「怎么防止公司变大以后变僵化」这种口语化提问,BM25 的词项一个都对不上。
from rank_bm25 import BM25Okapi corpus = [tokenize(r["body"]) for r in rows] bm25 = BM25Okapi(corpus, k1=1.5, b=0.75) def bm25_search(query, topk=20): scores = bm25.get_scores(tokenize(query)) return sorted(range(len(scores)), key=lambda i: -scores[i])[:topk]k1控制词频饱和速度,b控制长度归一化强度。教材默认值k1=1.5, b=0.75适合长短差异大的语料;如果每条正文长度都在几十字上下、差异不大,把b调到 0.3~0.5 更合适,否则短条文会被过度补偿、排名虚高。
两路召回的结果用 Reciprocal Rank Fusion 合并,不需要调权重,只用排名:
def rrf(rank_lists, k=60): fused = {} for lst in rank_lists: for rank, idx in enumerate(lst): fused[idx] = fused.get(idx, 0) + 1 / (k + rank + 1) return sorted(fused, key=fused.get, reverse=True)k=60是 RRF 论文里的常用平滑项,作用是把头部排名的优势压一压。设成 1 会让第一名几乎决定结果,设成几百则退化成简单计数。
不同查询该走哪条通道,可以按下面的规则分流:
| 查询形态 | 例子 | 首选通道 | 典型失效 |
|---|---|---|---|
| 条号 | 第二十三条、第 10 条 | SQL 直接查art_no | 中文数字没归一化,cn2int漏写「百」 |
| 术语/短语 | 终生效能费用比 | FTS5 trigram + BM25 | 查询串短于 3 字符 |
| 主题标签 | 讲了哪些经营政策 | GROUP BY topic | topic为空,正文里没括号主题词 |
| 口语长问句 | 怎么避免高速增长带来组织脆弱 | 向量召回 + RRF | 无标注数据,无法评估好坏 |
4.3 召回质量怎么验证
没有评估集,调参就是玄学。花半小时做 20 条「问题 → 正确条号」的标注,计算 Recall@3,收益远大于继续调k1。比如「高层为什么必须警惕长期高速增长」对应第十五条,「进入新领域要看什么」对应第十二条。评测脚本跑一遍,命中率低于 70% 就先去补词典和topic字段,而不是急着换更大的向量模型。
5. 条文版本化:用 sha1 做增量 diff 与快照
《华为基本法》这类文件会在润色、修订、重排中反复出新版,真正需要知道的是「哪一条被改过」,而不是整份文件重导一遍。做法是在article.sha1里存正文归一化后的摘要,重跑解析流程时对新旧两版做集合运算。
import hashlib, json def digest(text: str) -> str: norm = "".join(text.split()) # 先抹掉全部空白,规避排版差异 norm = norm.replace("(", "(").replace(")", ")") # 全半角统一 return hashlib.sha1(norm.encode("utf-8")).hexdigest()[:12] def diff(old_rows, new_rows): old = {r["art_no"]: r["sha1"] for r in old_rows} new = {r["art_no"]: r["sha1"] for r in new_rows} return { "added": sorted(set(new) - set(old)), "removed": sorted(set(old) - set(new)), "changed": sorted(k for k in set(old) & set(new) if old[k] != new[k]), } print(json.dumps(diff(prev_rows, curr_rows), ensure_ascii=False))归一化这一步是整个方案的生死线。不做空白和全半角处理,PDF 换一次导出引擎、行距调整一磅,所有条文的哈希都会变,changed列表直接刷满,等于没做 diff。
| 变更类型 | 判据 | 处理动作 |
|---|---|---|
| 新增条文 | art_no只在 new 中出现 | 插库,重建该条的 FTS 索引与向量 |
| 删除条文 | art_no只在 old 中出现 | 软删除,保留art_no防止历史引用断链 |
| 正文修改 | art_no相同但sha1不同 | 更新body与sha1,触发向量重算 |
| 纯排版变化 | sha1不变 | 什么都不做,只更新page |
最后一个技巧:把page字段和sha1一起维护,diff 报出「第二十三条已修改」时,直接翻到记录的 PDF 页码比对原文,而不是全文搜一遍。修完数据重跑一次连续性校验和短正文校验,两条 SQL 都干净了再对外发版。
本文还有配套的精品资源,点击获取