ms-swift + Python API：灵活控制训练与推理流程-深圳市維司達科技有限公司

ms-swift + Python API：灵活控制训练与推理流程

1. 为什么需要 Python API？——从命令行到工程化控制的跃迁

你有没有遇到过这些场景？

在自动化训练流水线中，想根据上一轮评估结果动态调整学习率，但命令行参数是写死的；
想在 Web 应用里让用户上传数据集、选择模型、点击“开始训练”，而不是让工程师手动敲命令；
需要将微调过程嵌入现有业务系统，比如电商客服模型每天凌晨自动用新对话日志微调一次；
做研究时想对比 5 种 LoRA 配置的效果，每次改参数、跑命令、等日志、手动记录，重复劳动占了大半时间。

这些都不是“能不能做”的问题，而是“好不好控”的问题。ms-swift 的命令行工具（swift sft/swift infer）开箱即用、上手极快，但它本质是一个封装好的终端程序——像一辆配置齐全的轿车，你坐进去就能开，但没法拆开发动机加装涡轮。

而 Python API，就是那把能打开引擎盖、拧动每一颗螺丝、甚至重写控制逻辑的万能扳手。

它不替代命令行，而是补全命令行做不到的事：
动态构建训练配置，而非静态传参
在训练循环中插入自定义钩子（比如每 10 步保存中间状态并触发质检）
将模型加载、预处理、推理、后处理串成可复用函数，无缝接入 Flask/FastAPI
用面向对象方式管理多个模型实例（比如同时跑一个 Qwen3-VL 做图文理解，一个 GLM4.5-V 做多模态生成）
真正实现“训练即代码”（Training as Code），版本可控、调试可见、协作可追溯

这不是给极客准备的玩具接口，而是为真实工程落地设计的控制中枢。接下来，我们就用最贴近生产环境的方式，带你亲手搭起这套灵活可控的训练与推理体系。

2. Python API 核心架构：三层抽象，各司其职

ms-swift 的 Python API 并非简单把命令行参数转成函数调用，而是按工程实践分层设计，每一层解决一类问题：

2.1 底层：模型与分词器工厂（Model & Tokenizer）

这是所有操作的起点，负责“把模型真正加载进内存”。

from swift.llm import get_model_tokenizer # 加载 Qwen3-7B-Instruct，自动识别架构、加载权重、配置 dtype model, tokenizer = get_model_tokenizer( 'Qwen/Qwen3-7B-Instruct', torch_dtype='bfloat16', # 指定计算精度 device_map='auto', # 自动分配 GPU/CPU max_length=4096, # 预设上下文长度 quantization_bit=0, # 不量化（0 表示不启用） )

关键点：

不关心路径细节：传入 ModelScope ID 或本地路径均可，自动处理下载/缓存/格式转换；
模板自动绑定：Qwen 系列自动匹配qwentemplate，Llama 系列匹配llama，无需手动指定；
安全兜底：若模型不支持 bfloat16，自动降级为 float16；若显存不足，自动启用device_map='balanced_low_0'。

✦ 小贴士：get_model_tokenizer返回的是标准 Hugging FacePreTrainedModel和PreTrainedTokenizer实例，意味着你后续可以完全使用 HF 生态的任何工具（如pipeline、Trainer、TextIteratorStreamer），零学习成本。

2.2 中层：轻量微调控制器（Swift Adapter）

这是 ms-swift 的核心创新之一——把 LoRA/QLoRA/DoRA 等适配器技术封装成可插拔模块。

from swift.llm import Swift # 为模型注入 LoRA 层（仅修改 adapter 参数，原模型权重冻结） lora_config = { 'r': 8, # rank 'lora_alpha': 32, # 缩放系数 'target_modules': 'all-linear', # 自动识别所有线性层 'modules_to_save': ['embed_tokens', 'lm_head'] # 需微调的非 LoRA 模块 } model = Swift.prepare_model(model, config=lora_config)

它解决了三个痛点：

统一入口：无论用 LoRA、QLoRA 还是 DoRA，都调Swift.prepare_model，只需换 config；
无感融合：训练时 adapter 参与计算，推理时model.forward()自动路由，无需手动开关；
多 adapter 切换：同一模型实例可加载多个 adapter（如adapter_a,adapter_b），通过model.set_adapter('adapter_a')切换，适合 A/B 测试。

2.3 上层：训练与推理引擎（Trainer & Engine）

这才是真正“控制流程”的部分——它把训练循环和推理服务抽象成可编程对象。

训练引擎：Seq2SeqTrainer（继承自 HF Trainer）

from swift.llm import Seq2SeqTrainer, TrainingArguments # 构建训练参数（完全兼容 Hugging Face TrainingArguments） training_args = TrainingArguments( output_dir='./output', num_train_epochs=1, per_device_train_batch_size=1, per_device_eval_batch_size=1, learning_rate=1e-4, logging_steps=5, save_steps=100, eval_steps=100, save_total_limit=2, report_to='none', # 关闭 wandb/tensorboard，避免干扰 ) # 构建数据集（支持 ModelScope/HF 数据集 ID 或本地路径） from swift.llm import load_dataset train_dataset, val_dataset = load_dataset( 'AI-ModelScope/alpaca-gpt4-data-zh#500', 'AI-ModelScope/alpaca-gpt4-data-en#500' ) # 自动编码：将文本对话转为 token ids，支持 packing、truncation、padding from swift.llm import EncodePreprocessor template = get_template(model.model_meta.template, tokenizer) # 获取对应 template train_dataset = EncodePreprocessor(template=template)(train_dataset, num_proc=4) val_dataset = EncodePreprocessor(template=template)(val_dataset, num_proc=4) # 启动训练（注意：传入的是 Swift 包装后的 model，不是原始 model） trainer = Seq2SeqTrainer( model=model, args=training_args, data_collator=template.data_collator, # 自动处理 batch padding train_dataset=train_dataset, eval_dataset=val_dataset, template=template, # 关键！让 trainer 知道如何解码输出 ) # 这里可以插入任意自定义逻辑 class CustomCallback(TrainerCallback): def on_step_end(self, args, state, control, **kwargs): if state.global_step % 50 == 0: # 每 50 步用当前模型生成样例，发到企业微信通知群 sample_output = generate_sample(model, tokenizer) send_to_wxwork(sample_output) trainer.add_callback(CustomCallback()) trainer.train()

推理引擎：PtEngine（PyTorch 原生）与 vLLMEngine（高性能）

from swift.llm import PtEngine, vLLMEngine, InferRequest, RequestConfig # 方式一：PtEngine —— 简单、可控、易调试（适合开发/测试） engine = PtEngine( model_id_or_path='Qwen/Qwen3-7B-Instruct', adapters=['./output/checkpoint-100'], # 支持多个 adapter 路径 torch_dtype='bfloat16', device_map='auto' ) # 构造请求（完全兼容 OpenAI Chat Completion 格式） infer_request = InferRequest( messages=[ {'role': 'system', 'content': '你是一个严谨的学术助手'}, {'role': 'user', 'content': '请用三句话总结 Transformer 架构的核心思想'} ], tools=None # 支持 tool calling ) request_config = RequestConfig( max_tokens=512, temperature=0.7, top_p=0.9, stream=False ) # 同步推理（返回完整响应） resp_list = engine.infer([infer_request], request_config) print(resp_list[0].choices[0].message.content) # 输出：Transformer 的核心是自注意力机制……（省略） # 方式二：vLLMEngine —— 高吞吐、低延迟（适合生产部署） vllm_engine = vLLMEngine( model_id_or_path='Qwen/Qwen3-7B-Instruct', adapters=['./output/checkpoint-100'], tensor_parallel_size=2, # 多卡并行 max_model_len=8192, gpu_memory_utilization=0.9 ) # 异步流式推理（返回 Generator，逐 token 输出） stream_resp = vllm_engine.infer([infer_request], request_config) for chunk in stream_resp: print(chunk.choices[0].delta.content, end='', flush=True) # 实时打印

✦ 对比总结：
PtEngine：像手摇咖啡机，每一步清晰可见，适合调试、教学、小规模服务；
vLLMEngine：像全自动意式咖啡机，吞吐高、延迟低，需预热但稳定可靠，适合线上 API；
两者 API 完全一致，切换只需改一行engine = ...，业务逻辑零改造。

3. 实战：构建一个“自动优化”的微调工作流

光讲概念不够直观。我们来做一个真实场景的端到端示例：电商商品文案生成模型的持续微调系统。

目标：每天凌晨，系统自动拉取昨日 1000 条用户好评，提取其中“描述商品卖点”的句子，微调模型使其生成更抓眼球的文案。

3.1 数据准备：动态清洗与构造

import json from datasets import Dataset from swift.llm import load_dataset def fetch_yesterday_reviews(): """模拟从数据库拉取昨日好评（实际对接 MySQL/API）""" # 返回 [{'review_id': 'xxx', 'text': '这个手机拍照太好了，夜景清晰...'}] pass def extract_selling_points(reviews): """用规则+小模型提取卖点短句（简化版）""" selling_points = [] for r in reviews: # 真实场景可用轻量 NER 模型或关键词匹配 if '拍照' in r['text'] or '清晰' in r['text']: selling_points.append(r['text'].split('，')[0] + '，超清晰！') return selling_points def build_instruction_dataset(selling_points): """构造 SFT 格式数据集""" instructions = [] for sp in selling_points: # 模板：输入卖点 → 输出爆款文案 instructions.append({ 'instruction': f'根据以下商品卖点，生成一条吸引人的电商文案，不超过30字：{sp}', 'input': '', 'output': f'【夜景神器】{sp}，随手一拍就是大片！' }) return Dataset.from_list(instructions) # 执行流程 yesterday_reviews = fetch_yesterday_reviews() selling_points = extract_selling_points(yesterday_reviews) train_dataset = build_instruction_dataset(selling_points) # 保存为 JSONL，供后续训练加载 train_dataset.to_json('./data/daily_sft.jsonl')

3.2 训练：带早停与自动评估的闭环

from swift.llm import Seq2SeqTrainer, TrainingArguments, load_dataset from transformers import EvalPrediction import numpy as np # 加载基础模型与 tokenizer model, tokenizer = get_model_tokenizer('Qwen/Qwen3-7B-Instruct') # 注入 LoRA model = Swift.prepare_model(model, config={ 'r': 16, 'lora_alpha': 64, 'target_modules': 'all-linear' }) # 加载数据集（支持本地文件路径） train_dataset = load_dataset('./data/daily_sft.jsonl') # 划分 9:1 训练/验证 train_dataset = train_dataset['train'].train_test_split(test_size=0.1) train_ds, eval_ds = train_dataset['train'], train_dataset['test'] # 编码 template = get_template('qwen', tokenizer) train_ds = EncodePreprocessor(template=template)(train_ds, num_proc=2) eval_ds = EncodePreprocessor(template=template)(eval_ds, num_proc=2) # 定义评估指标（这里用 BLEU 简化，实际可用 BERTScore） def compute_metrics(eval_preds: EvalPrediction): preds, labels = eval_preds decoded_preds = tokenizer.batch_decode(preds, skip_special_tokens=True) decoded_labels = tokenizer.batch_decode(labels, skip_special_tokens=True) # 简化：计算预测长度与标签长度的绝对误差均值（越小越好） pred_lens = [len(p) for p in decoded_preds] label_lens = [len(l) for l in decoded_labels] return {'length_error': np.mean(np.abs(np.array(pred_lens) - np.array(label_lens)))} # 训练参数（开启早停） training_args = TrainingArguments( output_dir='./output/daily_sft', num_train_epochs=3, per_device_train_batch_size=1, per_device_eval_batch_size=1, learning_rate=2e-4, warmup_ratio=0.1, logging_steps=1, save_steps=10, eval_steps=10, evaluation_strategy='steps', load_best_model_at_end=True, # 训练结束自动加载最优 checkpoint metric_for_best_model='length_error', # 以 length_error 最小为优 greater_is_better=False, save_total_limit=1, report_to='none' ) trainer = Seq2SeqTrainer( model=model, args=training_args, train_dataset=train_ds, eval_dataset=eval_ds, compute_metrics=compute_metrics, data_collator=template.data_collator, template=template ) # 开始训练（自动早停，最多 3 epoch，但若连续 5 次 eval 无提升则停止） trainer.train() # 保存最终 adapter（用于后续推理） Swift.save_adapters(model, './output/daily_sft/final_adapter')

3.3 推理：封装成 Web API 服务

# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from swift.llm import PtEngine, InferRequest, RequestConfig app = FastAPI(title="电商文案生成 API") # 全局加载一次模型（启动时） engine = PtEngine( model_id_or_path='Qwen/Qwen3-7B-Instruct', adapters=['./output/daily_sft/final_adapter'], device_map='auto' ) class GenerateRequest(BaseModel): selling_point: str max_length: int = 30 @app.post("/generate") def generate_text(req: GenerateRequest): try: infer_request = InferRequest( messages=[ {'role': 'system', 'content': '你是一个专业的电商文案策划师，生成简短有力、带emoji的爆款文案'}, {'role': 'user', 'content': f'根据卖点生成文案：{req.selling_point}'} ] ) request_config = RequestConfig( max_tokens=req.max_length, temperature=0.8, top_k=50 ) resp_list = engine.infer([infer_request], request_config) return {"text": resp_list[0].choices[0].message.content} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 启动：uvicorn app:app --reload

运行后，即可用 curl 测试：

curl -X POST "http://localhost:8000/generate" \ -H "Content-Type: application/json" \ -d '{"selling_point": "这款耳机音质纯净，低音震撼"}' # 返回：{"text": "🎧音质天花板！低音轰鸣，沉浸感拉满！"}

这个工作流的价值在于：
🔹全自动：从数据拉取、清洗、训练、评估、保存到 API 更新，全程无人值守；
🔹可迭代：每天新数据驱动模型进化，文案质量随时间提升；
🔹可监控：length_error指标可接入 Prometheus，异常升高自动告警；
🔹可回滚：final_adapter每日覆盖，但历史 checkpoint 保留，一键切回昨日版本。

4. 进阶技巧：解锁 Python API 的隐藏能力

命令行工具是“开箱即用”，Python API 是“按需定制”。以下这些能力，只有深入代码层才能释放：

4.1 多模态模型的精细控制：单独冻结/解冻视觉编码器

对 Qwen3-VL 这类图文模型，常需只微调语言模型（LLM），保持视觉编码器（ViT）不变：

from swift.llm import get_model_tokenizer, Swift model, tokenizer = get_model_tokenizer('Qwen/Qwen3-VL') # 查看模型结构（关键！） print(model) # 输出包含：model.vision_tower（ViT）、model.aligner（连接层）、model.language_model（LLM） # 方法一：冻结整个 vision_tower（推荐） for name, param in model.vision_tower.named_parameters(): param.requires_grad = False # 方法二：用 Swift 的模块级控制（更精准） model = Swift.prepare_model( model, config={ 'r': 8, 'target_modules': 'language_model.*', # 只对 LLM 中的线性层加 LoRA 'modules_to_save': ['vision_tower'] # 但保留 vision_tower 的原始权重可保存 } )

4.2 GRPO 强化学习：自定义奖励函数

ms-swift 内置 GRPO 等算法，但奖励函数常需业务定制。Python API 允许你完全接管：

from swift.llm import GRPOTrainer from transformers import TrainerCallback class EcommerceRewardCallback(TrainerCallback): def on_compute_loss(self, args, state, control, model, inputs, **kwargs): # inputs 包含 prompt + response（模型生成的文案） # 这里可调用自有风控模型、点击率预估模型、人工审核 API reward_score = call_click_rate_api(inputs['prompt'], inputs['response']) # 将 reward 注入 loss 计算（GRPOTrainer 会读取） inputs['reward'] = reward_score return None trainer = GRPOTrainer( model=model, args=training_args, # ... 其他参数 ) trainer.add_callback(EcommerceRewardCallback()) trainer.train()

4.3 混合后端推理：vLLM + PyTorch 协同

某些场景需“大部分请求走 vLLM，少数复杂请求走 PyTorch”：

class HybridEngine: def __init__(self): self.vllm_engine = vLLMEngine(...) # 高吞吐主力 self.pt_engine = PtEngine(...) # 精确控制备用 def infer(self, requests, config): # 规则：长 prompt（>2048 tokens）或含 tool calling 的走 PtEngine if any(len(r.messages[0]['content']) > 2048 for r in requests) or config.tools: return self.pt_engine.infer(requests, config) else: return self.vllm_engine.infer(requests, config) hybrid_engine = HybridEngine()