3.1 文本处理与语义检索

学习目标

• 掌握文本预处理的核心技术和实现方法
• 理解传统和深度学习文本表示方法的区别
• 学会构建高性能的语义检索系统
• 掌握BERT等预训练模型的应用技巧

核心概念

• 文本预处理：清洗、规范化、分词、词性标注
• 文本向量化：词袋模型、TF-IDF、词嵌入、深度学习表示
• 语义检索：LSI、主题模型、深度语义相似度计算
• 预训练模型：BERT、Sentence-BERT等在大规模语料上的应用

环境准备 / 前置知识

• Python 3.8+
• 机器学习库：scikit-learn、gensim
• 深度学习库：PyTorch、transformers
• 自然语言处理库：jieba、spacy
• 向量计算库：numpy、faiss

分步实战

步骤 1：文本预处理与清洗

完整的文本清洗流程，包括HTML标签去除、Unicode规范化、特殊字符处理等。

import re
import unicodedata
import jieba
from typing import List, Tuple, Set
import logging

class TextPreprocessor:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.stop_words = self._load_stop_words()
    
    def _load_stop_words(self, stop_words_file: str = 'stop_words.txt') -> Set[str]:
        """加载停用词表"""
        try:
            with open(stop_words_file, 'r', encoding='utf-8') as f:
                return set([line.strip() for line in f])
        except FileNotFoundError:
            # 内置停用词
            return {'的', '了', '和', '是', '在', '我', '有', '就', '不', '人', '都', '一', '个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'}
    
    def clean_html_tags(self, text: str) -> str:
        """去除HTML标签"""
        pattern = re.compile(r'<[^>]+>')
        return pattern.sub('', text)
    
    def normalize_unicode(self, text: str) -> str:
        """Unicode规范化为NFKC形式"""
        return unicodedata.normalize('NFKC', text)
    
    def remove_extra_whitespace(self, text: str) -> str:
        """去除多余空白字符"""
        # 统一换行符
        text = re.sub(r'\r\n', '\n', text)
        text = re.sub(r'\r', '\n', text)
        
        # 去除多余空行
        lines = [line.strip() for line in text.split('\n') if line.strip()]
        text = '\n'.join(lines)
        
        # 去除多余空格
        text = re.sub(r'\s+', ' ', text)
        
        return text.strip()
    
    def remove_special_chars(self, text: str, keep_chars: str = '.,;:!?()\"\'-') -> str:
        """去除特殊字符，保留标点符号"""
        pattern = f'[^\\w\\s{re.escape(keep_chars)}\\u4e00-\\u9fff]'
        return re.sub(pattern, ' ', text)
    
    def preprocess_text(self, text: str) -> str:
        """完整的文本预处理流程"""
        if not text:
            return ""
        
        try:
            # 1. HTML标签去除
            text = self.clean_html_tags(text)
            
            # 2. Unicode规范化
            text = self.normalize_unicode(text)
            
            # 3. 特殊字符处理
            text = self.remove_special_chars(text)
            
            # 4. 空白字符处理
            text = self.remove_extra_whitespace(text)
            
            self.logger.info(f"文本预处理完成，长度变化")
            return text
            
        except Exception as e:
            self.logger.error(f"文本预处理失败: {e}")
            return text
    
    def tokenize(self, text: str, with_pos: bool = False) -> List[str]:
        """中文分词"""
        words = jieba.cut(text, cut_all=False)
        if with_pos:
            return list(jieba.posseg.cut(text))
        else:
            # 过滤停用词
            return [word for word in words if word not in self.stop_words and len(word.strip()) > 0]
    
    def extract_keywords(self, text: str, top_k: int = 10) -> List[Tuple[str, float]]:
        """提取关键词"""
        words = self.tokenize(text)
        word_freq = {}
        for word in words:
            word_freq[word] = word_freq.get(word, 0) + 1
        
        # 按频率排序
        sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
        return sorted_words[:top_k]

# 使用示例
preprocessor = TextPreprocessor()
dirty_text = """
  <p>这是一个<b>测试</b>文本！    有一些多余空格。
  包含HTML标签、特殊字符$#^&和Unicode字符￥。
  </p>
"""
cleaned_text = preprocessor.preprocess_text(dirty_text)
print(f"清洗后文本: {cleaned_text}")
​

步骤 2：传统文本表示方法

实现词袋模型和TF-IDF特征提取：

from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from typing import List, Dict, Tuple
import numpy as np

class TraditionalTextVectorizer:
    def __init__(self, max_features: int = 10000):
        self.bow_vectorizer = CountVectorizer(max_features=max_features)
        self.tfidf_vectorizer = TfidfVectorizer(max_features=max_features)
        self.feature_names = None
        self.vocabulary = None
    
    def fit_bow(self, documents: List[str]) -> np.ndarray:
        """拟合词袋模型"""
        tf_matrix = self.bow_vectorizer.fit_transform(documents)
        self.feature_names = self.bow_vectorizer.get_feature_names_out()
        self.vocabulary = self.bow_vectorizer.vocabulary_
        return tf_matrix
    
    def fit_tfidf(self, documents: List[str]) -> np.ndarray:
        """拟合TF-IDF模型"""
        tfidf_matrix = self.tfidf_vectorizer.fit_transform(documents)
        self.feature_names = self.tfidf_vectorizer.get_feature_names_out()
        self.vocabulary = self.tfidf_vectorizer.vocabulary_
        return tfidf_matrix
    
    def get_top_terms(self, document_vector: np.ndarray, top_k: int = 10) -> List[Tuple[str, float]]:
        """获取文档中最重要_terms"""
        if not self.feature_names is None:
            # 获取最重要的词
            top_indices = np.argsort(document_vector)[-top_k:][::-1]
            top_terms = [(self.feature_names[i], document_vector[i]) for i in top_indices if document_vector[i] > 0]
            return top_terms
        return []
    
    def similarity(self, vector1: np.ndarray, vector2: np.ndarray) -> float:
        """计算余弦相似度"""
        dot_product = np.dot(vector1, vector2)
        norm1 = np.linalg.norm(vector1)
        norm2 = np.linalg.norm(vector2)
        return dot_product / (norm1 * norm2) if norm1 > 0 and norm2 > 0 else 0.0

# 使用示例
vectorizer = TraditionalTextVectorizer(max_features=5000)
documents = [
    "多模态知识库构建是一个重要的技术领域",
    "文本处理与语义检索是核心技术",
    "图像特征提取需要深度学习方法",
    "跨模态融合算法是研究热点"
]

# 训练模型
tfidf_matrix = vectorizer.fit_tfidf(documents)
print(f"TF-IDF矩阵形状: {tfidf_matrix.shape}")

# 获取文档关键词
doc_vector = tfidf_matrix[0].toarray()[0]
top_terms = vectorizer.get_top_terms(doc_vector, top_k=5)
print(f"文档关键词: {top_terms}")

# 计算相似度
similarity = vectorizer.similarity(tfidf_matrix[0].toarray()[0], tfidf_matrix[1].toarray()[0])
print(f"文档相似度: {similarity:.4f}")
​

步骤 3：深度学习文本表示

实现Word2Vec和BERT特征提取：

from gensim.models import Word2Vec
import torch
from transformers import AutoTokenizer, AutoModel
import numpy as np
from typing import List

class DeepTextVectorizer:
    def __init__(self, word2vec_size: int = 100, bert_model: str = 'bert-base-chinese'):
        self.word2vec_size = word2vec_size
        self.bert_model_name = bert_model
        self.word2vec_model = None
        self.bert_tokenizer = None
        self.bert_model = None
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    def train_word2vec(self, sentences: List[List[str]]):
        """训练Word2Vec模型"""
        self.word2vec_model = Word2Vec(
            sentences=sentences,
            vector_size=self.word2vec_size,
            window=5,
            min_count=1,
            workers=4
        )
        print(f"Word2Vec模型训练完成，词汇量: {len(self.word2vec_model.wv)}")
    
    def get_word_vector(self, word: str) -> np.ndarray:
        """获取词向量"""
        if self.word2vec_model and word in self.word2vec_model.wv:
            return self.word2vec_model.wv[word]
        else:
            return np.zeros(self.word2vec_size)
    
    def get_document_vector_w2v(self, document: str, method: str = 'average') -> np.ndarray:
        """获取文档向量"""
        words = document.split()
        word_vectors = [self.get_word_vector(word) for word in words]
        
        if not word_vectors:
            return np.zeros(self.word2vec_size)
        
        word_vectors = np.array(word_vectors)
        
        if method == 'average':
            return np.mean(word_vectors, axis=0)
        elif method == 'max':
            return np.max(word_vectors, axis=0)
        elif method == 'min':
            return np.min(word_vectors, axis=0)
        else:
            raise ValueError(f"不支持的向量聚合方法: {method}")
    
    def init_bert(self):
        """初始化BERT模型"""
        self.bert_tokenizer = AutoTokenizer.from_pretrained(self.bert_model_name)
        self.bert_model = AutoModel.from_pretrained(self.bert_model_name)
        self.bert_model.eval()
        self.bert_model.to(self.device)
    
    def extract_bert_embedding(self, text: str, aggregation: str = 'mean') -> np.ndarray:
        """提取BERT文本嵌入"""
        if self.bert_tokenizer is None:
            self.init_bert()
        
        # 分词
        sentences = text.split('。')
        sentence_embeddings = []
        
        for sentence in sentences:
            if sentence.strip():
                inputs = self.bert_tokenizer(
                    sentence,
                    return_tensors='pt',
                    truncation=True,
                    max_length=512,
                    padding=True
                )
                
                inputs = {k: v.to(self.device) for k, v in inputs.items()}
                
                with torch.no_grad():
                    outputs = self.bert_model(**inputs)
                
                # 使用[CLS]标记的输出作为句子表示
                cls_embedding = outputs.last_hidden_state[:, 0, :].cpu().numpy()
                sentence_embeddings.append(cls_embedding[0])
        
        if not sentence_embeddings:
            return np.zeros(self.bert_model.config.hidden_size)
        
        sentence_embeddings = np.array(sentence_embeddings)
        
        if aggregation == 'mean':
            return np.mean(sentence_embeddings, axis=0)
        elif aggregation == 'max':
            return np.max(sentence_embeddings, axis=0)
        elif aggregation == 'cls':
            return sentence_embeddings[0]
        else:
            raise ValueError(f"不支持的聚合方法: {aggregation}")
    
    def similarity(self, embedding1: np.ndarray, embedding2: np.ndarray) -> float:
        """计算余弦相似度"""
        dot_product = np.dot(embedding1, embedding2)
        norm1 = np.linalg.norm(embedding1)
        norm2 = np.linalg.norm(embedding2)
        return dot_product / (norm1 * norm2) if norm1 > 0 and norm2 > 0 else 0.0

# 使用示例
deep_vectorizer = DeepTextVectorizer(word2vec_size=100)

# 训练Word2Vec
sentences = [
    ["多模态", "知识库", "构建", "技术"],
    ["文本", "处理", "语义", "检索"],
    ["图像", "特征", "提取", "分析"]
]
deep_vectorizer.train_word2vec(sentences)

# 获取词向量
word_vector = deep_vectorizer.get_word_vector("多模态")
print(f"词向量形状: {word_vector.shape}")

# 获取文档向量
doc_vector = deep_vectorizer.get_document_vector_w2v("多模态知识库构建")
print(f"文档向量形状: {doc_vector.shape}")

# BERT特征提取
bert_embedding = deep_vectorizer.extract_bert_embedding("多模态知识库构建是一个重要的技术领域")
print(f"BERT嵌入形状: {bert_embedding.shape}")
​

步骤 4：构建语义检索系统

实现基于LSI和深度语义的检索系统：

from sklearn.decomposition import TruncatedSVD
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
from typing import List, Dict, Tuple

class SemanticSearchEngine:
    def __init__(self, n_components: int = 50):
        self.n_components = n_components
        self.vectorizer = TraditionalTextVectorizer(max_features=10000)
        self.svd = TruncatedSVD(n_components=n_components, random_state=42)
        self.documents = []
        self.tfidf_matrix = None
        self.lsi_matrix = None
        self.deep_vectorizer = DeepTextVectorizer()
    
    def fit(self, documents: List[str]):
        """拟合检索模型"""
        self.documents = documents
        
        # 传统方法
        self.tfidf_matrix = self.vectorizer.fit_tfidf(documents)
        
        # LSI降维
        self.lsi_matrix = self.svd.fit_transform(self.tfidf_matrix)
        
        # 初始化深度学习模型
        self.deep_vectorizer.init_bert()
    
    def search_lsi(self, query: str, top_k: int = 10) -> List[Dict]:
        """基于LSI的语义搜索"""
        # 预处理查询
        preprocessor = TextPreprocessor()
        clean_query = preprocessor.preprocess_text(query)
        
        # 转换为TF-IDF向量
        query_vector = self.vectorizer.tfidf_vectorizer.transform([clean_query])
        
        # 投影到LSI空间
        query_lsi = self.svd.transform(query_vector)
        
        # 计算相似度
        similarities = cosine_similarity(query_lsi, self.lsi_matrix)[0]
        
        # 获取top-k结果
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        
        results = []
        for idx, similarity in zip(top_indices, similarities[top_indices]):
            results.append({
                'document_idx': idx,
                'similarity': float(similarity),
                'document': self.documents[idx],
                'method': 'LSI'
            })
        
        return results
    
    def search_deep(self, query: str, top_k: int = 10) -> List[Dict]:
        """基于深度学习的语义搜索"""
        preprocessor = TextPreprocessor()
        clean_query = preprocessor.preprocess_text(query)
        
        # 提取查询的BERT特征
        query_embedding = self.deep_vectorizer.extract_bert_embedding(clean_query)
        
        # 提取所有文档的BERT特征
        document_embeddings = []
        for doc in self.documents:
            doc_embedding = self.deep_vectorizer.extract_bert_embedding(doc)
            document_embeddings.append(doc_embedding)
        
        document_embeddings = np.array(document_embeddings)
        
        # 计算相似度
        similarities = []
        for doc_embedding in document_embeddings:
            sim = self.deep_vectorizer.similarity(query_embedding, doc_embedding)
            similarities.append(sim)
        
        similarities = np.array(similarities)
        
        # 获取top-k结果
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        
        results = []
        for idx, similarity in zip(top_indices, similarities[top_indices]):
            results.append({
                'document_idx': idx,
                'similarity': float(similarity),
                'document': self.documents[idx],
                'method': 'Deep Learning'
            })
        
        return results
    
    def hybrid_search(self, query: str, top_k: int = 10, alpha: float = 0.5) -> List[Dict]:
        """混合搜索：LSI + 深度学习"""
        lsi_results = self.search_lsi(query, top_k * 2)
        deep_results = self.search_deep(query, top_k * 2)
        
        # 合并结果
        combined_scores = {}
        for result in lsi_results:
            idx = result['document_idx']
            if idx not in combined_scores:
                combined_scores[idx] = {'lsi_score': 0, 'deep_score': 0, 'document': result['document']}
            combined_scores[idx]['lsi_score'] = result['similarity']
        
        for result in deep_results:
            idx = result['document_idx']
            if idx not in combined_scores:
                combined_scores[idx] = {'lsi_score': 0, 'deep_score': 0, 'document': result['document']}
            combined_scores[idx]['deep_score'] = result['similarity']
        
        # 计算混合得分
        hybrid_results = []
        for idx, scores in combined_scores.items():
            hybrid_score = alpha * scores['lsi_score'] + (1 - alpha) * scores['deep_score']
            hybrid_results.append({
                'document_idx': idx,
                'lsi_score': scores['lsi_score'],
                'deep_score': scores['deep_score'],
                'hybrid_score': hybrid_score,
                'document': scores['document']
            })
        
        # 按混合得分排序
        hybrid_results.sort(key=lambda x: x['hybrid_score'], reverse=True)
        
        return hybrid_results[:top_k]

# 使用示例
search_engine = SemanticSearchEngine(n_components=50)

documents = [
    "多模态知识库构建是一个重要的技术领域，涉及文本、图像、音频等多种数据类型的处理",
    "文本处理与语义检索是核心技术，包括分词、词性标注、命名实体识别等技术",
    "图像特征提取需要深度学习方法，如CNN、ResNet等模型提取视觉特征",
    "跨模态融合算法是研究热点，通过注意力机制实现不同模态信息的有效整合",
    "向量数据库如FAISS、Milvus等为大规模向量相似性搜索提供了高效解决方案"
]

search_engine.fit(documents)

# 搜索测试
query = "图像特征提取"
lsi_results = search_engine.search_lsi(query, top_k=3)
print("LSI搜索结果:")
for result in lsi_results:
    print(f"文档{result['document_idx']}: {result['similarity']:.4f}")

deep_results = search_engine.search_deep(query, top_k=3)
print("\n深度学习搜索结果:")
for result in deep_results:
    print(f"文档{result['document_idx']}: {result['similarity']:.4f}")

hybrid_results = search_engine.hybrid_search(query, top_k=3)
print("\n混合搜索结果:")
for result in hybrid_results:
    print(f"文档{result['document_idx']}: LSI={result['lsi_score']:.4f}, Deep={result['deep_score']:.4f}, Hybrid={result['hybrid_score']:.4f}")
​

完整示例

基于上述技术的完整实现：

import numpy as np
from typing import List, Dict

class CompleteTextProcessingPipeline:
    def __init__(self):
        self.preprocessor = TextPreprocessor()
        self.vectorizer = TraditionalTextVectorizer(max_features=5000)
        self.search_engine = SemanticSearchEngine(n_components=30)
        self.documents = []
    
    def add_documents(self, documents: List[str]):
        """添加文档到索引"""
        self.documents = documents
        self.search_engine.fit(documents)
    
    def process_query(self, query: str, method: str = 'hybrid', top_k: int = 5) -> Dict:
        """处理查询"""
        clean_query = self.preprocessor.preprocess_text(query)
        
        results = {
            'query': query,
            'cleaned_query': clean_query,
            'method': method,
            'results': []
        }
        
        if method == 'lsi':
            search_results = self.search_engine.search_lsi(query, top_k)
        elif method == 'deep':
            search_results = self.search_engine.search_deep(query, top_k)
        else:  # hybrid
            search_results = self.search_engine.hybrid_search(query, top_k)
        
        results['results'] = search_results
        return results

# 使用示例
pipeline = CompleteTextProcessingPipeline()

documents = [
    "多模态知识库构建是人工智能领域的重要研究方向",
    "文本处理包括分词、词性标注、命名实体识别等自然语言处理技术",
    "图像特征提取使用SIFT、SURF、CNN等算法提取视觉特征",
    "音频处理涉及MFCC、频谱分析、声学特征提取等技术",
    "跨模态融合通过注意力机制实现不同模态信息的有效整合",
    "向量数据库如FAISS、Milvus为大规模向量相似性搜索提供解决方案",
    "深度学习模型如BERT、GPT在大规模语料上预训练后表现优异",
    "检索增强生成技术结合了检索和生成的优势，提升了回答质量"
]

pipeline.add_documents(documents)

# 测试查询
query = "深度学习模型"
result = pipeline.process_query(query, method='hybrid', top_k=3)

print(f"查询: {result['query']}")
print(f"清洗后: {result['cleaned_query']}")
print(f"使用方法: {result['method']}")
print("\n搜索结果:")
for i, doc_result in enumerate(result['results'], 1):
    print(f"{i}. 文档{doc_result['document_idx']}: {doc_result['document'][:50]}...")
    if 'hybrid_score' in doc_result:
        print(f"   混合得分: {doc_result['hybrid_score']:.4f}")
    else:
        print(f"   相似度: {doc_result['similarity']:.4f}")
​

常见问题 FAQ

Q1：如何选择合适的文本预处理方法？

A：预处理方法的选择取决于具体任务：

• 基础任务：去除HTML标签、规范化格式、去重
• 中文处理：jieba分词、停用词过滤
• 高级任务：命名实体识别、句法分析、语义标注

Q2：词袋模型和TF-IDF有什么区别？

A：主要区别在于权重计算：

• 词袋模型：仅统计词频，不考虑文档频率
• TF-IDF：结合词频和逆文档频率，更能区分文档间的差异

Q3：BERT模型比传统方法有什么优势？

A：BERT的优势包括：

• 深度语义理解，能捕捉上下文信息
• 预训练在大规模语料上，泛化能力强
• 解决一词多义问题，理解语义而非表面形式

Q4：如何处理大规模文本数据？

A：对于大规模数据处理，建议：

• 使用分布式计算框架如Spark
• 采用增量学习策略
• 使用近似最近邻搜索算法
• 优化内存使用和计算效率

最佳实践与避坑

• 实践1：预处理阶段确保数据质量，垃圾进垃圾出
• 实践2：根据任务特点选择合适的文本表示方法
• 实践3：语义检索时注意同义词扩展和语义扩展
• 实践4：深度学习模型需要足够的训练数据和计算资源
• 坑点1：避免在预处理过程中丢失重要语义信息
• 坑点2：注意内存使用，避免一次性处理过大的文本数据
• 坑点3：深度学习模型推理速度较慢，需要优化

本节小结

本节详细介绍了文本处理与语义检索的核心技术，涵盖了从传统方法到深度学习的完整技术栈。通过实际代码示例，展示了如何构建一个完整的文本处理和语义检索系统。主要技术要点包括：

1. 文本预处理：清洗、分词、规范化处理
2. 文本表示：词袋、TF-IDF、Word2Vec、BERT
3. 语义检索：LSI、深度语义搜索、混合检索
4. 工程实践：完整的处理流水线实现

这些技术为构建多模态知识库中的文本处理模块提供了坚实的基础。

延伸阅读

• 官方文档：Scikit-learn文本处理教程
• 相关章节：本教程3.2节图像特征提取与检索
• 进阶资源：《自然语言处理实战》书籍