基于ChatGPT的PDF文档智能解析与问答系统实战

1次阅读

共计 3339 个字符，预计需要花费 9 分钟才能阅读完成。

传统 PDF 处理方式通常面临以下挑战：

格式解析困难：PDF 本质上是呈现格式而非结构化数据，表格、数学公式等元素解析准确率低
语义理解缺失：基于关键词的检索无法理解查询意图，如 ” 预算增长趋势 ” 需要跨段落推理
处理效率低下：大规模文档处理时 CPU 密集型运算导致响应延迟
维护成本高：规则引擎需要持续更新以适应新文档类型

通过对比主流 Python 库的实验数据（测试 100 份含表格 / 公式的学术论文）：

库名称	文本提取准确率	表格保持度	处理速度(页 / 秒)
PyPDF2	82%	15%	120
pdfplumber	95%	78%	85
pdfminer.six	89%	65%	70

推荐方案：pdfplumber + 自定义后处理，对复杂文档可结合 Adobe Extract API

清洗阶段：
移除页眉页脚（正则匹配页码模式）
标准化 Unicode 字符（如转换弯引号为直引号）
修复断行错误（基于标点 + 大写字母的启发式规则）
分块策略：
滑动窗口法：512token 的窗口，128token 重叠（适合技术文档）
语义分块：使用 Sentence-BERT 检测话题转折点
表格特殊处理：转为 Markdown 格式保留结构

FAISS 本地部署：

import faiss
import numpy as np

d = 768  # 向量维度
index = faiss.IndexFlatIP(d)  # 内积相似度

# 添加向量示例
vectors = np.random.random((1000, d)).astype('float32')
index.add(vectors)  # 构建索引耗时约 200ms/ 万条

Pinecone 云服务 对比：
– 优点：自动扩展、支持混合搜索
– 缺点：网络延迟增加约 300ms，成本 $0.0004/query

提示工程模板：

def build_prompt(context, question):
    return f""" 基于以下文档内容回答问题，若信息不足请说明：文档片段：{context}

    问题：{question}

    要求：1. 优先使用文档信息
    2. 保持回答简洁
    3. 标注引用段落编号 """

import pdfplumber
from transformers import AutoTokenizer

class PDFProcessor:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")

    def extract_text(self, file_path):
        """PDF 文本提取与预处理"""
        full_text = []
        with pdfplumber.open(file_path) as pdf:
            for page in pdf.pages:
                text = page.extract_text()
                if text:
                    # 基础清洗
                    text = text.replace('\ufeff', '').strip()
                    full_text.append(text)
        return '\n'.join(full_text)

    def chunk_text(self, text, chunk_size=512, overlap=64):
        """基于 token 数的智能分块"""
        tokens = self.tokenizer.encode(text)
        chunks = []
        for i in range(0, len(tokens), chunk_size - overlap):
            chunk = tokens[i:i + chunk_size]
            chunks.append(self.tokenizer.decode(chunk))
        return chunks

from fastapi import FastAPI, UploadFile
import uvicorn

app = FastAPI()

@app.post("/process")
async def process_pdf(file: UploadFile):
    """处理上传的 PDF 文件"""
    if not file.filename.endswith('.pdf'):
        return {"error": "Invalid file type"}

    try:
        # 临时存储文件
        with open("temp.pdf", "wb") as f:
            f.write(await file.read())

        processor = PDFProcessor()
        text = processor.extract_text("temp.pdf")
        chunks = processor.chunk_text(text)

        return {"chunks": chunks}
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

并行解析：使用 multiprocessing 分页处理

from multiprocessing import Pool

def parse_page(page):
    return page.extract_text()

with pdfplumber.open(file_path) as pdf:
    with Pool(4) as p:
        texts = p.map(parse_page, pdf.pages)

增量索引：FAISS 支持 add_with_ids 实现增量更新

向量缓存：对相同文档块 MD5 哈希作 key
结果缓存：Redis 存储常见问答对，TTL 设置 1 小时
分级存储：热点文档保持内存，冷数据存磁盘

请求队列：Celery + RabbitMQ 实现任务分发

速率限制：

from fastapi import Request
from fastapi.middleware import Middleware
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/query")
@limiter.limit("5/minute")
async def query_endpoint(request: Request):
    ...

扫描件处理：组合使用 OCR（Tesseract）和文本校正
加密 PDF：预先检测并提示用户提供密码
特殊字体：提取时保留字体映射表

摘要生成：先用 GPT-3.5-turbo 生成章节摘要
层次检索：先匹配章节再定位具体段落
动态截断：根据重要性分数保留关键句

计费监控：

def calculate_cost(model, prompt_tokens, completion_tokens):
    """实时计算 OpenAI API 调用成本"""
    price_table = {"gpt-4": (0.03, 0.06),  # 输入 / 输出 每 1K tokens
        "gpt-3.5-turbo": (0.0015, 0.002)
    }
    in_price, out_price = price_table[model]
    return (prompt_tokens/1000)*in_price + (completion_tokens/1000)*out_price