第03章 HuggingFace生态：模型、Tokenizer与数据集的完整使用
第03章 HuggingFace生态：模型、Tokenizer与数据集的完整使用
“HuggingFace不只是一个模型仓库，它是LLM工程师的操作系统。” —— 大模型工程实践
如果说PyTorch是LLM工程的引擎，那HuggingFace就是驾驶舱。几乎所有的微调工作、模型加载、数据集管理都在这个生态里进行。
本章全面覆盖HuggingFace生态的核心工具：从Tokenizer到AutoModel，从datasets库到pipeline API，再到Hub上的模型管理。读完本章，你能把任何HuggingFace模型从Hub拉下来、处理数据集、推理出结果。
3.1 理解Tokenizer
LLM不接受文字，只接受数字。Tokenizer是这个翻译器。
from transformers import AutoTokenizer
# 加载Tokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
# 基本使用
text = "大模型工程实战：从理论到落地"
# 文字 → Token IDs
token_ids = tokenizer.encode(text)
print(f"Token IDs: {token_ids}")
# [151644, 46700, 5655, 66093, 102378, ...]
# Token IDs → 文字（解码）
decoded = tokenizer.decode(token_ids)
print(f"解码: {decoded}")
# 查看每个Token是什么
tokens = tokenizer.tokenize(text)
print(f"Tokens: {tokens}")
# ['大', '模型', '工程', '实战', '：', '从', '理论', '到', '落地']
# 注意：中文通常按词或子词分割，不一定是单字
# 计算Token数量（非常重要！）
def count_tokens(text: str, tokenizer) -> int:
"""计算文本的Token数量，不含特殊Token"""
return len(tokenizer.encode(text, add_special_tokens=False))
print(f"Token数量: {count_tokens(text, tokenizer)}")
Tokenizer的三个工程陷阱：
# 陷阱1：不同模型的Token数不同
from transformers import AutoTokenizer
text = "The quick brown fox jumps over the lazy dog"
for model_name in ["gpt2", "bert-base-uncased"]:
tok = AutoTokenizer.from_pretrained(model_name)
n = len(tok.encode(text))
print(f"{model_name}: {n} tokens")
# gpt2: 9 tokens
# bert-base-uncased: 11 tokens (可能不同)
# 陷阱2：聊天模型需要特定格式（chat template）
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
messages = [
{"role": "system", "content": "你是一个专业的合同审查助手。"},
{"role": "user", "content": "帮我分析这份合同的风险条款。"},
]
# 正确方式：用apply_chat_template
formatted = tokenizer.apply_chat_template(
messages,
tokenize=False, # 返回字符串，不要编码
add_generation_prompt=True, # 在末尾加上"Assistant:"提示
)
print(formatted)
# <|im_start|>system
# 你是一个专业的合同审查助手。<|im_end|>
# <|im_start|>user
# 帮我分析这份合同的风险条款。<|im_end|>
# <|im_start|>assistant
# ⚠️ 错误方式：直接拼接字符串
wrong_format = f"system: 你是专业助手\nuser: {messages[1]['content']}"
# 这种方式模型可能看不懂，输出质量大幅下降！
# 陷阱3：padding方向影响批处理
tokenizer.padding_side = "left" # 生成任务：左padding
tokenizer.padding_side = "right" # 训练任务：右padding
batch = ["短文本", "这是一段较长的文本，用于演示padding的方向"]
encoded = tokenizer(
batch,
padding=True, # 自动填充到相同长度
truncation=True, # 超长截断
max_length=50,
return_tensors="pt",
)
print(f"input_ids shape: {encoded['input_ids'].shape}")
print(f"attention_mask shape: {encoded['attention_mask'].shape}")
# attention_mask: 1=真实token, 0=padding token
3.2 加载和使用模型
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
# 标准加载方式：自动选择精度和设备
def load_chat_model(model_name: str, load_in_8bit: bool = False):
"""
加载聊天模型的标准方式
参数说明：
- torch_dtype=torch.bfloat16: 半精度，节省内存，速度快
- device_map="auto": 自动分配GPU（多卡时自动分配）
- load_in_8bit: 8位量化，进一步节省内存（需要bitsandbytes）
"""
tokenizer = AutoTokenizer.from_pretrained(model_name)
load_kwargs = {
"torch_dtype": torch.bfloat16,
"device_map": "auto",
}
if load_in_8bit:
load_kwargs["load_in_8bit"] = True
model = AutoModelForCausalLM.from_pretrained(model_name, **load_kwargs)
model.eval() # 推理模式：关闭dropout等
return tokenizer, model
# 最简单的推理
def chat(model, tokenizer, messages: list[dict], **kwargs) -> str:
"""
统一的聊天接口
"""
# 格式化输入
inputs = tokenizer.apply_chat_template(
messages,
tokenize=True,
add_generation_prompt=True,
return_tensors="pt",
).to(model.device)
# 生成
with torch.no_grad():
output_ids = model.generate(
inputs,
max_new_tokens=kwargs.get("max_new_tokens", 512),
temperature=kwargs.get("temperature", 0.7),
do_sample=kwargs.get("do_sample", True),
top_p=kwargs.get("top_p", 0.9),
repetition_penalty=kwargs.get("repetition_penalty", 1.1),
pad_token_id=tokenizer.eos_token_id,
)
# 只解码新生成的部分
generated_ids = output_ids[0][inputs.shape[1]:]
return tokenizer.decode(generated_ids, skip_special_tokens=True)
# 使用示例
tokenizer, model = load_chat_model("Qwen/Qwen2.5-7B-Instruct")
response = chat(model, tokenizer, [
{"role": "system", "content": "你是一个专业的代码审查助手。"},
{"role": "user", "content": "请审查这段Python代码的安全问题：\n```python\nquery = f'SELECT * FROM users WHERE id = {user_id}'\n```"},
])
print(response)
3.3 流式输出（Streaming）
from transformers import TextIteratorStreamer
from threading import Thread
def chat_streaming(model, tokenizer, messages: list[dict]):
"""
流式输出：生成时实时打印（适合Web应用）
"""
inputs = tokenizer.apply_chat_template(
messages,
tokenize=True,
add_generation_prompt=True,
return_tensors="pt",
).to(model.device)
# 创建流式输出器
streamer = TextIteratorStreamer(
tokenizer,
skip_special_tokens=True,
skip_prompt=True, # 不打印输入部分
)
# 在单独线程中运行生成
generation_kwargs = {
"input_ids": inputs,
"max_new_tokens": 512,
"streamer": streamer,
"do_sample": True,
"temperature": 0.7,
}
thread = Thread(target=model.generate, kwargs=generation_kwargs)
thread.start()
# 主线程实时处理输出
full_response = ""
for chunk in streamer:
print(chunk, end="", flush=True) # 实时打印
full_response += chunk
print() # 换行
thread.join()
return full_response
# FastAPI中使用流式输出
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/chat/stream")
async def chat_stream_endpoint(messages: list[dict]):
"""流式聊天API"""
async def generate():
inputs = tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True,
return_tensors="pt"
).to(model.device)
streamer = TextIteratorStreamer(
tokenizer, skip_special_tokens=True, skip_prompt=True
)
thread = Thread(
target=model.generate,
kwargs={"input_ids": inputs, "max_new_tokens": 512, "streamer": streamer}
)
thread.start()
for chunk in streamer:
yield f"data: {chunk}\n\n" # SSE格式
thread.join()
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
3.4 datasets库：数据集管理
HuggingFace的datasets库不只用来加载公开数据集，也是处理和管理训练数据的最佳工具。
from datasets import load_dataset, Dataset, DatasetDict
import pandas as pd
# 加载公开数据集
# 方式1：从Hub加载
dataset = load_dataset("tatsu-lab/alpaca") # 经典指令数据集
print(dataset)
# DatasetDict({
# train: Dataset({features: ['instruction', 'input', 'output', 'text'], num_rows: 52002})
# })
# 方式2：从本地CSV/JSON创建
df = pd.DataFrame({
"instruction": ["总结以下文本", "将以下内容翻译成英文"],
"input": ["这是一篇关于AI的文章...", "大模型改变了软件开发方式"],
"output": ["AI文章摘要...", "Large models have changed..."],
})
local_dataset = Dataset.from_pandas(df)
print(local_dataset)
# 方式3：从JSONL文件创建（最常用的微调数据格式）
import json
import tempfile
import os
# 创建示例数据
samples = [
{
"messages": [
{"role": "system", "content": "你是一个专业的合同分析助手"},
{"role": "user", "content": "这份合同的违约条款是否合理？"},
{"role": "assistant", "content": "根据合同内容分析..."},
]
}
]
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
for sample in samples:
f.write(json.dumps(sample, ensure_ascii=False) + "\n")
tmp_path = f.name
chat_dataset = load_dataset("json", data_files=tmp_path)
os.unlink(tmp_path)
数据集预处理管道：
from datasets import load_dataset
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
def format_instruction_sample(sample: dict) -> dict:
"""
将指令数据集格式化为chat template格式
输入格式：{"instruction": ..., "input": ..., "output": ...}
输出格式：{"text": "<|im_start|>..."}
"""
user_content = sample["instruction"]
if sample.get("input"):
user_content += f"\n\n{sample['input']}"
messages = [
{"role": "user", "content": user_content},
{"role": "assistant", "content": sample["output"]},
]
text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=False,
)
return {"text": text}
def tokenize_sample(sample: dict, max_length: int = 2048) -> dict:
"""tokenize并截断"""
result = tokenizer(
sample["text"],
truncation=True,
max_length=max_length,
padding=False, # 批处理时再padding
)
# 对于SFT，labels = input_ids（预测自身）
result["labels"] = result["input_ids"].copy()
return result
# 构建完整预处理管道
dataset = load_dataset("tatsu-lab/alpaca", split="train[:1000]")
# 并行处理（多核加速）
formatted = dataset.map(
format_instruction_sample,
num_proc=4, # 使用4个CPU核
remove_columns=dataset.column_names, # 删除原始列
desc="格式化数据",
)
tokenized = formatted.map(
tokenize_sample,
num_proc=4,
remove_columns=["text"],
desc="Tokenize",
)
# 划分训练/验证集
splits = tokenized.train_test_split(test_size=0.05, seed=42)
print(splits)
# DatasetDict({
# train: Dataset(950 rows),
# test: Dataset(50 rows)
# })
# 保存到磁盘（下次直接加载，避免重新处理）
splits.save_to_disk("./processed_alpaca")
# 下次加载
from datasets import load_from_disk
cached = load_from_disk("./processed_alpaca")
3.5 Pipeline API：最快的推理方式
对于简单任务，pipeline是最简洁的接口：
from transformers import pipeline
import torch
# 文本生成
generator = pipeline(
"text-generation",
model="Qwen/Qwen2.5-0.5B-Instruct",
device="cuda" if torch.cuda.is_available() else "cpu",
torch_dtype=torch.bfloat16,
)
output = generator(
"LLM工程师的核心技能包括",
max_new_tokens=100,
do_sample=True,
temperature=0.7,
)
print(output[0]["generated_text"])
# 文本分类
classifier = pipeline(
"text-classification",
model="distilbert-base-uncased-finetuned-sst-2-english",
)
result = classifier("This product is amazing!")
print(result) # [{'label': 'POSITIVE', 'score': 0.9998}]
# 批量处理（更高效）
texts = ["文本1", "文本2", "文本3"] * 10 # 30条文本
results = classifier(texts, batch_size=8) # 按批处理
print(f"处理了 {len(results)} 条")
# 特征提取（嵌入向量）
embedder = pipeline(
"feature-extraction",
model="BAAI/bge-m3", # 多语言embedding模型
device="cuda",
)
sentence = "这是一个需要embedding的句子"
embedding = embedder(sentence, return_tensors=True)[0][0] # (1, hidden_size)
print(f"Embedding维度: {embedding.shape}")
# 实际场景：批量生成embedding用于RAG
def batch_embed(texts: list[str], batch_size: int = 32) -> list:
"""批量生成embedding"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
outputs = embedder(batch, return_tensors=True)
# 取[CLS] token的embedding作为句子表示
embeddings = [out[0][0].tolist() for out in outputs]
all_embeddings.extend(embeddings)
return all_embeddings
3.6 模型Hub管理
from huggingface_hub import HfApi, snapshot_download, login
import os
# 登录（需要HuggingFace token）
# 方式1：环境变量
os.environ["HF_TOKEN"] = "hf_xxxxxxxx"
# 方式2：命令行
# huggingface-cli login
# 下载模型到本地（断网环境使用）
def download_model_for_offline(model_name: str, save_dir: str):
"""下载整个模型到本地"""
snapshot_download(
repo_id=model_name,
local_dir=save_dir,
ignore_patterns=["*.msgpack", "*.h5", "flax_model*"], # 跳过非PyTorch格式
)
print(f"模型已保存至: {save_dir}")
# 使用本地路径加载（断网环境）
from transformers import AutoModelForCausalLM, AutoTokenizer
def load_from_local(local_path: str):
tokenizer = AutoTokenizer.from_pretrained(
local_path,
local_files_only=True, # 禁止联网
)
model = AutoModelForCausalLM.from_pretrained(
local_path,
local_files_only=True,
torch_dtype=torch.bfloat16,
device_map="auto",
)
return tokenizer, model
# 上传微调后的模型到Hub（可选）
def upload_finetuned_model(
local_path: str,
repo_name: str,
private: bool = True,
):
"""上传微调模型到HuggingFace Hub"""
api = HfApi()
# 创建仓库
api.create_repo(
repo_id=repo_name,
private=private,
exist_ok=True, # 如果已存在不报错
)
# 上传
api.upload_folder(
folder_path=local_path,
repo_id=repo_name,
commit_message="Add fine-tuned model",
)
print(f"已上传至: https://huggingface.co/{repo_name}")
3.7 实战：构建私有文档问答系统的基础层
把本章知识综合起来，构建一个可以本地运行的文档问答基础：
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from sentence_transformers import SentenceTransformer
import torch
import numpy as np
from pathlib import Path
class LocalDocQA:
"""
完全本地运行的文档问答系统（无需联网）
- LLM：本地推理
- Embedding：本地模型
- 向量存储：内存中（演示用，生产用Qdrant）
"""
def __init__(
self,
llm_path: str = "Qwen/Qwen2.5-7B-Instruct",
embed_path: str = "BAAI/bge-m3",
):
print("加载LLM...")
self.tokenizer = AutoTokenizer.from_pretrained(llm_path)
self.model = AutoModelForCausalLM.from_pretrained(
llm_path,
torch_dtype=torch.bfloat16,
device_map="auto",
)
self.model.eval()
print("加载Embedding模型...")
self.embedder = SentenceTransformer(embed_path)
self.docs: list[str] = []
self.embeddings: np.ndarray = None
def add_documents(self, texts: list[str]):
"""添加文档到知识库"""
self.docs.extend(texts)
new_embeddings = self.embedder.encode(
texts,
batch_size=32,
show_progress_bar=True,
normalize_embeddings=True, # L2归一化，用于余弦相似度
)
if self.embeddings is None:
self.embeddings = new_embeddings
else:
self.embeddings = np.vstack([self.embeddings, new_embeddings])
print(f"知识库现有 {len(self.docs)} 条文档")
def retrieve(self, query: str, top_k: int = 3) -> list[str]:
"""检索最相关的文档"""
query_embedding = self.embedder.encode(
[query],
normalize_embeddings=True
)
# 余弦相似度
scores = np.dot(self.embeddings, query_embedding.T).flatten()
top_indices = np.argsort(scores)[-top_k:][::-1]
return [self.docs[i] for i in top_indices]
def answer(self, question: str, top_k: int = 3) -> str:
"""检索增强生成"""
# 检索相关文档
relevant_docs = self.retrieve(question, top_k=top_k)
context = "\n\n".join([f"[文档{i+1}]\n{doc}"
for i, doc in enumerate(relevant_docs)])
messages = [
{
"role": "system",
"content": "你是一个专业的问答助手。请基于提供的文档内容回答问题。"
"如果文档中没有相关信息，请明确说明。"
},
{
"role": "user",
"content": f"参考文档：\n{context}\n\n问题：{question}"
},
]
inputs = self.tokenizer.apply_chat_template(
messages, tokenize=True, add_generation_prompt=True,
return_tensors="pt"
).to(self.model.device)
with torch.no_grad():
output = self.model.generate(
inputs,
max_new_tokens=512,
temperature=0.3, # 问答任务：低温度更精确
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id,
)
return self.tokenizer.decode(
output[0][inputs.shape[1]:],
skip_special_tokens=True
)
# 使用示例
qa = LocalDocQA()
# 添加知识文档
qa.add_documents([
"LoRA（Low-Rank Adaptation）是一种参数高效微调技术，通过在原始权重矩阵旁边添加低秩矩阵来实现微调，只训练很少的参数。",
"QLoRA在LoRA基础上增加了4位量化，使得在消费级GPU上也能微调70B以上的模型。",
"RLHF（人类反馈强化学习）是训练ChatGPT等模型的关键技术，包括SFT、奖励模型训练、PPO优化三个阶段。",
])
# 提问
answer = qa.answer("LoRA和QLoRA有什么区别？")
print(answer)
本章小结
五个核心认知：
-
Tokenizer决定输入格式：不同模型有不同的分词方式和chat template，必须用
apply_chat_template，不要手动拼接字符串 -
AutoModel系列处理模型加载：
AutoModelForCausalLM、AutoTokenizer是标准接口，device_map="auto"自动处理多卡分布 -
datasets库是数据处理核心：
map(num_proc=N)并行处理，save_to_disk缓存避免重复计算，是微调数据准备的标准工具 -
pipeline API适合快速原型：5行代码完成推理，适合测试和Demo；生产环境用底层API控制更多细节
-
本地路径=离线部署基础：
local_files_only=True加上预先下载，可以在内网/离线环境运行任何HuggingFace模型
核心行动：
# 练习：加载一个本地模型，处理一批数据，保存结果
from transformers import pipeline
pipe = pipeline("text-generation", model="Qwen/Qwen2.5-0.5B-Instruct")
results = pipe(["你好", "什么是大模型工程？"], max_new_tokens=50)
print(results)
本章提示词模板
模板一：HuggingFace数据集处理咨询
我在处理微调训练数据时遇到问题：
[描述问题，例如：tokenize后的数据太长，截断后信息损失严重]
数据样本格式：
{
"instruction": "...",
"output": "..."
}
模型: [模型名]
max_length: [值]
平均样本长度（tokens）: [值]
最长样本长度（tokens）: [值]
请帮我：
1. 分析truncation策略（从头截断 vs 从尾截断 vs 按句子截断）
2. 给出数据清洗建议（如何过滤掉对训练没用的样本）
3. 提供完整的map()预处理代码
4. 如何验证预处理结果的正确性
模板二：模型推理优化咨询
我的本地推理服务遇到性能问题：
模型: [模型名，参数量]
GPU: [型号和显存]
当前批处理大小: [N]
当前平均延迟: [ms]
目标延迟: [ms]
请分析：
1. 当前配置下的理论最大吞吐量是多少？
2. 批处理大小应该调整到多少？（延迟vs吞吐量权衡）
3. 量化（int8/int4）能带来多少提升？有什么代价？
4. 是否应该切换到vLLM？预期收益是多少？
5. 给出完整的优化步骤（优先级顺序）