手把手教程：用 Transformers 部署 Gemma 小模型
版本: v0.0.2
重点: AI 编码过程 + Prompt 调试记录
难度: 小白友好
一、AI 编码工作流介绍
1.1 为什么记录 AI 编码过程？
- 学习 Prompt 技巧：如何向 AI 描述需求
- 调试能力：遇到问题怎么排查
- 迭代思维：从粗糙到完善的思考路径
1.2 我们的 AI 编码流程
需求 → Prompt → 代码 → 测试 → 报错 → 调试 → 优化 → 完成
二、第一步：创建项目骨架（AI 交互实录）
2.1 我的初始 Prompt
我需要一个 FastAPI 项目，用 Transformers 部署 unsloth/functiongemma-270m-it 模型。
要求：
1. 支持 OpenAI 兼容的 /v1/chat/completions 接口
2. 支持模型下载和初始化
3. 代码要模块化，分文件存放
4. 适合部署到 HuggingFace Space
请给出项目结构和每个文件的代码。
2.2 AI 的第一次回复（问题分析）
AI 给出了完整代码，但我发现：
- ❌ 没有考虑免费资源限制
- ❌ 没有错误处理细节
- ❌ 没有调试建议
2.3 我的优化 Prompt
很好，但需要改进：
1. 添加资源限制检测（内存/CPU）
2. 增加详细的错误处理和日志
3. 提供本地测试的 curl 命令
4. 说明如何在 HuggingFace Space 上调试
三、第二步：手写 utils/model.py（分步实现）
3.1 第一版 Prompt（简单需求）
写一个 Python 模块，检查 HuggingFace 模型是否已下载到本地缓存。
如果不存在，提示用户下载。
使用 transformers 库。
AI 生成的代码：
from pathlib import Path
from transformers import AutoTokenizer, AutoModelForCausalLM
def check_model(model_name):
cache_dir = "./my_model_cache"
model_path = Path(cache_dir) / f"models--{model_name.replace('/', '--')}"
snapshot_path = model_path / "snapshots"
if snapshot_path.exists() and any(snapshot_path.iterdir()):
return True
return False
3.2 测试与问题发现
运行测试：
python -c "from utils.model import check_model; print(check_model('unsloth/functiongemma-270m-it'))"
问题：函数只返回 True/False，没有详细信息。
3.3 第二版 Prompt（增加需求）
改进 check_model 函数：
1. 返回 (model_name, cache_dir, success) 三元组
2. 打印详细的检查过程
3. 如果模型不存在，提示用户如何下载
AI 优化后的代码：
from pathlib import Path
from transformers import AutoTokenizer, AutoModelForCausalLM
from fastapi import HTTPException
def check_model(model_name):
"""检查模型是否存在，返回详细信息"""
cache_dir = "./my_model_cache"
model_path = Path(cache_dir) / f"models--{model_name.replace('/', '--')}"
snapshot_path = model_path / "snapshots"
if snapshot_path.exists() and any(snapshot_path.iterdir()):
print(f"✅ 模型 {model_name} 已存在于 {cache_dir}")
try:
# 验证能否加载
tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
return model_name, cache_dir, True
except Exception as e:
print(f"⚠️ 模型文件损坏: {e}")
return model_name, cache_dir, False
print(f"❌ 模型 {model_name} 不存在")
print(f" 请先运行: curl -X POST /download -d '{{\"model\": \"{model_name}\"}}'")
return model_name, cache_dir, False
3.4 第三版 Prompt（处理特殊情况）
如果用户没有安装 transformers 库，或者网络不通怎么办？
添加 try-catch 和友好的错误提示。
最终版本（见完整代码）
四、第三步：调试 chat_response.py（真实踩坑记录）
4.1 初始 Prompt
写一个函数，调用 pipeline 生成响应，并返回 OpenAI 格式。
需要处理 tokenizer 和 max_new_tokens。
4.2 第一次运行报错
# 测试命令
python -c "
from utils.chat_response import create_chat_response
from utils.chat_request import ChatRequest
from transformers import pipeline
pipe = pipeline('text-generation', model='unsloth/functiongemma-270m-it')
tokenizer = pipe.tokenizer
request = ChatRequest(messages=[{'role': 'user', 'content': 'hi'}])
print(create_chat_response(request, pipe, tokenizer))
"
报错：
TypeError: 'NoneType' object is not callable
4.3 调试过程
我的思考：
- 问题在
pipe(messages, max_new_tokens=...) - 可能是 pipeline 返回格式不对
调试 Prompt：
Transformers 的 pipeline 返回什么格式？
如何正确调用 text-generation pipeline？
请给出完整示例。
AI 回答：
# 正确的调用方式
result = pipe(
messages,
max_new_tokens=100,
return_full_text=False # 关键参数
)
# 返回: [{'generated_text': '...'}]
4.4 修复后的代码
def create_chat_response(request, pipe, tokenizer):
"""创建聊天响应 - 修复版"""
if pipe is None:
return ChatResponse(...) # 降级处理
# 关键：正确调用 pipeline
max_new_tokens = request.max_tokens if request.max_tokens is not None else 500
result = pipe(request.messages, max_new_tokens=max_new_tokens)
# 解析结果
completion_text = result[0]['generated_text']
# 计算 token
prompt_tokens = sum(len(tokenizer.encode(msg["content"])) for msg in request.messages)
completion_tokens = len(tokenizer.encode(completion_text))
return ChatResponse(...)
4.5 第二次报错：格式转换问题
问题：Gemma 返回的格式是 assistant: 内容，需要提取纯内容。
调试 Prompt：
Gemma 模型返回的 generated_text 格式是：
"assistant: 你好，我是助手"
如何提取 "你好，我是助手" 这部分？
用正则表达式或字符串处理。
AI 给出的方案：
import re
def extract_assistant_content(text):
# 方法1：正则
match = re.search(r'assistant:\s*(.*)', text, re.DOTALL)
if match:
return match.group(1).strip()
# 方法2：分割
if 'assistant:' in text:
return text.split('assistant:')[-1].strip()
return text
我的选择：用正则，更精确。
五、第四步：主程序 app.py（整合调试）
5.1 全局变量设计（AI 交互）
我的 Prompt：
FastAPI 中如何在多个路由间共享全局变量？
比如我想在 startup 时加载模型，然后在所有路由中使用。
需要考虑线程安全吗？
AI 回答：
# 方式1：全局变量（适合单进程）
model_name = None
pipe = None
@app.on_event("startup")
async def init():
global pipe
pipe = load_model()
@app.get("/chat")
async def chat():
global pipe
return pipe.generate(...)
我的优化：
- 添加类型提示
- 添加注释说明
- 考虑多进程情况（虽然 Space 是单进程）
5.2 Startup 事件调试
问题：模型加载失败时，应用应该启动还是报错？
我的决策：
@app.on_event("startup")
async def startup_event():
try:
pipe, tokenizer, success = initialize_pipeline(default_model)
if success:
model_name = default_model
print("✅ 启动成功")
else:
print("⚠️ 等待模型下载")
# 不阻塞启动，允许先下载
except Exception as e:
print(f"❌ 启动失败: {e}")
# 但应用仍启动，只是模型不可用
理由：给用户容错空间，先启动服务再下载模型。
六、完整代码（手写版）
6.1 文件结构
my_gemma_service/
├── .env
├── app.py
├── utils/
│ ├── chat_request.py
│ ├── chat_response.py
│ └── model.py
├── requirements.txt
└── Dockerfile
6.2 逐文件手写（带注释）
.env
# 模型名称，可以修改为其他支持的模型
DEFAULT_MODEL_NAME="unsloth/functiongemma-270m-it"
utils/model.py
"""
模型管理模块
功能：检查、下载、初始化模型
"""
import os
from pathlib import Path
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
from huggingface_hub import login
from fastapi import HTTPException
from pydantic import BaseModel
class DownloadRequest(BaseModel):
"""下载请求模型"""
model: str
def check_model(model_name):
"""
检查模型是否已下载
返回: (model_name, cache_dir, success)
"""
cache_dir = "./my_model_cache"
model_path = Path(cache_dir) / f"models--{model_name.replace('/', '--')}"
snapshot_path = model_path / "snapshots"
if snapshot_path.exists() and any(snapshot_path.iterdir()):
print(f"✅ 模型 {model_name} 已存在于 {cache_dir}")
try:
# 验证能否加载 tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
return model_name, cache_dir, True
except Exception as e:
print(f"⚠️ 模型文件损坏: {e}")
return model_name, cache_dir, False
print(f"❌ 模型 {model_name} 不存在")
return model_name, cache_dir, False
def download_model(model_name):
"""
下载模型到本地缓存
"""
cache_dir = "./my_model_cache"
print(f"📥 开始下载: {model_name}")
print(f" 缓存目录: {cache_dir}")
# 如果需要登录（下载私有模型）
token = os.getenv("HUGGINGFACE_TOKEN")
if token:
try:
print(" 正在登录 HuggingFace...")
login(token=token)
print(" ✅ 登录成功")
except Exception as e:
print(f" ⚠️ 登录失败: {e}")
print(" 继续尝试下载公开模型...")
else:
print(" ℹ️ 未设置 HUGGINGFACE_TOKEN，仅下载公开模型")
try:
print(" 下载 tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
print(" ✅ Tokenizer 下载完成")
print(" 下载模型权重...")
model = AutoModelForCausalLM.from_pretrained(model_name, cache_dir=cache_dir)
print(" ✅ 模型下载完成")
print(f"✅ 模型 {model_name} 下载成功！")
return True, f"模型 {model_name} 下载成功"
except Exception as e:
print(f"❌ 下载失败: {e}")
print("\n可能原因：")
print("1. 网络连接问题")
print("2. 模型名称错误")
print("3. 需要 HUGGINGFACE_TOKEN")
return False, f"下载失败: {str(e)}"
def initialize_pipeline(model_name):
"""
初始化模型 pipeline
返回: (pipe, tokenizer, success)
"""
print(f"\n🔄 初始化 pipeline: {model_name}")
# 先检查模型
model_name, cache_dir, success = check_model(model_name)
if not success:
print("⚠️ 请先下载模型")
return None, None, False
try:
print(" 加载 tokenizer...")
tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
print(" 创建 pipeline...")
pipe = pipeline("text-generation", model=model_name, tokenizer=tokenizer)
print("✅ Pipeline 初始化完成！")
return pipe, tokenizer, True
except Exception as e:
print(f"❌ 初始化失败: {e}")
return None, None, False
utils/chat_request.py
"""
聊天请求验证模块
"""
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
class ChatRequest(BaseModel):
"""
OpenAI 兼容的聊天请求
所有字段都是可选的，有默认值
"""
model: Optional[str] = "unsloth/functiongemma-270m-it"
messages: List[Dict[str, Any]]
temperature: Optional[float] = 1.0
max_tokens: Optional[int] = None # None = 使用默认值 500
top_p: Optional[float] = 1.0
frequency_penalty: Optional[float] = 0.0
presence_penalty: Optional[float] = 0.0
utils/chat_response.py
"""
聊天响应生成模块
核心：调用 pipeline 并格式化输出
"""
from pydantic import BaseModel
from typing import List, Dict, Any
import time
import re
class ChatChoice(BaseModel):
index: int
message: Dict[str, str]
finish_reason: str
class ChatUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class ChatResponse(BaseModel):
id: str
object: str
created: int
model: str
choices: List[ChatChoice]
usage: ChatUsage
def convert_json_format(input_data):
"""
转换 pipeline 输出为统一格式
处理 Gemma 的特殊返回格式
"""
output_generations = []
for item in input_data:
generated_text_list = item.get('generated_text', [])
assistant_content = ""
for message in generated_text_list:
if message.get('role') == 'assistant':
assistant_content = message.get('content', '')
break
# 清理 Gemma 的特殊标记
clean_content = re.sub(r'</think>.*?yện\s*', '', assistant_content, flags=re.DOTALL).strip()
output_generations.append([
{
"text": clean_content,
"generationInfo": {"finish_reason": "stop"}
}
])
return {"generations": output_generations}
def create_chat_response(request, pipe, tokenizer):
"""
创建聊天响应 - 核心函数
"""
# 降级处理：模型未加载
if pipe is None:
return ChatResponse(
id=f"chatcmpl-{int(time.time())}",
object="chat.completion",
created=int(time.time()),
model=request.model,
choices=[ChatChoice(
index=0,
message={"role": "assistant", "content": "模型正在初始化中，请稍后..."},
finish_reason="stop"
)],
usage=ChatUsage(prompt_tokens=0, completion_tokens=0, total_tokens=0)
)
# 调用模型
max_new_tokens = request.max_tokens if request.max_tokens is not None else 500
result = pipe(request.messages, max_new_tokens=max_new_tokens)
# 格式转换
converted_result = convert_json_format(result)
completion_text = converted_result["generations"][0][0]["text"]
# Token 计算
prompt_tokens = sum(len(tokenizer.encode(msg.get("content", ""))) for msg in request.messages)
completion_tokens = len(tokenizer.encode(completion_text))
return ChatResponse(
id=f"chatcmpl-{int(time.time())}",
object="chat.completion",
created=int(time.time()),
model=request.model,
choices=[ChatChoice(
index=0,
message={"role": "assistant", "content": completion_text},
finish_reason="stop"
)],
usage=ChatUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens
)
)
app.py
"""
主程序：FastAPI 应用
"""
from fastapi import FastAPI, HTTPException
import os
from dotenv import load_dotenv
# 导入自定义模块
from utils.chat_request import ChatRequest
from utils.chat_response import create_chat_response, ChatResponse
from utils.model import check_model, initialize_pipeline, download_model, DownloadRequest
# 全局状态（单进程安全）
model_name = None
pipe = None
tokenizer = None
# 创建应用
app = FastAPI(
title="Gemma 函数调用服务",
description="基于 Transformers 的轻量级模型服务",
version="1.0.0"
)
@app.on_event("startup")
async def startup_event():
"""
应用启动时自动加载模型
失败时不阻塞启动，允许先下载
"""
global pipe, tokenizer, model_name
# 加载环境变量
load_dotenv()
# 获取默认模型
default_model = os.getenv("DEFAULT_MODEL_NAME", "unsloth/functiongemma-270m-it")
print(f"\n🚀 应用启动，正在加载模型: {default_model}")
try:
pipe, tokenizer, success = initialize_pipeline(default_model)
if success:
model_name = default_model
print(f"✅ 模型 {model_name} 加载成功！")
else:
print(f"⚠️ 模型未就绪，请先下载")
except Exception as e:
print(f"❌ 启动异常: {e}")
print(" 应用将继续启动，但模型功能不可用")
@app.get("/")
async def read_root():
"""
服务状态检查
"""
return {
"message": "Gemma 函数调用服务已启动！",
"current_model": model_name,
"status": "ready" if pipe else "waiting_for_model",
"docs": "http://localhost:7860/docs"
}
@app.post("/download")
async def download_model_endpoint(request: DownloadRequest):
"""
下载模型接口
下载后自动初始化
"""
global pipe, tokenizer, model_name
success, message = download_model(request.model)
if success:
# 自动初始化
pipe, tokenizer, init_success = initialize_pipeline(request.model)
if init_success:
model_name = request.model
return {
"status": "success",
"message": message,
"loaded": True,
"current_model": model_name
}
else:
return {
"status": "success",
"message": message,
"loaded": False,
"error": "下载成功但初始化失败"
}
else:
raise HTTPException(status_code=500, detail=message)
@app.post("/v1/chat/completions", response_model=ChatResponse)
async def chat_completions(request: ChatRequest):
"""
OpenAI 兼容的聊天接口
"""
global pipe, tokenizer, model_name
# 检查是否需要切换模型
if request.model != model_name:
print(f"\n🔄 切换模型: {model_name} → {request.model}")
pipe, tokenizer, success = initialize_pipeline(request.model)
if not success:
raise HTTPException(status_code=500, detail="模型初始化失败")
model_name = request.model
try:
return create_chat_response(request, pipe, tokenizer)
except Exception as e:
print(f"❌ 处理请求失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
# 运行命令: uvicorn app:app --host 0.0.0.0 --port 7860 --reload
requirements.txt
fastapi
uvicorn[standard]
transformers
torch
accelerate
python-dotenv
python-multipart
huggingface_hub
Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 复制依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 暴露端口
EXPOSE 7860
# 启动服务
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]
七、测试与调试（真实过程）
7.1 本地测试
启动服务：
uvicorn app:app --host 0.0.0.0 --port 7860 --reload
测试 1：检查状态
curl http://localhost:7860/
预期：
{
"message": "Gemma 函数调用服务已启动！",
"current_model": "unsloth/functiongemma-270m-it",
"status": "ready"
}
测试 2：函数调用
curl -X POST "http://localhost:7860/v1/chat/completions" \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "user", "content": "北京天气如何？"},
{"role": "system", "content": "使用 get_weather(city) 函数"}
],
"max_tokens": 100
}'
测试 3：下载模型（如果没下载）
curl -X POST "http://localhost:7860/download" \
-H "Content-Type: application/json" \
-d '{"model": "unsloth/functiongemma-270m-it"}'
7.2 常见问题调试
问题 1：ImportError: No module named 'transformers'
# 解决
pip install transformers
问题 2：OutOfMemoryError
# 解决：换更小的模型
# 修改 .env
DEFAULT_MODEL_NAME="TinyLlama/TinyLlama-1.1B-Chat-v1.0"
问题 3：下载超时
# 解决：用国内镜像
export HF_ENDPOINT=https://hf-mirror.com
八、Prompt 调试技巧总结
8.1 好的 Prompt 特征
✅ 具体明确：不说"写个函数"，而说"写个检查模型的函数，返回三元组"
✅ 分步迭代：先实现基础功能，再逐步优化
✅ 提供上下文：说明用途、环境、约束
✅ 要求示例：让 AI 给出测试代码
8.2 调试技巧
- 打印中间结果：在代码中加
print()看数据流 - 缩小范围：单独测试出问题的函数
- 对比测试：用已知正确的代码对比
- 分步验证：每改一步就测试一次
8.3 我的 Prompt 模板
任务：[具体要做什么]
背景：[为什么要做]
要求：
1. [具体要求1]
2. [具体要求2]
3. [具体要求3]
输出格式：[代码/解释/示例]
已知问题：[如果有]
九、部署到 HuggingFace Space
9.1 上传代码
git init
git add .
git commit -m "v0.0.2 - 完整可运行版本"
git remote add origin https://huggingface.co/spaces/你的用户名/你的Space名称
git push -u origin main
9.2 Space 配置
- SDK: Docker
- Port: 7860
- Environment: 无需配置（使用 .env 默认值）
9.3 监控日志
在 Space 页面查看构建和运行日志，如果有问题：
- 看构建日志（依赖安装）
- 看运行日志（模型加载）
- 看请求日志（API 调用）
十、总结
10.1 学到了什么？
- ✅ AI 编码流程：Prompt → 代码 → 调试 → 优化
- ✅ Prompt 技巧：具体、分步、迭代
- ✅ 调试方法：打印、缩小范围、对比
- ✅ 完整项目：从 0 到部署的全过程
10.2 代码量统计
model.py: 60 行chat_request.py: 10 行chat_response.py: 50 行app.py: 50 行- 总计: ~170 行
10.3 下一步
- 测试更多模型
- 添加更多函数调用示例
- 集成到实际应用
版本: v0.0.2
状态: ✅ 完整可运行
更新: 2026-01-01
重点: AI 编码过程 + Prompt 调试记录
有问题随时问！🚀