HuggingFace for LLM Engineering：从 0 跑通一个真正的 LLM Workflow

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def load_tokenizer(model_name: str):
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
    logger.info("tokenizer 加载完成：vocab=%d pad=%s", tokenizer.vocab_size, tokenizer.pad_token)
    return tokenizer

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def render_chat(tokenizer, messages, add_generation_prompt: bool = True) -> str:
    return tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=add_generation_prompt
    )

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def load_model(model_name: str, dtype: str = "float32", device: str = "cpu"):
    torch_dtype = _DTYPES.get(dtype, torch.float32)
    model = AutoModelForCausalLM.from_pretrained(model_name, dtype=torch_dtype)
    model.to(device)
    model.eval()
    n_params = sum(p.numel() for p in model.parameters())
    logger.info("模型加载完成：%s | %.1fM 参数 | %s | %s", model_name, n_params / 1e6, dtype, device)
    return model

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def next_token_logits(tokenizer, model, messages):
    enc = tokenizer.apply_chat_template(
        messages, tokenize=True, return_dict=True, return_tensors="pt"
    )
    with torch.no_grad():
        logits = model(**enc).logits
    return logits

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def batch_generate(tokenizer, model, prompts, max_new_tokens: int = 64):
    tokenizer.padding_side = "left"
    chats = [render_chat(tokenizer, [{"role": "user", "content": p}]) for p in prompts]
    batch = tokenizer(chats, return_tensors="pt", padding=True, truncation=True, max_length=512)
    batch = {k: v.to(model.device) for k, v in batch.items()}
    with torch.no_grad():
        output = model.generate(
            **batch, max_new_tokens=max_new_tokens,
            do_sample=False, pad_token_id=tokenizer.pad_token_id,
        )
    generated = output[:, batch["input_ids"].shape[1]:]
    return tokenizer.batch_decode(generated, skip_special_tokens=True)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def split_dataset(rows, eval_ratio: float = 0.25, seed: int = 0):
    if rows and all("split" in row for row in rows):
        train = [row["messages"] for row in rows if row["split"] == "train"]
        eval_ = [row["messages"] for row in rows if row["split"] == "eval"]
        return train, eval_
    rng = random.Random(seed)
    shuffled = rows[:]
    rng.shuffle(shuffled)
    n_eval = max(1, int(len(shuffled) * eval_ratio))
    return [r["messages"] for r in shuffled[n_eval:]], [r["messages"] for r in shuffled[:n_eval]]

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def build_labels(tokenizer, messages):
    input_ids = _as_ids(tokenizer.apply_chat_template(messages, tokenize=True))
    prompt_ids = _as_ids(tokenizer.apply_chat_template(
        messages[:-1], tokenize=True, add_generation_prompt=True
    ))
    labels = list(input_ids)
    n_prompt = len(prompt_ids)
    labels[:n_prompt] = [-100] * n_prompt
    return input_ids, labels

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def make_lora_model(model, cfg):
    lora_cfg = LoraConfig(
        r=cfg.lora_r,
        lora_alpha=cfg.lora_alpha,
        lora_dropout=cfg.lora_dropout,
        target_modules=cfg.target_modules,
        task_type="CAUSAL_LM",
    )
    return get_peft_model(model, lora_cfg)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def _collate(batch_ids, batch_labels, pad_id):
    max_len = max(len(ids) for ids in batch_ids)
    input_ids, labels, attention = [], [], []
    for ids, labs in zip(batch_ids, batch_labels):
        pad_n = max_len - len(ids)
        input_ids.append(ids + [pad_id] * pad_n)
        labels.append(labs + [-100] * pad_n)
        attention.append([1] * len(ids) + [0] * pad_n)
    return torch.tensor(input_ids), torch.tensor(labels), torch.tensor(attention)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def save_adapter(model, output_dir: str, cfg) -> None:
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    model.save_pretrained(out)
    cfg.save(out / "config.json")

def load_adapter(base_model, adapter_dir):
    from peft import PeftModel
    return PeftModel.from_pretrained(base_model, adapter_dir)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

# 正确基线评测模式（run_lab.py [7/8] 的写法）
tuned_base = load_model(cfg.model_name, dtype=cfg.dtype, device=cfg.device)
tuned_model = PeftModel.from_pretrained(tuned_base, cfg.output_dir)
clean_base = load_model(cfg.model_name, dtype=cfg.dtype, device=cfg.device)

base_eval = eval_loss(clean_base, tokenizer, eval_samples, cfg.max_length)
tuned_eval = eval_loss(tuned_model, tokenizer, eval_samples, cfg.max_length)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def eval_loss(model, tokenizer, samples, max_length: int = 256) -> float:
    model.eval()
    total, count = 0.0, 0
    with torch.no_grad():
        for messages in samples:
            ids, labels = build_labels(tokenizer, messages)
            ids, labels = ids[:max_length], labels[:max_length]
            if all(label == -100 for label in labels):
                continue
            out = model(input_ids=torch.tensor([ids]), labels=torch.tensor([labels]))
            total += float(out.loss)
            count += 1
    return total / count if count else float("nan")

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

完整实现：projects/hf-mini-lab/src/hf_lab/ + scripts/run_lab.py。 重点对照三个地方：

1. data.build_labels 的掩码边界；
2. train.train_lora 的 collate 与优化器参数过滤；
3. run_lab.py [7/8] 的「两次干净加载」模式。