学习进度 · 登录后可记录
模型项目地址:hf_mirrors/google/gemma-3-270m-it 课程难度:入门 · 需要 Python 基础和 PyTorch 经验 | 预估学时:20~30 分钟 部署架构:基于 transformers + torch_npu + Flask API 服务部署
本章结束后,你能够:
device_map="auto" 在昇腾上的常见陷阱本节课的目标是:在昇腾 NPU 上,部署 Gemma-3-270M-IT——Google 开源的仅 270M 参数的指令微调语言模型,并将其封装为兼容 OpenAI 格式的 Flask API 服务。
Gemma-3-270M-IT 是一个"极致轻量"的模型:270M 参数意味着模型权重仅约 540MB(bfloat16),昇腾 910B 单卡的 64GB HBM 中仅占不到 1%。它非常适合作为昇腾适配的入门模型——加载快、推理快、显存无压力,让你能专注于理解昇腾适配的核心机制,而非等待漫长的模型加载。
但"小"不代表"无需适配"——恰恰相反,Gemma-3-270M-IT 的部署过程揭示了昇腾适配中最基础、最重要的模式。
很多习惯了 CUDA 的开发者会认为:270M 这么小,直接 model.to("cuda") 改成 model.to("npu:0") 不就完事了?——方向对了,但魔鬼在细节里。
| 适配环节 | GPU 上的常见做法 | 昇腾 NPU 上的正确做法 |
|---|---|---|
| 模型加载 | device_map="auto" 自动分配 | 手动加载到 CPU 再 .to("npu:0"),避免 device_map 与 NPU 冲突 |
| 数据类型 | torch.float16 或 torch.float32 | torch.bfloat16,昇腾 910B 原生支持 BF16 且性能最优 |
| 输入张量 | .to("cuda") | .to("npu:0"),设备字符串不同 |
| 推理上下文 | with torch.no_grad(): | 同 GPU,必须使用,否则显存泄漏 |
| Tokenizer 设备 | 自动跟随模型设备 | 必须手动 .to("npu:0"),apply_chat_template 返回的张量不会自动迁移 |
┌──────────────────────────────────────────────────────┐
│ Flask API 服务层 │
│ ┌────────────────────────────────────────────────┐ │
│ │ Flask (port 8100) │ │
│ │ ├─ /v1/models → 模型列表 │ │
│ │ └─ /v1/chat/completions → 对话推理接口 │ │
│ └──────────────────┬───────────────────────────────┘ │
│ │ │
│ ┌──────────────────▼───────────────────────────┐ │
│ │ Chat 函数(核心推理链路) │ │
│ │ ├─ tokenizer.apply_chat_template → 输入编码 │ │
│ │ ├─ model.generate → 自回归生成 │ │
│ │ └─ clean_response → 特殊 token 清洗 │ │
│ └──────────────────┬───────────────────────────┘ │
│ │ │
├─────────────────────┼────────────────────────────────┤
│ torch_npu + CANN 8.5 │
│ ┌──────────────┐ ┌─▼──────────────┐ │
│ │ GE 图编译 │ │ CUBE 矩阵乘 │ │
│ │ (算子融合) │ │ (Attention) │ │
│ └──────────────┘ └────────────────┘ │
├──────────────────────────────────────────────────────┤
│ 昇腾 NPU 硬件层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ Scalar │ │ Vector │ │ CUBE 矩阵乘 │ │
│ │ 控制流 │ │ 激活函数 │ │ 自回归解码 │ │
│ └──────────┘ └──────────┘ └──────────────┘ │
└──────────────────────────────────────────────────────┘设计思路:Gemma-3-270M-IT 模型权重托管在 AtomGit 平台,可以使用 atomgit_hub 下载、或者在hugging face下载。模型仅约 540MB,下载速度较快。
# [工具安装] 安装 atomgit SDK
!pip install -U atomgit atomgit_hub -i https://mirrors.huaweicloud.com/repository/pypi/simple# [模型下载] 从 AtomGit 平台下载 Gemma-3-270M-IT 模型
from atomgit_hub import snapshot_download
# 下载模型到本地
model_repo = "hf_mirrors/google/gemma-3-270m-it"
local_dir = "/opt/atomgit/gemma-3-270m-it"
snapshot_download(model_repo, local_dir=local_dir)
print(f"模型已下载到: {local_dir}")设计思路:这是昇腾适配中最关键的一步。很多习惯了 CUDA 的开发者会直接使用 device_map="auto" 加载模型,但这在昇腾上可能导致设备分配冲突。正确做法是:先加载到 CPU,再手动 .to("npu:0")。同时,Gemma 系列模型使用 Chat Template 格式化对话,需要确保 tokenizer.chat_template 正确加载。
# [模型加载] 加载 Tokenizer 和模型到 NPU
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
MODEL_PATH = "/opt/atomgit/gemma-3-270m-it"
# 加载 tokenizer
tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
print(f"Tokenizer 类: {type(tokenizer).__name__}")
# 检查 chat template
if tokenizer.chat_template:
print(f"Chat template 已加载,长度: {len(tokenizer.chat_template)}")
else:
import os
template_path = os.path.join(MODEL_PATH, "chat_template.jinja")
if os.path.exists(template_path):
with open(template_path) as f:
tokenizer.chat_template = f.read()
print(f"已从文件加载 chat template,长度: {len(tokenizer.chat_template)}")
# 加载模型(不使用 device_map,完全手动控制)
print("\n正在加载模型到 CPU,请稍候...")
model = AutoModelForCausalLM.from_pretrained(
MODEL_PATH,
trust_remote_code=True,
torch_dtype=torch.bfloat16,
)
print("正在将模型移到 NPU...")
model = model.to("npu:0")
print("✅ 模型加载完成")
print(f"模型所在设备: {next(model.parameters()).device}")device_map="auto" 在 GPU 上会自动将模型分配到可用设备。但在昇腾上,device_map="auto" 的设备发现逻辑可能走 CUDA 路径,导致以下问题:
torch_npu 未正确初始化时报 No CUDA GPUs available手动 .to("npu:0") 是昇腾适配中最稳妥的设备迁移方式,确保模型所有参数都在同一个 NPU 上。
设计思路:将推理过程封装为可复用的 chat() 函数。这里有两个关键细节:一是输入张量必须手动 .to("npu:0")(apply_chat_template 返回的是 CPU 张量);二是需要 clean_response() 函数清洗模型输出中的特殊 token 残留,这是 Gemma 系列模型的常见问题。
# [推理封装] 封装 Chat 函数和输出清洗
import re
def clean_response(text):
"""清理模型输出中的特殊 token 和乱码"""
# 1. 移除已知的特殊 token
text = re.sub(r'<(end_of_turn|start_of_turn|eos|pad|bos|\/s|s>)[^>]*>', '', text)
text = text.replace("<end_of_turn>", "").replace("<start_of_turn>", "")
text = text.replace("<eos>", "").replace("<pad>", "").replace("<bos>", "")
# 2. 移除末尾的孤立非 ASCII 字符(tokenizer 未正确解码的残留)
text = text.strip()
while text and len(text) > 1:
last_char = text[-1]
if ord(last_char) > 127:
if len(text) > 1 and ord(text[-2]) <= 127:
text = text[:-1].strip()
else:
break
else:
break
return text.strip()
def chat(messages, max_new_tokens=256, temperature=0.7, top_p=0.95):
"""核心推理函数"""
inputs = tokenizer.apply_chat_template(
messages,
return_tensors="pt",
add_generation_prompt=True
)
# 关键:apply_chat_template 返回 CPU 张量,必须手动迁移到 NPU
inputs = inputs.to("npu:0")
with torch.no_grad():
outputs = model.generate(
inputs,
max_new_tokens=max_new_tokens,
do_sample=True,
temperature=temperature,
top_p=top_p,
pad_token_id=tokenizer.pad_token_id,
eos_token_id=tokenizer.eos_token_id,
)
# 截取生成部分(去除输入 prompt)
response_tokens = outputs[0][inputs.shape[1]:]
raw_response = tokenizer.decode(response_tokens, skip_special_tokens=True)
response = clean_response(raw_response)
return responseGemma 系列模型在生成时,偶尔会在回复末尾输出 <end_of_turn>、<start_of_turn> 等控制 token 的文本形式,以及一些孤立的非 ASCII 乱码字符(这是 tokenizer 解码的边界效应)。如果不清洗,这些残留会直接暴露给用户,影响输出质量。
设计思路:NPU 上的首次推理(冷启动)会触发算子编译,耗时较长。通过一次预热推理,让 GE 图引擎完成算子编译和缓存,后续推理即可直接命中缓存。
# [预热测试] 预热推理 + 性能测量
import time
messages = [{"role": "user", "content": "Tell me a short story about a robot"}]
# 预热(触发算子编译)
_ = chat(messages, max_new_tokens=50)
# 正式测试
start = time.time()
response = chat(messages, max_new_tokens=256)
elapsed = time.time() - start
tokens_generated = len(tokenizer.encode(response))
print(f"生成 {tokens_generated} tokens,耗时 {elapsed:.2f}s")
print(f"速度: {tokens_generated/elapsed:.1f} tokens/s")
print(f"\nResponse: {response[:500]}...")设计思路:将推理能力封装为 OpenAI 兼容的 HTTP API,便于其他应用调用。使用 Flask 轻量级框架,在后台线程中启动服务,不阻塞 Notebook。
# [API 服务] 启动 Flask API 服务
from flask import Flask, request, jsonify
import threading
app = Flask(__name__)
@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
data = request.json
messages = data.get("messages", [])
max_tokens = data.get("max_tokens", 256)
temperature = data.get("temperature", 0.7)
try:
response_text = chat(messages, max_new_tokens=max_tokens, temperature=temperature)
return jsonify({
"id": "chatcmpl-local",
"object": "chat.completion",
"created": int(time.time()),
"model": "gemma-3-270m-it",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": response_text},
"finish_reason": "stop"
}]
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/v1/models", methods=["GET"])
def list_models():
return jsonify({
"object": "list",
"data": [{"id": "gemma-3-270m-it", "object": "model"}]
})
# 在后台线程启动
def run_app():
app.run(host="0.0.0.0", port=8100, debug=False)
thread = threading.Thread(target=run_app, daemon=True)
thread.start()
print("✅ API 服务已启动在 http://localhost:8100")设计思路:通过 HTTP 请求验证 API 服务是否正常工作。
# [API 测试] 验证 API 服务
import requests
url = "http://localhost:8100/v1/chat/completions"
payload = {
"model": "gemma-3-270m-it",
"messages": [{"role": "user", "content": "hello, who are you"}],
"max_tokens": 128,
"temperature": 0.7
}
response = requests.post(url, json=payload, timeout=60)
print(response.json()["choices"][0]["message"]["content"])device_map="auto",手动控制设备迁移很多习惯了 CUDA 的开发者在这里容易踩坑:直接在 from_pretrained() 中加 device_map="auto",结果在昇腾上报 No CUDA GPUs available 或模型参数分散在 CPU/NPU 上导致推理极慢。
根本原因:device_map="auto" 的设备发现逻辑依赖 torch.cuda.device_count(),在昇腾环境下 torch_npu 尚未完全接管 CUDA 设备发现接口,导致分配逻辑走 CUDA 路径并失败。
解决方案:始终使用"先 CPU 后 NPU"的两步迁移模式:
# ❌ 错误:使用 device_map="auto"
model = AutoModelForCausalLM.from_pretrained(MODEL_PATH, device_map="auto")
# ✅ 正确:手动控制设备迁移
model = AutoModelForCausalLM.from_pretrained(MODEL_PATH, torch_dtype=torch.bfloat16)
model = model.to("npu:0")另一个高频踩坑点:tokenizer.apply_chat_template() 返回的是 CPU 张量,直接传入模型会触发跨设备数据拷贝或直接报错。
根本原因:transformers 的 tokenizer 输出始终在 CPU 上,不会自动跟随模型所在设备。这在 GPU 上通常不是问题(model.generate() 内部会自动处理),但在昇腾上 torch_npu 的自动迁移逻辑不够完善。
解决方案:每次调用 apply_chat_template() 后,手动 .to("npu:0"):
inputs = tokenizer.apply_chat_template(messages, return_tensors="pt", add_generation_prompt=True)
inputs = inputs.to("npu:0") # 关键:手动迁移到 NPU| 环境变量 | 默认值 | 说明 |
|---|---|---|
ASCEND_RT_VISIBLE_DEVICES | 0 | 昇腾 NPU 可见设备 ID |
CUDA_VISIBLE_DEVICES | (空) | 必须清空,防止 CUDA 回退 |
MODEL_PATH | /opt/atomgit/gemma-3-270m-it | 模型本地目录 |
API_PORT | 8100 | Flask API 服务监听端口 |
| 问题 | 原因 | 解决方案 |
|---|---|---|
No CUDA GPUs available | 使用了 device_map="auto" 或 CUDA_VISIBLE_DEVICES 未清空 | 改用手动 .to("npu:0"),清空 CUDA_VISIBLE_DEVICES |
ModuleNotFoundError: No module named 'atomgit_hub' | 安装后未重启内核 | 重启 Jupyter 内核后重试 |
输出包含 <end_of_turn> 等残留 | Gemma 模型特殊 token 未清洗 | 确保 clean_response() 函数正确处理 |
| 推理结果末尾有乱码 | tokenizer 解码边界效应 | clean_response() 中已处理孤立非 ASCII 字符 |
| 首次推理极慢 | NPU 算子编译冷启动 | 执行预热推理后再正式使用 |
| 端口 8100 被占用 | 旧服务未正确停止 | 重启内核或 pkill -f flask 杀掉旧进程 |
模型加载到 CPU 后 .to("npu:0") 报错 | torch_npu 未安装或未初始化 | 确认 import torch_npu 成功 |
GET /v1/models响应:
{"object": "list", "data": [{"id": "gemma-3-270m-it", "object": "model"}]}POST /v1/chat/completions
Content-Type: application/json请求体:
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
model | string | ✅ | 填写 gemma-3-270m-it |
messages | array | ✅ | 对话消息数组 |
messages[].role | string | ✅ | user / assistant / system |
messages[].content | string | ✅ | 消息文本内容 |
max_tokens | int | ❌ | 最大生成 token 数,默认 256 |
temperature | float | ❌ | 采样温度,默认 0.7 |
响应:兼容 OpenAI ChatCompletion 格式,回复文本在 choices[0].message.content。
登录后即可查看完整教程内容、运行代码和参
与学习互动
还没有账号?