← 返回首页
🤖

自然语言处理实践详解

📂 ai ⏱ 6 min 1139 words

自然语言处理实践详解

自然语言处理(Natural Language Processing,NLP)是让计算机理解、解释和生成人类语言的技术。

NLP基础

文本预处理

import numpy as np
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
from torchtext.data.utils import get_tokenizer
from collections import Counter
import re
import warnings
warnings.filterwarnings('ignore')

# 文本预处理类
class TextPreprocessor:
    def __init__(self, vocab_size=10000, max_length=200):
        self.vocab_size = vocab_size
        self.max_length = max_length
        self.word2idx = {}
        self.idx2word = {}
        self.vocab = None
    
    def clean_text(self, text):
        """清理文本"""
        # 转换为小写
        text = text.lower()
        # 移除特殊字符
        text = re.sub(r'[^a-zA-Z0-9\s]', '', text)
        # 移除多余空格
        text = re.sub(r'\s+', ' ', text).strip()
        return text
    
    def build_vocab(self, texts):
        """构建词汇表"""
        # 统计词频
        word_counts = Counter()
        for text in texts:
            cleaned = self.clean_text(text)
            words = cleaned.split()
            word_counts.update(words)
        
        # 选择最常见的词
        most_common = word_counts.most_common(self.vocab_size - 2)  # 保留特殊token
        
        # 构建词汇表
        self.word2idx = {'<PAD>': 0, '<UNK>': 1}
        for word, _ in most_common:
            self.word2idx[word] = len(self.word2idx)
        
        self.idx2word = {v: k for k, v in self.word2idx.items()}
        self.vocab = list(self.word2idx.keys())
        
        print(f"词汇表大小: {len(self.word2idx)}")
    
    def text_to_sequence(self, text):
        """将文本转换为序列"""
        cleaned = self.clean_text(text)
        words = cleaned.split()
        
        sequence = []
        for word in words:
            if word in self.word2idx:
                sequence.append(self.word2idx[word])
            else:
                sequence.append(self.word2idx['<UNK>'])
        
        # 填充或截断
        if len(sequence) < self.max_length:
            sequence = sequence + [self.word2idx['<PAD>']] * (self.max_length - len(sequence))
        else:
            sequence = sequence[:self.max_length]
        
        return sequence
    
    def texts_to_sequences(self, texts):
        """批量转换文本"""
        return [self.text_to_sequence(text) for text in texts]

# 创建示例文本数据
np.random.seed(42)
n_samples = 1000
texts = [
    "这是一个很好的产品,质量非常好",
    "服务态度很差,不推荐购买",
    "物流很快,包装也很好",
    "价格有点贵,但质量不错",
    "客服回复很及时,问题解决了",
    "产品和描述不符,很失望",
    "下次还会购买,非常满意",
    "质量太差了,用了一天就坏了",
    "性价比很高,推荐购买",
    "发货速度很快,第二天就到了"
]

# 扩展文本数据
expanded_texts = []
for _ in range(n_samples):
    text = np.random.choice(texts)
    # 添加一些变化
    if np.random.random() > 0.5:
        text = text + ",真的不错"
    expanded_texts.append(text)

# 创建标签(0: 负面, 1: 正面)
labels = np.random.randint(0, 2, n_samples)

print(f"样本数量: {len(expanded_texts)}")
print(f"标签分布: {np.bincount(labels)}")

文本分类

数据集准备

# 文本分类数据集
class TextClassificationDataset(Dataset):
    def __init__(self, texts, labels, preprocessor):
        self.texts = texts
        self.labels = labels
        self.preprocessor = preprocessor
        
        # 预处理文本
        self.sequences = preprocessor.texts_to_sequences(texts)
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        sequence = self.sequences[idx]
        label = self.labels[idx]
        
        return torch.LongTensor(sequence), torch.LongTensor([label])

# 预处理文本
preprocessor = TextPreprocessor(vocab_size=5000, max_length=100)
preprocessor.build_vocab(expanded_texts)

# 创建数据集
dataset = TextClassificationDataset(expanded_texts, labels, preprocessor)

# 划分数据集
train_size = int(0.8 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])

# 创建数据加载器
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)

print(f"训练集大小: {len(train_dataset)}")
print(f"验证集大小: {len(val_dataset)}")

简单文本分类模型

class TextCNN(nn.Module):
    def __init__(self, vocab_size, embed_size, num_classes, num_filters=100, filter_sizes=[3, 4, 5]):
        super(TextCNN, self).__init__()
        
        self.embedding = nn.Embedding(vocab_size, embed_size)
        
        self.convs = nn.ModuleList([
            nn.Conv1d(embed_size, num_filters, kernel_size=fs)
            for fs in filter_sizes
        ])
        
        self.dropout = nn.Dropout(0.5)
        self.fc = nn.Linear(num_filters * len(filter_sizes), num_classes)
    
    def forward(self, x):
        # 嵌入层
        x = self.embedding(x)  # [batch_size, seq_len, embed_size]
        x = x.permute(0, 2, 1)  # [batch_size, embed_size, seq_len]
        
        # 卷积层
        conv_outputs = []
        for conv in self.convs:
            conv_out = torch.relu(conv(x))  # [batch_size, num_filters, seq_len - fs + 1]
            conv_out = torch.max(conv_out, dim=2)[0]  # [batch_size, num_filters]
            conv_outputs.append(conv_out)
        
        # 拼接
        out = torch.cat(conv_outputs, dim=1)  # [batch_size, num_filters * len(filter_sizes)]
        
        # 全连接层
        out = self.dropout(out)
        out = self.fc(out)
        
        return out

# 创建模型
vocab_size = len(preprocessor.word2idx)
embed_size = 128
num_classes = 2

model = TextCNN(vocab_size, embed_size, num_classes).to(device)
print(f"模型参数: {sum(p.numel() for p in model.parameters()):,}")

训练模型

def train_text_classifier(model, train_loader, val_loader, epochs=10, lr=0.001):
    """训练文本分类模型"""
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=lr)
    
    train_losses = []
    val_accs = []
    
    for epoch in range(epochs):
        # 训练阶段
        model.train()
        running_loss = 0.0
        correct = 0
        total = 0
        
        for batch_X, batch_y in train_loader:
            batch_X, batch_y = batch_X.to(device), batch_y.to(device)
            batch_y = batch_y.squeeze()
            
            optimizer.zero_grad()
            outputs = model(batch_X)
            loss = criterion(outputs, batch_y)
            loss.backward()
            optimizer.step()
            
            running_loss += loss.item()
            _, predicted = torch.max(outputs.data, 1)
            total += batch_y.size(0)
            correct += (predicted == batch_y).sum().item()
        
        train_loss = running_loss / len(train_loader)
        train_acc = 100 * correct / total
        train_losses.append(train_loss)
        
        # 验证阶段
        val_acc = evaluate_text_classifier(model, val_loader)
        val_accs.append(val_acc)
        
        print(f'Epoch [{epoch+1}/{epochs}], '
              f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, '
              f'Val Acc: {val_acc:.2f}%')
    
    return train_losses, val_accs

def evaluate_text_classifier(model, val_loader):
    """评估文本分类模型"""
    model.eval()
    correct = 0
    total = 0
    
    with torch.no_grad():
        for batch_X, batch_y in val_loader:
            batch_X, batch_y = batch_X.to(device), batch_y.to(device)
            batch_y = batch_y.squeeze()
            
            outputs = model(batch_X)
            _, predicted = torch.max(outputs.data, 1)
            total += batch_y.size(0)
            correct += (predicted == batch_y).sum().item()
    
    return 100 * correct / total

# 训练模型
print("训练文本分类模型:")
train_losses, val_accs = train_text_classifier(model, train_loader, val_loader, epochs=10)

# 可视化训练过程
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))

ax1.plot(train_losses, 'b-', linewidth=2)
ax1.set_xlabel('Epoch')
ax1.set_ylabel('损失')
ax1.set_title('训练损失曲线')
ax1.grid(True, alpha=0.3)

ax2.plot(val_accs, 'r-', linewidth=2)
ax2.set_xlabel('Epoch')
ax2.set_ylabel('准确率')
ax2.set_title('验证准确率曲线')
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

使用预训练模型

BERT模型

from transformers import BertTokenizer, BertForSequenceClassification, AdamW
from transformers import get_linear_schedule_with_warmup

# 加载预训练的BERT模型
def load_bert_model(num_classes):
    """加载BERT模型"""
    tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
    model = BertForSequenceClassification.from_pretrained('bert-base-chinese', num_labels=num_classes)
    
    return tokenizer, model

# 创建BERT分类器
class BERTClassifier:
    def __init__(self, num_classes=2):
        self.num_classes = num_classes
        self.tokenizer = None
        self.model = None
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        
        self._load_model()
    
    def _load_model(self):
        """加载模型"""
        try:
            self.tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
            self.model = BertForSequenceClassification.from_pretrained(
                'bert-base-chinese', 
                num_labels=self.num_classes
            ).to(self.device)
            print("BERT模型加载成功")
        except Exception as e:
            print(f"BERT模型加载失败: {e}")
            print("使用模拟BERT模型")
            self._create_mock_model()
    
    def _create_mock_model(self):
        """创建模拟BERT模型"""
        self.model = nn.Sequential(
            nn.Embedding(21128, 768),  # BERT词汇表大小
            nn.Linear(768, 256),
            nn.ReLU(),
            nn.Linear(256, self.num_classes)
        ).to(self.device)
        
        # 创建模拟tokenizer
        class MockTokenizer:
            def __call__(self, texts, padding=True, truncation=True, max_length=128, return_tensors='pt'):
                if isinstance(texts, str):
                    texts = [texts]
                
                # 模拟tokenization
                sequences = []
                for text in texts:
                    # 简单的字符级tokenization
                    seq = [ord(c) % 21128 for c in text[:max_length]]
                    seq = seq + [0] * (max_length - len(seq))  # 填充
                    sequences.append(seq)
                
                return {
                    'input_ids': torch.tensor(sequences).to(self.device),
                    'attention_mask': torch.ones(len(texts), max_length).to(self.device)
                }
        
        self.tokenizer = MockTokenizer()
    
    def predict(self, texts):
        """预测"""
        self.model.eval()
        
        # 预处理
        inputs = self.tokenizer(texts, padding=True, truncation=True, max_length=128)
        
        # 预测
        with torch.no_grad():
            outputs = self.model(inputs['input_ids'])
            probabilities = torch.softmax(outputs, dim=1)
            predictions = torch.argmax(probabilities, dim=1)
        
        return predictions.cpu().numpy(), probabilities.cpu().numpy()

# 创建BERT分类器
bert_classifier = BERTClassifier(num_classes=2)

# 测试预测
sample_texts = ["这个产品很好", "服务态度很差"]
predictions, probabilities = bert_classifier.predict(sample_texts)

print("\nBERT预测结果:")
for text, pred, prob in zip(sample_texts, predictions, probabilities):
    print(f"文本: {text}")
    print(f"预测: {'正面' if pred == 1 else '负面'}")
    print(f"置信度: {prob.max():.2%}")
    print()

文本生成

简单文本生成

class TextGenerator:
    def __init__(self, vocab_size, embed_size, hidden_size):
        self.vocab_size = vocab_size
        self.embed_size = embed_size
        self.hidden_size = hidden_size
        
        # 创建模型
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.rnn = nn.LSTM(embed_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)
    
    def forward(self, x, hidden=None):
        # 嵌入层
        x = self.embedding(x)
        
        # RNN层
        if hidden is None:
            output, hidden = self.rnn(x)
        else:
            output, hidden = self.rnn(x, hidden)
        
        # 输出层
        output = self.fc(output)
        
        return output, hidden
    
    def generate(self, start_sequence, length=50, temperature=1.0):
        """生成文本"""
        self.eval()
        
        current_sequence = start_sequence
        
        with torch.no_grad():
            for _ in range(length):
                # 前向传播
                output, _ = self.forward(current_sequence)
                
                # 获取最后一个时间步的输出
                logits = output[:, -1, :] / temperature
                
                # 采样
                probabilities = torch.softmax(logits, dim=1)
                next_token = torch.multinomial(probabilities, 1)
                
                # 更新序列
                current_sequence = torch.cat([current_sequence, next_token], dim=1)
        
        return current_sequence

# 创建文本生成器
vocab_size = len(preprocessor.word2idx)
text_gen = TextGenerator(vocab_size, embed_size=128, hidden_size=256)

print("文本生成器已创建")
print("使用方法:text_gen.generate(start_sequence)")

实际应用

情感分析系统

class SentimentAnalysisSystem:
    def __init__(self):
        self.preprocessor = TextPreprocessor(vocab_size=5000, max_length=100)
        self.model = None
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        
        self._build_model()
    
    def _build_model(self):
        """构建模型"""
        # 使用简单的TextCNN
        self.model = TextCNN(
            vocab_size=5000,
            embed_size=128,
            num_classes=2
        ).to(self.device)
    
    def train(self, texts, labels, epochs=10):
        """训练模型"""
        # 构建词汇表
        self.preprocessor.build_vocab(texts)
        
        # 创建数据集
        dataset = TextClassificationDataset(texts, labels, self.preprocessor)
        
        # 划分数据集
        train_size = int(0.8 * len(dataset))
        val_size = len(dataset) - train_size
        train_dataset, val_dataset = torch.utils.data.random_split(
            dataset, [train_size, val_size]
        )
        
        # 创建数据加载器
        train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
        val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)
        
        # 训练
        print("训练情感分析模型:")
        train_losses, val_accs = train_text_classifier(
            self.model, train_loader, val_loader, epochs=epochs
        )
        
        return train_losses, val_accs
    
    def predict(self, text):
        """预测情感"""
        self.model.eval()
        
        # 预处理
        sequence = self.preprocessor.text_to_sequence(text)
        sequence_tensor = torch.LongTensor([sequence]).to(self.device)
        
        # 预测
        with torch.no_grad():
            output = self.model(sequence_tensor)
            probabilities = torch.softmax(output, dim=1)
            prediction = torch.argmax(probabilities, dim=1)
        
        result = {
            'text': text,
            'sentiment': '正面' if prediction.item() == 1 else '负面',
            'confidence': probabilities[0][prediction.item()].item(),
            'probabilities': {
                '负面': probabilities[0][0].item(),
                '正面': probabilities[0][1].item()
            }
        }
        
        return result

# 创建情感分析系统
sentiment_system = SentimentAnalysisSystem()

# 训练系统
train_losses, val_accs = sentiment_system.train(expanded_texts, labels, epochs=10)

# 测试预测
test_texts = [
    "这个产品真的很好用,强烈推荐!",
    "服务太差了,再也不买了",
    "质量不错,价格也合理",
    "物流很慢,等了很久"
]

print("\n情感分析结果:")
for text in test_texts:
    result = sentiment_system.predict(text)
    print(f"文本: {text}")
    print(f"情感: {result['sentiment']}")
    print(f"置信度: {result['confidence']:.2%}")
    print()

NLP最佳实践

  1. 文本预处理:清理文本、分词、去停用词
  2. 词汇表构建:选择合适的词汇表大小
  3. 嵌入层:使用预训练词向量或学习嵌入
  4. 模型选择:根据任务选择CNN、RNN或Transformer
  5. 迁移学习:使用预训练模型(BERT、GPT等)

自然语言处理是人工智能的重要分支,掌握NLP技术对于文本分类、情感分析、机器翻译等应用至关重要。