RAG 从入门到实战(三):PDF 解析与文本分块完全指南

# 基础 PDF 解析pip install pypdf==3.17.0# 高级 PDF 解析（推荐）pip install pdfplumber==0.10.3# 表格提取pip install tabula-py==2.8.2# OCR（可选，处理扫描件）pip install paddleocr==2.7.3

# -*- coding: utf-8 -*-"""PDF 解析基础教程运行：python pdf-parser.py"""from pypdf import PdfReader# ========== 1. 读取 PDF ==========pdf_path = "sample.pdf"  # 替换为你的 PDF 文件reader = PdfReader(pdf_path)print(f"📄 PDF 页数：{len(reader.pages)}")# ========== 2. 提取文本 ==========full_text = ""for i, page in enumerate(reader.pages, 1):    text = page.extract_text()    full_text += text    print(f"\n--- 第{i}页 ---")    print(text[:500])  # 只显示前 500 字符# ========== 3. 保存文本 ==========with open("output.txt", "w", encoding="utf-8") as f:    f.write(full_text)print(f"\n✅ 文本已保存到 output.txt")print(f"总字符数：{len(full_text)}")

# -*- coding: utf-8 -*-"""pdfplumber 高级 PDF 解析运行：python advanced-pdf-parser.py"""import pdfplumberimport pandas as pdpdf_path = "sample.pdf"# ========== 1. 提取文本（保留格式） ==========with pdfplumber.open(pdf_path) as pdf:    for i, page in enumerate(pdf.pages, 1):        text = page.extract_text()        print(f"\n=== 第{i}页 ===")        print(text)# ========== 2. 提取表格 ==========tables_data = []with pdfplumber.open(pdf_path) as pdf:    for i, page in enumerate(pdf.pages, 1):        tables = page.extract_tables()        for j, table in enumerate(tables):            print(f"\n📊 第{i}页 - 表格{j+1}")            # 转换为 DataFrame            df = pd.DataFrame(table)            print(df.head())            # 保存为 CSV            df.to_csv(f"table_page{i}_table{j+1}.csv", index=False)# ========== 3. 提取特定区域文本 ==========with pdfplumber.open(pdf_path) as pdf:    page = pdf.pages[0]    # 定义区域（左，上，右，下）    crop_box = (50, 100, 500, 300)    cropped = page.crop(crop_box)    text = cropped.extract_text()    print("\n🎯 指定区域文本:")    print(text)# ========== 4. 提取元数据 ==========with pdfplumber.open(pdf_path) as pdf:    metadata = pdf.metadata    print("\n📋 PDF 元数据:")    for key, value in metadata.items():        print(f"{key}: {value}")

# -*- coding: utf-8 -*-"""中文 PDF 解析（处理编码问题）运行：python chinese-pdf-parser.py"""from pypdf import PdfReaderimport redef clean_text(text):    """清洗文本"""    if not text:        return ""    # 移除多余空白    text = re.sub(r'\s+', ' ', text)    # 移除特殊字符    text = re.sub(r'[^\w\s\u4e00-\u9fff，。！？；：""''、]', '', text)    return text.strip()def parse_chinese_pdf(pdf_path):    """解析中文 PDF"""    reader = PdfReader(pdf_path)    pages_text = []    for page in reader.pages:        text = page.extract_text()        cleaned = clean_text(text)        if cleaned:            pages_text.append(cleaned)    return pages_text# 使用示例if __name__ == "__main__":    pdf_path = "chinese_document.pdf"    pages = parse_chinese_pdf(pdf_path)    print(f"✅ 成功解析 {len(pages)} 页")    # 合并所有页面    full_text = "\n".join(pages)    print(f"总字符数：{len(full_text)}")    # 保存    with open("chinese_output.txt", "w", encoding="utf-8") as f:        f.write(full_text)

# -*- coding: utf-8 -*-"""文本分块器对比运行：python text-splitter-demo.py"""from langchain.text_splitter import (    RecursiveCharacterTextSplitter,    CharacterTextSplitter,    TokenTextSplitter)# 示例文本sample_text = """人工智能（Artificial Intelligence，简称 AI）是计算机科学的一个分支，它企图了解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器。该领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。人工智能从诞生以来，理论和技术日益成熟，应用领域也不断扩大。可以设想，未来人工智能带来的科技产品，将会是人类智慧的"容器"。"""# ========== 1. 字符分块器 ==========print("="*60)print("字符分块器 (CharacterTextSplitter)")print("="*60)char_splitter = CharacterTextSplitter(    separator="\n",    chunk_size=50,    chunk_overlap=10)char_chunks = char_splitter.split_text(sample_text)for i, chunk in enumerate(char_chunks, 1):    print(f"\n块{i} ({len(chunk)}字符):")    print(chunk)# ========== 2. 递归字符分块器（推荐） ==========print("\n" + "="*60)print("递归字符分块器 (RecursiveCharacterTextSplitter)")print("="*60)rec_splitter = RecursiveCharacterTextSplitter(    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],    chunk_size=50,    chunk_overlap=10,    length_function=len)rec_chunks = rec_splitter.split_text(sample_text)for i, chunk in enumerate(rec_chunks, 1):    print(f"\n块{i} ({len(chunk)}字符):")    print(chunk)# ========== 3. Token 分块器 ==========print("\n" + "="*60)print("Token 分块器 (TokenTextSplitter)")print("="*60)token_splitter = TokenTextSplitter(    chunk_size=20,    chunk_overlap=5)token_chunks = token_splitter.split_text(sample_text)for i, chunk in enumerate(token_chunks, 1):    print(f"\n块{i}:")    print(chunk)

# -*- coding: utf-8 -*-"""中文优化的文本分块器运行：python chinese-text-splitter.py"""import refrom langchain.text_splitter import RecursiveCharacterTextSplitterclass ChineseTextSplitter(RecursiveCharacterTextSplitter):    """中文优化的文本分块器"""    def __init__(self, **kwargs):        # 中文分隔符优先级        separators = [            "\n\n",      # 段落            "\n",        # 换行            "。",        # 句号            "！",        # 感叹号            "？",        # 问号            "；",        # 分号            "，",        # 逗号            "、",        # 顿号            " ",         # 空格            ""           # 最后按字符        ]        super().__init__(            separators=separators,            chunk_size=kwargs.get('chunk_size', 200),            chunk_overlap=kwargs.get('chunk_overlap', 20),            length_function=len        )# 使用示例if __name__ == "__main__":    text = """    人工智能是计算机科学的一个分支。它研究如何使计算机能够模拟人类智能。    机器学习是人工智能的核心技术之一。深度学习是机器学习的重要分支。    自然语言处理让计算机能够理解和生成人类语言。计算机视觉让计算机能够"看"懂图像。    """    splitter = ChineseTextSplitter(chunk_size=50, chunk_overlap=10)    chunks = splitter.split_text(text)    print(f"分块数量：{len(chunks)}\n")    for i, chunk in enumerate(chunks, 1):        print(f"块{i}: {chunk}")

# 中文文档splitter = RecursiveCharacterTextSplitter(    chunk_size=200,      # 200 字符    chunk_overlap=20,    # 10% 重叠    separators=["\n\n", "\n", "。", "！", "？", "，", ""])

# -*- coding: utf-8 -*-"""Embedding 模型使用示例运行：python embedding-demo.py"""from sentence_transformers import SentenceTransformerimport numpy as npfrom sklearn.metrics.pairwise import cosine_similarity# ========== 1. 加载模型 ==========print("正在加载模型...")model = SentenceTransformer('shibing624/text2vec-base-chinese')print("✅ 模型加载完成")# ========== 2. 编码文本 ==========sentences = [    "人工智能是未来",    "AI 技术发展前景广阔",    "今天天气不错",    "机器学习需要大量数据"]print("\n正在编码...")embeddings = model.encode(sentences)print(f"✅ 编码完成，向量维度：{embeddings.shape}")# ========== 3. 计算相似度 ==========print("\n📊 文本相似度矩阵:")similarity = cosine_similarity(embeddings)for i, s1 in enumerate(sentences):    print(f"\n'{s1}' 与其他文本的相似度:")    for j, s2 in enumerate(sentences):        if i != j:            print(f"  - '{s2}': {similarity[i][j]:.4f}")# ========== 4. 保存和加载 ==========import pickle# 保存with open("embeddings.pkl", "wb") as f:    pickle.dump(embeddings, f)print("\n✅ Embedding 已保存")# 加载with open("embeddings.pkl", "rb") as f:    loaded_embeddings = pickle.load(f)print("✅ Embedding 已加载")

# 批量编码（更快）sentences = ["文本 1", "文本 2", ..., "文本 N"]# ❌ 慢：逐个编码embeddings = [model.encode(s) for s in sentences]# ✅ 快：批量编码embeddings = model.encode(sentences, batch_size=32, show_progress_bar=True)

# -*- coding: utf-8 -*-"""完整的 RAG 数据处理流水线运行：python data-pipeline.py"""from pypdf import PdfReaderfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom sentence_transformers import SentenceTransformerfrom langchain.vectorstores import Chromafrom langchain.schema import Documentimport osclass RAGDataPipeline:    """RAG 数据处理流水线"""    def __init__(self, pdf_path, output_dir="./rag_data"):        self.pdf_path = pdf_path        self.output_dir = output_dir        os.makedirs(output_dir, exist_ok=True)        # 初始化组件        self.splitter = RecursiveCharacterTextSplitter(            chunk_size=200,            chunk_overlap=20,            separators=["\n\n", "\n", "。", "！", "？", "，", ""]        )        print("正在加载 Embedding 模型...")        self.embeddings = SentenceTransformer('shibing624/text2vec-base-chinese')        print("✅ 模型加载完成")    def extract_text_from_pdf(self):        """从 PDF 提取文本"""        print(f"\n📄 正在解析 PDF: {self.pdf_path}")        reader = PdfReader(self.pdf_path)        pages_text = []        for i, page in enumerate(reader.pages, 1):            text = page.extract_text()            if text:                pages_text.append({                    "page": i,                    "text": text.strip()                })        print(f"✅ 提取 {len(pages_text)} 页文本")        return pages_text    def split_text(self, pages_text):        """文本分块"""        print("\n✂️  正在分块...")        chunks = []        for page_data in pages_text:            page_chunks = self.splitter.split_text(page_data["text"])            for chunk in page_chunks:                chunks.append(Document(                    page_content=chunk,                    metadata={"page": page_data["page"]}                ))        print(f"✅ 分块完成：共 {len(chunks)} 个文本块")        return chunks    def create_vectorstore(self, chunks):        """创建向量数据库"""        print("\n🗄️  正在创建向量数据库...")        db = Chroma.from_documents(            documents=chunks,            embedding=self.embeddings,            persist_directory=self.output_dir        )        print(f"✅ 向量数据库创建完成")        print(f"📁 存储位置：{self.output_dir}")        return db    def run(self):        """运行完整流水线"""        print("="*60)        print("RAG 数据处理流水线")        print("="*60)        # 1. 提取文本        pages_text = self.extract_text_from_pdf()        # 2. 分块        chunks = self.split_text(pages_text)        # 3. 创建向量库        db = self.create_vectorstore(chunks)        # 4. 测试检索        print("\n🔍 测试检索...")        query = "什么是人工智能"        results = db.similarity_search(query, k=2)        print(f"\n查询：'{query}'")        print("\n检索结果:")        for i, doc in enumerate(results, 1):            print(f"\n[{i}] (第{doc.metadata['page']}页)")            print(doc.page_content)        print("\n" + "="*60)        print("✅ 流水线执行完成！")        print("="*60)        return db# 使用示例if __name__ == "__main__":    # 替换为你的 PDF 文件    pdf_file = "sample.pdf"    if os.path.exists(pdf_file):        pipeline = RAGDataPipeline(pdf_file)        db = pipeline.run()    else:        print(f"❌ 文件不存在：{pdf_file}")        print("请将 PDF 文件放到当前目录后重试")

# 尝试 pdfplumberimport pdfplumberwith pdfplumber.open("file.pdf") as pdf:    text = pdf.pages[0].extract_text()

# 首次加载会下载模型（慢）# 后续使用本地缓存（快）# 手动预下载from sentence_transformers import SentenceTransformerSentenceTransformer('shibing624/text2vec-base-chinese')