判别模型与生成模型

# 判别模型直接建模 P(Y|X)
# 例如：逻辑回归
import numpy as np

def logistic_regression(X, weights, bias):
    """逻辑回归：典型的判别模型"""
    z = np.dot(X, weights) + bias
    return 1 / (1 + np.exp(-z))  # 直接输出 P(Y=1|X)

# 目标：最大化条件概率
# Loss = -Σ log P(Y_i|X_i)

# 生成模型建模 P(X,Y) = P(X|Y) * P(Y)
# 例如：朴素贝叶斯
from scipy.stats import norm

class NaiveBayes:
    """朴素贝叶斯：典型的生成模型"""
    
    def fit(self, X, y):
        self.classes = np.unique(y)
        self.class_priors = {}
        self.feature_params = {}
        
        for c in self.classes:
            # 学习先验概率 P(Y)
            self.class_priors[c] = np.mean(y == c)
            
            # 学习似然概率 P(X|Y)
            X_c = X[y == c]
            self.feature_params[c] = {
                'mean': np.mean(X_c, axis=0),
                'std': np.std(X_c, axis=0)
            }
    
    def predict_proba(self, X):
        """使用贝叶斯公式：P(Y|X) ∝ P(X|Y) * P(Y)"""
        probs = {}
        
        for c in self.classes:
            # P(Y=c)
            prior = self.class_priors[c]
            
            # P(X|Y=c) - 假设特征独立
            likelihood = np.prod(
                norm.pdf(X, 
                        self.feature_params[c]['mean'],
                        self.feature_params[c]['std']),
                axis=1
            )
            
            probs[c] = prior * likelihood
        
        return probs

from sklearn.linear_model import LogisticRegression
import matplotlib.pyplot as plt

# 逻辑回归：学习决策边界
def demo_logistic_regression():
    # 生成示例数据
    np.random.seed(42)
    X1 = np.random.multivariate_normal([2, 2], [[1, 0], [0, 1]], 100)
    X2 = np.random.multivariate_normal([6, 6], [[1, 0], [0, 1]], 100)
    
    X = np.vstack([X1, X2])
    y = np.hstack([np.zeros(100), np.ones(100)])
    
    # 训练模型
    model = LogisticRegression()
    model.fit(X, y)
    
    # 可视化决策边界
    plt.figure(figsize=(10, 8))
    plt.scatter(X1[:, 0], X1[:, 1], c='red', alpha=0.6, label='类别 0')
    plt.scatter(X2[:, 0], X2[:, 1], c='blue', alpha=0.6, label='类别 1')
    
    # 绘制决策边界
    x_min, x_max = X[:, 0].min() - 1, X[:, 0].max() + 1
    y_min, y_max = X[:, 1].min() - 1, X[:, 1].max() + 1
    xx, yy = np.meshgrid(np.arange(x_min, x_max, 0.1),
                         np.arange(y_min, y_max, 0.1))
    
    mesh_points = np.c_[xx.ravel(), yy.ravel()]
    Z = model.predict_proba(mesh_points)[:, 1]
    Z = Z.reshape(xx.shape)
    
    plt.contour(xx, yy, Z, levels=[0.5], colors='black', linestyles='--', linewidths=2)
    plt.title('逻辑回归 - 判别模型\n直接学习决策边界')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()

demo_logistic_regression()

from sklearn.svm import SVC

# SVM：找到最优分离超平面
svm_model = SVC(kernel='linear', probability=True)

# 特点：
# - 最大化间隔
# - 只关心决策边界附近的支持向量
# - 不对数据分布做假设

from sklearn.neural_network import MLPClassifier

# 神经网络：学习复杂的非线性映射
nn_model = MLPClassifier(hidden_layer_sizes=(100, 50), max_iter=1000)

# 特点：
# - 强大的表达能力
# - 端到端学习 P(Y|X)
# - 不提供概率解释

from sklearn.naive_bayes import GaussianNB
from sklearn.datasets import make_classification

def demo_naive_bayes():
    # 生成示例数据
    X, y = make_classification(n_samples=200, n_features=2, n_redundant=0, 
                             n_informative=2, n_clusters_per_class=1, random_state=42)
    
    # 训练朴素贝叶斯
    nb_model = GaussianNB()
    nb_model.fit(X, y)
    
    # 可视化类别分布
    plt.figure(figsize=(15, 5))
    
    # 子图1：原始数据
    plt.subplot(1, 3, 1)
    for class_label in np.unique(y):
        mask = y == class_label
        plt.scatter(X[mask, 0], X[mask, 1], alpha=0.6, label=f'类别 {class_label}')
    plt.title('原始数据分布')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    # 子图2：学习到的类条件分布
    plt.subplot(1, 3, 2)
    x_range = np.linspace(X[:, 0].min() - 1, X[:, 0].max() + 1, 100)
    y_range = np.linspace(X[:, 1].min() - 1, X[:, 1].max() + 1, 100)
    xx, yy = np.meshgrid(x_range, y_range)
    
    # 绘制类条件分布的等高线
    for class_label in np.unique(y):
        X_class = X[y == class_label]
        mean = np.mean(X_class, axis=0)
        cov = np.cov(X_class.T)
        
        pos = np.dstack((xx, yy))
        rv = multivariate_normal(mean, cov)
        plt.contour(xx, yy, rv.pdf(pos), alpha=0.6, label=f'P(X|Y={class_label})')
    
    plt.title('学习到的类条件分布 P(X|Y)')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    # 子图3：决策边界
    plt.subplot(1, 3, 3)
    mesh_points = np.c_[xx.ravel(), yy.ravel()]
    Z = nb_model.predict_proba(mesh_points)[:, 1]
    Z = Z.reshape(xx.shape)
    
    plt.contourf(xx, yy, Z, alpha=0.4, cmap='RdYlBu')
    plt.contour(xx, yy, Z, levels=[0.5], colors='black', linestyles='--')
    
    for class_label in np.unique(y):
        mask = y == class_label
        plt.scatter(X[mask, 0], X[mask, 1], alpha=0.6, label=f'类别 {class_label}')
    
    plt.title('朴素贝叶斯决策边界')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

from scipy.stats import multivariate_normal
demo_naive_bayes()

# HMM：时序数据的生成模型
class SimpleHMM:
    """简单的隐马尔可夫模型示例"""
    
    def __init__(self, n_states, n_observations):
        self.n_states = n_states
        self.n_observations = n_observations
        
        # 模型参数
        self.transition_probs = None  # P(S_t+1|S_t)
        self.emission_probs = None    # P(O_t|S_t)
        self.initial_probs = None     # P(S_1)
    
    def generate_sequence(self, length):
        """生成观测序列"""
        states = []
        observations = []
        
        # 初始状态
        current_state = np.random.choice(self.n_states, p=self.initial_probs)
        states.append(current_state)
        
        for t in range(length):
            # 生成观测
            obs = np.random.choice(self.n_observations, 
                                 p=self.emission_probs[current_state])
            observations.append(obs)
            
            # 状态转移
            if t < length - 1:
                current_state = np.random.choice(self.n_states,
                                               p=self.transition_probs[current_state])
                states.append(current_state)
        
        return states, observations

from sklearn.mixture import GaussianMixture

def demo_gaussian_mixture():
    """高斯混合模型演示"""
    
    # 生成混合数据
    np.random.seed(42)
    component1 = np.random.multivariate_normal([2, 2], [[1, 0], [0, 1]], 150)
    component2 = np.random.multivariate_normal([6, 6], [[1.5, 0.5], [0.5, 1.5]], 100)
    component3 = np.random.multivariate_normal([2, 6], [[1, -0.5], [-0.5, 1]], 120)
    
    X = np.vstack([component1, component2, component3])
    
    # 训练GMM
    gmm = GaussianMixture(n_components=3, random_state=42)
    gmm.fit(X)
    
    # 生成新样本
    new_samples, _ = gmm.sample(100)
    
    # 可视化
    plt.figure(figsize=(15, 5))
    
    # 原始数据
    plt.subplot(1, 3, 1)
    plt.scatter(X[:, 0], X[:, 1], alpha=0.6, c='blue')
    plt.title('原始数据')
    plt.grid(True, alpha=0.3)
    
    # 学习到的组件
    plt.subplot(1, 3, 2)
    colors = ['red', 'green', 'blue']
    for i in range(3):
        mean = gmm.means_[i]
        cov = gmm.covariances_[i]
        weight = gmm.weights_[i]
        
        # 绘制95%置信椭圆
        eigenvals, eigenvecs = np.linalg.eigh(cov)
        angle = np.degrees(np.arctan2(eigenvecs[1, 0], eigenvecs[0, 0]))
        width, height = 2 * np.sqrt(eigenvals)
        
        from matplotlib.patches import Ellipse
        ellipse = Ellipse(mean, width, height, angle=angle, 
                         facecolor=colors[i], alpha=0.3, 
                         label=f'组件{i+1} (权重={weight:.2f})')
        plt.gca().add_patch(ellipse)
        plt.scatter(*mean, color=colors[i], s=100, marker='x')
    
    plt.scatter(X[:, 0], X[:, 1], alpha=0.4, c='black', s=20)
    plt.title('学习到的混合成分')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    # 生成的新样本
    plt.subplot(1, 3, 3)
    plt.scatter(new_samples[:, 0], new_samples[:, 1], alpha=0.6, c='red')
    plt.title('生成的新样本')
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

demo_gaussian_mixture()

import torch
import torch.nn as nn

class DiscriminativeNet(nn.Module):
    """判别式神经网络"""
    
    def __init__(self, input_dim, hidden_dim, num_classes):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_dim, num_classes),
            nn.Softmax(dim=1)
        )
    
    def forward(self, x):
        """直接输出 P(Y|X)"""
        return self.network(x)

# 应用场景：
# - 图像分类 (CNN)
# - 文本分类 (BERT, RoBERTa)
# - 目标检测 (YOLO, R-CNN)

class SimpleVAE(nn.Module):
    """变分自编码器：生成模型"""
    
    def __init__(self, input_dim, latent_dim):
        super().__init__()
        
        # 编码器：学习 q(z|x)
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 512),
            nn.ReLU(),
            nn.Linear(512, 256),
            nn.ReLU()
        )
        
        self.fc_mu = nn.Linear(256, latent_dim)
        self.fc_logvar = nn.Linear(256, latent_dim)
        
        # 解码器：学习 p(x|z)
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 512),
            nn.ReLU(),
            nn.Linear(512, input_dim),
            nn.Sigmoid()
        )
    
    def encode(self, x):
        h = self.encoder(x)
        mu = self.fc_mu(h)
        logvar = self.fc_logvar(h)
        return mu, logvar
    
    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std
    
    def decode(self, z):
        return self.decoder(z)
    
    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        recon_x = self.decode(z)
        return recon_x, mu, logvar
    
    def generate(self, num_samples, device):
        """生成新样本"""
        with torch.no_grad():
            z = torch.randn(num_samples, self.fc_mu.out_features).to(device)
            samples = self.decode(z)
        return samples

# 其他生成模型：
# - GAN (生成对抗网络)
# - Flow-based models
# - Diffusion models

advantages_discriminative = {
    "预测准确性": "通常在分类任务上表现更好",
    "计算效率": "不需要建模完整的数据分布",
    "参数更少": "只需要学习决策边界",
    "对数据要求低": "不需要对数据分布做强假设",
    "处理高维数据": "在高维空间中表现良好"
}

disadvantages_discriminative = {
    "无法生成数据": "不能生成新的样本",
    "缺少概率解释": "难以量化不确定性",
    "处理缺失数据困难": "需要完整的特征输入",
    "领域适应性差": "难以迁移到新领域",
    "可解释性有限": "特别是深度模型"
}

advantages_generative = {
    "数据生成": "可以生成新的样本",
    "概率解释": "提供完整的概率框架",
    "处理缺失数据": "可以推断缺失的特征",
    "异常检测": "可以识别与训练分布不同的样本",
    "可解释性": "提供数据生成过程的洞察"
}

disadvantages_generative = {
    "计算复杂": "需要估计完整的联合分布",
    "参数更多": "模型复杂度高",
    "对假设敏感": "需要对数据分布做假设",
    "训练困难": "特别是高维数据",
    "预测性能": "在纯分类任务上可能不如判别模型"
}

def should_use_discriminative_model(task_characteristics):
    """判断是否应该使用判别模型"""
    
    use_discriminative = []
    
    if task_characteristics.get('primary_goal') == 'classification':
        use_discriminative.append("主要目标是分类准确性")
    
    if task_characteristics.get('data_completeness') == 'complete':
        use_discriminative.append("数据完整，无缺失值")
    
    if task_characteristics.get('computational_budget') == 'limited':
        use_discriminative.append("计算资源有限")
    
    if task_characteristics.get('interpretability_need') == 'low':
        use_discriminative.append("不需要强可解释性")
    
    return use_discriminative

# 示例场景
scenarios_discriminative = [
    "图像分类任务",
    "文本情感分析",
    "医疗诊断分类",
    "欺诈检测",
    "推荐系统的点击预测"
]

def should_use_generative_model(task_characteristics):
    """判断是否应该使用生成模型"""
    
    use_generative = []
    
    if task_characteristics.get('need_data_generation'):
        use_generative.append("需要生成新数据")
    
    if task_characteristics.get('missing_data') == 'frequent':
        use_generative.append("数据经常缺失")
    
    if task_characteristics.get('uncertainty_quantification'):
        use_generative.append("需要量化不确定性")
    
    if task_characteristics.get('anomaly_detection'):
        use_generative.append("需要异常检测")
    
    if task_characteristics.get('domain_understanding'):
        use_generative.append("需要理解数据生成过程")
    
    return use_generative

# 示例场景
scenarios_generative = [
    "数据增强和合成",
    "异常检测",
    "密度估计",
    "缺失值插补",
    "无监督学习",
    "强化学习中的环境建模"
]

class SemiSupervisedModel:
    """结合判别和生成模型的半监督学习"""
    
    def __init__(self):
        self.discriminative_model = LogisticRegression()
        self.generative_model = GaussianMixture(n_components=2)
    
    def fit(self, X_labeled, y_labeled, X_unlabeled):
        """使用标记和未标记数据训练"""
        
        # 步骤1：用标记数据训练判别模型
        self.discriminative_model.fit(X_labeled, y_labeled)
        
        # 步骤2：用判别模型为未标记数据生成伪标签
        pseudo_labels = self.discriminative_model.predict(X_unlabeled)
        
        # 步骤3：用所有数据训练生成模型
        X_all = np.vstack([X_labeled, X_unlabeled])
        y_all = np.hstack([y_labeled, pseudo_labels])
        
        # 为每个类别训练生成模型
        self.class_models = {}
        for class_label in np.unique(y_all):
            X_class = X_all[y_all == class_label]
            self.class_models[class_label] = GaussianMixture(n_components=1)
            self.class_models[class_label].fit(X_class)
        
        # 步骤4：迭代优化
        for iteration in range(5):
            # 使用生成模型重新评估未标记数据的标签
            new_pseudo_labels = self._predict_with_generative(X_unlabeled)
            
            # 重新训练判别模型
            X_combined = np.vstack([X_labeled, X_unlabeled])
            y_combined = np.hstack([y_labeled, new_pseudo_labels])
            self.discriminative_model.fit(X_combined, y_combined)
    
    def _predict_with_generative(self, X):
        """使用生成模型进行预测"""
        predictions = []
        
        for x in X:
            class_likelihoods = {}
            for class_label, model in self.class_models.items():
                likelihood = model.score_samples([x])[0]
                class_likelihoods[class_label] = likelihood
            
            predicted_class = max(class_likelihoods, key=class_likelihoods.get)
            predictions.append(predicted_class)
        
        return np.array(predictions)

class SimpleGAN:
    """生成对抗网络：同时使用生成和判别模型"""
    
    def __init__(self, input_dim, latent_dim):
        # 生成器：生成模型
        self.generator = nn.Sequential(
            nn.Linear(latent_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 256),
            nn.ReLU(),
            nn.Linear(256, input_dim),
            nn.Tanh()
        )
        
        # 判别器：判别模型
        self.discriminator = nn.Sequential(
            nn.Linear(input_dim, 256),
            nn.LeakyReLU(0.2),
            nn.Linear(256, 128),
            nn.LeakyReLU(0.2),
            nn.Linear(128, 1),
            nn.Sigmoid()
        )
    
    def train_step(self, real_data, latent_dim):
        """训练步骤：对抗训练"""
        batch_size = real_data.size(0)
        
        # 训练判别器
        # 真实数据
        real_labels = torch.ones(batch_size, 1)
        real_output = self.discriminator(real_data)
        d_loss_real = nn.BCELoss()(real_output, real_labels)
        
        # 生成数据
        noise = torch.randn(batch_size, latent_dim)
        fake_data = self.generator(noise)
        fake_labels = torch.zeros(batch_size, 1)
        fake_output = self.discriminator(fake_data.detach())
        d_loss_fake = nn.BCELoss()(fake_output, fake_labels)
        
        d_loss = d_loss_real + d_loss_fake
        
        # 训练生成器
        fake_output = self.discriminator(fake_data)
        g_loss = nn.BCELoss()(fake_output, real_labels)  # 欺骗判别器
        
        return d_loss, g_loss

# 判别方法：BERT分类器
from transformers import BertTokenizer, BertForSequenceClassification

class DiscriminativeTextClassifier:
    """基于BERT的判别式文本分类器"""
    
    def __init__(self):
        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
        self.model = BertForSequenceClassification.from_pretrained('bert-base-uncased')
    
    def predict(self, text):
        """直接预测文本类别"""
        inputs = self.tokenizer(text, return_tensors='pt', truncation=True, padding=True)
        outputs = self.model(**inputs)
        probabilities = torch.softmax(outputs.logits, dim=-1)
        return probabilities

# 生成方法：基于语言模型的分类
class GenerativeTextClassifier:
    """基于生成模型的文本分类器"""
    
    def __init__(self):
        # 为每个类别训练一个语言模型
        self.class_models = {}
    
    def fit(self, texts, labels):
        """为每个类别学习文本生成分布"""
        for label in set(labels):
            class_texts = [t for t, l in zip(texts, labels) if l == label]
            # 训练该类别的语言模型 P(text|class)
            self.class_models[label] = self._train_language_model(class_texts)
    
    def predict(self, text):
        """使用贝叶斯公式分类"""
        class_scores = {}
        for label, model in self.class_models.items():
            # 计算 P(text|class) * P(class)
            likelihood = model.score(text)
            prior = 1.0 / len(self.class_models)  # 假设均匀先验
            class_scores[label] = likelihood * prior
        
        return max(class_scores, key=class_scores.get)

# 判别方法：CNN分类器
class DiscriminativeImageClassifier:
    """卷积神经网络图像分类器"""
    
    def __init__(self, num_classes):
        self.model = nn.Sequential(
            nn.Conv2d(3, 32, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(32, 64, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Flatten(),
            nn.Linear(64 * 8 * 8, 128),
            nn.ReLU(),
            nn.Linear(128, num_classes)
        )
    
    def forward(self, x):
        """直接输出类别概率"""
        return torch.softmax(self.model(x), dim=1)

# 生成方法：VAE + 分类
class GenerativeImageClassifier:
    """基于VAE的生成式图像分类器"""
    
    def __init__(self, latent_dim, num_classes):
        self.vaes = {}  # 每个类别一个VAE
        self.num_classes = num_classes
    
    def fit(self, images, labels):
        """为每个类别训练VAE"""
        for class_label in range(self.num_classes):
            class_images = images[labels == class_label]
            self.vaes[class_label] = self._train_vae(class_images)
    
    def predict(self, image):
        """基于重构误差进行分类"""
        reconstruction_errors = {}
        
        for class_label, vae in self.vaes.items():
            # 计算重构误差作为似然的代理
            recon_image, _, _ = vae(image)
            error = torch.mean((image - recon_image) ** 2)
            reconstruction_errors[class_label] = error.item()
        
        # 选择重构误差最小的类别
        return min(reconstruction_errors, key=reconstruction_errors.get)

# 现代LLM既可以看作判别模型也可以看作生成模型
class ModernLLM:
    """现代大型语言模型的双重性质"""
    
    def discriminative_use(self, text, candidates):
        """判别式用法：从候选中选择"""
        prompt = f"给定文本: {text}\n从以下选项中选择最合适的: {candidates}"
        # 返回最可能的选项
        pass
    
    def generative_use(self, prompt):
        """生成式用法：生成新文本"""
        # 基于prompt生成新内容
        pass
    
    def few_shot_classification(self, examples, query):
        """少样本分类：结合两种特性"""
        prompt = "根据以下例子进行分类:\n"
        for text, label in examples:
            prompt += f"文本: {text} -> 类别: {label}\n"
        prompt += f"文本: {query} -> 类别: "
        # 生成类别标签
        pass

class DiffusionModel:
    """扩散模型：新兴的生成模型"""
    
    def __init__(self):
        # 前向过程：逐步添加噪声
        self.noise_schedule = self._create_noise_schedule()
        
        # 反向过程：学习去噪
        self.denoising_network = self._create_denoising_network()
    
    def forward_process(self, x0, t):
        """前向扩散过程"""
        noise = torch.randn_like(x0)
        alpha_t = self.noise_schedule[t]
        xt = torch.sqrt(alpha_t) * x0 + torch.sqrt(1 - alpha_t) * noise
        return xt, noise
    
    def reverse_process(self, xt, t):
        """反向去噪过程"""
        predicted_noise = self.denoising_network(xt, t)
        # 根据预测的噪声恢复原始数据
        return self._denoise_step(xt, predicted_noise, t)
    
    def generate(self, shape):
        """从噪声生成新样本"""
        x = torch.randn(shape)
        
        for t in reversed(range(self.num_timesteps)):
            x = self.reverse_process(x, t)
        
        return x

def model_selection_guide():
    """模型选择指南"""
    
    decision_tree = {
        "任务类型": {
            "纯分类/回归": "考虑判别模型",
            "数据生成": "选择生成模型",
            "异常检测": "生成模型更合适",
            "缺失值处理": "生成模型优势明显"
        },
        
        "数据特点": {
            "高维度": "判别模型通常更好",
            "小样本": "生成模型可能有优势",
            "数据完整": "判别模型足够",
            "频繁缺失": "生成模型必要"
        },
        
        "计算资源": {
            "有限": "优选判别模型",
            "充足": "可考虑复杂生成模型"
        },
        
        "可解释性": {
            "高要求": "简单生成模型(如朴素贝叶斯)",
            "低要求": "任意复杂模型"
        }
    }
    
    return decision_tree

# 最佳实践
best_practices = [
    "从简单模型开始，逐步增加复杂度",
    "根据具体任务选择合适的建模方式",
    "考虑使用预训练模型进行迁移学习",
    "评估模型时要考虑多个指标",
    "在实际部署前进行充分的验证"
]