自然语言处理与机器学习：从基础到实践

本文系统梳理 NLP 与 ML 领域的核心知识点，结合代码示例帮助读者深入理解关键技术。

---

目录

1. 机器学习基础
2. 文本预处理技术
3. 特征提取方法
4. 深度学习与NLP
5. 预训练语言模型
6. 实战案例：情感分析

---

1. 机器学习基础

1.1 监督学习 vs 无监督学习

类型 特点 NLP应用
监督学习 需要标注数据 文本分类、命名实体识别
无监督学习 无需标注数据 主题建模、聚类分析
半监督学习 少量标注+大量未标注 大规模文本分类

1.2 常用评估指标

from sklearn.metrics import accuracy_score, precision_recall_fscore_support
import numpy as np

def evaluate_model(y_true, y_pred):
    """
    计算分类模型的核心评估指标
    """
    accuracy = accuracy_score(y_true, y_pred)
    precision, recall, f1, _ = precision_recall_fscore_support(
        y_true, y_pred, average='weighted'
    )
    
    print(f"准确率 (Accuracy): {accuracy:.4f}")
    print(f"精确率 (Precision): {precision:.4f}")
    print(f"召回率 (Recall): {recall:.4f}")
    print(f"F1分数: {f1:.4f}")
    
    return {
        'accuracy': accuracy,
        'precision': precision,
        'recall': recall,
        'f1': f1
    }

# 示例使用
y_true = [0, 1, 1, 0, 1, 1, 0, 0, 1, 0]
y_pred = [0, 1, 0, 0, 1, 1, 0, 1, 1, 0]
metrics = evaluate_model(y_true, y_pred)

---

2. 文本预处理技术

文本预处理是 NLP pipeline 的第一步，质量直接影响模型效果。

2.1 完整预处理流程

import re
import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
from nltk.stem import PorterStemmer, WordNetLemmatizer

# 下载必要资源
# nltk.download('punkt')
# nltk.download('stopwords')
# nltk.download('wordnet')
# nltk.download('averaged_perceptron_tagger')

class TextPreprocessor:
    """文本预处理类：集成多种清洗技术"""
    
    def __init__(self, language='english'):
        self.stop_words = set(stopwords.words(language))
        self.stemmer = PorterStemmer()
        self.lemmatizer = WordNetLemmatizer()
    
    def clean_text(self, text):
        """基础清洗：去除特殊字符、转小写"""
        # 去除URL
        text = re.sub(r'http\S+|www\S+|https\S+', '', text, flags=re.MULTILINE)
        # 去除@提及和#标签
        text = re.sub(r'@\w+|#\w+', '', text)
        # 去除非字母字符
        text = re.sub(r'[^a-zA-Z\s]', '', text)
        # 转小写
        return text.lower().strip()
    
    def tokenize(self, text):
        """分词"""
        return word_tokenize(text)
    
    def remove_stopwords(self, tokens):
        """去除停用词"""
        return [token for token in tokens if token not in self.stop_words]
    
    def stem(self, tokens):
        """词干提取（Stemming）"""
        return [self.stemmer.stem(token) for token in tokens]
    
    def lemmatize(self, tokens):
        """词形还原（Lemmatization）- 比Stemming更精确"""
        return [self.lemmatizer.lemmatize(token) for token in tokens]
    
    def preprocess(self, text, use_lemmatization=True):
        """完整预处理流程"""
        text = self.clean_text(text)
        tokens = self.tokenize(text)
        tokens = self.remove_stopwords(tokens)
        
        if use_lemmatization:
            tokens = self.lemmatize(tokens)
        else:
            tokens = self.stem(tokens)
        
        return tokens

# 使用示例
preprocessor = TextPreprocessor()
sample_text = "Check out this amazing NLP tutorial at https://example.com! @nlp_expert #MachineLearning"
processed = preprocessor.preprocess(sample_text)
print(f"原始文本: {sample_text}")
print(f"预处理后: {processed}")

2.2 中文文本预处理

import jieba
import re

class ChineseTextPreprocessor:
    """中文文本预处理"""
    
    def __init__(self):
        # 中文停用词表
        self.stopwords = set([
            '的', '了', '在', '是', '我', '有', '和', '就', '不', '人',
            '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去',
            '你', '会', '着', '没有', '看', '好', '自己', '这'
        ])
    
    def clean(self, text):
        """清洗中文文本"""
        # 去除HTML标签
        text = re.sub(r'<[^>]+>', '', text)
        # 去除URL
        text = re.sub(r'http[s]?://\S+', '', text)
        # 去除非中文字符（保留中文、英文、数字）
        text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', ' ', text)
        return text.strip()
    
    def segment(self, text):
        """中文分词"""
        return list(jieba.cut(text))
    
    def preprocess(self, text):
        """完整预处理"""
        text = self.clean(text)
        tokens = self.segment(text)
        tokens = [t for t in tokens if t.strip() and t not in self.stopwords]
        return tokens

# 示例
# preprocessor = ChineseTextPreprocessor()
# text = "自然语言处理是人工智能的重要分支！"
# print(preprocessor.preprocess(text))

---

3. 特征提取方法

3.1 词袋模型 (Bag of Words)

from sklearn.feature_extraction.text import CountVectorizer

corpus = [
    'This is the first document.',
    'This document is the second document.',
    'And this is the third one.',
    'Is this the first document?'
]

# 词袋模型
vectorizer = CountVectorizer()
X = vectorizer.fit_transform(corpus)

print("词汇表:", vectorizer.get_feature_names_out())
print("特征矩阵:\n", X.toarray())

3.2 TF-IDF 向量化

from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

class TfidfFeatureExtractor:
    """TF-IDF特征提取器"""
    
    def __init__(self, max_features=5000, ngram_range=(1, 2)):
        self.vectorizer = TfidfVectorizer(
            max_features=max_features,
            ngram_range=ngram_range,
            stop_words='english',
            min_df=2,  # 忽略在少于2个文档中出现的词
            max_df=0.8  # 忽略在超过80%文档中出现的词
        )
    
    def fit_transform(self, texts):
        """拟合并转换文本"""
        return self.vectorizer.fit_transform(texts)
    
    def transform(self, texts):
        """转换新文本"""
        return self.vectorizer.transform(texts)
    
    def get_top_features(self, text, n=10):
        """获取文本中TF-IDF值最高的特征词"""
        tfidf_matrix = self.vectorizer.transform([text])
        feature_array = np.array(self.vectorizer.get_feature_names_out())
        tfidf_sorting = np.argsort(tfidf_matrix.toarray()).flatten()[::-1]
        
        top_n = feature_array[tfidf_sorting][:n]
        scores = tfidf_matrix.toarray().flatten()[tfidf_sorting][:n]
        
        return list(zip(top_n, scores))

# 使用示例
corpus = [
    "Machine learning is amazing for NLP tasks",
    "Deep learning revolutionizes natural language processing",
    "NLP enables computers to understand human language"
]

extractor = TfidfFeatureExtractor(max_features=100)
tfidf_matrix = extractor.fit_transform(corpus)
print(f"TF-IDF矩阵形状: {tfidf_matrix.shape}")

3.3 Word2Vec 词嵌入

from gensim.models import Word2Vec
from nltk.tokenize import word_tokenize

# 准备训练数据（已分词的句子）
sentences = [
    ['natural', 'language', 'processing', 'is', 'fascinating'],
    ['machine', 'learning', 'powers', 'modern', 'nlp'],
    ['deep', 'learning', 'improves', 'language', 'understanding'],
    ['word', 'embeddings', 'capture', 'semantic', 'meaning'],
    ['neural', 'networks', 'process', 'text', 'efficiently']
]

# 训练Word2Vec模型
model = Word2Vec(
    sentences=sentences,
    vector_size=100,    # 词向量维度
    window=5,           # 上下文窗口大小
    min_count=1,        # 最小词频
    workers=4,          # 训练线程数
    sg=1                # 1=Skip-gram, 0=CBOW
)

# 获取词向量
vector = model.wv['learning']
print(f"'learning'的词向量维度: {vector.shape}")

# 找相似词
similar_words = model.wv.most_similar('learning', topn=3)
print(f"与'learning'最相似的词: {similar_words}")

# 计算词向量相似度
similarity = model.wv.similarity('machine', 'deep')
print(f"'machine'与'deep'的相似度: {similarity:.4f}")

---

4. 深度学习与NLP

4.1 RNN 与 LSTM 基础

import torch
import torch.nn as nn

class LSTMTextClassifier(nn.Module):
    """基于LSTM的文本分类器"""
    
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers=2):
        super().__init__()
        
        # 词嵌入层
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        
        # LSTM层
        self.lstm = nn.LSTM(
            embedding_dim,
            hidden_dim,
            num_layers=n_layers,
            bidirectional=True,
            dropout=0.3,
            batch_first=True
        )
        
        # 全连接层
        self.fc = nn.Linear(hidden_dim * 2, output_dim)
        self.dropout = nn.Dropout(0.3)
    
    def forward(self, text):
        # text: [batch_size, seq_len]
        
        embedded = self.embedding(text)  # [batch_size, seq_len, embedding_dim]
        
        # LSTM输出
        lstm_out, (hidden, cell) = self.lstm(embedded)
        
        # 拼接双向LSTM的最后隐藏状态
        hidden = torch.cat((hidden[-2, :, :], hidden[-1, :, :]), dim=1)
        
        hidden = self.dropout(hidden)
        output = self.fc(hidden)
        
        return output

# 模型实例化示例
vocab_size = 10000
embedding_dim = 128
hidden_dim = 256
output_dim = 2  # 二分类

model = LSTMTextClassifier(vocab_size, embedding_dim, hidden_dim, output_dim)
print(model)

4.2 自注意力机制

import torch
import torch.nn as nn
import math

class SelfAttention(nn.Module):
    """缩放点积自注意力机制"""
    
    def __init__(self, embed_size, heads):
        super().__init__()
        self.embed_size = embed_size
        self.heads = heads
        self.head_dim = embed_size // heads
        
        assert self.head_dim * heads == embed_size, "Embedding size must be divisible by heads"
        
        # Q, K, V 线性变换
        self.values = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.keys = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.queries = nn.Linear(self.head_dim, self.head_dim, bias=False)
        self.fc_out = nn.Linear(heads * self.head_dim, embed_size)
    
    def forward(self, values, keys, query, mask=None):
        N = query.shape[0]  # batch size
        value_len, key_len, query_len = values.shape[1], keys.shape[1], query.shape[1]
        
        # 分割成多个头
        values = values.reshape(N, value_len, self.heads, self.head_dim)
        keys = keys.reshape(N, key_len, self.heads, self.head_dim)
        queries = query.reshape(N, query_len, self.heads, self.head_dim)
        
        # 线性变换
        values = self.values(values)
        keys = self.keys(keys)
        queries = self.queries(queries)
        
        # 计算注意力分数: Q @ K^T
        energy = torch.einsum("nqhd,nkhd->nhqk", [queries, keys])
        
        if mask is not None:
            energy = energy.masked_fill(mask == 0, float("-1e20"))
        
        # 缩放并softmax
        attention = torch.softmax(energy / math.sqrt(self.head_dim), dim=3)
        
        # 加权求和: attention @ V
        out = torch.einsum("nhql,nlhd->nqhd", [attention, values])
        out = out.reshape(N, query_len, self.heads * self.head_dim)
        
        return self.fc_out(out)

# 测试
embed_size = 256
heads = 8
seq_len = 10
batch_size = 4

attention = SelfAttention(embed_size, heads)
x = torch.randn(batch_size, seq_len, embed_size)
output = attention(x, x, x)
print(f"输入形状: {x.shape}")
print(f"输出形状: {output.shape}")

---

5. 预训练语言模型

5.1 使用 Transformers 库

from transformers import (
    BertTokenizer, 
    BertForSequenceClassification,
    pipeline,
    AutoModel, 
    AutoTokenizer
)
import torch

class BertTextAnalyzer:
    """基于BERT的文本分析器"""
    
    def __init__(self, model_name='bert-base-chinese'):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.tokenizer = BertTokenizer.from_pretrained(model_name)
        self.model = BertForSequenceClassification.from_pretrained(
            model_name, 
            num_labels=2
        ).to(self.device)
    
    def predict(self, texts):
        """批量预测"""
        self.model.eval()
        
        if isinstance(texts, str):
            texts = [texts]
        
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=512,
            return_tensors='pt'
        ).to(self.device)
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            predictions = torch.argmax(outputs.logits, dim=-1)
        
        return predictions.cpu().numpy()
    
    def get_embeddings(self, text):
        """获取文本的BERT嵌入向量"""
        self.model.eval()
        
        inputs = self.tokenizer(
            text,
            return_tensors='pt',
            truncation=True,
            max_length=512
        ).to(self.device)
        
        with torch.no_grad():
            outputs = self.model.bert(**inputs)
            # 使用[CLS] token的表示作为句子嵌入
            embeddings = outputs.last_hidden_state[:, 0, :]
        
        return embeddings.cpu().numpy()

# 使用Hugging Face Pipeline进行快速推理
def quick_sentiment_analysis():
    """快速情感分析"""
    classifier = pipeline(
        'sentiment-analysis',
        model='distilbert-base-uncased-finetuned-sst-2-english'
    )
    
    texts = [
        "I love this product! It's amazing.",
        "This is the worst experience ever.",
        "The movie was okay, nothing special."
    ]
    
    results = classifier(texts)
    for text, result in zip(texts, results):
        print(f"文本: {text}")
        print(f"情感: {result['label']}, 置信度: {result['score']:.4f}\n")

# 运行示例
# quick_sentiment_analysis()

5.2 微调预训练模型

from transformers import AdamW, get_linear_schedule_with_warmup
from torch.utils.data import DataLoader, Dataset

class TextDataset(Dataset):
    """自定义文本数据集"""
    
    def __init__(self, texts, labels, tokenizer, max_len=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]
        
        encoding = self.tokenizer.encode_plus(
            text,
            add_special_tokens=True,
            max_length=self.max_len,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

def train_epoch(model, data_loader, optimizer, scheduler, device):
    """训练一个epoch"""
    model.train()
    total_loss = 0
    
    for batch in data_loader:
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['labels'].to(device)
        
        optimizer.zero_grad()
        
        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            labels=labels
        )
        
        loss = outputs.loss
        total_loss += loss.item()
        
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        scheduler.step()
    
    return total_loss / len(data_loader)

# 微调配置示例
"""
# 准备数据
train_texts = ["I love this", "This is bad", "Amazing product", "Terrible service"]
train_labels = [1, 0, 1, 0]  # 1=positive, 0=negative

# 创建数据集
dataset = TextDataset(train_texts, train_labels, tokenizer)
data_loader = DataLoader(dataset, batch_size=2, shuffle=True)

# 优化器和学习率调度
optimizer = AdamW(model.parameters(), lr=2e-5, correct_bias=False)
epochs = 3
total_steps = len(data_loader) * epochs
scheduler = get_linear_schedule_with_warmup(
    optimizer,
    num_warmup_steps=0,
    num_training_steps=total_steps
)

# 训练循环
for epoch in range(epochs):
    avg_loss = train_epoch(model, data_loader, optimizer, scheduler, device)
    print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.4f}")
"""

---

6. 实战案例：情感分析

6.1 完整项目代码

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, confusion_matrix
import seaborn as sns
import matplotlib.pyplot as plt
from collections import Counter
import re

class SentimentAnalyzer:
    """完整的情感分析系统"""
    
    def __init__(self):
        self.vectorizer = TfidfVectorizer(
            max_features=5000,
            ngram_range=(1, 2),
            min_df=2
        )
        self.classifier = LogisticRegression(max_iter=1000, C=10)
        self.preprocessor = TextPreprocessor()
    
    def preprocess_data(self, texts):
        """批量预处理文本"""
        processed = []
        for text in texts:
            tokens = self.preprocessor.preprocess(text)
            processed.append(' '.join(tokens))
        return processed
    
    def train(self, texts, labels):
        """训练模型"""
        print("预处理文本...")
        processed_texts = self.preprocess_data(texts)
        
        print("提取TF-IDF特征...")
        X = self.vectorizer.fit_transform(processed_texts)
        
        print("训练分类器...")
        self.classifier.fit(X, labels)
        
        # 输出特征重要性
        feature_names = self.vectorizer.get_feature_names_out()
        coefs = self.classifier.coef_[0]
        top_positive = np.argsort(coefs)[-10:]
        top_negative = np.argsort(coefs)[:10]
        
        print("\n最积极的特征词:")
        for idx in reversed(top_positive):
            print(f"  {feature_names[idx]}: {coefs[idx]:.4f}")
        
        print("\n最消极的特征词:")
        for idx in top_negative:
            print(f"  {feature_names[idx]}: {coefs[idx]:.4f}")
    
    def predict(self, texts):
        """预测新文本"""
        processed = self.preprocess_data(texts)
        X = self.vectorizer.transform(processed)
        return self.classifier.predict(X)
    
    def predict_proba(self, texts):
        """预测概率"""
        processed = self.preprocess_data(texts)
        X = self.vectorizer.transform(processed)
        return self.classifier.predict_proba(X)
    
    def evaluate(self, texts, labels):
        """评估模型"""
        predictions = self.predict(texts)
        
        print("\n分类报告:")
        print(classification_report(labels, predictions, target_names=['Negative', 'Positive']))
        
        # 绘制混淆矩阵
        cm = confusion_matrix(labels, predictions)
        plt.figure(figsize=(8, 6))
        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
                   xticklabels=['Negative', 'Positive'],
                   yticklabels=['Negative', 'Positive'])
        plt.xlabel('Predicted')
        plt.ylabel('Actual')
        plt.title('Confusion Matrix')
        plt.tight_layout()
        plt.savefig('confusion_matrix.png', dpi=150)
        plt.show()

# 模拟数据演示
def demo_sentiment_analysis():
    """情感分析演示"""
    
    # 模拟数据集
    data = {
        'text': [
            "This movie was absolutely fantastic! Loved every minute.",
            "Terrible waste of time. The plot made no sense.",
            "Best film I've seen this year. Highly recommended!",
            "Boring and predictable. Don't bother watching.",
            "Amazing performances by the entire cast!",
            "Worst acting ever. Complete disaster.",
            "A masterpiece of cinema. Beautifully shot.",
            "So disappointing. Expected much better.",
            "Incredible story, great direction!",
            "Dull and lifeless. Fell asleep halfway through."
        ],
        'label': [1, 0, 1, 0, 1, 0, 1, 0, 1, 0]  # 1=positive, 0=negative
    }
    
    df = pd.DataFrame(data)
    
    # 划分训练集和测试集
    train_df, test_df = train_test_split(df, test_size=0.3, random_state=42)
    
    # 创建并训练模型
    analyzer = SentimentAnalyzer()
    analyzer.train(train_df['text'].tolist(), train_df['label'].tolist())
    
    # 评估
    analyzer.evaluate(test_df['text'].tolist(), test_df['label'].tolist())
    
    # 预测新样本
    new_reviews = [
        "An absolutely wonderful experience!",
        "Complete waste of money and time."
    ]
    predictions = analyzer.predict(new_reviews)
    probabilities = analyzer.predict_proba(new_reviews)
    
    print("\n新样本预测:")
    for review, pred, prob in zip(new_reviews, predictions, probabilities):
        sentiment = "Positive" if pred == 1 else "Negative"
        confidence = prob[pred]
        print(f"文本: {review}")
        print(f"预测: {sentiment} (置信度: {confidence:.4f})\n")

# 运行演示
# demo_sentiment_analysis()

6.2 模型保存与加载

import pickle
import joblib

def save_model(analyzer, filepath='sentiment_model.pkl'):
    """保存训练好的模型"""
    model_data = {
        'vectorizer': analyzer.vectorizer,
        'classifier': analyzer.classifier,
        'preprocessor': analyzer.preprocessor
    }
    with open(filepath, 'wb') as f:
        pickle.dump(model_data, f)
    print(f"模型已保存到 {filepath}")

def load_model(filepath='sentiment_model.pkl'):
    """加载训练好的模型"""
    with open(filepath, 'rb') as f:
        model_data = pickle.load(f)
    
    analyzer = SentimentAnalyzer()
    analyzer.vectorizer = model_data['vectorizer']
    analyzer.classifier = model_data['classifier']
    analyzer.preprocessor = model_data['preprocessor']
    
    print(f"模型已从 {filepath} 加载")
    return analyzer

---

总结

本文系统梳理了 NLP 与机器学习的核心知识点：

技术领域 关键要点
文本预处理 清洗 → 分词 → 去停用词 → 词干提取/词形还原
特征提取 Bag of Words → TF-IDF → Word2Vec
深度学习 RNN → LSTM → Attention → Transformer
预训练模型 BERT → Fine-tuning → 下游任务适配

学习路径建议

入门 → 掌握Python基础 + 机器学习概念
  ↓
进阶 → 深入理解NLP预处理与经典算法
  ↓
提高 → 学习深度学习框架(PyTorch/TensorFlow)
  ↓
精通 → 掌握Transformer架构与预训练模型微调

---

如有问题，欢迎在评论区交流讨论。