把固定换到学习时踩过的坑

你是一个专业的金融合规专家，回答问题时需要：
1. 先理解业务场景
2. 分析相关法规要求
3. 给出具体步骤
4. 提示潜在风险
...

pip install torch transformers peft accelerate

[
    {
        "instruction": "回答问题，要求先分析场景，再给出具体步骤",
        "input": "某公司购买了一项专利，需要如何做会计处理？",
        "output": "首先判断专利的来源..."
    },
    ...
]

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PromptTuningConfig, get_peft_model

# 加载基础模型
model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 配置提示微调
prompt_tuning_config = PromptTuningConfig(
    task_type="CAUSAL_LM",           # 因果语言模型
    prompt_tuning_init="TEXT",       # 从文本初始化
    prompt_tuning_init_text="你是一个专业的金融合规专家，",
    num_virtual_tokens=50,           # 虚拟 token 数量
    tokenizer_name_or_path=model_name
)

# 应用提示微调
model = get_peft_model(model, prompt_tuning_config)
model.print_trainable_parameters()
# 输出：trainable params: 38,500 || all params: 6,738,415,616 || trainable%: 0.00057%

# 训练过程（简化）
def train(model, train_dataloader, num_epochs=5):
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3)
    model.train()

    for epoch in range(num_epochs):
        for batch in train_dataloader:
            outputs = model(**batch)
            loss = outputs.loss
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

    return model

# 推理示例
def generate(model, prompt, max_length=512):
    model.eval()
    inputs = tokenizer(prompt, return_tensors="pt")
    with torch.no_grad():
        outputs = model.generate(
            inputs.input_ids,
            max_length=max_length,
            num_return_sequences=1,
            temperature=0.7
        )
    return tokenizer.decode(outputs[0], skip_special_tokens=True)

from tqdm import tqdm

def train_with_monitoring(model, train_dataloader, eval_dataloader):
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3)

    for epoch in range(num_epochs):
        # 训练
        train_loss = 0
        model.train()
        for batch in tqdm(train_dataloader, desc=f"Epoch {epoch}"):
            outputs = model(**batch)
            loss = outputs.loss
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()
            train_loss += loss.item()

        # 评估
        model.eval()
        eval_loss = 0
        exact_match = 0
        for batch in eval_dataloader:
            with torch.no_grad():
                outputs = model.generate(**batch, max_length=256)
                # 计算 exact match 等指标
                ...

        print(f"Epoch {epoch}: train_loss={train_loss:.4f}, eval_loss={eval_loss:.4f}, exact_match={exact_match:.2f}")

# 调整最大长度
max_input_length = 512 - num_virtual_tokens - output_length

# 对长问题进行截断或分段
def preprocess_question(question, max_length=256):
    if len(question) > max_length:
        # 保留关键信息
        return summarize_question(question, max_length)
    return question

# 加载不同场景的 prompt
compliance_prompt = load_prompt_tuning("compliance")
risk_prompt = load_prompt_tuning("risk")

# 切换场景
def switch_model(model, prompt_weights):
    model.prompt_encoder.embedding.weight.data = prompt_weights

# 使用
switch_model(model, compliance_prompt)
result = model.generate("某公司需要...")

# 结合 RAG
from transformers import RagRetriever

def retrieve_and_generate(query):
    # 检索相关文档
    docs = retriever.retrieve(query, top_k=3)
    # 构建增强输入
    context = "\n".join(docs)
    augmented_input = f"参考文档：\n{context}\n\n问题：{query}"
    # 生成答案
    return model.generate(augmented_input)