AI 在医疗健康中的应用

# 医疗 AI 系统架构
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional, Dict
import numpy as np

@dataclass
class PatientRecord:
    patient_id: str
    age: int
    gender: str
    vital_signs: Dict[str, float]
    lab_results: Dict[str, float]
    imaging_data: Optional[np.ndarray] = None
    medical_history: List[str] = None

@dataclass
class PredictionResult:
    diagnosis: str
    confidence: float
    explanation: str
    recommended_actions: List[str]

class MedicalAIModel(ABC):
    @abstractmethod
    def predict(self, patient: PatientRecord) -> PredictionResult:
        pass

    @abstractmethod
    def explain(self, patient: PatientRecord) -> str:
        pass

    @abstractmethod
    def get_uncertainty(self, patient: PatientRecord) -> float:
        pass

# 医疗数据评估指标
import numpy as np
from sklearn.metrics import confusion_matrix, roc_auc_score

def medical_metrics(y_true: np.ndarray, y_pred: np.ndarray,
                    y_prob: np.ndarray) -> dict:
    tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
    sensitivity = tp / (tp + fn)
    specificity = tn / (tn + fp)
    precision = tp / (tp + fp)
    f1 = 2 * precision * sensitivity / (precision + sensitivity)
    npv = tn / (tn + fn)
    auc = roc_auc_score(y_true, y_prob)
    youden = sensitivity + specificity - 1
    return {
        "sensitivity": round(sensitivity, 4),
        "specificity": round(specificity, 4),
        "precision": round(precision, 4),
        "f1_score": round(f1, 4),
        "npv": round(npv, 4),
        "auc_roc": round(auc, 4),
        "youden_index": round(youden, 4)
    }

# U-Net 医学图像分割（PyTorch）
import torch
import torch.nn as nn

class DoubleConv(nn.Module):
    def __init__(self, in_ch, out_ch):
        super().__init__()
        self.conv = nn.Sequential(
            nn.Conv2d(in_ch, out_ch, 3, padding=1),
            nn.BatchNorm2d(out_ch),
            nn.ReLU(inplace=True),
            nn.Conv2d(out_ch, out_ch, 3, padding=1),
            nn.BatchNorm2d(out_ch),
            nn.ReLU(inplace=True)
        )
    def forward(self, x):
        return self.conv(x)

class UNet(nn.Module):
    def __init__(self, in_channels=1, num_classes=2):
        super().__init__()
        self.enc1 = DoubleConv(in_channels, 64)
        self.enc2 = DoubleConv(64, 128)
        self.enc3 = DoubleConv(128, 256)
        self.enc4 = DoubleConv(256, 512)
        self.pool = nn.MaxPool2d(2)
        self.bottleneck = DoubleConv(512, 1024)
        self.up4 = nn.ConvTranspose2d(1024, 512, 2, stride=2)
        self.dec4 = DoubleConv(1024, 512)
        self.up3 = nn.ConvTranspose2d(512, 256, 2, stride=2)
        self.dec3 = DoubleConv(512, 256)
        self.up2 = nn.ConvTranspose2d(256, 128, 2, stride=2)
        self.dec2 = DoubleConv(256, 128)
        self.up1 = nn.ConvTranspose2d(128, 64, 2, stride=2)
        self.dec1 = DoubleConv(128, 64)
        self.out = nn.Conv2d(64, num_classes, 1)
    def forward(self, x):
        e1 = self.enc1(x)
        e2 = self.enc2(self.pool(e1))
        e3 = self.enc3(self.pool(e2))
        e4 = self.enc4(self.pool(e3))
        b = self.bottleneck(self.pool(e4))
        d4 = self.dec4(torch.cat([self.up4(b), e4], dim=1))
        d3 = self.dec3(torch.cat([self.up3(d4), e3], dim=1))
        d2 = self.dec2(torch.cat([self.up2(d3), e2], dim=1))
        d1 = self.dec1(torch.cat([self.up1(d2), e1], dim=1))
        return self.out(d1)

# 医学影像数据增强与预处理
import albumentations as A
import numpy as np

class MedicalImageAugmentor:
    def __init__(self, train=True):
        if train:
            self.transform = A.Compose([
                A.HorizontalFlip(p=0.5),
                A.RandomBrightnessContrast(brightness_limit=0.15, contrast_limit=0.15, p=0.5),
                A.ShiftScaleRotate(shift_limit=0.05, scale_limit=0.1, rotate_limit=15, p=0.5),
                A.GaussNoise(var_limit=(10.0, 50.0), p=0.3),
                A.Normalize(mean=[0.5], std=[0.5])
            ])
        else:
            self.transform = A.Compose([A.Normalize(mean=[0.5], std=[0.5])])
    def __call__(self, image, mask=None):
        if mask is not None:
            transformed = self.transform(image=image, mask=mask)
            return transformed["image"], transformed["mask"]
        return self.transform(image=image)["image"]

def load_dicom_image(filepath: str) -> np.ndarray:
    import pydicom
    ds = pydicom.dcmread(filepath)
    image = ds.pixel_array.astype(np.float32)
    image = image * ds.RescaleSlope + ds.RescaleIntercept
    image = np.clip(image, -1000, 1000)
    image = (image + 1000) / 2000.0
    return image

# 临床 NER - 医疗实体识别
import torch
import torch.nn as nn
from transformers import AutoModel

class ClinicalNER(nn.Module):
    def __init__(self, model_name="emilyalsentzer/Bio_ClinicalBERT", num_labels=13, dropout=0.1):
        super().__init__()
        self.bert = AutoModel.from_pretrained(model_name)
        self.dropout = nn.Dropout(dropout)
        self.classifier = nn.Linear(self.bert.config.hidden_size, num_labels)
        self.id2label = {
            0: "O", 1: "B-DISEASE", 2: "I-DISEASE",
            3: "B-DRUG", 4: "I-DRUG",
            5: "B-SYMPTOM", 6: "I-SYMPTOM",
            7: "B-PROCEDURE", 8: "I-PROCEDURE",
            9: "B-LAB", 10: "I-LAB",
            11: "B-ANATOMY", 12: "I-ANATOMY"
        }
    def forward(self, input_ids, attention_mask, labels=None):
        outputs = self.bert(input_ids, attention_mask=attention_mask)
        sequence_output = self.dropout(outputs.last_hidden_state)
        logits = self.classifier(sequence_output)
        if labels is not None:
            loss_fn = nn.CrossEntropyLoss()
            loss = loss_fn(logits.view(-1, 13), labels.view(-1))
            return {"loss": loss, "logits": logits}
        return {"logits": logits}

    def predict_entities(self, text, tokenizer):
        inputs = tokenizer(text, return_tensors="pt", truncation=True)
        with torch.no_grad():
            outputs = self(**inputs)
        preds = torch.argmax(outputs["logits"], dim=-1)[0].tolist()
        tokens = tokenizer.convert_ids_to_tokens(inputs["input_ids"][0])
        return self._decode_biluo(tokens, preds)

# 时序电子病历 - 疾病风险预测
import torch
import torch.nn as nn
import math

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

class EHRTransformer(nn.Module):
    def __init__(self, n_codes=10000, d_model=256, nhead=8, num_layers=4, num_classes=5):
        super().__init__()
        self.embedding = nn.Embedding(n_codes, d_model)
        self.pos_encoder = PositionalEncoding(d_model)
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model, nhead=nhead, dim_feedforward=512,
            dropout=0.1, batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.classifier = nn.Sequential(
            nn.Linear(d_model, 128), nn.ReLU(), nn.Dropout(0.3),
            nn.Linear(128, num_classes)
        )
    def forward(self, visits, mask=None):
        x = self.embedding(visits) * math.sqrt(256)
        x = self.pos_encoder(x)
        x = self.transformer(x, src_key_padding_mask=mask)
        lengths = mask.eq(False).sum(dim=1) - 1
        last_visit = x[torch.arange(x.size(0)), lengths]
        return self.classifier(last_visit)

# 分子图神经网络（PyTorch + DGL）
import torch
import torch.nn as nn

class AtomEncoder(nn.Module):
    def __init__(self, emb_dim=64):
        super().__init__()
        self.atomic_number_emb = nn.Embedding(100, emb_dim)
        self.degree_emb = nn.Embedding(6, emb_dim)
        self.formal_charge_emb = nn.Embedding(22, emb_dim)
        self.hybridization_emb = nn.Embedding(5, emb_dim)
    def forward(self, atom_features):
        h = (
            self.atomic_number_emb(atom_features["atomic_number"])
            + self.degree_emb(atom_features["degree"])
            + self.formal_charge_emb(atom_features["formal_charge"])
            + self.hybridization_emb(atom_features["hybridization"])
        )
        return h

class BondEncoder(nn.Module):
    def __init__(self, emb_dim=64):
        super().__init__()
        self.bond_type_emb = nn.Embedding(4, emb_dim)
        self.conjugated_emb = nn.Embedding(2, emb_dim)
        self.in_ring_emb = nn.Embedding(2, emb_dim)
    def forward(self, bond_features):
        return (
            self.bond_type_emb(bond_features["bond_type"])
            + self.conjugated_emb(bond_features["conjugated"])
            + self.in_ring_emb(bond_features["in_ring"])
        )

class MPNN(nn.Module):
    def __init__(self, node_dim=64, edge_dim=64, n_layers=4, out_dim=1):
        super().__init__()
        self.atom_encoder = AtomEncoder(node_dim)
        self.bond_encoder = BondEncoder(edge_dim)
        self.message_layers = nn.ModuleList([
            nn.Sequential(
                nn.Linear(node_dim * 2 + edge_dim, node_dim * 2),
                nn.ReLU(),
                nn.Linear(node_dim * 2, node_dim)
            ) for _ in range(n_layers)
        ])
        self.readout = nn.Sequential(
            nn.Linear(node_dim, 64), nn.ReLU(), nn.Linear(64, out_dim)
        )
    def forward(self, graph):
        h = self.atom_encoder(graph.ndata)
        e = self.bond_encoder(graph.edata)
        for msg_layer in self.message_layers:
            graph.ndata["h"] = h
            graph.edata["e"] = e
            graph.update_all(
                message_func=lambda edges: {"m": msg_layer(torch.cat([edges.src["h"], edges.dst["h"], edges.data["e"]], dim=-1))},
                reduce_func=lambda nodes: {"h_new": torch.sum(nodes.mailbox["m"], dim=1)}
            )
            h = h + graph.ndata["h_new"]
        with graph.local_scope():
            graph.ndata["h"] = h
            return self.readout(dgl.sum_nodes(graph, "h"))

# 分子生成 - 基于 VAE 的分子设计
import torch
import torch.nn as nn
import torch.nn.functional as F

class MoleculeVAE(nn.Module):
    def __init__(self, vocab_size=64, max_len=120, hidden_dim=256, latent_dim=56):
        super().__init__()
        self.max_len = max_len
        self.vocab_size = vocab_size
        self.hidden_dim = hidden_dim
        self.latent_dim = latent_dim
        self.encoder = nn.GRU(input_size=vocab_size, hidden_size=hidden_dim, num_layers=3, dropout=0.2, batch_first=True)
        self.fc_mu = nn.Linear(hidden_dim, latent_dim)
        self.fc_logvar = nn.Linear(hidden_dim, latent_dim)
        self.decoder = nn.GRU(input_size=vocab_size, hidden_size=hidden_dim, num_layers=3, dropout=0.2, batch_first=True)
        self.fc_out = nn.Linear(hidden_dim, vocab_size)
    def encode(self, smiles_onehot):
        _, h = self.encoder(smiles_onehot)
        mu = self.fc_mu(h[-1])
        logvar = self.fc_logvar(h[-1])
        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):
        batch_size = z.size(0)
        h = z.unsqueeze(0).repeat(3, 1, 1)
        outputs = []
        x = torch.zeros(batch_size, 1, self.vocab_size, device=z.device)
        for _ in range(self.max_len):
            out, h = self.decoder(x, h)
            logits = self.fc_out(out[:, -1, :])
            outputs.append(logits)
            next_token = F.gumbel_softmax(logits, hard=True)
            x = next_token.unsqueeze(1)
        return torch.stack(outputs, dim=1)
    def forward(self, smiles_onehot):
        mu, logvar = self.encode(smiles_onehot)
        z = self.reparameterize(mu, logvar)
        return self.decode(z), mu, logvar

# DNA 序列分类模型
import torch
import torch.nn as nn
import torch.nn.functional as F

class DNABertClassifier(nn.Module):
    def __init__(self, vocab_size=6, max_seq=1000, n_classes=2, embed_dim=128):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.conv1 = nn.Conv1d(embed_dim, 256, kernel_size=8, padding=3)
        self.conv2 = nn.Conv1d(256, 512, kernel_size=4, padding=1)
        self.conv3 = nn.Conv1d(512, 256, kernel_size=4, padding=1)
        self.bn1 = nn.BatchNorm1d(256)
        self.bn2 = nn.BatchNorm1d(512)
        self.bn3 = nn.BatchNorm1d(256)
        self.pool = nn.AdaptiveMaxPool1d(1)
        self.classifier = nn.Sequential(
            nn.Linear(256, 128), nn.ReLU(), nn.Dropout(0.4),
            nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.3),
            nn.Linear(64, n_classes)
        )
    def forward(self, x):
        x = self.embedding(x).transpose(1, 2)
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.max_pool1d(x, 4)
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.max_pool1d(x, 4)
        x = F.relu(self.bn3(self.conv3(x)))
        x = self.pool(x).squeeze(-1)
        return self.classifier(x)

def encode_dna(sequence: str, max_len: int = 1000) -> torch.Tensor:
    base_to_idx = {"A": 0, "C": 1, "G": 2, "T": 3, "N": 4}
    encoded = [base_to_idx.get(b.upper(), 4) for b in sequence[:max_len]]
    while len(encoded) < max_len:
        encoded.append(4)
    return torch.tensor(encoded[:max_len])

# 药物基因组学 - 药物反应预测
import torch
import torch.nn as nn
from sklearn.preprocessing import StandardScaler

class DrugResponsePredictor(nn.Module):
    def __init__(self, n_genes=1000, n_drug_features=256, hidden_dim=512):
        super().__init__()
        self.gene_encoder = nn.Sequential(
            nn.Linear(n_genes, 1024), nn.BatchNorm1d(1024), nn.ReLU(), nn.Dropout(0.3),
            nn.Linear(1024, 256), nn.ReLU(),
        )
        self.drug_encoder = nn.Sequential(
            nn.Linear(n_drug_features, 256), nn.ReLU(),
            nn.Linear(256, 128), nn.ReLU(),
        )
        self.predictor = nn.Sequential(
            nn.Linear(256 + 128, 256), nn.ReLU(), nn.Dropout(0.2),
            nn.Linear(256, 128), nn.ReLU(),
            nn.Linear(128, 1)
        )
    def forward(self, gene_expr, drug_feat):
        g = self.gene_encoder(gene_expr)
        d = self.drug_encoder(drug_feat)
        combined = torch.cat([g, d], dim=1)
        return self.predictor(combined)

def predict_drug_sensitivity(model, gene_data, drug_data, scaler):
    model.eval()
    gene_norm = torch.tensor(scaler.transform(gene_data), dtype=torch.float32)
    drug_tensor = torch.tensor(drug_data, dtype=torch.float32)
    with torch.no_grad():
        ic50_pred = model(gene_norm, drug_tensor)
    sensitivity = -torch.log10(ic50_pred.squeeze())
    return sensitivity.numpy()

# 联邦学习 - 多医院联合训练
import torch
import torch.nn as nn
import copy

class FederatedAveraging:
    def __init__(self, model_class, n_clients=5, lr=0.01):
        self.global_model = model_class()
        self.n_clients = n_clients
        self.lr = lr
    def client_update(self, client_data, client_labels, local_epochs=5):
        model = copy.deepcopy(self.global_model)
        optimizer = torch.optim.SGD(model.parameters(), lr=self.lr)
        criterion = nn.CrossEntropyLoss()
        for epoch in range(local_epochs):
            optimizer.zero_grad()
            outputs = model(client_data)
            loss = criterion(outputs, client_labels)
            loss.backward()
            optimizer.step()
        return model.state_dict()
    def federated_round(self, client_datasets):
        client_weights = []
        for data, labels in client_datasets:
            local_weights = self.client_update(data, labels)
            client_weights.append(local_weights)
        global_weights = copy.deepcopy(self.global_model.state_dict())
        for key in global_weights:
            global_weights[key] = torch.stack([w[key] for w in client_weights]).mean(dim=0)
        self.global_model.load_state_dict(global_weights)
        return global_weights

# 差分隐私 - DP-SGD 实现
import torch
from torch.nn.utils import clip_grad_norm_
import numpy as np

class DPSGDTrainer:
    def __init__(self, model, noise_multiplier=1.0, max_grad_norm=1.0, batch_size=64):
        self.model = model
        self.noise_multiplier = noise_multiplier
        self.max_grad_norm = max_grad_norm
        self.batch_size = batch_size
    def compute_dp_noise(self):
        return self.noise_multiplier * self.max_grad_norm
    def train_step_dp(self, data, labels, optimizer):
        optimizer.zero_grad()
        outputs = self.model(data)
        loss = torch.nn.functional.cross_entropy(outputs, labels, reduction="none")
        for i in range(len(loss)):
            loss[i].backward(retain_graph=True)
        clip_grad_norm_(self.model.parameters(), self.max_grad_norm)
        noise_std = self.compute_dp_noise()
        for param in self.model.parameters():
            noise = torch.normal(0, noise_std, param.grad.shape, device=param.device)
            param.grad += noise
        optimizer.step()
        return loss.mean().item()
    def estimate_privacy_budget(self, n_steps, dataset_size):
        q = self.batch_size / dataset_size
        sigma = self.noise_multiplier
        epsilon = np.sqrt(2 * n_steps * np.log(1.25 / 1e-5)) * q / sigma
        return epsilon

# 胸部 X 光分类 - 数据处理与模型
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms
import pandas as pd
import numpy as np
from PIL import Image

class ChestXRayDataset(Dataset):
    def __init__(self, csv_path, image_dir, mode="train"):
        self.df = pd.read_csv(csv_path)
        self.image_dir = image_dir
        self.mode = mode
        self.labels = [
            "Atelectasis", "Cardiomegaly", "Effusion",
            "Infiltration", "Mass", "Nodule", "Pneumonia",
            "Pneumothorax"
        ]
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.RandomHorizontalFlip(p=0.5),
            transforms.RandomAffine(degrees=10, translate=(0.1, 0.1)),
            transforms.ColorJitter(brightness=0.1, contrast=0.1),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485], std=[0.229])
        ])
        self.val_transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485], std=[0.229])
        ])
    def __len__(self):
        return len(self.df)
    def __getitem__(self, idx):
        row = self.df.iloc[idx]
        img = Image.open(f"{self.image_dir}/{row['Image Index']}").convert("RGB")
        if self.mode == "train":
            img = self.transform(img)
        else:
            img = self.val_transform(img)
        labels = torch.tensor([row[label] for label in self.labels], dtype=torch.float32)
        return img, labels

class PneumoniaClassifier(nn.Module):
    def __init__(self, num_classes=8, pretrained=True):
        super().__init__()
        from torchvision.models import efficientnet_b0
        weights = "DEFAULT" if pretrained else None
        self.backbone = efficientnet_b0(weights=weights)
        n_features = self.backbone.classifier[1].in_features
        self.backbone.classifier = nn.Sequential(
            nn.Dropout(0.3),
            nn.Linear(n_features, num_classes)
        )
    def forward(self, x):
        return self.backbone(x)

# 训练与评估 - 类别不平衡 + Grad-CAM
import torch
import torch.nn.functional as F
from sklearn.metrics import roc_auc_score

class FocalLoss(nn.Module):
    def __init__(self, alpha=1.0, gamma=2.0):
        super().__init__()
        self.alpha = alpha
        self.gamma = gamma
    def forward(self, logits, targets):
        bce_loss = F.binary_cross_entropy_with_logits(logits, targets, reduction="none")
        pt = torch.exp(-bce_loss)
        focal_loss = self.alpha * (1 - pt) ** self.gamma * bce_loss
        return focal_loss.mean()

class GradCAM:
    def __init__(self, model, target_layer):
        self.model = model
        self.target_layer = target_layer
        self.gradients = None
        self.activations = None
        target_layer.register_forward_hook(self._save_activation)
        target_layer.register_full_backward_hook(self._save_gradient)
    def _save_activation(self, module, input, output):
        self.activations = output.detach()
    def _save_gradient(self, module, grad_input, grad_output):
        self.gradients = grad_output[0].detach()
    def __call__(self, x, class_idx=None):
        output = self.model(x)
        if class_idx is None:
            class_idx = torch.argmax(output, dim=1)
        self.model.zero_grad()
        output[:, class_idx].backward()
        weights = F.adaptive_avg_pool2d(self.gradients, 1)
        cam = (weights * self.activations).sum(dim=1, keepdim=True)
        cam = F.relu(cam)
        cam = F.interpolate(cam, size=(224, 224), mode="bilinear")
        cam = cam - cam.min()
        cam = cam / (cam.max() + 1e-8)
        return cam.squeeze()

def train_epoch(model, dataloader, criterion, optimizer, device):
    model.train()
    total_loss = 0
    all_labels, all_preds = [], []
    for images, labels in dataloader:
        images, labels = images.to(device), labels.to(device)
        outputs = model(images)
        loss = criterion(outputs, labels)
        optimizer.zero_grad()
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()
        total_loss += loss.item()
        all_labels.append(labels.cpu())
        all_preds.append(torch.sigmoid(outputs).cpu())
    all_labels = torch.cat(all_labels)
    all_preds = torch.cat(all_preds)
    aucs = []
    for i in range(all_labels.size(1)):
        try:
            auc = roc_auc_score(all_labels[:, i], all_preds[:, i])
            aucs.append(auc)
        except ValueError:
            aucs.append(0.0)
    return total_loss / len(dataloader), np.mean(aucs)