2.2 文档处理与向量化 — AI知识库搭建全攻略

本节导读：深入掌握文档预处理、分块策略和向量化技术的核心原理，学习如何将非结构化文本转换为高质量的向量表示，为AI知识库的检索和生成奠定基础。

学习目标

• 掌握文档预处理的标准流程和技术要点
• 理解文档分块策略的设计原理和最佳实践
• 学习不同向量化方法的特点和适用场景
• 掌握文本清洗和标准化技术
• 了解文档质量评估和优化方法

核心概念

文档处理与向量化是构建AI知识库的关键环节，直接影响检索的准确性和生成的质量。文档处理的本质是将非结构化文本转换为结构化的数据表示，通过向量化技术使机器能够理解和处理文本内容。

文档预处理技术

文档预处理是向量化前的必要步骤，包括：

• 文本清洗：去除HTML标签、特殊字符、停用词等
• 标准化：统一大小写、处理标点符号、规范化数字
• 结构提取：识别文档结构（标题、段落、列表等）
• 编码处理：处理不同字符编码的文本

向量化技术分类

向量化方法可分为三类：

• 统计方法：TF-IDF、Bag of Words等基于词频的表示
• 分布式表示：Word2Vec、GloVe等基于上下文的词向量
• 深度表示：BERT、Sentence Transformers等基于神经网络的表达

环境准备 / 前置知识

技术依赖

• Python 3.8+
• 文本处理库：NLTK、spaCy、正则表达式
• 向量化库：Sentence Transformers、scikit-learn
• 数据处理：pandas、numpy

版本兼容性

• spaCy 3.x
• Sentence Transformers 2.0+
• NLTK 3.8+

工具安装

# 安装基础文本处理库
pip install nltk spacy scikit-learn pandas numpy

# 安装向量化工具
pip install sentence-transformers torch

# 安装中文处理工具
pip install jieba pangu

# 下载NLTK数据
python -c "import nltk; nltk.download('punkt'); nltk.download('stopwords'); nltk.download('wordnet')"

# 下载spaCy模型
python -m spacy download en_core_web_lg
python -m spacy download zh_core_web_lg
​

分步实战

步骤 1：文档预处理管道

构建完整的文档预处理流程：

import re
import html
import unicodedata
from typing import List, Dict, Any
import nltk
import spacy
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize, sent_tokenize
from nltk.stem import WordNetLemmatizer
import jieba
from pangu import spacing_text

class DocumentPreprocessor:
    """文档预处理器"""
    
    def __init__(self, language='en'):
        self.language = language
        self.setup_nlp_tools()
        
    def setup_nlp_tools(self):
        """设置NLP工具"""
        if self.language == 'en':
            # 英文处理
            self.nlp = spacy.load('en_core_web_lg')
            self.stop_words = set(stopwords.words('english'))
            self.lemmatizer = WordNetLemmatizer()
        elif self.language == 'zh':
            # 中文处理
            self.nlp = spacy.load('zh_core_web_lg')
            self.stop_words = set(['的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'])
            
    def clean_text(self, text: str) -> str:
        """基础文本清洗"""
        if not text:
            return ""
        
        # HTML解码
        text = html.unescape(text)
        
        # 移除HTML标签
        text = re.sub(r'<[^>]+>', '', text)
        
        # 移除URL
        text = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '', text)
        
        # 移除邮箱地址
        text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '', text)
        
        # 规范化Unicode
        text = unicodedata.normalize('NFKD', text)
        
        # 移除控制字符
        text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
        
        # 统一换行符
        text = re.sub(r'\r\n|\r', '\n', text)
        
        # 移除多余空格
        text = re.sub(r'\s+', ' ', text).strip()
        
        return text
    
    def normalize_text(self, text: str) -> str:
        """文本标准化"""
        if self.language == 'en':
            # 英文标准化
            text = text.lower()  # 转小写
            text = re.sub(r'[^\w\s]', ' ', text)  # 移除标点
            text = re.sub(r'\s+', ' ', text).strip()  # 统一空格
        elif self.language == 'zh':
            # 中文标准化
            # 全角转半角
            text = unicodedata.normalize('NFKC', text)
            # 添加空格（Pangu库）
            text = spacing_text(text)
            # 移除多余空格
            text = re.sub(r'\s+', ' ', text).strip()
            
        return text
    
    def tokenize_sentences(self, text: str) -> List[str]:
        """句子切分"""
        if self.language == 'en':
            sentences = sent_tokenize(text, language='english')
        else:
            # 中文句子切分
            sentences = re.split(r'[。！？；\n]+', text)
            sentences = [s.strip() for s in sentences if s.strip()]
        
        return sentences
    
    def tokenize_words(self, text: str) -> List[str]:
        """词语切分"""
        if self.language == 'en':
            # 英文词语切分
            words = word_tokenize(text.lower())
            words = [word for word in words if word.isalpha() and len(word) > 1]
        else:
            # 中文词语切分
            words = jieba.lcut(text)
            words = [word.strip() for word in words if word.strip() and len(word) > 1]
        
        # 移除停用词
        words = [word for word in words if word not in self.stop_words]
        
        return words
    
    def preprocess_document(self, text: str) -> Dict[str, Any]:
        """完整的文档预处理"""
        # 基础清洗
        cleaned_text = self.clean_text(text)
        
        # 标准化
        normalized_text = self.normalize_text(cleaned_text)
        
        # 句子切分
        sentences = self.tokenize_sentences(cleaned_text)
        
        # 词语切分
        words = self.tokenize_words(normalized_text)
        
        return {
            'original_text': text,
            'cleaned_text': cleaned_text,
            'normalized_text': normalized_text,
            'sentences': sentences,
            'words': words,
            'word_count': len(words),
            'sentence_count': len(sentences)
        }

# 使用示例
preprocessor = DocumentPreprocessor('en')

# 测试文档
test_document = """
    <h1>Introduction to Natural Language Processing</h1>
    
    <p>Natural Language Processing (NLP) is a subfield of linguistics, computer science, and AI 
    concerned with the interactions between computers and human language.</p>
    
    <p>Here are some key aspects of NLP:</p>
    <ul>
        <li>Text classification</li>
        <li>Sentiment analysis</li>
        <li>Named entity recognition</li>
        <li>Machine translation</li>
    </ul>
    
    <p>For more information, contact us at nlp@example.com or visit https://example.com/nlp</p>
"""

# 预处理文档
result = preprocessor.preprocess_document(test_document)
print(f"原始文档词数: {result['word_count']}")
print(f"句子数量: {result['sentence_count']}")
print(f"清洗后文本: {result['cleaned_text'][:100]}...")
​

步骤 2：文档分块策略

实现智能文档分块算法：

import math
from typing import List, Tuple
from dataclasses import dataclass

@dataclass
class Chunk:
    """文档块数据结构"""
    content: str
    start_pos: int
    end_pos: int
    chunk_id: int
    metadata: dict

class DocumentChunker:
    """智能文档分块器"""
    
    def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        
    def chunk_by_sentences(self, sentences: List[str], metadata: dict = None) -> List[Chunk]:
        """按句子分块"""
        if metadata is None:
            metadata = {}
            
        chunks = []
        current_chunk = []
        current_size = 0
        chunk_id = 0
        start_pos = 0
        
        for i, sentence in enumerate(sentences):
            # 计算当前句子的大小
            sentence_size = len(sentence.split())
            
            # 如果添加当前句子会超过块大小
            if current_size + sentence_size > self.chunk_size and current_chunk:
                # 创建当前块
                chunk_content = ' '.join(current_chunk)
                chunk = Chunk(
                    content=chunk_content,
                    start_pos=start_pos,
                    end_pos=start_pos + len(chunk_content),
                    chunk_id=chunk_id,
                    metadata={**metadata, 'sentence_count': len(current_chunk)}
                )
                chunks.append(chunk)
                
                # 准备下一个块，包含重叠内容
                overlap_sentences = current_chunk[-2:] if self.chunk_overlap > 0 else []
                current_chunk = overlap_sentences
                current_size = sum(len(s.split()) for s in overlap_sentences)
                start_pos = chunk.end_pos - len(' '.join(overlap_sentences))
                chunk_id += 1
            
            # 添加当前句子到块
            current_chunk.append(sentence)
            current_size += sentence_size
        
        # 添加最后一个块
        if current_chunk:
            chunk_content = ' '.join(current_chunk)
            chunk = Chunk(
                content=chunk_content,
                start_pos=start_pos,
                end_pos=start_pos + len(chunk_content),
                chunk_id=chunk_id,
                metadata={**metadata, 'sentence_count': len(current_chunk)}
            )
            chunks.append(chunk)
        
        return chunks

# 使用示例
chunker = DocumentChunker(chunk_size=300, chunk_overlap=30)

# 测试文档
test_sentences = [
    "Natural language processing is a subfield of artificial intelligence.",
    "It focuses on the interaction between computers and human language.",
    "NLP techniques are used in various applications including text classification.",
    "Sentiment analysis is one of the most common NLP tasks.",
    "Named entity recognition helps identify important entities in text.",
    "Machine translation systems can translate text between languages.",
    "Question answering systems can understand and respond to questions.",
    "Text summarization algorithms can generate concise summaries of long documents."
]

# 按句子分块
sentence_chunks = chunker.chunk_by_sentences(test_sentences)
print(f"句子分块数量: {len(sentence_chunks)}")
for i, chunk in enumerate(sentence_chunks[:3]):
    print(f"块 {i+1}: {chunk.content[:50]}...")
​

步骤 3：向量化技术实现

实现多种向量化方法：

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
from sklearn.decomposition import TruncatedSVD
from sentence_transformers import SentenceTransformer

class DocumentVectorizer:
    """文档向量化器"""
    
    def __init__(self, method='sentence-transformers'):
        self.method = method
        self.vectorizer = None
        self.dimension = None
        
        if method == 'tfidf':
            self.vectorizer = TfidfVectorizer(
                max_features=5000,
                stop_words='english',
                ngram_range=(1, 2),
                min_df=2,
                max_df=0.8
            )
        elif method == 'sentence-transformers':
            self.model = SentenceTransformer('all-MiniLM-L6-v2')
            self.dimension = 384
    
    def tfidf_vectorize(self, documents: List[str]) -> np.ndarray:
        """TF-IDF向量化"""
        if not self.vectorizer:
            self.vectorizer = TfidfVectorizer(
                max_features=5000,
                stop_words='english',
                ngram_range=(1, 2),
                min_df=2,
                max_df=0.8
            )
        
        # 拟合并转换文档
        tfidf_matrix = self.vectorizer.fit_transform(documents)
        
        # 转换为数组
        tfidf_array = tfidf_matrix.toarray()
        
        # 应用降维
        if tfidf_array.shape[1] > 1000:
            svd = TruncatedSVD(n_components=1000, random_state=42)
            tfidf_array = svd.fit_transform(tfidf_array)
        
        return tfidf_array
    
    def sentence_transformer_vectorize(self, documents: List[str]) -> np.ndarray:
        """Sentence Transformers向量化"""
        if not hasattr(self, 'model'):
            self.model = SentenceTransformer('all-MiniLM-L6-v2')
        
        # 生成嵌入
        embeddings = self.model.encode(documents, convert_to_numpy=True)
        
        return embeddings
    
    def vectorize_documents(self, documents: List[str]) -> np.ndarray:
        """根据选择的向量化方法向量化文档"""
        if self.method == 'tfidf':
            return self.tfidf_vectorize(documents)
        elif self.method == 'sentence-transformers':
            return self.sentence_transformer_vectorize(documents)
        else:
            raise ValueError(f"不支持的向量化方法: {self.method}")

# 使用示例
vectorizer = DocumentVectorizer(method='sentence-transformers')

# 测试文档
test_documents = [
    "Natural language processing is a subfield of artificial intelligence.",
    "Machine learning algorithms can learn patterns from data.",
    "Deep learning models use neural networks with multiple layers."
]

# 向量化文档
embeddings = vectorizer.vectorize_documents(test_documents)
print(f"向量化结果形状: {embeddings.shape}")
print(f"向量维度: {embeddings.shape[1]}")
​

常见问题 FAQ

Q1：如何选择合适的文档分块策略？

A：选择分块策略需要考虑：

• 文档类型：技术文档适合按主题分块，对话文本适合按对话轮次分块
• 检索需求：语义检索适合语义分块，关键词检索适合按段落分块
• 嵌入模型：不同嵌入模型有最佳输入长度，需要适配模型特性
• 性能要求：更细粒度的分块检索精度更高但速度更慢

Q2：如何平衡文档预处理的质量和效率？

A：质量与效率的平衡策略：

• 批量处理：对大量文档进行批量预处理，提高效率
• 缓存机制：缓存预处理结果，避免重复处理
• 渐进式处理：先进行基础清洗，必要时进行深度处理
• 并行处理：利用多进程或多线程并行处理文档

Q3：不同向量化方法如何选择？

A：向量化方法选择指南：

• TF-IDF：适合短文本、关键词检索、资源受限场景
• Word2Vec：适合语义相似度计算、词级别任务
• BERT类模型：适合长文本、语义理解、高质量场景
• Sentence Transformers：适合句子级语义、平衡精度和效率

Q4：如何处理多语言文档的向量化？

A：多语言向量化策略：

• 多语言模型：使用支持多种语言的预训练模型（如LaBSE）
• 语言特定模型：为每种语言使用专门的模型，然后融合结果
• 翻译后处理：将所有文档翻译成同一种语言再向量化
• 混合嵌入：使用语言嵌入标识语言信息，然后融合特征

最佳实践与避坑

实践 1：分块大小优化

• 根据嵌入模型输入限制调整分块大小（通常512-1024 tokens）
• 在分块间保持适度重叠（10-20%）
• 考虑文档逻辑结构，避免在重要概念处断开

实践 2：预处理质量控制

• 建立预处理质量检查机制
• 定期更新停用词词典和专业术语词典
• 保留重要标点符号，避免过度清洗

实践 3：向量化方法评估

• 建立向量化质量评估框架
• 使用标准数据集测试不同方法的性能
• 考虑领域适应性，针对性选择或微调模型

坑点 1：过度预处理

• 过度清洗会丢失重要语义信息
• 移除标点可能影响句子理解
• 保留必要的结构信息

坑点 2：分块边界不合理

• 在关键概念中间断开会影响检索质量
• 缺乏重叠可能导致信息丢失
• 忽略文档逻辑结构

坑点 3：向量维度过高

• 高维向量会增加存储和计算成本
• 可能导致维度灾难
• 适当降维可以提高效率

本节小结

本节详细介绍了文档处理与向化的核心技术，包括文档预处理、智能分块和向量化方法。通过实际的代码示例，我们学习了如何构建完整的文档处理流水线，以及如何选择和优化不同的处理策略。

文档处理是AI知识库质量的关键基础，需要根据具体应用场景选择合适的处理策略。在实际应用中，需要平衡质量、效率和成本，持续优化处理流程。

在下一节中，我们将学习检索增强生成（RAG）技术，这是AI知识库系统的核心功能。

延伸阅读

• 官方文档：Sentence Transformers官方文档，详细介绍预训练模型的使用方法
• 相关章节：本教程2.3节检索增强生成(RAG)，学习如何利用向量化结果进行检索
• 技术论文：BERT和GPT模型的原理与应用，深入理解深度学习模型的语义表示能力

关键词：AI知识库搭建全攻略, 文档预处理, 文档分块, 向量化, 文本清洗, 智能分块, TF-IDF, Sentence Transformers, NLP, 文档质量
难度：进阶
预计阅读：35分钟