新增Gemma4教程 (#503)

* Gemma4教程

提交Gemma4教程

* 优化gemma4教程
This commit is contained in:
陈榆
2026-04-13 20:41:18 +08:00
committed by GitHub
parent 23dd98a696
commit 0ae047b221
33 changed files with 7314 additions and 0 deletions
View File
File diff suppressed because one or more lines are too long
@@ -0,0 +1,112 @@
# ollama + open-webui 部署 Gemma 4 E4B-it 模型
> 权重与能力说明见 [google/gemma-4-E4B-it](https://huggingface.co/google/gemma-4-E4B-it) 与 [Gemma 4 博文](https://huggingface.co/blog/gemma4)。本地推理也可使用 llama.cpp`llama-server -hf ggml-org/gemma-4-E4B-it-GGUF`。
Ollama 是一个开源的大语言模型服务工具,旨在帮助用户快速在**本地**运行大模型。
Open WebUI 是一个**可扩展、功能丰富、用户友好的自托管WebUI**,旨在完全离线操作。它支持各种LLM运行程序,包括Ollama和OpenAI兼容的API。
本教程使用 Ollama **本地部署**与 Hugging Face 上 `google/gemma-4-E4B-it` 对应的 **`gemma4:e4b`** 标签(以 [Ollama Library](https://ollama.com/library/gemma4) 为准),并用 Open WebUI 部署 Web 界面。
## 环境准备
```
ubuntu 22.04
python 3.12
pytorch 2.5.1
cuda 12.4
```
本文默认学习者已安装好如上环境,如未安装请自行安装。
## 安装 ollama
### 1. macOS和Windows系统安装
macOS用户通过[此安装包链接](https://ollama.com/download/Ollama-darwin.zip)下载安装ollama
Windows用户通过[此安装包链接](https://ollama.com/download/OllamaSetup.exe)下载安装ollama
### 2. Linux系统安装
方案一:在终端输入以下命令,**自动安装ollama**
```bash
curl -fsSL https://ollama.com/install.sh | sh
```
方案二:在终端输入以下命令,**手动安装ollama**
```bash
curl -L https://ollama.com/download/ollama-linux-amd64.tgz -o ollama-linux-amd64.tgz
sudo tar -C /usr -xzf ollama-linux-amd64.tgz
```
如果出现无法下载安装包的情况,修改GitHub镜像源之后再下载安装
```bash
curl -L https://git.886.be/https://github.com/ollama/ollama/releases/download/v0.6.0/ollama-linux-amd64.tgz -o ollama-linux-amd64.tgz
sudo tar -C /usr -xzf ollama-linux-amd64.tgz
```
> 考虑到部分同学配置环境可能会遇到一些问题,我们在 AutoDL 平台准备了 gemma-4-E4B-it 的环境镜像,点击下方链接并直接创建 Autodl 示例即可。
> ***https://www.codewithgpu.com/i/datawhalechina/self-llm/self-llm-gemma4***
## 运行 ollama
```bash
ollama serve
```
## 下载并运行 Gemma 4 E4BOllama`gemma4:e4b`
```bash
ollama run gemma4:e4b
```
## 查看模型运行状态,以检测是否运行模型
```bash
ollama ps
```
## 下载 open-webui
```bash
# 升级 pip
python -m pip install --upgrade pip
# 更换 pypi 源加速库的安装
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
pip install open-webui==0.5.20
```
## 运行 open-webui
```bash
open-webui serve
```
openwebui默认在8080端口运行,如需修改服务端口,请输入如下命令:
```bash
open-webui serve --port 6006
```
如果出现 `Connection to huggingface.co timed out` 等报错,添加环境变量修改镜像源后再运行服务:
```bash
export HF_ENDPOINT=https://hf-mirror.com
```
## 访问 open-webui
打开浏览器,访问 http://localhost:6006 即可访问 open-webui。
在开启 Ollama 服务并运行 `gemma4:e4b` 后,Open WebUI 会**自动检测**本地 Ollama 并选用该模型。
![03-1](./images/03-1.png)
## 测试 Gemma 4 E4B-it 可用性
![03-2](./images/03-2.png)
@@ -0,0 +1,107 @@
# 04-Gemma4-E4B-it evalscope智商情商评测
> 评测对象可与 Ollama 模型 **`gemma4:e4b`**(对应 Hugging Face **[google/gemma-4-E4B-it](https://huggingface.co/google/gemma-4-E4B-it)**)对齐;概述见 [Gemma 4 博文](https://huggingface.co/blog/gemma4)。
## 大模型评测是什么
- 大语言模型评测是指对大语言模型(LLM)在多种任务和场景下的性能进行全面评估的过程。评测的目的是衡量模型的通用能力、特定领域表现、效率、鲁棒性、安全性等多方面性能,以便优化模型设计、指导技术选型和推动模型在实际应用中的部署。
- 评测的主要内容
通用能力:评估模型在语言理解、生成、推理等方面的基础能力。
特定领域表现:针对特定任务(如数学推理、代码生成、情感分析等)的性能评估。
效率与资源消耗:包括模型的训练和推理时间、计算资源需求等。
鲁棒性与可靠性:评估模型在面对噪声、对抗攻击或输入扰动时的稳定性。
伦理与安全性:检测模型是否会产生有害内容、是否存在偏见或歧视。
- EvalScope是魔搭社区官方推出的模型评测与性能基准测试框架,内置多个常用测试基准和评测指标,如MMLU、CMMLU、C-Eval、GSM8K、ARC、HellaSwag、TruthfulQA、MATH和HumanEval等;支持多种类型的模型评测,包括LLM、多模态LLM、embedding模型和reranker模型。EvalScope还适用于多种评测场景,如端到端RAG评测、竞技场模式和模型推理性能压测等。此外,通过ms-swift训练框架的无缝集成,可一键发起评测,实现了模型训练到评测的全链路支持。
官网地址:https://evalscope.readthedocs.io/zh-cn/latest/get_started
# evalscope评测使用方法
## 环境准备
本文基础环境如下:
```
----------------
ubuntu 22.04
python 3.12
Cuda 12.4
PyTorch 2.5.1
----------------
```
2. **pip安装evalscope**
```
pip install evalscope # 安装 Native backend (默认)
# 额外选项
pip install evalscope[opencompass] # 安装 OpenCompass backend
pip install evalscope[vlmeval] # 安装 VLMEvalKit backend
pip install evalscope[rag] # 安装 RAGEval backend
pip install evalscope[perf] # 安装 模型压测模块 依赖
pip install evalscope[all] # 安装所有 backends (Native, OpenCompass, VLMEvalKit, RAGEval)
```
> 考虑到部分同学配置环境可能会遇到一些问题,我们在 AutoDL 平台准备了 gemma-4-E4B-it 的环境镜像,点击下方链接并直接创建 Autodl 示例即可。
> ***https://www.codewithgpu.com/i/datawhalechina/self-llm/self-llm-gemma4***
## 模型评测方法
1. **创建ollama服务器**
这里首先使用ollama创建兼容 OpenAI API 接口的服务器,然后使用evalscope进行评测。当然接入其他的api也是可以的。
```bash
curl -L https://git.886.be/https://github.com/ollama/ollama/releases/download/v0.6.0/ollama-linux-amd64.tgz -o ollama-linux-amd64.tgz
sudo tar -C /usr -xzf ollama-linux-amd64.tgz
```
```bash
ollama serve #运行ollama服务器
```
新建一个bash窗口
```bash
ollama run gemma4:e4b
```
此时可以在控制台直接与模型对话。
1. **执行评测**
新建eval_api.py文件,并输入以下代码:
```
# 导入执行任务的函数和任务配置类
from evalscope.run import run_task
from evalscope.config import TaskConfig
"""
以下为多个AI服务的API端点地址,用于配置任务:
- siliconflow: https://api.siliconflow.cn/v1/chat/completions
- dashscope: https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions
- modelscope: https://api-inference.modelscope.cn/v1/chat/completions
- xunfei: https://maas-api.cn-huabei-1.xf-yun.com/v1/chat/completions
"""
# 配置任务参数
task_cfg = TaskConfig(
model='gemma4:e4b', # 指定使用的模型
api_url='http://localhost:11434/v1/chat/completions', # 指定API端点,这里使用的是ollama默认的api接口
api_key='sk-xxxxxxx', # API密钥(需替换为实际密钥,ollama 的api_key
eval_type='service', # 指定评估类型为服务模式
datasets=['iquiz'], # 指定使用的数据集(这个测试集可以快速测试模型的智商和情商)
generation_config={ # 文本生成配置
'max_tokens': 4096, # 最大令牌数
'max_new_tokens': 4096, # 最大新生成令牌数
'temperature': 1.0, # 温度参数,这里设置为1.0,模型的输出随机性较大,所以可能会有些实验误差
},
work_dir='outputs/Gemma4-E4B-it', # 输出目录
)
# 执行任务
run_task(task_cfg=task_cfg)
```
新建一个bash窗口,也就是控制台中执行。
控制台运行`python eval_api.py`命令即可。
等待3分钟左右评测就完成啦,控制台输出的结果如下图所示:
![](./images/04-01.png)
实验结果可能有误差,因为在评测任务配置中我们把temperature调到了1.0,如果调小一些,可能会得到更精确的结果。
可以看到模型的得分还是不错的,模型评测的文件保存在`/root/outputs/Gemma4-E4B-it/20250315_164601/reviews/Gemma4-E4B-it`目录下。
![](./images/04-05.png)
## evalscope简介:
- 支持多种模型评测backend,包括OpenAI API、OpenCompass、VLMEvalKit、RAGEval等。
![](./images/04-02.png)
- 支持自定义评测任务和数据集,支持多种评测指标。
![](./images/04-03.png)
模型评测对于验证和优化大模型如Gemma4-E4B-it至关重要。通过评测,我们可以全面了解模型的性能、能力边界及潜在问题,确保其在实际应用中的表现符合预期,并推动持续改进。此外,评测还能检测模型的公平性和安全性,提升用户体验,并为不同模型间的对比分析提供客观依据。最终,评测结果为后续版本迭代提供了关键数据支持,保障模型在实际场景中的可靠性和有效性。
File diff suppressed because it is too large Load Diff
+379
View File
@@ -0,0 +1,379 @@
# 05-gemma-4-E4B-it LoRA 微调
> 基座模型:**[google/gemma-4-E4B-it](https://huggingface.co/google/gemma-4-E4B-it)**Gemma 4 多模态指令版,详见 [Gemma 4 发布博文](https://huggingface.co/blog/gemma4))。以下 LoRA 示例以**纯文本对话**数据为主,与官方模型卡中的 `AutoModelForCausalLM` 文本路径一致;若需图像/音频管线请改用 `AutoModelForMultimodalLM`。
本节我们简要介绍如何基于 transformers、peft 等框架,使用由笔者合作开源的 [Chat-甄嬛](https://github.com/KMnO4-zx/huanhuan-chat) 项目中的**嬛嬛数据集**作为微调数据集,对 gemma-4-E4B-it 模型进行 LoRA 微调, 以构建一个能够模拟甄嬛对话风格的个性化 LLM , 数据集路径为[`../../dataset/huanhuan.json`](../../dataset/huanhuan.json)。
> **LoRA** 是一种高效微调方法,深入了解其原理可参见博客:[知乎|深入浅出 LoRA](https://zhuanlan.zhihu.com/p/650197598)。
> 本教程会在同目录下给大家提供一个 [**notebook** 文件 (05-gemma-4-E4B-it LoRA.ipynb)](05-gemma-4-E4B-it%20LoRA.ipynb) ,来帮助大家更好的学习。
## 环境配置
实验所依赖的基础开发环境如下:
```
----------------
ubuntu 22.04
Python 3.12.3
cuda 12.4
pytorch 2.5.1
----------------
```
> 本文默认学习者已安装好以上 Pytorch(cuda) 环境,如未安装请自行安装。
首先 `pip` 换源加速下载并安装依赖包:
```shell
# 升级pip
python -m pip install --upgrade pip
# 更换 pypi 源加速库的安装
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
# LoRA微调 相关依赖
pip install peft==0.14.0 # 用于 LoRA 微调
# 通用依赖(Gemma 4 需较新 transformers,见模型卡)
pip install -U transformers accelerate
pip install huggingface_hub>=0.28.0
pip install sentencepiece==0.2.0 # 用于处理文本数据
pip install accelerate==1.5.1 # 用于分布式训练和混合精度训练
pip install datasets==3.3.2 # 用于加载和处理数据集
```
> 考虑到部分同学配置环境可能会遇到一些问题,我们在 AutoDL 平台准备了 gemma-4-E4B-it 的环境镜像,点击下方链接并直接创建 Autodl 示例即可。
> ***https://www.codewithgpu.com/i/datawhalechina/self-llm/self-llm-gemma4***
## 模型下载
`/root/autodl-tmp` 路径下新建 `model_download.py` 文件并在其中粘贴以下代码,并保存文件。
```python
from huggingface_hub import snapshot_download
model_dir = snapshot_download("google/gemma-4-E4B-it", cache_dir="./")
```
> 注意:记得修改 `cache_dir` 为你的模型下载路径;若访问 hub 较慢可设置环境变量 `HF_ENDPOINT`(如镜像站)。
在终端运行 `python /root/autodl-tmp/model_download.py` 执行下载,权重体积以 Hugging Face 模型页显示为准(Safetensors 约 8B 量级参数),下载时间取决于带宽。
## 指令集构建
LLM 的微调一般指指令微调过程。所谓指令微调,是说我们使用的微调数据形如:
```json
{
"instruction": "回答以下用户问题,仅输出答案。",
"input": "1+1等于几?",
"output": "2"
}
```
其中,`instruction` 是用户指令,告知模型其需要完成的任务;`input` 是用户输入,是完成用户指令所必须的输入内容;`output` 是模型应该给出的输出。
即我们的核心训练目标是让模型具有理解并遵循用户指令的能力。因此,在指令集构建时,我们应针对我们的目标任务,针对性构建任务指令集。
例如,在本节我们使用由笔者合作开源的 [**Chat-甄嬛**](https://github.com/KMnO4-zx/huanhuan-chat) 项目作为示例,我们的目标是构建一个能够模拟甄嬛对话风格的个性化 LLM,因此我们构造的指令形如:
```json
{
"instruction": "你是谁?",
"input": "",
"output": "家父是大理寺少卿甄远道。"
}
```
我们所构造的全部指令数据集会被保存在根目录下。
## 数据格式化
`LoRA` 训练的数据是需要经过格式化、编码之后再输入给模型进行训练的,如果是熟悉 `Pytorch` 模型训练流程的同学会知道,我们一般需要将输入文本编码为 `input_ids`,将输出文本编码为 `labels`,编码之后的结果都是多维的向量。
为了得到 InternLM3-8b-Instruct 的 Prompt Template,使用 tokenizer 构建 messages 并打印, 查看 chat_template 的输出格式
```python
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": '你好呀'},
{"role": "assistant", "content": '有什么可以帮你的?'}
]
# 使用chat_template将messages格式化并打印
print(tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True))
## 得到输出结果如下
#<bos><start_of_turn>user
#You are a helpful assistant.
#你好呀<end_of_turn>
#<start_of_turn>model
#有什么可以帮你的?<end_of_turn>
#<start_of_turn>model
```
然后我们就可以定义预处理函数 `process_func`,这个函数用于对每一个样本,编码其输入、输出文本并返回一个编码后的字典,方便模型使用:
```python
system_prompt = '现在你要扮演皇帝身边的女人--甄嬛'
def process_func(example):
MAX_LENGTH = 384 # 分词器会将一个中文字切分为多个token,因此需要放开一些最大长度,保证数据的完整性
input_ids, attention_mask, labels = [], [], []
# 构建指令部分的输入, 可参考上面的输出格式进行调整和补充
instruction = tokenizer(
f"<s><|im_start|>system\n{system_prompt}<|im_end|>\n"
f"<|im_start|>user\n{example['instruction'] + example['input']}<|im_end|>\n"
f"<|im_start|>assistant\n",
add_special_tokens=False
)
# 构建模型回复部分的输入
response = tokenizer(
f"{example['output']}",
add_special_tokens=False
)
# 拼接指令和回复部分的 input_ids
input_ids = instruction["input_ids"] + response["input_ids"] + [tokenizer.pad_token_id]
# 拼接指令和回复部分的 attention_mask
attention_mask = instruction["attention_mask"] + response["attention_mask"] + [1] # 因为 EOS token 也需要关注,所以补充为 1
# 构建标签
# 对于指令部分,使用 -100 忽略其损失计算;对于回复部分,保留其 input_ids 作为标签
labels = [-100] * len(instruction["input_ids"]) + response["input_ids"] + [tokenizer.pad_token_id]
# 如果总长度超过最大长度,进行截断
if len(input_ids) > MAX_LENGTH:
input_ids = input_ids[:MAX_LENGTH]
attention_mask = attention_mask[:MAX_LENGTH]
labels = labels[:MAX_LENGTH]
return {
"input_ids": input_ids,
"attention_mask": attention_mask,
"labels": labels
}
```
> 补充: gemma-4-E4B-it 采用的 `Prompt Template`格式如下:
```text
<bos><start_of_turn>user
You are a helpful assistant.
你好呀<end_of_turn>
<start_of_turn>model
有什么可以帮你的?<end_of_turn>
<start_of_turn>model
```
## 加载 tokenizer 和半精度模型 (model)
`tokenizer` 是将文本转换为模型 (`model`) 能理解的数字的工具,`model` 是根据这些数字生成文本的核心部分。
以半精度形式加载 `model`, 如果你的显卡比较新的话,可以用 `torch.bfolat` 形式加载。对于自定义模型,必须指定 `trust_remote_code=True` ,以确保加载自定义代码时不会报错。
```python
model_path = '/root/autodl-tmp/google/gemma-4-E4B-it'
tokenizer = AutoTokenizer.from_pretrained(model_path, use_fast=False, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map="auto",
torch_dtype=torch.bfloat16,
attn_implementation="eager",
)
```
> 注意:此处要记得修改为自己的模型路径哦~
如果想要查看模型结构,可以打印模型:
```python
print(model)
# 输出结果(结构名称随 transformers 版本可能略有不同,以本地打印为准)例如:
'''
Gemma4ForConditionalGeneration(
(vision_tower): SiglipVisionModel(
(vision_model): SiglipVisionTransformer(
(embeddings): SiglipVisionEmbeddings(
(patch_embedding): Conv2d(3, 1152, kernel_size=(14, 14), stride=(14, 14), padding=valid)
(position_embedding): Embedding(4096, 1152)
)
(encoder): SiglipEncoder(
(layers): ModuleList(
(0-26): 27 x SiglipEncoderLayer(
(self_attn): SiglipSdpaAttention(
(k_proj): Linear(in_features=1152, out_features=1152, bias=True)
(v_proj): Linear(in_features=1152, out_features=1152, bias=True)
(q_proj): Linear(in_features=1152, out_features=1152, bias=True)
(out_proj): Linear(in_features=1152, out_features=1152, bias=True)
)
(layer_norm1): LayerNorm((1152,), eps=1e-06, elementwise_affine=True)
(mlp): SiglipMLP(
(activation_fn): PytorchGELUTanh()
(fc1): Linear(in_features=1152, out_features=4304, bias=True)
(fc2): Linear(in_features=4304, out_features=1152, bias=True)
)
(layer_norm2): LayerNorm((1152,), eps=1e-06, elementwise_affine=True)
)
)
)
(post_layernorm): LayerNorm((1152,), eps=1e-06, elementwise_affine=True)
)
)
(multi_modal_projector): Gemma4MultiModalProjector(
(mm_soft_emb_norm): Gemma4RMSNorm((1152,), eps=1e-06)
(avg_pool): AvgPool2d(kernel_size=4, stride=4, padding=0)
)
(language_model): Gemma4ForCausalLM(
(model): Gemma4TextModel(
(embed_tokens): Gemma4TextScaledWordEmbedding(262208, 2560, padding_idx=0)
(layers): ModuleList(
(0-33): 34 x Gemma4DecoderLayer(
(self_attn): Gemma4Attention(
(q_proj): Linear(in_features=2560, out_features=2048, bias=False)
(k_proj): Linear(in_features=2560, out_features=1024, bias=False)
(v_proj): Linear(in_features=2560, out_features=1024, bias=False)
(o_proj): Linear(in_features=2048, out_features=2560, bias=False)
(q_norm): Gemma4RMSNorm((256,), eps=1e-06)
(k_norm): Gemma4RMSNorm((256,), eps=1e-06)
)
(mlp): Gemma4MLP(
(gate_proj): Linear(in_features=2560, out_features=10240, bias=False)
(up_proj): Linear(in_features=2560, out_features=10240, bias=False)
(down_proj): Linear(in_features=10240, out_features=2560, bias=False)
(act_fn): PytorchGELUTanh()
)
(input_layernorm): Gemma4RMSNorm((2560,), eps=1e-06)
(post_attention_layernorm): Gemma4RMSNorm((2560,), eps=1e-06)
(pre_feedforward_layernorm): Gemma4RMSNorm((2560,), eps=1e-06)
(post_feedforward_layernorm): Gemma4RMSNorm((2560,), eps=1e-06)
)
)
(norm): Gemma4RMSNorm((2560,), eps=1e-06)
(rotary_emb): Gemma4RotaryEmbedding()
(rotary_emb_local): Gemma4RotaryEmbedding()
)
(lm_head): Linear(in_features=2560, out_features=262208, bias=False)
)
)
'''
```
上面打印了 `Gemma4Model` 的模型结构, 可以看到里面的 `self_attn``mlp` 是两个主要的模块, 因此可以考虑将这两个模块作为 **LoRA** 微调 的 `target_modules` , 包括 `q_proj`, `k_proj`, `v_proj`, `o_proj` 以及 `gate_proj``up_proj``down_proj`
通常我们只对 `self_attn` 模块中的 `q_proj`, `k_proj`, `v_proj`, `o_proj`进行微调, 本教程里我们也将对这四个模块进行微调演示, 感兴趣的同学可以自行尝试添加对 `mlp` 中的三个 `proj` 模块进行微调。
## 定义 LoraConfig
`LoraConfig`类用于设置 LoRA 微调参数,虽然可以设置很多参数,但主要的参数没多少,简单讲一讲,感兴趣的同学可以直接看源码。
- `task_type`:模型类型
- `target_modules`:需要训练的模型层的名字,主要就是 `attention`部分的层,不同的模型对应的层的名字不同,可以传入数组,也可以字符串,也可以正则表达式。
- `r``LoRA`的秩,具体可以看 `LoRA`原理。
- `lora_alpha``LoRA alaph` ,具体作用参见 `LoRA` 原理。
- `lora_dropout`: `LoRA` 层的 `Dropout` 比例,用于防止过拟合,具体作用参见 `LoRA` 原理。
`LoRA`的缩放是啥嘞?当然不是 `r`(秩),这个缩放就是 `lora_alpha/r`, 在这个 `LoraConfig`中缩放就是 4 倍。
```python
from peft import LoraConfig, TaskType, get_peft_model
config = LoraConfig(
task_type=TaskType.CAUSAL_LM,
target_modules=["q_proj", "k_proj","v_proj", "o_proj"], # 可以自行添加更多微调的target_modules
inference_mode=False, # 训练模式
r=8, # LoRA 秩
lora_alpha=32, # LoRA alaph,具体作用参见 LoRA 原理
lora_dropout=0.1 # Dropout 比例
)
```
## 自定义 TrainingArguments 参数
`TrainingArguments`类用于设置微调训练过程中的配置参数,这个类的源码也介绍了每个参数的具体作用,当然大家可以来自行探索,这里就简单说几个常用的。
- `output_dir`:模型的输出路径
- `per_device_train_batch_size`:顾名思义 `batch_size`,批量大小
- `gradient_accumulation_steps`: 梯度累加,如果你的显存比较小,那可以把 `batch_size` 设置小一点,梯度累加增大一些。
- `logging_steps`:多少步,输出一次 `log`
- `num_train_epochs`:顾名思义 `epoch`,训练轮次
- `gradient_checkpointing`:梯度检查,这个一旦开启,模型就必须执行 `model.enable_input_require_grads()`,这个原理大家可以自行探索,这里就不细说了。
```python
args = TrainingArguments(
output_dir="/root/autodl-tmp/gemma-4-E4B-it_lora_output",
per_device_train_batch_size=1,
gradient_accumulation_steps=4,
logging_steps=10,
num_train_epochs=3,
save_steps=100,
learning_rate=1e-4,
save_on_each_node=True,
gradient_checkpointing=True
)
```
## 使用 Trainer 训练
我们使用 `Trainer` 类来管理训练过程。`TrainingArguments` 用于设置训练参数,`Trainer` 则负责实际的训练逻辑。
```python
trainer = Trainer(
model=model, # 要训练的模型
args=args, # 训练参数
train_dataset=tokenized_id, # 训练数据集
data_collator=DataCollatorForSeq2Seq(tokenizer=tokenizer, padding=True), # 数据整理器
)
trainer.train() # 开始训练
```
## 加载 LoRA 权重推理
训练好了之后可以使用如下方式加载 `LoRA`权重进行推理:
```python
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
from peft import PeftModel
model_path = '/root/autodl-tmp/google/gemma-4-E4B-it'
lora_path = '/root/autodl-tmp/google/gemma-4-E4B-it_lora_output/checkpoint-2790' # 这里改成 LoRA 输出对应 checkpoint 地址和最终的 epoch 数值 2796
# 加载tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
# 加载模型
model = AutoModelForCausalLM.from_pretrained(model_path,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True).eval()
# 加载lora权重
model = PeftModel.from_pretrained(model, model_id=lora_path)
prompt = "你是谁?"
system_prompt = "现在你要扮演皇帝身边的女人--甄嬛"
print("prompt: ", prompt)
print("system_prompt: ", system_prompt)
inputs = tokenizer.apply_chat_template([{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}],
add_generation_prompt=True,
tokenize=True,
return_tensors="pt",
return_dict=True
).to(model.device) # 将 inputs 移动到模型所在的设备,确保设备一致性
gen_kwargs = {"max_length": 2500, "do_sample": True, "top_k": 1}
with torch.no_grad():
outputs = model.generate(**inputs, **gen_kwargs)
outputs = outputs[:, inputs['input_ids'].shape[1]:]
print("output: ", tokenizer.decode(outputs[0], skip_special_tokens=True))
```
> 注意修改为自己的模型路径哦~
> 如果显示 `Some parameters are on the meta device because they were offloaded to the cpu.` 的报错,需要将实例关机,重启后单独运行本条代码。
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,726 @@
# Gemma 4 E4B GRPO微调教程
> 基座模型:[google/gemma-4-E4B-it](https://huggingface.co/google/gemma-4-E4B-it)Gemma 4 系列说明见 [Hugging Face 博文](https://huggingface.co/blog/gemma4)。Unsloth 与训练脚本请以当前库文档为准。
> 话不多说,直接开始!
本文使用的测试环境为单张 A100,显存 80GB,可根据需求切换不同参数量的模型;Gemma 4 **E4B-it** 在 24GB 级显存上可按量化与序列长度实测调整。
使用的框架为 Unsloth
![06-01](./images/06-01.png)
Unsloth 是一个极其强调资源节省的框架,把所有的资源节省做到了极致,具体来讲Unsloth能够将 Llama-3、Mistral、Phi-4 和 Gemma 等大型语言模型的微调速度提升 2 倍,内存占用减少 70%,并且准确率没有任何下降!
官方文档非常全面,详细指导了如何训练自己的定制模型。其中涵盖了安装和更新 Unsloth、创建数据集、运行和部署模型等基本要素。 Unsloth 让大家在本地或在 Google Colab 和 Kaggle 等平台上训练像 Llama 3 这样的模型变得极其简单。Unsloth简化了整个训练工作流程,包括模型加载、量化、训练、评估、运行、保存、导出,以及与 Ollama、llama.cpp 和 vLLM 等推理引擎的集成。
Unsloth定期与 Hugging Face、Google 和 Meta 的团队合作,以修复 LLM 训练和模型中的错误。因此,当使用 Unsloth 进行训练或使用模型时,可以期待获得最准确的结果。 Unsloth 具有高度可定制性,允许更改聊天模板或数据集格式等内容。Unsloth还为视觉、文本转语音 (TTS)、BERT、强化学习 (RL) 等提供了预构建的脚本!此外,Unsloth支持所有训练方法和所有基于 Transformer 的模型。
Unsloth 可显著加快 Gemma 系列微调并降低显存占用;使用 Unsloth 时,Gemma 4 **E4B-it** 可在约 24GB VRAM 环境下结合量化与配置进行尝试(请以 Unsloth 与 transformers 实际支持为准)。
unsloth为Gemma4提供了Dynamic 2.0量化方法,在5-shot MMLU和KL散度基准测试中提供最佳性能。这意味着可以运行和微调量化后的Gemma4 LLM,同时保持最小的精度损失。unsloth还上传了支持原生长上下文的Gemma4版本。
## 教程概览
本教程将指导您完成 **Gemma 4 E4B 模型的 GRPOGroup Relative Policy Optimization)微调**,这是一种先进的强化学习技术,专门用于提升大语言模型在特定任务上的表现。
### 什么是GRPO
GRPOGroup Relative Policy Optimization)是一种强化学习优化技术,通过设计多个奖励函数来评估模型输出的不同方面,从而指导模型学习期望的行为模式。在数学推理任务中,GRPO可以帮助模型:
- 学会按照特定格式输出答案
- 提高推理过程的逻辑性
- 增强答案的准确性
- 改善输出的结构化程度
### 本教程的学习内容
1. **环境设置**: 安装Unsloth和相关依赖
2. **模型加载**: 加载Gemma 4 E4B预训练模型
3. **LoRA配置**: 设置高效的参数微调
4. **数据处理**: 处理GSM8K数学推理数据集
5. **格式设计**: 定义结构化的输出格式
6. **奖励函数**: 设计多维度评估体系
7. **GRPO训练**: 执行强化学习微调
8. **效果验证**: 测试微调后的模型
9. **模型保存**: 保存训练结果
10. **可视化监控**: 使用SwanLab跟踪训练过程
## 步骤1: 安装必要的依赖包
**安装软件包**
```Python
# pip install --no-deps unsloth vllm==0.8.5.post1
```
## 步骤2: 加载预训练模型和分词器
**准备模型,设置参数**
```Python
# 导入必要的库
from unsloth import FastModel # Unsloth的快速模型加载接口
import torch # PyTorch深度学习框架
# 设置最大序列长度
# 这个参数决定了模型能处理的最大文本长度(以token为单位)
max_seq_length = 1024
# 加载预训练模型和分词器
# 这里我们加载Gemma 4 E4B的指令微调版本
model, tokenizer = FastModel.from_pretrained(
model_name = "google/gemma-4-E4B-it", # 模型路径
max_seq_length = max_seq_length, # 最大序列长度
load_in_4bit = False, # 不使用4位量化,保持精度
load_in_8bit = False, # 不使用8位量化,保持精度
full_finetuning = False, # 使用LoRA微调,不进行全参数微调
)
```
## 步骤3: 配置LoRALow-Rank Adaptation
```Python
# 配置LoRALow-Rank Adaptation)参数
# 将基础模型转换为PEFTParameter Efficient Fine-Tuning)模型
model = FastModel.get_peft_model(
model,
# 层级配置:决定哪些层参与微调
finetune_vision_layers = False, # 关闭视觉层微调(仅文本任务)
finetune_language_layers = True, # 开启语言层微调(必须)
finetune_attention_modules = True, # 开启注意力模块微调(对GRPO很重要)
finetune_mlp_modules = True, # 开启MLP模块微调(建议保持开启)
# LoRA核心参数
r = 8, # LoRA的秩:控制适应层大小,值越大精度越高但可能过拟合
lora_alpha = 8, # LoRA的缩放因子:建议设置为r的值或略大
lora_dropout = 0, # LoRA的dropout率:防止过拟合,这里设为0
bias = "none", # 偏置项设置:不训练偏置项
random_state = 3407, # 随机种子:确保结果可复现
)
```
## 步骤4: 加载和探索GSM8K数据集
**设置CoT思考模版【让模型具备思考能力的必经之路】**
```Python
# 加载GSM8K数据集
from datasets import load_dataset
# 从本地路径加载GSM8K数据集的训练集
# GSM8K是一个包含小学数学推理问题的数据集
dataset = load_dataset("openai/gsm8k", "main", split = "train")
# 查看数据集基本信息
print(f"数据集大小: {len(dataset)} 条记录")
print(f"数据集特征: {dataset.features}")
dataset
```
```Python
# 定义函数提取最终答案
# GSM8K数据集中,最终答案位于####符号之后
def extract_hash_answer(text):
"""
从GSM8K答案中提取最终数值答案
Args:
text (str): 包含推理过程和最终答案的完整文本
Returns:
str or None: 提取的最终答案,如果没有####标记则返回None
"""
if "####" not in text:
return None
# 分割文本,取####后面的部分并去除空格
return text.split("####")[1].strip()
# 测试提取函数
final_answer = extract_hash_answer(dataset[0]["answer"])
print(f"提取的最终答案: {final_answer}")
# 验证提取结果
final_answer
```
## 步骤5: 设计输出格式和系统提示词
```Python
# 定义输出格式的标记符号
# 这些标记帮助我们识别和评估模型输出的不同部分
reasoning_start = "<start_working_out>" # 推理过程开始标记
reasoning_end = "<end_working_out>" # 推理过程结束标记
solution_start = "<SOLUTION>" # 最终答案开始标记
solution_end = "</SOLUTION>" # 最终答案结束标记
# 创建系统提示词
# 这个提示词指导模型按照我们期望的格式输出答案
system_prompt = f"""You are given a problem.
Think about the problem and provide your working out.
Place it between {reasoning_start} and {reasoning_end}.
Then, provide your solution between {solution_start}{solution_end}"""
print("系统提示词内容:")
print(system_prompt)
print("\n这个提示词告诉模型:")
print("1. 需要思考问题")
print("2. 将推理过程放在指定标记之间")
print("3. 将最终答案放在SOLUTION标记之间")
system_prompt
```
**加载数据集【这里使用一个数学推理的数据集】**
## 步骤6: 转换数据集格式
```Python
# 转换数据集格式
# 将原始数据转换为对话格式,便于模型训练
dataset = dataset.map(lambda x: {
# 构建对话prompt,包含系统提示和用户问题
"prompt" : [
{"role": "system", "content": system_prompt}, # 系统消息:指导输出格式
{"role": "user", "content": x["question"]}, # 用户消息:具体的数学问题
],
# 提取标准答案,用于后续的奖励计算
"answer": extract_hash_answer(x["answer"]),
})
print("转换后的数据格式示例:")
print("1. prompt包含系统提示和用户问题")
print("2. answer是提取的数值答案")
print(f"3. 数据集大小保持不变: {len(dataset)}")
# 查看转换后的第一个样本
dataset[0]
```
## 步骤7: 设计奖励函数系统
**奖励函数部分**
GRPO的核心是通过奖励函数来指导模型学习。我们将设计4个奖励函数来评估模型输出的不同方面:
### 7.1 首先定义正则表达式来匹配期望的格式
```Python
# 导入正则表达式库
import re
# 定义正则表达式来匹配期望的输出格式
# 这个正则表达式确保模型输出包含所有必需的标记并按正确顺序排列
match_format = re.compile(
rf"^[\s]{{0,}}" # 开头可以有任意数量的空白字符
rf"{reasoning_start}.+?{reasoning_end}.*?" # 推理过程部分(非贪婪匹配)
rf"{solution_start}(.+?){solution_end}" # 解决方案部分(捕获组获取答案)
rf"[\s]{{0,}}$", # 结尾可以有任意数量的空白字符
flags = re.MULTILINE | re.DOTALL # 多行模式,.匹配换行符
)
print("正则表达式说明:")
print("1. 匹配从<start_working_out>到<end_working_out>的推理过程")
print("2. 匹配从<SOLUTION>到</SOLUTION>的最终答案")
print("3. 捕获SOLUTION标记内的内容作为答案")
print("4. 允许前后有空白字符")
```
### 7.2 奖励函数1: 精确格式匹配
```Python
def match_format_exactly(completions, **kwargs):
"""
奖励函数1: 检查输出是否严格遵循指定格式
Args:
completions: 模型生成的完成文本列表
**kwargs: 其他参数(未使用)
Returns:
list: 每个完成文本的奖励分数列表
"""
scores = []
for completion in completions:
score = 0
response = completion[0]["content"]
# 如果输出完全匹配期望格式,给予高分奖励
if match_format.search(response) is not None:
score += 3.0
scores.append(score)
return scores
print("奖励函数1说明:")
print("- 检查输出是否包含完整的推理过程和解决方案格式")
print("- 格式正确: +3.0分")
print("- 格式不正确: 0分")
```
### 7.3 奖励函数2: 近似格式匹配
```Python
def match_format_approximately(completions, **kwargs):
"""
奖励函数2: 检查格式标记的出现次数
这个函数更宽松,检查各个格式标记是否恰好出现1次。
如果某个标记出现1次,获得奖励;如果出现0次或多次,会被惩罚。
Args:
completions: 模型生成的完成文本列表
**kwargs: 其他参数(未使用)
Returns:
list: 每个完成文本的奖励分数列表
"""
scores = []
for completion in completions:
score = 0
response = completion[0]["content"]
# 检查每个标记的出现次数,理想情况下每个标记应该恰好出现1次
score += 0.5 if response.count(reasoning_start) == 1 else -0.5
score += 0.5 if response.count(reasoning_end) == 1 else -0.5
score += 0.5 if response.count(solution_start) == 1 else -0.5
score += 0.5 if response.count(solution_end) == 1 else -0.5
scores.append(score)
return scores
print("奖励函数2说明:")
print("- 检查每个格式标记的出现次数")
print("- 每个标记出现1次: +0.5分")
print("- 每个标记出现0次或多次: -0.5分")
print("- 总分范围: -2.0 到 +2.0")
```
### 7.4 奖励函数3: 答案正确性检查
```Python
def check_answer(prompts, completions, answer, **kwargs):
"""
奖励函数3: 检查答案的正确性
这个函数实现了多层次的答案评估机制,从严格匹配到近似匹配。
Args:
prompts: 输入提示列表
completions: 模型生成的完成文本列表
answer: 标准答案列表
**kwargs: 其他参数
Returns:
list: 每个完成文本的奖励分数列表
"""
question = prompts[0][-1]["content"]
responses = [completion[0]["content"] for completion in completions]
# 从模型输出中提取答案
extracted_responses = [
guess.group(1)
if (guess := match_format.search(r)) is not None else None \
for r in responses
]
scores = []
for guess, true_answer in zip(extracted_responses, answer):
score = 0
# 如果无法提取答案,得分为0
if guess is None:
scores.append(0)
continue
# 完全匹配:最高奖励
if guess == true_answer:
score += 3.0
# 去除空格后匹配:高奖励
elif guess.strip() == true_answer.strip():
score += 1.5
else:
# 数值接近性检查:对于数值答案,允许一定误差
try:
ratio = float(guess) / float(true_answer)
if ratio >= 0.9 and ratio <= 1.1: # 10%误差内
score += 0.5
elif ratio >= 0.8 and ratio <= 1.2: # 20%误差内
score += 0.25
else:
score -= 1.0 # 错误答案惩罚
except:
score -= 0.5 # 无法转换为数值的惩罚
scores.append(score)
return scores
print("奖励函数3说明:")
print("- 完全匹配: +3.0分")
print("- 去空格匹配: +1.5分")
print("- 10%误差内: +0.5分")
print("- 20%误差内: +0.25分")
print("- 错误答案: -1.0分")
print("- 无法解析: -0.5分")
```
### 7.5 奖励函数4: 数值提取检查
```Python
# 定义用于提取数字的正则表达式
# 这个正则表达式专门用于从SOLUTION标记中提取数值
match_numbers = re.compile(
rf"{solution_start}.*?([\d\.]{{1,}})", # 匹配SOLUTION标记内的数字(包括小数)
flags = re.MULTILINE | re.DOTALL # 多行模式
)
# 测试数字提取功能
test_solution = "<SOLUTION> 0.34 </SOLUTION>"
extracted_numbers = match_numbers.findall(test_solution)
print(f"测试文本: {test_solution}")
print(f"提取的数字: {extracted_numbers}")
print("✓ 数字提取正则表达式工作正常" if extracted_numbers else "✗ 数字提取失败")
extracted_numbers
```
```Python
def check_numbers(prompts, completions, answer, **kwargs):
"""
奖励函数4: 检查数值提取能力
这个函数专门检查模型是否能在SOLUTION标记内输出有效的数值,
并与标准答案进行精确数值比较。
Args:
prompts: 输入提示列表
completions: 模型生成的完成文本列表
answer: 标准答案列表
**kwargs: 其他参数
Returns:
list: 每个完成文本的奖励分数列表
"""
question = prompts[0][-1]["content"]
responses = [completion[0]["content"] for completion in completions]
# 使用数字提取正则表达式从响应中提取数值
extracted_responses = [
guess.group(1)
if (guess := match_numbers.search(r)) is not None else None \
for r in responses
]
scores = []
# 打印调试信息(训练时会显示)
print('*'*20, f"Question:\n{question}",
f"\nAnswer:\n{answer[0]}",
f"\nResponse:\n{responses[0]}",
f"\nExtracted:\n{extracted_responses[0]}")
for guess, true_answer in zip(extracted_responses, answer):
# 如果无法提取数字,得分为0
if guess is None:
scores.append(0)
continue
# 尝试将提取的答案和标准答案转换为数值进行比较
try:
true_answer_num = float(true_answer.strip())
guess_num = float(guess.strip())
# 数值完全匹配时给予奖励,否则为0
scores.append(1.5 if guess_num == true_answer_num else 0.0)
except:
# 转换失败时得分为0
scores.append(0)
continue
return scores
print("奖励函数4说明:")
print("- 专门检查SOLUTION标记内的数值提取")
print("- 数值完全匹配: +1.5分")
print("- 无法提取数值或不匹配: 0分")
print("- 用于确保模型输出包含有效数字")
```
## 步骤8: 配置GRPO训练参数
**GRPO部分**
```Python
# 设置提示词的最大长度
max_prompt_length = 256
# 导入GRPO相关的配置和训练器
from trl import GRPOConfig, GRPOTrainer
# 创建GRPO训练配置
training_args = GRPOConfig(
# 优化器参数
learning_rate = 5e-6, # 学习率:GRPO通常使用较小的学习率
adam_beta1 = 0.9, # Adam优化器的beta1参数
adam_beta2 = 0.99, # Adam优化器的beta2参数
weight_decay = 0.1, # 权重衰减,防止过拟合
optim = "adamw_torch_fused", # 使用融合的AdamW优化器,更高效
# 学习率调度
warmup_ratio = 0.1, # 学习率预热比例
lr_scheduler_type = "cosine", # 余弦学习率调度
# 训练批次设置
per_device_train_batch_size = 1, # 每个设备的批次大小
gradient_accumulation_steps = 1, # 梯度累积步数(可以增加到4获得更平滑的训练)
num_generations = 4, # 每个提示生成的候选数量(显存不足时可减少)
# 序列长度控制
max_prompt_length = max_prompt_length, # 提示的最大长度
max_completion_length = max_seq_length - max_prompt_length, # 完成文本的最大长度
# 训练控制
max_steps = 50, # 最大训练步数(演示用,实际训练建议更多)
save_steps = 50, # 保存模型的步数间隔
max_grad_norm = 0.1, # 梯度裁剪阈值
# 日志和监控
logging_steps = 1, # 日志记录间隔
report_to = "swanlab", # 这里改成swanlab
output_dir = "outputs", # 输出目录
)
print("GRPO训练配置已设置完成!")
print(f"- 最大训练步数: {training_args.max_steps}")
print(f"- 每步生成候选数: {training_args.num_generations}")
print(f"- 学习率: {training_args.learning_rate}")
print(f"- 使用SwanLab进行可视化监控")
```
## 步骤9: 执行GRPO训练
```Python
# 创建GRPO训练器
# 训练器整合了模型、奖励函数、训练参数和数据集
trainer = GRPOTrainer(
model = model, # 要训练的模型
processing_class = tokenizer, # 分词器(用于文本处理)
# 奖励函数列表:这些函数将评估模型输出质量
reward_funcs = [
match_format_exactly, # 奖励函数1:严格格式匹配
match_format_approximately, # 奖励函数2:近似格式匹配
check_answer, # 奖励函数3:答案正确性
check_numbers, # 奖励函数4:数值提取
],
args = training_args, # 训练配置参数
train_dataset = dataset, # 训练数据集
)
print("GRPO训练器创建完成!")
print("包含的奖励函数:")
print("1. match_format_exactly - 检查完整格式")
print("2. match_format_approximately - 检查标记使用")
print("3. check_answer - 检查答案正确性")
print("4. check_numbers - 检查数值提取")
print("\n开始训练...")
# 开始GRPO训练
# 注意:训练过程中会显示大量调试信息,包括问题、答案和模型输出
trainer.train()
```
## 步骤10: 测试训练后的模型
**训练完毕后调用模型**
```Python
# 构建测试消息
# 使用训练时相同的系统提示词,但提出一个新问题
messages = [
{"role": "system", "content": system_prompt}, # 使用相同的格式指导
{"role": "user", "content": "What is the sqrt of 101?"}, # 新的数学问题
]
# 将消息转换为模型输入格式
text = tokenizer.apply_chat_template(
messages,
add_generation_prompt = True, # 添加生成提示,告诉模型开始回答
tokenize = False, # 先不分词,保持文本格式
)
print("测试问题: What is the sqrt of 101?")
print("期望输出格式:")
print("- 包含 <start_working_out> ... <end_working_out> 的推理过程")
print("- 包含 <SOLUTION> ... </SOLUTION> 的最终答案")
print("\n模型输出:")
# 导入文本流输出器,用于实时显示生成过程
from transformers import TextStreamer
# 生成回答
_ = model.generate(
**tokenizer(text, return_tensors = "pt").to("cuda"), # 将输入转换为张量并移到GPU
max_new_tokens = 64, # 限制输出长度(可以根据需要增加)
# Gemma-4推荐的生成参数
temperature = 1.0, # 控制输出的随机性
top_p = 0.95, # 核采样参数
top_k = 64, # top-k采样参数
# 实时输出流
streamer = TextStreamer(tokenizer, skip_prompt = True), # 跳过输入提示,只显示生成内容
)
```
## 步骤11: 保存训练后的模型
**保存模型**
```Python
# 保存LoRA适配器(推荐方式)
# 这种方式只保存训练过程中新增的LoRA权重,文件很小
print("正在保存LoRA适配器...")
model.save_pretrained("gemma-4") # 保存模型(包含LoRA权重)
tokenizer.save_pretrained("gemma-4") # 保存分词器
print("✓ LoRA适配器和分词器已保存到 'gemma-4' 目录")
print("保存内容:")
print("- adapter_config.json: LoRA配置文件")
print("- adapter_model.safetensors: LoRA权重文件")
print("- tokenizer相关文件")
print("\n使用方法:")
print("1. 先加载原始Gemma 4 E4B模型")
print("2. 再加载这个LoRA适配器")
print("3. 即可获得微调后的模型")
```
```Python
# 可选:保存完整的微调模型
# 将LoRA权重合并到原模型中,生成一个完整的模型文件
if False: # 设置为True以执行保存
print("正在保存完整的微调模型...")
model.save_pretrained_merged("gemma-4-finetune", tokenizer)
print("✓ 完整模型已保存到 'gemma-4-finetune' 目录")
print("注意:完整模型文件很大(几GB),但使用时不需要原始模型")
```
```Python
# 可选:保存为GGUF格式
# GGUF格式支持量化,文件更小,推理速度更快
if False: # 设置为True以执行保存
print("正在保存GGUF格式模型...")
model.save_pretrained_gguf(
"gemma-4-finetune",
quantization_type = "Q8_0", # 量化类型:目前支持Q8_0, BF16, F16
)
print("✓ GGUF格式模型已保存")
print("特点:")
print("- 文件更小(通过量化压缩)")
print("- 推理速度更快")
print("- 适合部署到边缘设备")
print("- 可以用llama.cpp等工具加载")
```
## Swanlab
![06-02](./images/06-02.png)
> ++[SwanLab](https://github.com/swanhubx/swanlab)++ 是一个开源的模型训练记录工具,面向 AI 研究者,提供了训练可视化、自动日志记录、超参数记录、实验对比、多人协同等功能。在 `SwanLab` 上,研究者能基于直观的可视化图表发现训练问题,对比多个实验找到研究灵感,并通过在线链接的分享与基于组织的多人协同训练,打破团队沟通的壁垒。
### 为什么要记录训练?
相较于软件开发,模型训练更像一个实验科学。一个品质优秀的模型背后,往往是成千上万次实验。研究者需要不断尝试、记录、对比,积累经验,才能找到最佳的模型结构、超参数与数据配比。在这之中,如何高效进行记录与对比,对于研究效率的提升至关重要。
### 在哪里用?
建议先在 ++[SwanLab 官网](https://swanlab.cn/)++ 注册账号,然后在GRPO训练初始化阶段选择
```Python
from trl import GRPOConfig, GRPOTrainer
training_args = GRPOConfig(
# 优化器参数
learning_rate = 5e-6, # 学习率:GRPO通常使用较小的学习率
adam_beta1 = 0.9, # Adam优化器的beta1参数
adam_beta2 = 0.99, # Adam优化器的beta2参数
weight_decay = 0.1, # 权重衰减,防止过拟合
optim = "adamw_torch_fused", # 使用融合的AdamW优化器,更高效
# 学习率调度
warmup_ratio = 0.1, # 学习率预热比例
lr_scheduler_type = "cosine", # 余弦学习率调度
# 训练批次设置
per_device_train_batch_size = 1, # 每个设备的批次大小
gradient_accumulation_steps = 1, # 梯度累积步数(可以增加到4获得更平滑的训练)
num_generations = 4, # 每个提示生成的候选数量(显存不足时可减少)
# 序列长度控制
max_prompt_length = max_prompt_length, # 提示的最大长度
max_completion_length = max_seq_length - max_prompt_length, # 完成文本的最大长度
# 训练控制
max_steps = 50, # 最大训练步数(演示用,实际训练建议更多)
save_steps = 50, # 保存模型的步数间隔
max_grad_norm = 0.1, # 梯度裁剪阈值
# 日志和监控
logging_steps = 1, # 日志记录间隔
report_to = "swanlab", # 这里改成swanlab
output_dir = "outputs", # 输出目录
)
```
### 本试验的试验记录
#### GRPO阶段
![06-03](./images/06-03.png)
400个step之后loss会有明显变化
## 教程总结
🎉 恭喜!你已经成功完成了Gemma 4 E4B的GRPO微调教程。
### 本教程涵盖的核心概念:
1. **GRPO微调**: 使用奖励函数指导模型学习特定输出格式
2. **LoRA技术**: 高效的参数微调方法,节省显存和时间
3. **奖励函数设计**: 多层次评估体系,从格式到内容的全面评价
4. **结构化输出**: 训练模型按照特定格式输出推理过程和答案
5. **SwanLab监控**: 实时跟踪训练进度和指标变化
### 学到的技能:
- ✅ 设置GRPO训练环境
- ✅ 设计多维度奖励函数
- ✅ 配置LoRA参数进行高效微调
- ✅ 处理数学推理数据集
- ✅ 监控和分析训练过程
- ✅ 保存和部署微调模型
### 进一步探索:
1. **调整奖励函数**: 设计更复杂的评估机制
2. **扩展数据集**: 使用更大或不同类型的数据集
3. **优化参数**: 尝试不同的LoRA配置和训练参数
4. **模型评估**: 在测试集上系统评估模型性能
5. **应用部署**: 将模型集成到实际应用中
### 注意事项:
- 本教程使用了较少的训练步数作为演示,实际应用中建议使用更多步数
- 可以根据显存情况调整批次大小和生成数量
- SwanLab提供了丰富的可视化功能,建议深入探索
感谢你的学习!如果有任何问题,欢迎查看SwanLab的实验记录或重新运行代码。
# 总结
Congratulations!看到了这,你已经初步实现了一个简单的RL实战,掌握了使用 Unsloth 对 Gemma4 这类大模型进行 GRPO 微调的具体操作步骤,更能体会到 Unsloth 在大幅提升训练速度、显著降低显存占用方面的强大优势,从而使在有限资源下进行复杂强化学习实验成为可能!如果支持我们的工作希望得到你的star!!这是我们持续更新的最大动力!!!
# 相关链接
- 完整可运行的代码:[GitHub](https://github.com/datawhalechina/self-llm/blob/master/models/Gemma4/6-gemma4-E4B-itGRPO微调及通过swanlab可视化.ipynb)
- 综述:https://arxiv.org/abs/2001.06921
- deepseek-r1https://arxiv.org/abs/2501.12948
- 数学原理:https://blog.csdn.net/weixin\_38991876/article/details/146474767
- Unslothhttps://docs.unsloth.ai/
@@ -0,0 +1,113 @@
# gemma-4-E4B-it AMD环境准备
> 与 **AMD Ryzen AI / Lemonade** 搭配部署时,多模态权重说明见 [google/gemma-4-E4B-it](https://huggingface.co/google/gemma-4-E4B-it)GGUF 见 [ggml-org/gemma-4-E4B-it-GGUF](https://huggingface.co/ggml-org/gemma-4-E4B-it-GGUF)。概述:[Gemma 4 博文](https://huggingface.co/blog/gemma4)。
## 环境准备
本文基础环境如下:
```
----------------
Windows11
CPU AI 395
内存 128G
----------------
```
> 非常感谢 AMD University Program 对本开源项目的大力支持,本项目的环境都在此主机下完成
## 芯片介绍
AMD Strix Halo处理器可以说是一款划时代的产品,尤其是旗舰型号锐龙 AI Max+ 395,拥有史上最强集显,可以轻松媲美桌面级RTX 4060独立显卡。全新"Zen5"架构 CPU、RDNA3.5 架构 GPU、XDNA2架构 NPU,其中NPU AI引擎拥有高达50 TOPS的算力。锐龙 AI Max系列可以搭配最多128GB LPDDR5X-8000统一内存,带宽高达256GB/s,分配最多96GB作为专属显存,从而能在本地直接运行例如GPT-OSS-120B这种千亿参数的完整大模型,尤其是对于MoE专家模型可谓得天独厚。
## NPU 安装与配置
### 先决条件
Ryzen AI 软件支持带有神经网络处理单元(NPU)的 AMD 处理器。请参考发布说明以获取完整的支持配置列表。
在安装 Ryzen AI 软件之前,系统必须安装以下依赖项:
| 依赖项 | 版本要求 |
|--------|---------|
| Windows 11 | build >= 22621.3527 |
| Visual Studio | 2022 |
| cmake | version >= 3.26 |
| Python 发行版(推荐 Miniforge | 最新版本 |
**重要提示**
* Visual Studio 2022 Community:确保安装了"使用 C++ 的桌面开发"工作负载
* Miniforge:确保在系统 PATH 环境变量中设置以下路径之一:
- `path\to\miniforge3\condabin`
- `path\to\miniforge3\Scripts\`
- `path\to\miniforge3\`
(系统 PATH 变量应在"环境变量"窗口的"系统变量"部分设置),安装程序请都是用管理员权限打开使用!!!!
### 安装 NPU 驱动程序
1. **下载 NPU 驱动程序**
- 下载并安装 NPU 驱动程序版本:32.0.203.280 或更新版本
- 下载链接:
* [NPU Driver (Version 32.0.203.280)](https://ryzenai.docs.amd.com/en/latest/inst.html#install-npu-drivers)
* [NPU Driver (Version 32.0.203.304)](https://ryzenai.docs.amd.com/en/latest/inst.html#install-npu-drivers)
2. **安装步骤**
- 解压下载的 ZIP 文件
- 以管理员模式打开终端
- 执行 `.\npu_sw_installer.exe` 文件
3. **验证安装**
- 打开任务管理器 -> 性能 -> NPU0
- 确保 NPU MCDM 驱动程序已正确安装:
* 版本:32.0.203.280,日期:5/16/2025
* 或版本:32.0.203.304,日期:10/07/2025
### 安装 Ryzen AI 软件
1. **下载安装程序**
- 下载 Ryzen AI 软件安装程序:`ryzenai-lt-1.6.1.exe`
2. **运行安装向导**
- 启动 EXE 安装程序并按照安装向导的说明操作:
* 接受许可协议条款
* 提供 Ryzen AI 安装的目标文件夹(默认:`C:\Program Files\RyzenAI\1.6.1`
* 指定 conda 环境的名称(默认:`ryzen-ai-1.6.1`
3. **完成安装**
- Ryzen AI 软件包现在已安装在安装程序创建的 conda 环境中
> **注意**NuGet 包可在 [ryzen-ai-1.6.1-nuget.zip](https://ryzenai.docs.amd.com/en/latest/inst.html#install-npu-drivers) 下载
### 测试安装
Ryzen AI 软件安装文件夹包含用于验证软件是否正确安装的测试。此安装测试位于 `quicktest` 子文件夹中。
1. **打开 Conda 命令提示符**
- 在 Windows 开始菜单中搜索"Miniforge Prompt"
2. **激活 Conda 环境**
```bash
conda activate <env_name>
```
其中 `<env_name>` 是安装程序创建的 conda 环境名称(默认为 `ryzen-ai-1.6.1`
3. **运行测试**
```bash
cd %RYZEN_AI_INSTALLATION_PATH%/quicktest
python quicktest.py
```
4. **验证结果**
- `quicktest.py` 脚本会设置环境并运行一个简单的 CNN 模型
- 成功运行时,您将看到类似以下的输出,这表明模型正在 NPU 上运行,并且 Ryzen AI 软件的安装成功:
```
[Vitis AI EP] No. of Operators : NPU 398 VITIS_EP_CPU 2
[Vitis AI EP] No. of Subgraphs : NPU 1 Actually running on NPU 1
Test Passed
```
> **注意**Ryzen AI 软件安装文件夹的完整路径存储在 `RYZEN_AI_INSTALLATION_PATH` 环境变量中。
---
@@ -0,0 +1,171 @@
# 8-gemma4-E4B-it 模型服务部署
> **GGUF 仓库**llama.cpp / Lemonade 等):[ggml-org/gemma-4-E4B-it-GGUF](https://huggingface.co/ggml-org/gemma-4-E4B-it-GGUF)。简介见 [Gemma 4 博文](https://huggingface.co/blog/gemma4)。
## 基础环境准备
本文基础环境如下:
```
----------------
Windows11
CPU AI 395
内存 128G
----------------
```
> 请确定AMD芯片的版本,目前支持AI 395和 AI 370
下载 lemonade-server 进行安装
![](./images/11-01.png)
> NPU 需要配备 AMD Ryzen AI 300 系列的 Windows 11 电脑及驱动安装。请先下载并安装 NPU 驱动程序,再继续操作,请参考 7-gemma4-E4B-it AMD环境准备.md
首先 `pip` 换源加速下载并安装依赖包
```shell
# 升级pip
python -m pip install --upgrade pip
# 更换 pypi 源加速库的安装
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
pip install -U huggingface_hub
pip install lemonade-sdk[dev]
```
## 模型下载
在 Windows Powershell 下输入以下配置镜像站
> $env:HF_ENDPOINT = "https://hf-mirror.com"
使用 `huggingface_hub` 中的 `snapshot_download` 函数下载模型,第一个参数为模型名称,参数 `cache_dir` 为模型的下载路径。
新建 `model_download.py` 文件并在其中输入以下内容,粘贴代码后请及时保存文件,如下图所示。并运行 `python model_download.py ggml-org/gemma-4-E4B-it-GGUF "C:\Users\aup\.cache\huggingface\hub\gemma-4-E4B-it-GGUF"` 执行下载。
```python
#!/usr/bin/env python
"""
使用 Python API 下载 Hugging Face 模型
"""
import os
from huggingface_hub import snapshot_download
def download_model(repo_id: str, local_dir: str = None, resume_download: bool = True):
"""
下载 Hugging Face 模型
Args:
repo_id: 模型仓库ID,例如 "ggml-org/gemma-4-E4B-it-GGUF"
local_dir: 本地保存目录,如果为None则使用默认缓存目录
resume_download: 是否支持断点续传
"""
print(f"开始下载模型: {repo_id}")
print(f"保存位置: {local_dir if local_dir else '默认缓存目录'}")
try:
# 如果指定了本地目录,使用它;否则使用默认缓存
if local_dir:
cache_dir = os.path.dirname(local_dir) if os.path.dirname(local_dir) else None
local_dir_use = local_dir
else:
cache_dir = None
local_dir_use = None
# 下载模型
downloaded_path = snapshot_download(
repo_id=repo_id,
local_dir=local_dir_use,
cache_dir=cache_dir,
resume_download=resume_download,
local_files_only=False
)
print(f"\n✓ 模型下载完成!")
print(f"保存路径: {downloaded_path}")
return downloaded_path
except Exception as e:
print(f"\n✗ 下载失败: {str(e)}")
raise
if __name__ == "__main__":
import sys
# 默认下载的模型
repo_id = "ggml-org/gemma-4-E4B-it-GGUF"
# 如果提供了命令行参数,使用它作为模型ID
if len(sys.argv) > 1:
repo_id = sys.argv[1]
# 可选:指定本地保存目录
local_dir = None
if len(sys.argv) > 2:
local_dir = sys.argv[2]
download_model(repo_id, local_dir)
```
## 支持的硬件加速器
![](./images/11-05.png)
## 支持的推理引擎
![](./images/11-06.png)
## 启动服务
> 在菜单栏里输入 Lemonade Server 启动,点击菜单里中对应的图标
![](./images/11-02.png)
然后选中对应的模型即可启动,可以对外输出兼容OpenAI的服务接口
![](./images/11-03.png)
## 服务测试
新建 `test.py` 文件并在其中输入以下内容,粘贴代码后请及时保存文件。以下代码有很详细的注释,大家如有不理解的地方,欢迎提出 issue 。
```python
# Client library provided by OpenAI to automate request
# and response processing with the server
from openai import OpenAI
# The base_url points to an LLM server, which can either be
# local (localhost address) or cloud-based (web address)
base_url = f"http://localhost:8000/api/v1"
# The `client` instance here provides APIs to request
# LLM invocations from the server
client = OpenAI(
base_url=base_url,
api_key="lemonade", # required, but unused in Lemonade
)
# The `messages` list provides the history of messages from
# the system, assistant, and user roles
messages = [
{"role":"system", "content":"You are a helpful assistant."},
{"role":"user", "content":"Hi, how are you?"},
]
# This is the API call that sends the `messages` history to
# the server's specific LLM `model`
# It returns a `completion`, which is OpenAI's way of referring
# to the LLM's reponse to the messages
completion = client.chat.completions.create(
model="gemma-4-E4B-it-GGUF",
messages=messages,
)
# This code gets the LLM's response from the `completion`
# and prints it to the screen
response = completion.choices[0].message.content
print(response)
```
返回结果如下
![](./images/11-04.png)
+185
View File
@@ -0,0 +1,185 @@
# api.py — 与教程 01 一致,已适配 Pydantic v2 / FastAPI,并支持本地模型路径
from __future__ import annotations
import logging
import os
import time
from contextlib import asynccontextmanager
from typing import List, Literal, Optional
import torch
import uvicorn
from fastapi import Body, FastAPI, HTTPException
from pydantic import BaseModel, Field, model_validator
from transformers import AutoModelForMultimodalLM, AutoProcessor
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
DEVICE = "cuda"
DEVICE_ID = os.environ.get("CUDA_DEVICE_ID", "0")
CUDA_DEVICE = f"{DEVICE}:{DEVICE_ID}" if DEVICE_ID else DEVICE
MODEL_PATH = os.environ.get("GEMMA_MODEL_PATH", "/dataset/gemma-4-E4B-it")
model = None
processor = None
DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant."
def torch_gc():
if torch.cuda.is_available():
with torch.cuda.device(CUDA_DEVICE):
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
class ContentItem(BaseModel):
type: Literal["text", "image"]
text: Optional[str] = Field(None, description="文本内容(当 type 为 text 时必填)")
image: Optional[str] = Field(None, description="图片 URL 或 base64(当 type 为 image 时必填)")
@model_validator(mode="after")
def validate_content(self):
if self.type == "text":
if not self.text or not str(self.text).strip():
raise ValueError("文本类型必须提供 text 字段")
elif self.type == "image":
img = self.image or ""
if not str(img).startswith(("http://", "https://", "data:image")):
raise ValueError("图片必须是有效的 URL 或 base64 编码字符串")
return self
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: List[ContentItem]
class ProcessRequest(BaseModel):
messages: List[Message] = Field(..., min_length=1, description="对话历史记录")
max_new_tokens: int = Field(1000, ge=10, le=4096, description="生成的最大 token 数")
class ProcessResponse(BaseModel):
response: str
status: int
time: int
processing_time: float
tokens_generated: int
def load_models():
global model, processor
if not os.path.isdir(MODEL_PATH):
raise FileNotFoundError(f"模型目录不存在: {MODEL_PATH}")
try:
logger.info("正在加载模型: %s", MODEL_PATH)
model = AutoModelForMultimodalLM.from_pretrained(
MODEL_PATH,
dtype="auto",
device_map="auto",
trust_remote_code=True,
).eval()
logger.info("正在加载处理器...")
processor = AutoProcessor.from_pretrained(MODEL_PATH, trust_remote_code=True)
logger.info("模型加载完成 device=%s", getattr(model, "device", "?"))
except Exception as e:
logger.error("模型加载失败: %s", e)
raise
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
load_models()
yield
except Exception as e:
logger.error("服务初始化失败: %s", e)
raise
finally:
torch_gc()
app = FastAPI(lifespan=lifespan)
def _normalize_content_items(items):
out = []
for it in items:
if it.get("type") == "text":
out.append({"type": "text", "text": it.get("text") or ""})
elif it.get("type") == "image":
img = it.get("image") or it.get("url")
if not img:
continue
if str(img).startswith(("http://", "https://")):
out.append({"type": "image", "url": img})
else:
out.append({"type": "image", "image": img})
return out
@app.post("/chat/completions", response_model=ProcessResponse)
async def generate_response(payload: ProcessRequest = Body(...)):
start_time = time.time()
try:
processed_messages = []
system_prompt = DEFAULT_SYSTEM_PROMPT
for msg in payload.messages:
if msg.role == "system":
system_prompt = " ".join(
[item.text or "" for item in msg.content if item.type == "text"]
)
else:
d = msg.model_dump()
d["content"] = _normalize_content_items(d["content"])
processed_messages.append(d)
messages = [
{"role": "system", "content": [{"type": "text", "text": system_prompt}]},
*processed_messages,
]
inputs = processor.apply_chat_template(
messages,
add_generation_prompt=True,
tokenize=True,
return_tensors="pt",
return_dict=True,
).to(model.device)
input_len = inputs["input_ids"].shape[-1]
max_token_num = min(4096, int(payload.max_new_tokens))
with torch.inference_mode():
generation = model.generate(
**inputs,
max_new_tokens=max_token_num,
do_sample=False,
)
response_ids = generation[0][input_len:]
raw = processor.decode(response_ids, skip_special_tokens=False)
try:
parsed = processor.parse_response(raw)
decoded = parsed.get("content", raw) if isinstance(parsed, dict) else raw
except Exception:
decoded = processor.decode(response_ids, skip_special_tokens=True)
ntok = int(response_ids.numel()) if hasattr(response_ids, "numel") else len(response_ids)
return ProcessResponse(
response=str(decoded),
status=200,
time=int(time.time()),
processing_time=time.time() - start_time,
tokens_generated=ntok,
)
except HTTPException:
raise
except Exception as e:
logger.error("处理请求时出错: %s", e)
raise HTTPException(status_code=500, detail=str(e)) from e
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=6006)
View File
Binary file not shown.

After

Width:  |  Height:  |  Size: 96 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 131 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 47 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 91 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 122 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 200 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 197 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 152 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 108 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 550 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 209 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 657 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 193 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.0 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 220 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 729 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 83 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 518 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 553 KiB

+261
View File
@@ -0,0 +1,261 @@
#!/usr/bin/env python3
"""
Gemma4 教程冒烟测试(不默认加载整模权重,避免 OOM;可按需开启网络拉取 Processor)。
用法:
/path/to/self-llm/bin/python verify_gemma4_tutorials.py
GEMMA_PULL_PROCESSOR=1 ... # 从 Hub 拉取 Gemma4Processor(需已安装 torchvision,与 torch 同 CUDA 版本)
说明:本脚本不做整模加载/推理;真机部署请参考各 .md 并自行下载权重。
"""
import json
import os
import sys
import time
import traceback
from pathlib import Path
from collections import UserDict
from typing import Any
from unittest.mock import MagicMock
REPO_ROOT = Path(__file__).resolve().parents[2]
DATASET = REPO_ROOT / "dataset" / "huanhuan.json"
MODEL_ID = "google/gemma-4-E4B-it"
def _ok(name: str, detail: str = "") -> None:
print(f"[PASS] {name}" + (f"{detail}" if detail else ""))
def _fail(name: str, err: BaseException) -> None:
print(f"[FAIL] {name}: {err}")
traceback.print_exc()
def test_imports() -> bool:
import numpy
import torch
import transformers
from transformers import AutoModelForMultimodalLM, AutoProcessor
assert hasattr(torch, "cuda")
_ok(
"import 栈",
f"numpy={numpy.__version__}, torch={torch.__version__}, "
f"cuda={torch.cuda.is_available()}, transformers={transformers.__version__}",
)
_ = AutoModelForMultimodalLM
_ = AutoProcessor
return True
def test_pull_processor() -> bool:
if os.environ.get("GEMMA_PULL_PROCESSOR", "").strip() not in ("1", "true", "yes"):
print("[SKIP] Processor Hub 拉取(设置 GEMMA_PULL_PROCESSOR=1 可开启)")
return True
from transformers import AutoProcessor
t0 = time.time()
proc = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True)
dt = time.time() - t0
_ok("AutoProcessor.from_pretrained", f"{type(proc).__name__}, {dt:.1f}s")
return True
def test_fastapi_smoke_with_mock() -> bool:
"""教程 01 路由逻辑:用 Mock 模型验证请求/响应(避免真推理)。"""
from contextlib import asynccontextmanager
from fastapi import Body, FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field, model_validator
from typing import List, Literal, Optional
class ContentItem(BaseModel):
type: Literal["text", "image"]
text: Optional[str] = Field(None, description="文本")
image: Optional[str] = Field(None, description="图片 URL 或 base64")
@model_validator(mode="after")
def _v(self):
if self.type == "text" and not (self.text and str(self.text).strip()):
raise ValueError("文本类型必须提供 text 字段")
if self.type == "image":
img = self.image or ""
if not img.startswith(("http://", "https://", "data:image")):
raise ValueError("图片必须是有效的 URL 或 base64(data:image)")
return self
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: List[ContentItem]
class ProcessRequest(BaseModel):
messages: List[Message] = Field(..., min_length=1)
max_new_tokens: int = Field(1000, ge=10, le=4096)
class ProcessResponse(BaseModel):
response: str
status: int
time: int
processing_time: float
tokens_generated: int
import torch
class _Batch(UserDict):
def to(self, _device):
return self
mock_model = MagicMock()
mock_model.device = "cpu"
mock_model.generate = MagicMock(return_value=torch.tensor([[1, 2, 99, 100]]))
mock_processor = MagicMock()
mock_processor.apply_chat_template = MagicMock(
return_value=_Batch(
input_ids=torch.tensor([[1, 2]]),
attention_mask=torch.tensor([[1, 1]]),
)
)
mock_processor.decode = MagicMock(return_value="<mock>")
mock_processor.parse_response = MagicMock(side_effect=Exception("no parse"))
model_ref = {"m": mock_model, "p": mock_processor}
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
app = FastAPI(lifespan=lifespan)
DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant."
def _normalize_content_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
out = []
for it in items:
if it.get("type") == "text":
out.append({"type": "text", "text": it.get("text") or ""})
elif it.get("type") == "image":
img = it.get("image") or it.get("url")
if not img:
continue
if str(img).startswith(("http://", "https://")):
out.append({"type": "image", "url": img})
else:
out.append({"type": "image", "image": img})
return out
@app.post("/chat/completions", response_model=ProcessResponse)
async def generate_response(chat: ProcessRequest = Body(...)):
start_time = time.time()
try:
model, processor = model_ref["m"], model_ref["p"]
processed_messages = []
system_prompt = DEFAULT_SYSTEM_PROMPT
for msg in chat.messages:
if msg.role == "system":
system_prompt = " ".join([item.text or "" for item in msg.content if item.type == "text"])
else:
d = msg.model_dump()
d["content"] = _normalize_content_items(d["content"])
processed_messages.append(d)
messages = [
{"role": "system", "content": [{"type": "text", "text": system_prompt}]},
*processed_messages,
]
inputs = processor.apply_chat_template(
messages,
add_generation_prompt=True,
tokenize=True,
return_tensors="pt",
return_dict=True,
).to(model.device)
input_len = inputs["input_ids"].shape[-1]
max_token_num = min(4096, int(chat.max_new_tokens))
with torch.inference_mode():
generation = model.generate(**inputs, max_new_tokens=max_token_num, do_sample=False)
response_ids = generation[0][input_len:]
raw = processor.decode(response_ids, skip_special_tokens=False)
try:
parsed = processor.parse_response(raw)
decoded = parsed.get("content", raw) if isinstance(parsed, dict) else raw
except Exception:
decoded = processor.decode(response_ids, skip_special_tokens=True)
return ProcessResponse(
response=str(decoded),
status=200,
time=int(time.time()),
processing_time=time.time() - start_time,
tokens_generated=int(len(response_ids)),
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
payload = {
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "你好,只做连通性测试"},
],
}
],
"max_new_tokens": 64,
}
with TestClient(app) as client:
r = client.post("/chat/completions", json=payload)
assert r.status_code == 200, r.text
body = r.json()
assert "response" in body
_ok("FastAPI /chat/completions (Mock)", f"tokens_generated={body.get('tokens_generated')}")
return True
def test_lora_dataset() -> bool:
if not DATASET.is_file():
print(f"[SKIP] 数据集不存在: {DATASET}")
return True
raw = json.loads(DATASET.read_text(encoding="utf-8"))
assert isinstance(raw, list) and len(raw) > 0
first = raw[0]
for k in ("instruction", "input", "output"):
assert k in first
_ok("05 LoRA 数据集", f"{DATASET.name} 条数={len(raw)}")
return True
def test_evalscope_import() -> bool:
from evalscope.config import TaskConfig
from evalscope.run import run_task
_ = TaskConfig
_ = run_task
_ok("evalscope 导入", "TaskConfig / run_task 可用(完整评测需 Ollama 等服务)")
return True
def main() -> int:
tests = [
("环境导入", test_imports),
("05 嬛嬛数据集", test_lora_dataset),
("01 FastAPI Mock", test_fastapi_smoke_with_mock),
("04 evalscope", test_evalscope_import),
("Hub Processor", test_pull_processor),
]
failed = 0
for name, fn in tests:
try:
fn()
except Exception as e:
failed += 1
_fail(name, e)
print("---")
print(f"完成: {len(tests) - failed}/{len(tests)} 通过")
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())