1
# Prompt Engineering 方式
2
prompt = """
3
你是一个金融分析师。请分析以下财报：
4

5
财报内容：{financial_report}
6

7
请按以下格式输出：
8
1. 营收分析：
9
2. 利润分析：
10
3. 风险提示：
11
"""
12

13
response = llm.generate(prompt)

1
# Fine-tuning 方式
2
# 训练数据示例
3
training_data = [
4
    {
5
        "messages": [
6
            {"role": "system", "content": "你是一个专业的金融分析师。"},
7
            {"role": "user", "content": "分析这份财报：{report_1}"},
8
            {"role": "assistant", "content": "1. 营收分析：...\n2. 利润分析：...\n3. 风险提示：..."}
9
        ]
10
    }
11
]
12

13
# Fine-tune 后的模型直接理解任务
14
response = llm.generate("分析这份财报：{report_new}")

1
# 不同微调方法的显存估算
2
def estimate_vram(model_size_b: float, method: str, batch_size: int = 1):
3
    """
4
    模型大小单位：Billions 参数
5
    """
6
    # 基础模型显存
7
    base_vram = model_size_b * 2  # FP16
8

9
    if method == "full":
10
        # 全量微调：模型 + 梯度 + 优化器 + 激活值
11
        return base_vram * 4 + batch_size * model_size_b * 2
12

13
    elif method == "lora":
14
        # LoRA：只更新 LoRA 参数
15
        lora_params = model_size_b * 0.01  # ~1%
16
        return base_vram + lora_params * 2 + batch_size * model_size_b * 0.1
17

18
    elif method == "qlora":
19
        # QLoRA：NF4 量化 + LoRA
20
        base_vram = model_size_b * 0.5  # 4-bit 量化
21
        lora_params = model_size_b * 0.01
22
        return base_vram + lora_params * 2 + batch_size * model_size_b * 0.05
23

24
    elif method == "ptuning":
25
        # PTuning：只训练 prompt embedding 和 MLP
26
        prompt_params = model_size_b * 0.001
27
        return base_vram + prompt_params * 2 + batch_size * model_size_b * 0.1

1
# LoRA 核心公式
2
class LoRALinear(torch.nn.Module):
3
    def __init__(self, original_layer, rank: int = 4, alpha: float = 1.0):
4
        super().__init__()
5
        self.original = original_layer
6
        self.original.weight.requires_grad = False  # 冻结
7

8
        # LoRA 参数
9
        d, k = original_layer.weight.shape
10
        self.rank = rank
11
        self.alpha = alpha
12

13
        # A: 随机初始化（先用随机小值）
14
        self.lora_A = torch.nn.Parameter(torch.randn(rank, k) * 0.01)
15
        # B: 零初始化（保证初始时与原模型一致）
16
        self.lora_B = torch.nn.Parameter(torch.zeros(d, rank))
17

18
    def forward(self, x):
19
        # 原模型输出
20
        original_output = self.original(x)
21

22
        # LoRA 增量
23
        lora_output = (x @ self.lora_A.T @ self.lora_B.T) * (self.alpha / self.rank)
24

25
        return original_output + lora_output

1
from peft import LoraConfig, get_peft_model, TaskType
2

3
# LoRA 配置
4
lora_config = LoraConfig(
5
    task_type=TaskType.CAUSAL_LM,  # 任务类型
6
    r=8,                             # 秩（rank）
7
    lora_alpha=16,                   # 缩放因子
8
    lora_dropout=0.05,               # Dropout
9
    target_modules=[                  # 应用 LoRA 的模块
10
        "q_proj", "v_proj",          # Attention
11
        "k_proj", "o_proj",
12
        "gate_proj", "up_proj", "down_proj"  # FFN
13
    ],
14
    bias="none",                      # 不训练 bias
15
)
16

17
# 将 LoRA 应用到模型
18
model = get_peft_model(base_model, lora_config)
19
model.print_trainable_parameters()
20
# 输出：trainable params: 4,194,304 || all_params: 6,738,415,616 || trainable%: 0.062

1
# LoRA 超参数搜索
2
lora_experiments = [
3
    {"r": 4, "alpha": 8, "target": ["q_proj", "v_proj"]},
4
    {"r": 8, "alpha": 16, "target": ["q_proj", "v_proj", "k_proj"]},
5
    {"r": 16, "alpha": 32, "target": ["q_proj", "v_proj", "k_proj", "o_proj"]},
6
    {"r": 8, "alpha": 16, "target": "all-linear"},  # 所有线性层
7
]

1
from bitsandbytes import BitsAndBytesConfig
2

3
# QLoRA 配置
4
bnb_config = BitsAndBytesConfig(
5
    # NF4 量化（4-bit NormalFloat）
6
    load_in_4bit=True,
7
    bnb_4bit_quant_type="nf4",
8
    # 双重量化
9
    bnb_4bit_use_double_quant=True,
10
    # 计算dtype
11
    bnb_4bit_compute_dtype=torch.bfloat16,
12
)
13

14
# 加载量化模型
15
model = AutoModelForCausalLM.from_pretrained(
16
    model_name,
17
    quantization_config=bnb_config,
18
    device_map="auto"
19
)

1
class NF4Quantizer:
2
    """
3
    NF4 (4-bit NormalFloat) 量化
4
    核心思想：数据分布近似正态时，NF4 比普通 4-bit 量化更优
5
    """
6
    def __init__(self):
7
        # NF4 的 16 个量化中心（对应 4-bit = 16 个值）
8
        self.quant_centers = self._get_nf4_centers()
9

10
    def _get_nf4_centers(self):
11
        """NF4 量化中心（基于正态分布分位数）"""
12
        import scipy.stats as stats
13
        # 8 个正值 + 8 个负值（对称）
14
        positive = [stats.norm.ppf((i + 0.5) / 16) for i in range(8)]
15
        negative = [-p for p in positive]
16
        return sorted(negative + positive)
17

18
    def quantize(self, tensor: torch.Tensor):
19
        """量化到 NF4"""
20
        flat = tensor.flatten()
21
        quantized = torch.zeros_like(flat, dtype=torch.uint8)
22

23
        for i, val in enumerate(flat):
24
            # 找最近的量化中心
25
            distances = [abs(val - c) for c in self.quant_centers]
26
            quantized[i] = argmin(distances)
27

28
        return quantized
29

30
    def dequantize(self, quantized, shape):
31
        """反量化"""
32
        flat = quantized.flatten()
33
        return torch.tensor([
34
            self.quant_centers[q] for q in flat
35
        ]).reshape(shape)

1
from peft import PromptTuningConfig, PromptTuningInit
2

3
# PTuning 配置
4
ptuning_config = PromptTuningConfig(
5
    task_type=TaskType.CAUSAL_LM,
6
    prompt_tuning_init=PromptTuningInit.TEXT,  # 或 RANDOM
7
    prompt_tuning_init_text="请用专业的方式回答用户问题：",  # 初始 prompt
8
    num_virtual_tokens=20,  # 虚拟 token 数量
9
    embedding_projection_dim=1024,
10
)

1
# P-Tuning v2：在每层都添加 learnable prefix
2
ptuning_v2_config = PromptTuningConfig(
3
    task_type=TaskType.CAUSAL_LM,
4
    num_virtual_tokens=16,
5
    num_layers=32,  # 参与的所有层数
6
    num_heads=12,
7
    hidden_size=768,
8
)

1
# 数据格式转换
2
def prepare_training_data(raw_data: list, tokenizer) -> list:
3
    """将对话数据转换为训练格式"""
4
    formatted = []
5

6
    for item in raw_data:
7
        # 拼接对话
8
        text = ""
9
        for msg in item["messages"]:
10
            if msg["role"] == "system":
11
                text += f"系统：{msg['content']}\n"
12
            elif msg["role"] == "user":
13
                text += f"用户：{msg['content']}\n"
14
            elif msg["role"] == "assistant":
15
                text += f"助手：{msg['content']}\n"
16

17
        # Tokenize
18
        encoding = tokenizer(
19
            text,
20
            truncation=True,
21
            max_length=2048,
22
            padding="max_length"
23
        )
24

25
        formatted.append({
26
            "input_ids": encoding["input_ids"],
27
            "attention_mask": encoding["attention_mask"],
28
            "labels": encoding["input_ids"].copy()
29
        })
30

31
    return formatted
32

33
# 数据清洗检查
34
data_quality_checks = {
35
    "max_length": "不超过模型上下文限制",
36
    "min_length": "过滤过于简短的样本",
37
    "dedup": "去除重复数据",
38
    "format": "确保对话格式完整（user/assistant 配对）",
39
    "lang": "确保目标语言一致"
40
}

1
from transformers import TrainingArguments
2

3
training_args = TrainingArguments(
4
    # 基础配置
5
    output_dir="./output",
6
    num_train_epochs=3,
7
    per_device_train_batch_size=4,
8
    gradient_accumulation_steps=4,  # 伪 batch = 16
9

10
    # 优化器配置
11
    optim="paged_adamw_32bit",  # 节省显存
12
    learning_rate=2e-4,
13
    weight_decay=0.001,
14

15
    # 学习率调度
16
    lr_scheduler_type="cosine",
17
    warmup_ratio=0.03,
18

19
    # 显存优化
20
    gradient_checkpointing=True,  # 用计算换显存
21
    fp16=False,
22
    bf16=True,  # A100 支持 bf16
23

24
    # 日志与保存
25
    logging_steps=10,
26
    save_strategy="epoch",
27
    save_total_limit=3,
28

29
    # 其他
30
    remove_unused_columns=False,
31
    group_by_length=True,  # 相近长度打包加速
32
)

1
class EWCRegularizer:
2
    """
3
    Elastic Weight Consolidation
4
    防止灾难性遗忘：对重要参数添加惩罚
5
    """
6
    def __init__(self, model, dataloader, fisher_diagonal=None):
7
        self.model = model
8
        self.fisher_diagonal = fisher_diagonal or self._compute_fisher(dataloader)
9
        self.params_old = {n: p.clone() for n, p in model.named_parameters()}
10

11
    def _compute_fisher(self, dataloader):
12
        """计算 Fisher Information Matrix 对角线"""
13
        fisher = {}
14
        for name, param in self.model.named_parameters():
15
            fisher[name] = torch.zeros_like(param)
16

17
        self.model.eval()
18
        for batch in dataloader:
19
            self.model.zero_grad()
20
            output = self.model(**batch)
21
            loss = output.loss
22
            loss.backward()
23

24
            for name, param in self.model.named_parameters():
25
                if param.grad is not None:
26
                    fisher[name] += param.grad.data ** 2
27

28
        # 归一化
29
        for name in fisher:
30
            fisher[name] /= len(dataloader)
31

32
        return fisher
33

34
    def penalty(self):
35
        """EWC 惩罚项"""
36
        loss = 0
37
        for name, param in self.model.named_parameters():
38
            loss += (self.fisher_diagonal[name] *
39
                    (param - self.params_old[name]) ** 2).sum()
40
        return loss