2.2 文本向量化方法

文本向量的基本概念

什么是文本向量？

文本向量是将文本信息转换为数值向量的过程，是AI搜索系统的核心步骤。与词向量不同，文本向量可以表示句子、段落甚至整个文档。

文本向量的特点：

1. 固定维度：无论文本长度，输出向量维度固定
2. 语义保持：向量空间中的距离反映语义相似性
3. 数值表示：所有信息转换为数值，便于数学运算
4. 连续性：相似文本在向量空间中距离相近

文本向量的重要性

在AI搜索系统中，文本向量化扮演着关键角色：

文本查询 → 向量化 → 向量检索 → 结果排序
​

重要性体现在：

1. 语义理解：从字面匹配到语义理解
2. 高效计算：向量运算比文本匹配快得多
3. 多模态融合：统一的数据表示形式
4. 机器学习兼容：为下游任务提供特征输入

基础向量化流程

向量化处理步骤

一个完整的文本向量化流程包括以下几个步骤：

1. 文本预处理

def text_preprocessing(text):
    # 文本清洗
    text = re.sub(r'[^\w\s]', '', text)
    
    # 分词处理
    tokens = jieba.cut(text)
    
    # 去除停用词
    tokens = [token for token in tokens if token not in stop_words]
    
    # 词形还原
    tokens = [lemmatize(token) for token in tokens]
    
    return tokens
​

2. 编码转换

def encode_tokens(tokens, vocabulary):
    """将词语转换为索引"""
    encoded = [vocabulary.get(token, vocabulary['<UNK>']) 
               for token in tokens]
    return encoded
​

3. 填充与截断

def pad_sequence(encoded, max_length=512, pad_token=0):
    """序列填充到固定长度"""
    if len(encoded) > max_length:
        encoded = encoded[:max_length]
    else:
        encoded = encoded + [pad_token] * (max_length - len(encoded))
    return encoded
​

4. 向量化转换

def tokens_to_vectors(encoded, embeddings):
    """索引转换为向量"""
    vectors = [embeddings[idx] for idx in encoded]
    return np.array(vectors)
​

向量归一化处理

def normalize_vectors(vectors):
    """向量归一化"""
    # L2归一化
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    normalized = vectors / norms
    return normalized

# 使用示例
vectors = np.array([[1, 2, 3], [4, 5, 6]])
normalized_vectors = normalize_vectors(vectors)
print(f"原始向量: {vectors}")
print(f"归一化向量: {normalized_vectors}")
​

统计方法向量化

TF-IDF向量化

TF-IDF（Term Frequency-Inverse Document Frequency）是一种经典的文本表示方法。

TF-IDF计算公式：

TF-IDF(t,d) = TF(t,d) × IDF(t)
​

其中：

• TF(t,d)：词语t在文档d中的频率
• IDF(t)：逆文档频率，衡量词语的稀有程度

from sklearn.feature_extraction.text import TfidfVectorizer

class TFIDFVectorizer:
    def __init__(self, max_features=10000):
        self.vectorizer = TfidfVectorizer(
            max_features=max_features,
            stop_words='english',
            ngram_range=(1, 2),
            min_df=2,
            max_df=0.8
        )
        self.vocabulary = None
        
    def fit(self, documents):
        """训练TF-IDF模型"""
        tfidf_matrix = self.vectorizer.fit_transform(documents)
        self.vocabulary = self.vectorizer.vocabulary_
        return tfidf_matrix
    
    def transform(self, documents):
        """转换文档为TF-IDF向量"""
        return self.vectorizer.transform(documents)
    
    def get_feature_names(self):
        """获取特征名称"""
        return self.vectorizer.get_feature_names_out()
​

TF-IDF的优缺点：

优点：

• 计算简单高效
• 能反映词语的重要性
• 支持多文档分析
• 没有OOV问题

缺点：

• 无法处理语义相似性
• 忽略词语顺序
• 无法处理上下文信息
• 高维特征稀疏

Word2Vec统计向量化

Word2Vec可以将统计信息整合到向量表示中：

class StatisticalWord2Vec:
    def __init__(self, word2vec_model):
        self.model = word2vec_model
        self.idf_scores = {}
        
    def fit_idf(self, documents):
        """计算IDF分数"""
        from sklearn.feature_extraction.text import TfidfVectorizer
        
        # 计算TF-IDF
        tfidf_vectorizer = TfidfVectorizer()
        tfidf_matrix = tfidf_vectorizer.fit_transform(documents)
        
        # 获取IDF分数
        idf_values = tfidf_vectorizer.idf_
        feature_names = tfidf_vectorizer.get_feature_names_out()
        
        self.idf_scores = dict(zip(feature_names, idf_values))
    
    def get_weighted_vector(self, word):
        """获取加权词向量"""
        base_vector = self.model.wv[word]
        idf_weight = self.idf_scores.get(word, 1.0)
        return base_vector * idf_weight
​

深度学习方法向量化

CNN文本向量化

卷积神经网络可以捕捉文本中的局部特征：

import torch
import torch.nn as nn

class TextCNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.convs = nn.ModuleList([
            nn.Conv1d(embed_dim, 128, kernel_size=k)
            for k in [3, 4, 5]
        ])
        self.dropout = nn.Dropout(0.5)
        self.fc = nn.Linear(128 * 3, num_classes)
        
    def forward(self, x):
        x = self.embedding(x)  # [batch, seq_len, embed_dim]
        x = x.transpose(1, 2)   # [batch, embed_dim, seq_len]
        
        conv_outputs = []
        for conv in self.convs:
            conv_out = conv(x)      # [batch, 128, new_seq_len]
            conv_out = torch.relu(conv_out)
            pooled = torch.max(conv_out, dim=2)[0]  # [batch, 128]
            conv_outputs.append(pooled)
        
        x = torch.cat(conv_outputs, dim=1)  # [batch, 384]
        x = self.dropout(x)
        x = self.fc(x)
        
        return x
​

Transformer文本向量化

Transformer是目前最先进的文本表示方法：

import torch
import torch.nn as nn

class TextTransformer(nn.Module):
    def __init__(self, vocab_size, embed_dim, num_heads, num_layers, max_len):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.positional_encoding = PositionalEncoding(embed_dim, max_len)
        self.transformer_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(
                d_model=embed_dim,
                nhead=num_heads,
                dim_feedforward=2048,
                dropout=0.1
            ),
            num_layers=num_layers
        )
        self.pool = nn.AdaptiveAvgPool1d(1)
        
    def forward(self, x):
        # 嵌入
        x = self.embedding(x)
        
        # 位置编码
        x = x.transpose(0, 1)  # [seq_len, batch, embed_dim]
        x = self.positional_encoding(x)
        
        # Transformer编码
        x = self.transformer_encoder(x)  # [seq_len, batch, embed_dim]
        
        # 池化
        x = x.transpose(1, 2)  # [batch, embed_dim, seq_len]
        x = self.pool(x).squeeze(-1)  # [batch, embed_dim]
        
        return x

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        position = torch.arange(max_len).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2) * 
                           -(torch.log(torch.tensor(10000.0)) / d_model))
        pe = torch.zeros(max_len, 1, d_model)
        pe[:, 0, 0::2] = torch.sin(position * div_term)
        pe[:, 0, 1::2] = torch.cos(position * div_term)
        self.register_buffer('pe', pe)
        
    def forward(self, x):
        return x + self.pe[:x.size(0)]
​

高级向量化技术

句子向量化

句子向量化是将整个句子编码为固定长度向量的过程：

class SentenceVectorizer:
    def __init__(self, model_name='bert-base-chinese'):
        from transformers import BertTokenizer, BertModel
        import torch
        
        self.tokenizer = BertTokenizer.from_pretrained(model_name)
        self.model = BertModel.from_pretrained(model_name)
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model.to(self.device)
        
    def encode_sentence(self, sentence):
        """编码单个句子"""
        inputs = self.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.model(**inputs)
        
        # 使用[CLS]标记的向量作为句子表示
        sentence_vector = outputs.last_hidden_state[:, 0, :].cpu().numpy()
        
        return sentence_vector[0]
    
    def encode_sentences(self, sentences):
        """批量编码句子"""
        return [self.encode_sentence(sent) for sent in sentences]
    
    def batch_encode(self, sentences, batch_size=16):
        """批量编码（高效版）"""
        all_vectors = []
        
        for i in range(0, len(sentences), batch_size):
            batch_sentences = sentences[i:i+batch_size]
            
            inputs = self.tokenizer(
                batch_sentences,
                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.model(**inputs)
            
            batch_vectors = outputs.last_hidden_state[:, 0, :].cpu().numpy()
            all_vectors.extend(batch_vectors)
        
        return all_vectors
​

文档向量化

长文档需要分层处理：

class DocumentVectorizer:
    def __init__(self, sentence_vectorizer):
        self.sentence_vectorizer = sentence_vectorizer
        
    def encode_document(self, document):
        """编码整个文档"""
        # 分段
        paragraphs = self.split_paragraphs(document)
        
        if len(paragraphs) == 0:
            return np.zeros(self.sentence_vectorizer.model.config.hidden_size)
        
        # 编码段落
        paragraph_vectors = [self.encode_paragraph(paragraph) 
                          for paragraph in paragraphs]
        
        # 文档向量 = 段落向量的聚合
        document_vector = self.aggregate_paragraphs(paragraph_vectors)
        
        return document_vector
    
    def split_paragraphs(self, text):
        """分段"""
        import re
        paragraphs = re.split(r'\n\n', text)
        paragraphs = [p.strip() for p in paragraphs if p.strip()]
        return paragraphs
    
    def aggregate_paragraphs(self, paragraph_vectors):
        """聚合段落向量"""
        if len(paragraph_vectors) == 1:
            return paragraph_vectors[0]
        
        # 简单平均
        return np.mean(paragraph_vectors, axis=0)
    
    def advanced_aggregate(self, paragraph_vectors):
        """高级聚合方法"""
        if len(paragraph_vectors) == 1:
            return paragraph_vectors[0]
        
        # 计算段落重要性
        importance_scores = [self.calculate_paragraph_importance(vec) 
                          for vec in paragraph_vectors]
        
        # 加权平均
        weights = np.array(importance_scores)
        weights = weights / weights.sum()
        
        weighted_avg = np.average(paragraph_vectors, axis=0, weights=weights)
        
        return weighted_avg
    
    def calculate_paragraph_importance(self, paragraph_vector):
        """计算段落重要性"""
        # 基于向量的范数（通常重要段落有更大的范数）
        norm = np.linalg.norm(paragraph_vector)
        
        # 归一化到0-1范围
        normalized_norm = norm / (np.linalg.norm(paragraph_vector) + 1e-8)
        
        return normalized_norm
​

向量聚合技术

不同聚合策略

class VectorAggregator:
    def __init__(self, vectors):
        self.vectors = np.array(vectors)
        
    def mean_aggregation(self):
        """均值聚合"""
        return np.mean(self.vectors, axis=0)
    
    def max_aggregation(self):
        """最大值聚合"""
        return np.max(self.vectors, axis=0)
    
    def weighted_aggregation(self, weights):
        """加权聚合"""
        weights = np.array(weights)
        weights = weights / weights.sum()
        return np.average(self.vectors, axis=0, weights=weights)
    
    def attention_aggregation(self, query_vector=None):
        """注意力聚合"""
        if query_vector is None:
            query_vector = np.mean(self.vectors, axis=0)
        
        # 计算注意力权重
        similarities = [cosine_similarity(vec, query_vector) for vec in self.vectors]
        weights = np.array(similarities)
        weights = weights / weights.sum()
        
        return np.average(self.vectors, axis=0, weights=weights)
​

聚合策略对比