第十章：案例研究与最佳实践

用户输入 → 预处理 → 安全检查 → 模型推理 → 后处理 → 响应输出
     ↑                                              ↓
     └─────────── 上下文管理（会话状态）──────────┘

会话状态结构：
{
  session_id: str,
  user_id: str,
  messages: List[Message],
  context_summary: str,  # 自动生成的上下文摘要
  metadata: {
    created_at: timestamp,
    last_active: timestamp,
    turn_count: int,
    tokens_used: int
  }
}

基础人设 (200 tokens)
  ├── 角色定义
  ├── 能力边界
  └── 行为准则

动态上下文 (variable)
  ├── 用户画像
  ├── 会话历史摘要
  └── 相关知识注入

任务指令 (100 tokens)
  ├── 当前任务描述
  └── 输出格式要求

原始数据 → 清洗 → 标准化 → 质量评分 → 去重 → 平衡采样 → 训练集
           ↓        ↓         ↓         ↓        ↓
        噪声过滤  格式统一  启发式+模型  MinHash  类别均衡

training_config:
  base_model: "llama-2-7b"
  learning_rate: 2e-5
  warmup_steps: 1000
  total_steps: 50000
  batch_size: 128
  gradient_accumulation: 4
  max_length: 2048

  # 关键优化
  gradient_checkpointing: true
  mixed_precision: "bf16"
  flash_attention: true

  # 正则化
  weight_decay: 0.01
  dropout: 0.1
  label_smoothing: 0.1

def dynamic_sampling_weight(category_loss):
    """损失越大的类别，采样权重越高"""
    weights = np.exp(category_loss / temperature)
    return weights / weights.sum()

输入 → Encoder → [CLS] token → Linear → Scalar Reward
         ↓
    共享 SFT 模型参数（冻结前 N-2 层）

ppo_config:
  kl_penalty: 0.1
  clip_range: 0.2
  value_loss_coef: 0.5
  entropy_coef: 0.01

  # 采样策略
  rollout_batch_size: 512
  minibatch_size: 64
  ppo_epochs: 4

  # 奖励设计
  reward_components:
    preference_score: 0.6
    kl_penalty: 0.2
    length_penalty: 0.1
    safety_score: 0.1

rewards = (rewards - rewards.mean()) / (rewards.std() + 1e-8)

if kl_divergence > target_kl * 1.5:
    kl_coef *= 1.5
elif kl_divergence < target_kl * 0.5:
    kl_coef *= 0.7

输入安全检查 → 生成过程干预 → 输出安全过滤 → 人工审核（采样）
      ↓              ↓               ↓              ↓
  敏感词过滤    安全引导生成     毒性检测      异常上报

         负载均衡器
              ↓
    ┌─────────┴─────────┐
    ↓                   ↓
推理服务器集群      缓存层(Redis)
    ↓                   ↓
    └─────────┬─────────┘
              ↓
         模型服务
    (TorchServe/Triton)

视觉编码器 ──┐
            ├→ Cross-Attention → Transformer Layers → Gated Fusion → 输出
文本编码器 ──┘                         ↑
                                    语音编码器

class ModalityAligner(nn.Module):
    def __init__(self, visual_dim, text_dim, audio_dim, hidden_dim):
        super().__init__()
        # 投影到统一维度
        self.visual_proj = nn.Linear(visual_dim, hidden_dim)
        self.text_proj = nn.Linear(text_dim, hidden_dim)
        self.audio_proj = nn.Linear(audio_dim, hidden_dim)

        # 可学习的模态 embedding
        self.modality_embeddings = nn.Parameter(
            torch.randn(3, hidden_dim)
        )

    def forward(self, visual=None, text=None, audio=None):
        features = []
        if visual is not None:
            features.append(self.visual_proj(visual) + self.modality_embeddings[0])
        if text is not None:
            features.append(self.text_proj(text) + self.modality_embeddings[1])
        if audio is not None:
            features.append(self.audio_proj(audio) + self.modality_embeddings[2])
        return torch.cat(features, dim=1)

transform = Compose([
    RandomResizedCrop(224, scale=(0.8, 1.0)),
    RandomHorizontalFlip(p=0.5),
    ColorJitter(brightness=0.4, contrast=0.4),
    ToTensor(),
    Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD)
])

视频帧采样策略：

- 均匀采样：每秒 1 帧
- 关键帧采样：场景变化检测
- 密集采样：动作识别任务（8 fps）

stage1_config:
  frozen_modules: ["text_encoder", "visual_encoder"]
  trainable: ["projection_layers", "alignment_modules"]
  learning_rate: 1e-4
  tasks:

    - image_text_matching: 0.3
    - masked_language_modeling: 0.3
    - image_text_contrastive: 0.4

stage2_config:
  frozen_modules: ["visual_encoder.layers[:-2]"]
  learning_rate: 5e-5
  tasks:

    - visual_question_answering: 0.25
    - image_captioning: 0.25
    - visual_reasoning: 0.25
    - audio_understanding: 0.25

class MultiModalEvaluator:
    def __init__(self):
        self.metrics = {
            'accuracy': AccuracyMetric(),
            'bleu': BLEUMetric(),
            'clip_score': CLIPScoreMetric(),
            'perplexity': PerplexityMetric()
        }

    def evaluate(self, predictions, references, modalities):
        results = {}
        for modality in modalities:
            modal_preds = predictions[modality]
            modal_refs = references[modality]

            if modality == 'text':
                results[f'{modality}_bleu'] = self.metrics['bleu'](modal_preds, modal_refs)
                results[f'{modality}_ppl'] = self.metrics['perplexity'](modal_preds)
            elif modality == 'vision':
                results[f'{modality}_acc'] = self.metrics['accuracy'](modal_preds, modal_refs)
                results[f'{modality}_clip'] = self.metrics['clip_score'](modal_preds, modal_refs)

        # 跨模态一致性
        results['cross_modal_alignment'] = self.compute_alignment(predictions)
        return results

def dynamic_forward(self, inputs):
    active_encoders = []
    if inputs.get('image') is not None:
        active_encoders.append(self.visual_encoder)
    if inputs.get('audio') is not None:
        active_encoders.append(self.audio_encoder)

    # 只计算必要的编码器
    features = [enc(inp) for enc, inp in zip(active_encoders, inputs.values())]
    return self.fusion_layer(features)

deployment_config:
  model_parallel: 2  # 模型并行度
  data_parallel: 4   # 数据并行度

  serving:
    max_batch_size: 32
    dynamic_batching: true
    timeout_ms: 5000

  optimization:
    use_flash_attention: true
    use_xformers: true
    compile_mode: "reduce-overhead"

疾病实体 ──[症状关系]──> 症状实体
    ↓                        ↑
[治疗关系]              [检查关系]
    ↓                        ↑
药物实体 ←──[相互作用]──→ 检查实体

knowledge_triples = [
    ("糖尿病", "常见症状", "多饮多尿"),
    ("二甲双胍", "治疗", "2型糖尿病"),
    ("二甲双胍", "禁忌症", "肾功能不全"),
    ("HbA1c", "诊断标准", ">6.5%")
]

def validate_medical_knowledge(text, knowledge_base):
    """验证医学内容的准确性"""
    claims = extract_medical_claims(text)

    for claim in claims:
        # 1. 检查与知识库的一致性
        kb_consistency = check_kb_consistency(claim, knowledge_base)

        # 2. 交叉引用验证
        citations = find_citations(claim)
        citation_quality = evaluate_citation_quality(citations)

        # 3. 专家审核标记
        if kb_consistency < 0.8 or citation_quality < 0.7:
            claim.mark_for_expert_review()

    return claims

cpt_config:
  base_model: "llama-2-7b"
  learning_rate: 5e-5
  total_steps: 100000

  data_mixture:
    medical_textbooks: 0.3
    clinical_guidelines: 0.2
    medical_papers: 0.2
    general_corpus: 0.3  # 防止遗忘

  curriculum:

    - phase: "basic"
      steps: 30000
      focus: "medical_terminology"

    - phase: "intermediate"
      steps: 40000
      focus: "disease_pathology"

    - phase: "advanced"
      steps: 30000
      focus: "clinical_reasoning"

class MedicalKnowledgeDistillation:
    def __init__(self, teacher_model, student_model):
        self.teacher = teacher_model
        self.student = student_model
        self.temp = 5.0

    def distillation_loss(self, inputs, alpha=0.7):
        with torch.no_grad():
            teacher_logits = self.teacher(inputs)

        student_logits = self.student(inputs)

        # KL divergence loss
        kl_loss = F.kl_div(
            F.log_softmax(student_logits / self.temp, dim=-1),
            F.softmax(teacher_logits / self.temp, dim=-1),
            reduction='batchmean'
        ) * (self.temp ** 2)

        # Combined with task loss
        task_loss = F.cross_entropy(student_logits, labels)

        return alpha * kl_loss + (1 - alpha) * task_loss

def uncertainty_sampling(model, unlabeled_pool, n_samples=1000):
    """基于不确定性的主动学习"""
    uncertainties = []

    for sample in unlabeled_pool:
        with torch.no_grad():
            logits = model(sample)
            probs = F.softmax(logits, dim=-1)

            # 熵作为不确定性度量
            entropy = -(probs * probs.log()).sum(dim=-1)
            uncertainties.append(entropy.item())

    # 选择不确定性最高的样本
    indices = np.argsort(uncertainties)[-n_samples:]
    return [unlabeled_pool[i] for i in indices]

初级标注员（医学生）
    ↓ [基础标注]
质量检查点 1
    ↓ [通过率 > 90%]
中级审核员（住院医师）
    ↓ [临床验证]
质量检查点 2
    ↓ [分歧案例]
高级专家（主治医师）
    ↓ [最终确认]
入库

def generate_synthetic_cases(templates, knowledge_base, n_cases=10000):
    """生成合成病例数据"""
    cases = []

    for _ in range(n_cases):
        template = random.choice(templates)

        # 填充疾病信息
        disease = sample_disease(knowledge_base)
        symptoms = get_symptoms(disease, knowledge_base)
        treatments = get_treatments(disease, knowledge_base)

        # 生成病例描述
        case = template.format(
            age=random.randint(20, 80),
            gender=random.choice(['男', '女']),
            symptoms=', '.join(symptoms[:3]),
            duration=random.randint(1, 30),
            diagnosis=disease,
            treatment=treatments[0]
        )

        cases.append(case)

    return cases

class EWC:
    def __init__(self, model, dataset, importance=1000):
        self.model = model
        self.importance = importance
        self.params = {n: p.clone() for n, p in model.named_parameters()}
        self.fisher = self._compute_fisher(dataset)

    def _compute_fisher(self, dataset):
        """计算 Fisher 信息矩阵"""
        fisher = {}
        model.eval()

        for data in dataset:
            model.zero_grad()
            output = model(data)
            loss = F.cross_entropy(output, data.labels)
            loss.backward()

            for n, p in model.named_parameters():
                if n not in fisher:
                    fisher[n] = p.grad.data.clone() ** 2
                else:
                    fisher[n] += p.grad.data.clone() ** 2

        for n in fisher:
            fisher[n] /= len(dataset)

        return fisher

    def penalty(self):
        """计算 EWC 惩罚项"""
        loss = 0
        for n, p in self.model.named_parameters():
            if n in self.fisher:
                loss += (self.fisher[n] * (p - self.params[n]) ** 2).sum()
        return self.importance * loss

class ExperienceReplay:
    def __init__(self, buffer_size=10000):
        self.buffer = deque(maxlen=buffer_size)
        self.priorities = deque(maxlen=buffer_size)

    def add(self, experience, priority=1.0):
        self.buffer.append(experience)
        self.priorities.append(priority)

    def sample(self, batch_size, alpha=0.6):
        """优先级采样"""
        probs = np.array(self.priorities) ** alpha
        probs /= probs.sum()

        indices = np.random.choice(
            len(self.buffer), 
            batch_size, 
            p=probs
        )

        return [self.buffer[i] for i in indices]

def evaluate_clinical_reasoning(model, test_cases):
    """评估临床推理能力"""
    metrics = {
        'diagnosis_accuracy': 0,
        'treatment_appropriateness': 0,
        'safety_score': 0
    }

    for case in test_cases:
        # 生成诊断
        diagnosis = model.generate_diagnosis(case.symptoms)
        metrics['diagnosis_accuracy'] += (
            diagnosis == case.gold_diagnosis
        )

        # 生成治疗方案
        treatment = model.suggest_treatment(diagnosis)
        metrics['treatment_appropriateness'] += evaluate_treatment(
            treatment, case.gold_treatment
        )

        # 安全性检查
        contraindications = check_contraindications(
            treatment, case.patient_info
        )
        metrics['safety_score'] += (len(contraindications) == 0)

    # 归一化
    for key in metrics:
        metrics[key] /= len(test_cases)

    return metrics

def identify_new_knowledge(recent_papers, existing_kb):
    """识别需要更新的知识"""
    updates = {
        'new_diseases': [],
        'updated_treatments': [],
        'revised_guidelines': []
    }

    for paper in recent_papers:
        entities = extract_medical_entities(paper)

        for entity in entities:
            if entity.type == 'disease' and entity not in existing_kb:
                updates['new_diseases'].append(entity)
            elif entity.type == 'treatment':
                if has_significant_change(entity, existing_kb):
                    updates['updated_treatments'].append(entity)

    return updates

update_pipeline:
  frequency: "monthly"

  steps:

    - name: "collect_updates"
      sources: ["pubmed", "clinical_trials", "fda_approvals"]

    - name: "validate_updates"
      validators: ["expert_review", "consistency_check"]

    - name: "prepare_training_data"
      augmentation: true
      balance_with_existing: 0.3

    - name: "incremental_training"
      method: "ewc"
      epochs: 5
      learning_rate: 1e-5

    - name: "evaluation"
      benchmarks: ["medqa", "safety_tests"]
      threshold: 0.95  # 相对于前版本

    - name: "deployment"
      strategy: "gradual_rollout"
      monitoring_period: "7d"