AI Agent8-从 PDF 到知识:工程标准文档的自动化处理管道

工程标准以 PDF 形式存在——扫描版、文本版、双栏排版、含大量表格。如何把一本 200 页的标准 PDF 自动变成可检索的知识？本文拆解 PJL的文档处理管道：从 PDF 解析、切片策略到 ChromaDB入库，以及针对三种不同数据源的适配器设计。

一、管道的现实挑战

PJL的数据源分为三类，每类挑战不同：

PJL的处理管道需要适配这三种输入，最终产出结构化的知识切片。

二、管道架构

PDF/DOCX/XLSX 文件

│

▼

┌──────────────┐

│ 文件类型路由 │  ← 根据扩展名分发

└──┬───┬───┬──┘

│   │   │

▼   ▼   ▼

PDF  DOCX XLSX

│   │   │

▼   ▼   ▼

┌──────────────────┐

│ 文本提取器 │

│ ├ pdf_parser │  ← PyMuPDF / pdfplumber

│ ├ docx_parser │  ← python-docx

│ └ xlsx_parser │  ← openpyxl

└────────┬─────────┘

│

▼

┌──────────────────┐

│ 文本清洗 │

│ ├ 去水印/页眉页脚 │

│ ├ 表格规范化 │

│ └ 编码统一 UTF-8 │

└────────┬─────────┘

│

▼

┌──────────────────┐

│ 智能切片 │

│ ├ chunk_size=512 │

│ ├ overlap=64 │

│ └ 自然断点优先 │

└────────┬─────────┘

│

▼

┌──────────────────┐

│ 向量化 + 入库 │

│ ├ Ollama embed │  ← nomic-embed-text

│ ├ ChromaDB upsert │

│ └ SQLite 索引 │

└──────────────────┘

三、PDF 解析：PyMuPDF + pdfplumber 双引擎

PJL使用PyMuPDF（fitz）作为主力解析器，pdfplumber作为表格提取的补充：

import fitz # PyMuPDF

import pdfplumber

def parse_pdf(file_path: str) -> str:

"""双引擎 PDF 解析"""

text_parts = []

# 引擎1：PyMuPDF 提取全文

doc = fitz.open(file_path)

for page in doc:

page_text = page.get_text("text")

# 检测页眉页脚（位于页面顶部/底部 10% 区域）

page_height = page.rect.height

blocks = page.get_text("blocks")

main_text = []

for block in blocks:

y0 = block[1]

y1 = block[3]

if y0 > page_height * 0.1 and y1 < page_height * 0.9:

main_text.append(block[4])

text_parts.append("\n".join(main_text))

# 引擎2：pdfplumber 提取表格

with pdfplumber.open(file_path) as pdf:

for page in pdf.pages:

tables = page.extract_tables()

for table in tables:

if table:

text_parts.append(format_table(table))

return "\n\n".join(text_parts)

双引擎的理由：PyMuPDF速度快、内存小，适合提取正文；pdfplumber的表格提取质量远优于PyMuPDF（后者容易把行列切乱）。

四、数据源适配器：三种来源的统一抽象

PJL设计了适配器模式封装不同数据源：

class DataSourceBase(ABC):

"""数据源适配器基类"""

@abstractmethod

async def fetch(self) -> list[str]:

"""拉取：获取文件路径列表"""

pass

@abstractmethod

async def parse(self, file_path: str) -> str:

"""解析：从文件提取文本"""

pass

@abstractmethod

async def filter(self, text: str) -> bool:

"""过滤：判断是否包含有效工程内容"""

pass

def classify(self, text: str) -> str:

"""分类：识别文档类型（标准/规范/设计说明/设备手册）"""

keywords = {

'standard': ['GB ', 'JGJ ', 'CJJ ', '标准', '规范'],

'design_manual': ['设计手册', '选型指南'],

'construction': ['施工', '验收', '安装'],

}

for category, kws in keywords.items():

if any(kw in text[:500] for kw in kws):

return category

return 'unknown'

适配器一：local_watcher

监控 `watch_folder/` 目录，自动检测新增文件：

class LocalWatcher(DataSourceBase):

def __init__(self, watch_path: str, poll_interval: int = 30):

self.watch_path = watch_path

self.poll_interval = poll_interval

self.seen_files = set()

async def fetch(self) -> list[str]:

"""扫描目录，返回新文件列表"""

current_files = set(Path(self.watch_path).glob("*.*"))

new_files = [str(f) for f in current_files - self.seen_files

if f.suffix.lower() in {'.pdf', '.docx', '.xlsx', '.txt'}]

self.seen_files.update(current_files)

return new_files

适配器二：web_collector

从三个权威网站抓取最新标准发布信息：

class WebCollector(DataSourceBase):

SOURCES = {

"国标委": "https://std.samr.gov.cn/gb/",

"住建部": "https://www.mohurd.gov.cn/",

"铁路标准": "https://www.tdzl.org.cn/",

}

async def fetch(self) -> list[str]:

"""抓取最新标准列表（仅文件名+链接，不下载全文）"""

results = []

for name, url in self.SOURCES.items():

try:

async with httpx.AsyncClient(timeout=10) as client:

resp = await client.get(url)

# 解析 HTML，提取标准编号和标题

results.extend(self._parse_standard_list(resp.text, name))

except Exception as e:

logger.warning(f"抓取 {name} 失败: {e}")

return results

注：web_collector仅采集元数据（标准编号、名称、发布日期），不自动下载全文——这涉及版权问题。全文仍需用户手动上传到 `watch_folder/`。

适配器三：collection_scheduler

管理三个适配器的调度：

class CollectionScheduler:

"""采集调度器：协调多个数据源"""

def __init__(self):

self.sources = [

LocalWatcher("D:/AI_PJL/data/watch_folder"),

WebCollector(),

]

async def run_cycle(self):

for source in self.sources:

files = await source.fetch()

for file_path in files:

text = await source.parse(file_path)

if await source.filter(text):

category = source.classify(text)

await self._feed_evolution_pipeline(file_path, text, category)

五、切片策略的实战调优

PJL的切片策略经过三次迭代：

512 token 约等于 300-400 个中文字符——大致是一段标准条款的长度。64 token 的重叠保证了相邻条款的连贯性。

# 实际切片逻辑

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(

chunk_size=512,

chunk_overlap=64,

separators=["\n\n", "\n", "。", "；", "，", " ", ""],

length_function=lambda x: len(x)  # 中文字符计数

)

chunks = splitter.split_text(clean_text)

为什么用 `len()` 而不是 tiktoken？因为 nomic-embed-text的 tokenizer 不是 tiktoken 兼容的，而且对中文来说，字符数 ≈ token 数（每个中文字符约 1.2 token）。用字符计数足够。

六、踩坑记录

坑1：扫描版 PDF 的文本提取。`knowledge_base/incoming/` 中的重庆标准 PDF 是扫描版，PyMuPDF提取出空白文本。pdfplumber同样无法处理。解决方案：标记为"需 OCR 预处理"，在 UI 上提示用户使用 OCR 工具（如 Adobe Acrobat）先转换为可搜索 PDF。

坑2：表格提取的语义丢失。从 PDF 表格提取出的数据是纯文本，丢失了行列关系。例如提取出的"路幅宽度 40m 路幅宽度 32m"无法区分是两行还是两列。PJL在表格格式化阶段用Markdown表格语法重建结构。

坑3：SQLite写入性能。107004 个切片如果逐条 INSERT，耗时超过 30 分钟。解决方案：批量写入（每 500 条一组），并使用 WAL 模式：

sqlite.execute("PRAGMA journal_mode=WAL")

sqlite.execute("PRAGMA synchronous=NORMAL")

sqlite.executemany(

"INSERT INTO knowledge_chunks (doc_id, chunk_index, content) VALUES (?, ?, ?)",

chunk_batch

)

七、下篇预告

知识入库后，还需要工程计算来验证设计的物理可行性。PJL的计算引擎封装了 49 个函数，覆盖暖通/热力学/传热/流体/制冷/蓄能/热泵/能耗。下一篇将解读这 49 个函数的设计哲学——为什么不追求高精度数值方法，而是选择工程实用主义的简化模型。

*下一篇：《隧道空气动力学计算引擎：49个函数的设计哲学》*