Generative AI with Large Language Models 学习笔记

> 课程来源: DeepLearning.AI + AWS > 课程链接: https://www.coursera.org/learn/generative-ai-with-llms/ > 学习时间: 2026-04-11 > 讲师团队: Antje Barth, Mike Chambers, Shelbee Eigenbrode, Chris Fregly > 课程评级: 4.8/5（95%学员推荐）

---

目录

1. 课程概述
2. Week 1: 生成式AI用例、项目生命周期和模型预训练
3. Week 2: 微调和评估大型语言模型
4. Week 3: 强化学习和LLM驱动的应用
5. 行为准则提炼
6. 应用到看宝AI的思考

1. 课程概述

1.1 课程定位

本课程是DeepLearning.AI与AWS联合推出的深度学习课程，旨在帮助开发者建立对生成式AI和LLM的系统性理解，不仅要知道"怎么用"，更要理解"为什么"。

1.2 学习目标

完成本课程后，你将能够： - 深入理解生成式AI的核心原理 - 描述基于LLM的典型项目生命周期 - 详细解释Transformer架构的工作机制 - 使用经验缩放定律优化模型目标函数 - 应用最先进的训练、调整和推理方法

1.3 课程结构

1.4 前置要求

- Python编程经验 - 机器学习基础知识（监督/非监督学习、损失函数） - 最好已完成DeepLearning.AI的ML或DL专项课程

---

2. Week 1: 生成式AI用例、项目生命周期和模型预训练

2.1 生成式AI与LLM基础

什么是生成式AI？

LLM的核心能力

大型语言模型展现出几种关键能力：

1. 涌现能力（Emergent Abilities）：模型规模增大后突然出现的新能力
2. 上下文学习（In-Context Learning）：无需微调即可从prompt中学习新任务
3. 思维链推理（Chain-of-Thought）：通过中间步骤提升复杂推理能力

2.2 Transformer架构详解

为什么Transformer如此重要？

在Transformer出现之前，文本生成主要依赖RNN/LSTM，存在以下问题： - 梯度消失/爆炸 - 难以捕捉长距离依赖 - 训练难以并行

Transformer通过注意力机制彻底解决了这些问题。

Transformer核心组件

┌─────────────────────────────────────────────────────────────┐
│                      Transformer 架构                       │
├─────────────────────────────────────────────────────────────┤
│  输入: "The cat sat on the"                                  │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                     编码器 (Encoder)                  │    │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  │    │
│  │  │ Layer 1 │→│ Layer 2 │→│ Layer 3 │→│ Layer N │  │    │
│  │  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘  │    │
│  │       ↓            ↓            ↓            ↓        │    │
│  │  [多头自注意力 + 前馈网络 + 残差连接 + 层归一化]       │    │
│  └─────────────────────────────────────────────────────┘    │
│                           ↓                                  │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                     解码器 (Decoder)                  │    │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐  │    │
│  │  │ Layer 1 │→│ Layer 2 │→│ Layer 3 │→│ Layer N │  │    │
│  │  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘  │    │
│  │       ↓            ↓            ↓            ↓        │    │
│  │  [掩码自注意力 + 编码器-解码器注意力 + 前馈网络]     │    │
│  └─────────────────────────────────────────────────────┘    │
│                           ↓                                  │
│  输出: "mat" (下一个token的预测概率)                        │
└─────────────────────────────────────────────────────────────┘

注意力机制的本质

注意力机制的核心思想："我应该关注输入的哪些部分？"

# 注意力机制的简化实现
def attention(Q, K, V):
    """
    Q: Query - 我正在查找什么
    K: Key - 我有什么信息
    V: Value - 信息的实际内容
    """
    d_k = Q.shape[-1]
    
    # 1. 计算相似度
    scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)
    
    # 2. Softmax归一化得到注意力权重
    attention_weights = F.softmax(scores, dim=-1)
    
    # 3. 加权求和
    output = torch.matmul(attention_weights, V)
    
    return output, attention_weights

多头注意力（Multi-Head Attention）

将注意力分成多个"头"，每个头学习不同的注意力模式：

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        self.num_heads = num_heads
        self.d_model = d_model
        self.d_k = d_model // num_heads
        
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)
    
    def forward(self, x):
        batch_size = x.size(0)
        
        # 线性变换并分头
        Q = self.W_q(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        V = self.W_v(x).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        
        # 注意力计算
        attn_output, _ = attention(Q, K, V)
        
        # 合并多头
        concat = attn_output.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
        
        return self.W_o(concat)

位置编码（Positional Encoding）

由于Transformer没有循环结构，需要显式注入位置信息：

def positional_encoding(seq_len, d_model):
    """正弦/余弦位置编码"""
    PE = torch.zeros(seq_len, d_model)
    position = torch.arange(0, seq_len, dtype=torch.float).unsqueeze(1)
    div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
    
    PE[:, 0::2] = torch.sin(position * div_term)
    PE[:, 1::2] = torch.cos(position * div_term)
    
    return PE

2.3 文本生成原理

自回归生成

LLM通过自回归方式生成文本：每次预测一个token，然后将预测的token加入输入继续预测。

def generate_text(model, prompt, max_length=100, temperature=0.7, top_k=50):
    """
    文本生成函数
    
    参数:
        model: 预训练语言模型
        prompt: 输入提示
        max_length: 最大生成长度
        temperature: 采样温度（越高越随机）
        top_k: 只考虑top_k个最高概率token
    """
    model.eval()
    input_ids = tokenizer.encode(prompt, return_tensors='pt')
    
    with torch.no_grad():
        for _ in range(max_length):
            # 前向传播
            outputs = model(input_ids)
            logits = outputs.logits[:, -1, :] / temperature
            
            # Top-K 采样
            if top_k > 0:
                indices_to_remove = logits < torch.topk(logits, top_k)[0][:, -1, None]
                logits[indices_to_remove] = float('-inf')
            
            # 概率分布采样
            probs = F.softmax(logits, dim=-1)
            next_token = torch.multinomial(probs, num_samples=1)
            
            # 终止条件
            if next_token == tokenizer.eos_token_id:
                break
            
            input_ids = torch.cat([input_ids, next_token], dim=1)
    
    return tokenizer.decode(input_ids[0], skip_special_tokens=True)

生成配置参数

2.4 提示工程（Prompt Engineering）

提示工程的核心概念

关键技巧

# ❌ 模糊提示
prompt = "写一首诗"
✅ 清晰具体
prompt = """写一首关于秋天的七言绝句，要求：
包含"落叶"和"丰收"两个意象
押ang韵
表达对时光流逝的感慨
字数控制在28字以内"""

few_shot_prompt = """将以下中文句子转换为繁体中文：
示例：
简体：机器学习是人工智能的子领域
繁体：機器學習是人工智慧的子領域
简体：深度学习使用多层神经网络
繁体："""

cot_prompt = """解决以下数学问题，请分步骤思考：
问题：小明有15个苹果，给了小红7个，又买了10个，请问小明现在有多少苹果？
思考步骤：
小明开始有15个苹果
给出7个：15 - 7 = 8个
又买了10个：8 + 10 = 18个

答案：18个

问题：小红有20元，买了一本书花了12元，又得到妈妈给的15元，请问小红现在有多少钱？"""

structured_prompt = """分析以下文本的情感，以JSON格式输出：
{
    "sentiment": "positive/negative/neutral",
    "confidence": 0.0-1.0,
    "key_phrases": ["关键短语1", "关键短语2"],
    "reasoning": "判断理由"
}
文本：这家餐厅的食物非常美味，但服务态度有待提高。"""

2.5 LLM预训练详解

预训练的任务

这是最基础的预训练任务：

class PretrainingDataset(Dataset):
    def __init__(self, texts, tokenizer, max_length=512):
        self.encoded = tokenizer(
            texts,
            truncation=True,
            max_length=max_length,
            return_overflowing_tokens=True,
            padding='max_length'
        )
    
    def __getitem__(self, idx):
        input_ids = self.encoded['input_ids'][idx]
        
        # 输入是除最后一个token外的所有token
        # 标签是除第一个token外的所有token
        return {
            'input_ids': torch.tensor(input_ids[:-1]),
            'labels': torch.tensor(input_ids[1:])
        }

随机遮盖一些token，模型需要预测被遮盖的内容。

预训练的计算挑战

2.6 缩放定律（Scaling Laws）

Kaplan等人的发现

OpenAI的研究表明：模型性能与模型参数量、数据集大小、计算量呈幂律关系。

┌─────────────────────────────────────────────────────────────┐
│                    缩放定律曲线                              │
│                                                             │
│  Loss ↑                                                      │
│   │        ╭─────────────────────────────────              │
│   │       ╱                                           ↑      │
│   │      ╱                                      ↑           │
│   │     ╱                                 ↑                  │
│   │    ╱                            ↑                         │
│   │   ╱                       ↑                               │
│   │  ╱                  ↑                                        │
│   │ ╱             ↑                                              │
│   │╱    ↑                                                        │
│   └────────────────────────────────────────────────→         │
│   小模型                                          大模型       │
│                                                             │
│  结论：模型越大、数据越多、计算量越大，性能通常越好            │
└─────────────────────────────────────────────────────────────┘

计算最优模型

> 在固定计算预算下，模型大小和数据量应该同比缩放，而非只增大模型。

实际应用

# 估算训练成本
def estimate_training_cost(params, tokens):
    """
    估算训练所需的FLOPS
    
    参数:
        params: 模型参数量（亿）
        tokens: 训练token数（亿）
    """
    # 粗略估算公式
    flops = 6  params  tokens * 2  # 前向+反向传播
    a100_hours = flops / (3.9e20)  # A100 312 TFLOPS (BF16)
    
    return {
        'total_flops': flops,
        'a100_gpu_hours': a100_hours,
        'cost_estimate_usd': a100_hours * 2.0  # 按$2/小时估算
    }
示例：估算70B模型的训练成本
estimate = estimate_training_cost(params=700, tokens=1400)
print(f"需要约 {estimate['a100_gpu_hours']/24:.0f} 天的A100算力")

2.7 领域适应预训练

为什么需要领域适应？

通用预训练模型在特定领域可能表现不佳，需要进行领域适应训练。

# 领域适应预训练示例
def domain_adaptive_pretraining(
    model,
    domain_corpus,
    tokenizer,
    output_dir,
    learning_rate=2e-5,
    epochs=3,
    batch_size=8
):
    """
    领域适应预训练流程
    
    参数:
        model: 基础预训练模型
        domain_corpus: 领域文本数据
        tokenizer: 分词器
    """
    # 1. 使用领域数据进行持续预训练
    training_args = TrainingArguments(
        output_dir=output_dir,
        learning_rate=learning_rate,
        per_device_train_batch_size=batch_size,
        num_train_epochs=epochs,
        warmup_steps=100,
        logging_steps=50,
        save_steps=1000,
        save_total_limit=2,
        fp16=True,
        dataloader_num_workers=4
    )
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=domain_corpus,
        data_collator=data_collator
    )
    
    trainer.train()
    
    # 2. 保存领域适配后的模型
    model.save_pretrained(f"{output_dir}/final_model")

---

3. Week 2: 微调和评估大型语言模型

3.1 指令微调（Instruction Fine-tuning）

什么是指令微调？

预训练阶段:  "今天天气" + "很好" → 学习语言规律
   ↓
指令微调阶段: "今天天气怎么样？" → "今天天气晴朗，温度适宜..."
   ↓
学习遵循指令

指令数据的格式

# 典型的指令微调数据格式
instruction_data = [
    {
        "instruction": "将以下句子改写成被动语态",
        "input": "科学家发现了新物种",
        "output": "新物种被科学家发现了"
    },
    {
        "instruction": "总结以下文章的要点",
        "input": "文章全文...",
        "output": "1. 要点1\n2. 要点2..."
    },
    {
        "instruction": "解释量子计算的基本原理",
        "input": "",
        "output": "量子计算是一种基于量子力学原理的计算方式..."
    }
]

单任务微调 vs 多任务微调

3.2 模型评估

评估的挑战

LLM评估面临三大挑战：

1. 多维度：语言质量、事实准确性、逻辑一致性等
2. 主观性：不同评估者可能有不同看法
3. 任务多样性：不同任务需要不同评估指标

评估指标分类

from rouge_score import rouge_scorer
from bleu import sentence_bleu
def evaluate_summary(prediction, reference):
    """评估摘要质量"""
    # ROUGE分数
    scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
    rouge_scores = scorer.score(reference, prediction)
    
    # BLEU分数
    bleu_score = sentence_bleu([reference.split()], prediction.split())
    
    return {
        'rouge1': rouge_scores['rouge1'].fmeasure,
        'rouge2': rouge_scores['rouge2'].fmeasure,
        'rougeL': rouge_scores['rougeL'].fmeasure,
        'bleu': bleu_score
    }

3.3 基准测试（Benchmarks）

常用基准

基准的局限性

- 数据泄露：训练数据可能包含测试集 - 过拟合评估：模型可能针对基准优化 - 不完整覆盖：无法覆盖所有真实场景

3.4 参数高效微调（PEFT）

为什么需要PEFT？

全参数微调的挑战： - 计算成本高（需要全量梯度计算） - 存储成本高（每个任务需要保存一份完整模型） - 灾难性遗忘（可能丢失预训练知识）

PEFT的核心思想

> "只训练一小部分参数，保持大部分预训练知识不变"

┌─────────────────────────────────────────────────────────────┐
│                   全参数微调 vs PEFT                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  全参数微调:                                                │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 预训练模型 (7B参数) → 梯度更新全部参数 → 新模型     │   │
│  └─────────────────────────────────────────────────────┘   │
│  成本: 7B参数 × 4字节 = 28GB显存                           │
│                                                             │
│  PEFT (LoRA为例):                                          │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 预训练模型 (冻结)  +  小适配器 (新参数) → 新模型    │   │
│  └─────────────────────────────────────────────────────┘   │
│  成本: 7B参数 × 4字节 + ~35M适配器参数 = ~0.5GB显存        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

3.5 LoRA技术详解

LoRA的核心原理

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM
class LoRALinear(nn.Module):
    """LoRA适配器"""
    def __init__(self, original_layer, rank=4, alpha=1.0):
        super().__init__()
        self.original = original_layer
        self.original.weight.requires_grad = False  # 冻结原权重
        
        # LoRA的核心：用两个低秩矩阵
        in_features = original_layer.in_features
        out_features = original_layer.out_features
        
        self.lora_A = nn.Parameter(torch.zeros(in_features, rank))
        self.lora_B = nn.Parameter(torch.zeros(rank, out_features))
        
        # 初始化
        nn.init.normal_(self.lora_A, std=1/rank)
        nn.init.zeros_(self.lora_B)
        
        self.scaling = alpha / rank
    
    def forward(self, x):
        # 原始输出 + LoRA调整
        return self.original(x) + (x @ self.lora_A @ self.lora_B) * self.scaling

LoRA配置示例

from peft import LoraConfig, get_peft_model, TaskType
LoRA配置
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=8,                      # 秩，越大参数量越多，效果通常更好
    lora_alpha=16,            # 缩放因子
    lora_dropout=0.05,        # Dropout概率
    target_modules=[          # 应用LoRA的模块
        "q_proj", "v_proj",   # 注意力投影
        "k_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj"  # FFN层
    ],
    bias="none",              # 不训练bias
)
应用LoRA
base_model = AutoModelForCausalLM.from_pretrained("base_model")
peft_model = get_peft_model(base_model, lora_config)
打印可训练参数
peft_model.print_trainable_parameters()
输出: trainable params: 4,194,304 || all params: 6,738,415,616 || trainable%: 0.062%

LoRA的优势

1. 显存高效：只需存储适配器参数（通常<1%的原始模型大小）
2. 训练快速：梯度只计算适配器部分
3. 易于切换：可快速加载/卸载不同任务的适配器
4. 可组合：支持多个任务适配器组合

3.6 软提示（Soft Prompts）

什么是软提示？

软提示（Prompt Tuning / Prefix Tuning）通过学习可优化的"虚拟token"来引导模型行为。

from peft import PromptTuningConfig, get_peft_model
Prompt Tuning配置
prompt_config = PromptTuningConfig(
    task_type=TaskType.CAUSAL_LM,
    num_virtual_tokens=20,        # 虚拟token数量
    prompt_tuning_init="TEXT",    # 初始化方式
    prompt_tuning_init_text="你是一个乐于助人的AI助手。",  # 初始文本
)
Prefix Tuning配置
prefix_config = PrefixTuningConfig(
    task_type=TaskType.CAUSAL_LM,
    num_virtual_tokens=20,
    encoder_hidden_size=128,
)

软提示 vs 硬提示

3.7 实验室：微调对话摘要模型

这是课程的核心实践环节，使用FLAN-T5模型进行对话摘要微调。

# 实验室核心代码框架
from transformers import (
    AutoModelForSeq2SeqLM,
    AutoTokenizer,
    Seq2SeqTrainingArguments,
    Seq2SeqTrainer,
    DataCollatorForSeq2Seq
)
1. 加载模型和分词器
model_name = "google/flan-t5-base"
model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)
2. 准备数据
def preprocess_function(examples):
    inputs = ["Summarize: " + doc for doc in examples["dialogue"]]
    targets = examples["summary"]
    
    model_inputs = tokenizer(
        inputs,
        max_length=512,
        truncation=True,
        padding="max_length"
    )
    
    labels = tokenizer(
        targets,
        max_length=128,
        truncation=True,
        padding="max_length"
    )
    
    model_inputs["labels"] = labels["input_ids"]
    return model_inputs
3. 训练配置
training_args = Seq2SeqTrainingArguments(
    output_dir="./results",
    evaluation_strategy="epoch",
    learning_rate=2e-5,
    per_device_train_batch_size=8,
    per_device_eval_batch_size=8,
    num_train_epochs=3,
    weight_decay=0.01,
    predict_with_generate=True,
    fp16=True,
)
4. 创建Trainer
trainer = Seq2SeqTrainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_train_dataset,
    eval_dataset=tokenized_eval_dataset,
    tokenizer=tokenizer,
    data_collator=DataCollatorForSeq2Seq(tokenizer, model=model),
)
5. 训练
trainer.train()
6. 评估
results = trainer.evaluate()
print(f"BLEU Score: {results['eval_bleu']:.2f}")

---

4. Week 3: 强化学习和LLM驱动的应用

4.1 模型对齐（Alignment）

为什么需要对齐？

预训练模型可能产生： - 有害内容：攻击性、歧视性内容 - 无用信息：冗长、不相关的回答 - 不遵循指令：答非所问

对齐的目标：让模型输出符合人类价值观和期望

4.2 RLHF详解

RLHF的三个阶段

┌─────────────────────────────────────────────────────────────────┐
│                    RLHF三阶段流程                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  阶段1: 监督微调 (SFT)                                          │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  人类写的 指令-回答对  →  微调预训练模型               │   │
│  └─────────────────────────────────────────────────────────┘   │
│                           ↓                                     │
│  阶段2: 奖励模型训练                                            │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  同一指令的多个回答 + 人类偏好排序  →  奖励模型        │   │
│  └─────────────────────────────────────────────────────────┘   │
│                           ↓                                     │
│  阶段3: RL微调                                                  │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │  SFT模型 + 奖励模型 + KL约束  →  最终对齐模型          │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

阶段1: 监督微调（SFT）

# SFT训练数据格式
sft_data = [
    {
        "prompt": "解释量子纠缠",
        "response": "量子纠缠是量子力学中的一种现象..."
    },
    {
        "prompt": "写一首关于月亮的诗",
        "response": "明月悬高空，清辉洒人间..."
    }
]
使用标准语言建模训练
def sft_train(model, dataset, tokenizer):
    """监督微调训练"""
    training_args = TrainingArguments(
        output_dir="./sft_model",
        learning_rate=1e-5,
        per_device_train_batch_size=4,
        num_train_epochs=3,
        gradient_checkpointing=True,
        fp16=True,
    )
    
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        data_collator=DataCollatorForLanguageModeling(tokenizer),
    )
    
    trainer.train()
    return trainer.model

阶段2: 奖励模型训练

class RewardModel(nn.Module):
    """奖励模型"""
    def __init__(self, base_model):
        super().__init__()
        self.base_model = base_model
        # 添加奖励头
        self.reward_head = nn.Linear(base_model.config.hidden_size, 1)
    
    def forward(self, input_ids, attention_mask, labels=None):
        outputs = self.base_model(
            input_ids=input_ids,
            attention_mask=attention_mask
        )
        
        # 使用最后一个token的隐状态预测奖励
        last_hidden_state = outputs.last_hidden_state[:, -1, :]
        reward = self.reward_head(last_hidden_state)
        
        return reward
def train_reward_model(reward_model, preference_data):
    """
    训练奖励模型
    
    preference_data格式:
    [
        {
            "prompt": "...",
            "chosen": "更好的回答",
            "rejected": "较差的回答"
        },
        ...
    ]
    """
    optimizer = torch.optim.AdamW(reward_model.parameters(), lr=1e-5)
    
    for batch in preference_data:
        # 获取prompt和两个回答的奖励
        reward_chosen = reward_model(**batch["chosen"])
        reward_rejected = reward_model(**batch["rejected"])
        
        # Bradley-Terry损失
        loss = -torch.log(torch.sigmoid(reward_chosen - reward_rejected)).mean()
        
        loss.backward()
        optimizer.step()
        optimizer.zero_grad()

阶段3: RL微调（PPO）

from trl import PPOTrainer, PPOConfig
PPO配置
ppo_config = PPOConfig(
    learning_rate=1e-5,
    batch_size=32,
    mini_batch_size=4,
    gradient_accumulation_steps=8,
    ppo_epochs=4,
    target_kl=0.1,  # KL散度约束
)
创建PPO Trainer
ppo_trainer = PPOTrainer(
    config=ppo_config,
    model=sft_model,
    ref_model=ref_model,  # 用于KL约束的参考模型
    reward_model=reward_model,
    tokenizer=tokenizer,
)
RLHF训练循环
for batch in dataloader:
    # 1. 生成响应
    query_tensors = batch["input_ids"]
    response_tensors = ppo_trainer.generate(query_tensors)
    
    # 2. 获取奖励
    texts = [tokenizer.decode(r) for r in response_tensors]
    rewards = [reward_model(**encode(t)) for t in texts]
    
    # 3. PPO更新
    ppo_trainer.step(query_tensors, response_tensors, rewards)

KL散度约束

为什么需要KL约束？

> 如果只优化奖励，模型会"作弊"——产生看似高分但实际无意义的内容。

总奖励 = 奖励模型分数 - β × KL(πθ || πref)
其中:
- πθ: 当前策略
- πref: 参考模型（通常是SFT模型）
- β: KL惩罚系数（通常0.1-0.2）

def compute_reward_with_kl(model, query, response, ref_model, beta=0.1):
    """带KL约束的奖励计算"""
    # 获取奖励
    reward_score = reward_model(query, response)
    
    # 计算KL散度
    kl_penalty = compute_kl_divergence(model, ref_model, query, response)
    
    # 最终奖励
    final_reward = reward_score - beta * kl_penalty
    
    return final_reward

4.3 RLHF的挑战与解决方案

挑战1: 奖励黑客（Reward Hacking）

模型发现奖励模型的"漏洞"，产生高分但无意义的内容。

挑战2: 训练不稳定性

PPO训练可能出现： - 策略崩溃 - 奖励爆炸 - KL散度失控

挑战3: 人类反馈成本高

高质量人类标注成本高昂。

4.4 Chain-of-Thought推理

什么是Chain-of-Thought？

通过要求模型展示中间推理步骤，提升复杂问题的解答能力。

# Chain-of-Thought示例
cot_prompts = [
    {
        "prompt": """问题：小明有5个苹果，小红给了他3个，他又买了2个，他一共有多少苹果？
        
请分步骤思考：""",
        "expected": "步骤1: 5 + 3 = 8\n步骤2: 8 + 2 = 10\n答案: 10个"
    },
    {
        "prompt": """问题：如果所有的A都是B，而所有的B都是C，那么所有的A都是C吗？
请分析：""",
        "expected": "是的，所有A都是C。这是三段论的传递性..."
    }
]
def cot_inference(model, question):
    """Chain-of-Thought推理"""
    prompt = question + "\n\n请分步骤思考："
    response = model.generate(prompt)
    
    # 提取答案部分（通常在"答案："或"Therefore"之后）
    return response

CoT的适用场景

4.5 ReAct框架

ReAct = Reasoning + Acting

┌─────────────────────────────────────────────────────────────────┐
│                    ReAct执行流程                                 │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  问题: 北京今天的天气如何？需要带伞吗？                           │
│                                                                 │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ Loop:                                                     │   │
│  │   1. Thought: 需要先查天气，再决定是否带伞               │   │
│  │   2. Action: search_weather(location="北京")              │   │
│  │   3. Observation: 今天北京晴转多云，有30%降雨概率       │   │
│  │   4. Thought: 降雨概率不高，但多云，建议带伞备用          │   │
│  │   5. Final Answer: ...                                  │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

ReAct实现

from typing import List, Dict, Any
class ReActAgent:
    """ReAct推理代理"""
    
    def __init__(self, llm, tools: List[Dict]):
        self.llm = llm
        self.tools = {t["name"]: t for t in tools}
        self.tools_description = "\n".join([
            f"{t['name']}: {t['description']}" 
            for t in tools
        ])
    
    def think(self, question: str, max_iterations=5) -> str:
        """ReAct推理循环"""
        
        context = f"问题: {question}\n\n可用工具:\n{self.tools_description}\n\n"
        loop_count = 0
        
        while loop_count < max_iterations:
            loop_count += 1
            
            # 1. 让模型决定下一步
            prompt = context + "\n\n请决定下一步操作："
            decision = self.llm.generate(prompt)
            
            # 2. 解析决策
            if "Final Answer:" in decision:
                # 提取最终答案
                return decision.split("Final Answer:")[-1].strip()
            
            # 3. 执行工具
            if "Action:" in decision:
                action_line = decision.split("Action:")[1].split("\n")[0].strip()
                action_name = action_line.split("(")[0].strip()
                action_args = action_line.split("(")[1].split(")")[0].strip()
                
                if action_name in self.tools:
                    result = self.tools[action_name]"function")
                    context += f"\n{decision}\nObservation: {result}\n"
                else:
                    context += f"\n{decision}\nObservation: 工具不存在\n"
            else:
                context += f"\n{decision}\n"
        
        return "达到最大迭代次数，未能给出答案"

常用工具定义

# 工具定义示例
tools = [
    {
        "name": "search_web",
        "description": "搜索互联网获取最新信息",
        "function": lambda query: search_web(query)
    },
    {
        "name": "calculator",
        "description": "执行数学计算",
        "function": lambda expression: eval(expression)
    },
    {
        "name": "get_date",
        "description": "获取当前日期",
        "function": lambda: datetime.now().strftime("%Y-%m-%d")
    }
]

4.6 LLM应用架构

典型应用架构

┌─────────────────────────────────────────────────────────────────┐
│                    LLM应用架构                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   ┌─────────────┐     ┌─────────────┐     ┌─────────────┐      │
│   │   用户请求   │────→│   网关/API   │────→│   业务逻辑   │      │
│   └─────────────┘     └─────────────┘     └──────┬──────┘      │
│                                                  │              │
│   ┌─────────────┐     ┌─────────────┐     ┌──────┴──────┐      │
│   │   缓存层    │←────│   Prompt    │←────│     LLM    │      │
│   │   (Redis)   │     │   管理      │     │  (API/本地) │      │
│   └─────────────┘     └─────────────┘     └──────┬──────┘      │
│                                                  │              │
│   ┌─────────────┐     ┌─────────────┐     ┌──────┴──────┐      │
│   │   知识库    │←────│   RAG      │←────│   外部工具  │      │
│   │  (向量库)   │     │   检索     │     │ (搜索/计算) │      │
│   └─────────────┘     └─────────────┘     └─────────────┘      │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

核心组件说明

4.7 负责任AI

为什么重要？

LLM可能产生： - 有害内容：暴力、仇恨言论 - 虚假信息：看似合理但错误的内容 - 隐私泄露：训练数据中的敏感信息 - 偏见歧视：对特定群体的不公平对待

负责任AI实践

class SafetyChecker:
    """安全检查器"""
    
    def __init__(self):
        self.harmful_patterns = [...]  # 有害内容模式
        self.bias_categories = [...]    # 偏见类别
    
    def check(self, text: str) -> Dict[str, Any]:
        """检查内容安全性"""
        results = {
            "is_safe": True,
            "harmful_content": [],
            "bias_flags": [],
            "confidence": 1.0
        }
        
        # 检查有害内容
        for pattern in self.harmful_patterns:
            if pattern in text:
                results["is_safe"] = False
                results["harmful_content"].append(pattern)
        
        # 检查偏见
        for category in self.bias_categories:
            if self.detect_bias(text, category):
                results["bias_flags"].append(category)
        
        return results

def safe_generate(prompt: str, model) -> str:
    """安全的文本生成"""
    # 前处理：检查输入
    input_check = safety_checker.check(prompt)
    if not input_check["is_safe"]:
        return "抱歉，我无法处理这个请求。"
    
    # 生成
    response = model.generate(prompt)
    
    # 后处理：检查输出
    output_check = safety_checker.check(response)
    if not output_check["is_safe"]:
        return "抱歉，这个回答可能不合适，请换个话题。"
    
    return response

def explain_decision(prompt: str, response: str, model) -> str:
    """解释模型决策"""
    explanation_prompt = f"""
    请解释为什么这样回答这个问题。
    
    问题: {prompt}
    回答: {response}
    
    请从以下角度分析：
    1. 回答的主要依据
    2. 可能的局限性
    3. 建议的改进方向
    """
    
    return model.generate(explanation_prompt)

---

5. 行为准则提炼

5.1 理解模型能力边界

# 在使用模型能力前，先测试边界
def test_model_capability(model, test_cases):
    """测试模型特定能力"""
    results = []
    for case in test_cases:
        output = model.generate(case["input"])
        results.append({
            "input": case["input"],
            "expected": case["expected"],
            "actual": output,
            "correct": case"validator"
        })
    
    success_rate = sum(r["correct"] for r in results) / len(results)
    print(f"能力测试成功率: {success_rate:.1%}")
    
    # 只有成功率足够高才在实际场景使用
    if success_rate < 0.8:
        print("警告: 此能力不稳定，建议添加回退策略")
    
    return results

5.2 迭代式Prompt优化

# 迭代优化Prompt
def optimize_prompt(base_prompt, test_cases, model):
    """Prompt迭代优化流程"""
    
    best_prompt = base_prompt
    best_score = 0
    
    # 尝试不同的Prompt变体
    variations = [
        "增加具体示例",
        "添加格式要求",
        "加入角色设定",
        "说明输出限制",
        "添加思考步骤引导"
    ]
    
    for variation in variations:
        test_prompt = apply_variation(base_prompt, variation)
        score = evaluate_prompt(test_prompt, test_cases, model)
        
        if score > best_score:
            best_score = score
            best_prompt = test_prompt
            print(f"✓ 改进 ({variation}): {score:.1%}")
        else:
            print(f"✗ 无改进 ({variation})")
    
    return best_prompt, best_score

5.3 构建弹性处理机制

class ResilientLLMWrapper:
    """带弹性的LLM封装"""
    
    def __init__(self, model, max_retries=3):
        self.model = model
        self.max_retries = max_retries
    
    def generate(self, prompt, fallback_response=None):
        """带重试和降级的生成"""
        
        for attempt in range(self.max_retries):
            try:
                response = self.model.generate(prompt)
                
                # 验证输出质量
                if self.validate_response(response):
                    return response
                else:
                    print(f"输出验证失败，重试 ({attempt+1}/{self.max_retries})")
                    
            except Exception as e:
                print(f"生成异常: {e}，重试 ({attempt+1}/{self.max_retries})")
        
        # 所有重试都失败，使用降级响应
        if fallback_response:
            print("使用降级响应")
            return fallback_response
        
        return "抱歉，我现在无法回答这个问题，请稍后再试。"
    
    def validate_response(self, response):
        """验证响应质量"""
        # 检查长度
        if len(response) < 10 or len(response) > 10000:
            return False
        
        # 检查是否为错误标记
        error_markers = ["sorry", "cannot", "unable", "error"]
        if any(marker in response.lower() for marker in error_markers):
            return False
        
        return True

5.4 利用上下文学习

def create_few_shot_prompt(task_type, query, examples):
    """创建Few-Shot提示"""
    
    if not examples:
        # 没有示例，使用零样本
        return query
    
    # 构建示例部分
    example_text = ""
    for i, ex in enumerate(examples):
        example_text += f"\n\n示例 {i+1}:\n输入: {ex['input']}\n输出: {ex['output']}"
    
    # 组装完整Prompt
    prompt = f"""任务类型: {task_type}
{examples}
现在请完成以下任务:
输入: {query}
输出:"""
    
    return prompt
使用示例
task = "情感分类"
query = "这部电影太精彩了！"
examples = [
    {"input": "我今天很开心", "output": "正面"},
    {"input": "这服务太差了", "output": "负面"},
    {"input": "一般般吧", "output": "中性"}
]
prompt = create_few_shot_prompt(task, query, examples)

5.5 结合外部工具增强能力

class ToolAugmentedAgent:
    """工具增强的Agent"""
    
    def __init__(self, llm):
        self.llm = llm
        self.tools = {
            "calculator": self.calculate,
            "search": self.search_web,
            "date": self.get_current_date,
        }
    
    def solve(self, query):
        """带工具的复杂问题解决"""
        
        # 检测是否需要工具
        needs_tools = self.detect_tool_requirements(query)
        
        if not needs_tools:
            # 不需要工具，直接回答
            return self.llm.generate(query)
        
        # 需要工具，构建ReAct风格的提示
        tool_desc = "\n".join([
            f"- {name}: {desc}" 
            for name, desc in self.tools.items()
        ])
        
        prompt = f"""问题: {query}
可用工具:
{tool_desc}
请分步骤解决，使用工具获取精确信息。"""
        
        # 迭代执行
        context = ""
        for _ in range(5):  # 最多5步
            step = self.llm.generate(prompt + context)
            
            if "Final Answer:" in step:
                return step.split("Final Answer:")[-1]
            
            # 执行工具
            if "Action:" in step:
                # 提取并执行工具
                ...
                
        return "无法完成"

5.6 持续评估和监控

class LLMMonitor:
    """LLM输出监控"""
    
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "success_count": 0,
            "failure_count": 0,
            "quality_scores": [],
            "latency_samples": []
        }
    
    def track_request(self, request_id, prompt, response, latency):
        """跟踪请求"""
        self.metrics["total_requests"] += 1
        
        # 记录延迟
        self.metrics["latency_samples"].append(latency)
        
        # 评估质量
        quality = self.evaluate_quality(prompt, response)
        self.metrics["quality_scores"].append(quality)
        
        # 标记成功/失败
        if quality > 0.5:
            self.metrics["success_count"] += 1
        else:
            self.metrics["failure_count"] += 1
        
        # 检测异常
        if quality < 0.3:
            self.alert_low_quality(request_id, prompt, response)
    
    def get_dashboard(self):
        """生成监控面板"""
        return {
            "total_requests": self.metrics["total_requests"],
            "success_rate": self.metrics["success_count"] / max(1, self.metrics["total_requests"]),
            "avg_latency": np.mean(self.metrics["latency_samples"]),
            "p95_latency": np.percentile(self.metrics["latency_samples"], 95),
            "avg_quality": np.mean(self.metrics["quality_scores"]),
            "quality_distribution": {
                "excellent": sum(1 for q in self.metrics["quality_scores"] if q > 0.9),
                "good": sum(1 for q in self.metrics["quality_scores"] if 0.7 < q <= 0.9),
                "fair": sum(1 for q in self.metrics["quality_scores"] if 0.5 < q <= 0.7),
                "poor": sum(1 for q in self.metrics["quality_scores"] if q <= 0.5)
            }
        }

---

6. 应用到看宝AI的思考

6.1 当前看宝AI的能力评估

基于课程学习的视角，评估看宝AI：

6.2 具体改进建议

建议1: 引入Few-Shot Learning提升任务表现

当前问题：复杂任务需要大量Prompt描述，效果不稳定。

改进方案：

# 为看宝AI的核心场景创建Few-Shot示例库
KANBAN_EXAMPLES = [
    {
        "task": "日程创建",
        "examples": [
            {
                "input": "帮我安排明天下午3点和王总开会",
                "output": "已创建日程：\n- 时间：明天 15:00\n- 主题：王总会议\n- 类型：会议"
            },
            {
                "input": "提醒我周六买礼物",
                "output": "已创建提醒：\n- 日期：周六\n- 时间：上午 10:00（默认）\n- 内容：购买礼物"
            }
        ]
    },
    # ... 更多任务场景
]

建议2: 使用LoRA进行轻量级微调

针对"看宝AI"的使用场景（个人助手、任务管理），可以考虑：

# 看宝AI专属微调方向
peft_tasks = [
    {
        "name": "中文任务理解",
        "description": "提升对中文口语化表达的理解",
        "rank": 4
    },
    {
        "name": "日程管理",
        "description": "准确提取和结构化日程信息",
        "rank": 8
    },
    {
        "name": "语气适配",
        "description": "调整为温暖、鼓励的对话风格",
        "rank": 4
    }
]

建议3: 构建ReAct风格的工具调用能力

当前看宝AI的工具使用较为简单，可以引入ReAct框架：

class KanbaoReActAgent:
    """
    看宝AI的ReAct实现
    
    工具池:
    - calendar: 日历管理
    - reminder: 提醒设置
    - notes: 笔记记录
    - search: 网络搜索
    - memory: 记忆查询
    """
    
    def __init__(self, llm):
        self.llm = llm
        self.tools = self._init_tools()
    
    def process(self, user_message):
        """ReAct处理用户消息"""
        # 思考 → 行动 → 观察 → 决策循环
        context = f"用户: {user_message}\n工具: {self.tool_descriptions}\n"
        
        for step in range(5):
            thought = self.llm.think(context, step)
            
            if thought.requires_action():
                result = self.execute_tool(thought.action)
                context += f"思考: {thought.reasoning}\n"
                context += f"行动: {thought.action}\n"
                context += f"观察: {result}\n"
            else:
                return thought.final_answer
        
        return "抱歉，我需要更多时间思考这个问题。"

建议4: 引入质量监控和反馈机制

class KanbaoQualityMonitor:
    """看宝AI质量监控"""
    
    def __init__(self):
        self.conversation_logs = []
        self.user_feedback = []
        self.quality_metrics = {}
    
    def log_interaction(self, user_input, ai_response, context):
        """记录交互"""
        interaction = {
            "timestamp": datetime.now(),
            "user_input": user_input,
            "ai_response": ai_response,
            "context": context,
            "auto_quality_score": self.auto_evaluate(ai_response)
        }
        
        self.conversation_logs.append(interaction)
        
        # 定期分析模式
        if len(self.conversation_logs) % 100 == 0:
            self.analyze_patterns()
    
    def analyze_patterns(self):
        """分析交互模式，识别改进点"""
        # 低质量回复模式
        low_quality_patterns = self.find_low_quality_patterns()
        
        # 用户重复请求模式（可能意味着理解偏差）
        misunderstanding_patterns = self.find_misunderstanding_patterns()
        
        # 建议Prompt优化
        prompt_suggestions = self.generate_prompt_suggestions()
        
        return {
            "low_quality_patterns": low_quality_patterns,
            "misunderstanding_patterns": misunderstanding_patterns,
            "prompt_suggestions": prompt_suggestions
        }

6.3 长期发展路线

看宝AI发展路线图
├── 短期（1-3个月）
│   ├── 完善Few-Shot示例库
│   ├── 建立基础质量监控
│   └── 优化Prompt模板
│
├── 中期（3-6个月）
│   ├── LoRA微调优化任务理解
│   ├── 引入ReAct框架
│   ├── 建立用户反馈收集机制
│   └── 开始RLHF实验
│
└── 长期（6-12个月）
    ├── 完整RLHF对齐
    ├── 多工具协同能力
    ├── 个性化适配
    └── 持续学习和更新

---

附录

A. 关键术语表

B. 推荐资源

- 原始论文: Attention is All You Need (Transformer) - 缩放定律: Scaling Laws for Neural Language Models (Kaplan et al.) - Chinchilla: Training Compute-Optimal Large Language Models - InstructGPT: Training language models to follow instructions with human feedback - LoRA: LoRA: Low-Rank Adaptation of Large Language Models - RLHF: Augmenting RL with Human Feedback - Chain-of-Thought: Chain-of-Thought Prompting Elicits Reasoning

C. 实践工具

- Hugging Face Transformers: 模型加载和微调 - PEFT: 参数高效微调库 - TRL: RLHF训练库 - LangChain: LLM应用开发框架 - LlamaIndex: RAG知识检索框架

---