微调流水线全景

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ 数据准备  │───▶│ 训练配置  │───▶│ 训练执行  │───▶│  评估    │───▶│ 部署发布  │
│ 清洗/标注 │    │ LoRA/QLoRA│    │ GPU 集群  │    │ 自动化   │    │ 灰度/AB   │
└──────────┘    └──────────┘    └──────────┘    └──────────┘    └──────────┘
      │                                              │                │
      ▼                                              ▼                ▼
 ┌──────────┐                                   ┌──────────┐   ┌──────────┐
 │ 数据版本  │                                   │ 模型注册  │   │ 监控告警  │
 │ DVC/MLflow│                                  │ MLflow   │   │ 回滚机制  │
 └──────────┘                                   └──────────┘   └──────────┘

一、数据准备

1.1 数据清洗

import re
import json
from datasets import Dataset

class DataCleaner:
    def __init__(self, min_length=10, max_length=8192):
        self.min_length = min_length
        self.max_length = max_length

    def clean(self, samples: list[dict]) -> list[dict]:
        cleaned = []
        for s in samples:
            text = s.get("text", "")
            # 去除 HTML 标签
            text = re.sub(r'<[^>]+>', '', text)
            # 去除多余空白
            text = re.sub(r'\s+', ' ', text).strip()
            # 长度过滤
            if self.min_length <= len(text) <= self.max_length:
                # 去重（基于内容哈希）
                cleaned.append({**s, "text": text})
        # 去重
        seen = set()
        unique = []
        for s in cleaned:
            h = hash(s["text"][:200])
            if h not in seen:
                seen.add(h)
                unique.append(s)
        return unique

    def to_chat_format(self, samples: list[dict]) -> list[dict]:
        """转换为 chatml 格式"""
        formatted = []
        for s in samples:
            formatted.append({
                "messages": [
                    {"role": "system", "content": s.get("system", "你是一个有用的助手")},
                    {"role": "user", "content": s["input"]},
                    {"role": "assistant", "content": s["output"]}
                ]
            })
        return formatted

1.2 数据增强

class DataAugmenter:
    """使用大模型生成训练数据变体"""

    AUGMENT_PROMPT = """基于以下示例，生成 3 个语义相同但表达不同的变体：
    原文：{original}
    要求：
    1. 保持意图一致
    2. 变化表达方式（句式/用词）
    3. 不要改变关键信息
    输出 JSON 数组格式。"""

    async def augment(self, sample: dict, llm_client) -> list[dict]:
        prompt = self.AUGMENT_PROMPT.format(original=sample["input"])
        resp = await llm_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        variants = json.loads(resp.choices[0].message.content)
        return [
            {"input": v["input"], "output": sample["output"]}
            for v in variants.get("variants", [])
        ]

1.3 数据集分割

from sklearn.model_selection import train_test_split

def split_dataset(data: list[dict], train=0.8, val=0.1, test=0.1):
    train_data, temp = train_test_split(data, test_size=1-train, random_state=42)
    val_data, test_data = train_test_split(temp, test_size=test/(test+val), random_state=42)
    return {"train": train_data, "val": val_data, "test": test_data}

二、训练配置

2.1 训练方法对比

方法显存需求训练速度效果适用场景
Full Fine-tune极高（全部参数）慢最好数据充足、预算充足
LoRA低（0.1-1% 参数）快接近全量通用首选
QLoRA极低（4bit 量化）中略低于 LoRA显存受限
P-Tuning v2低快中等特定任务

2.2 LoRA 训练配置

from peft import LoraConfig, get_peft_model, TaskType
from transformers import AutoModelForCausalLM, TrainingArguments
from trl import SFTTrainer

def setup_lora_training(model_name="Qwen/Qwen2.5-7B"):
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype="auto",
        device_map="auto",
    )

    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        r=64,                    # LoRA 秩，越大效果越好但显存越多
        lora_alpha=128,          # 通常为 r 的 2 倍
        lora_dropout=0.05,
        target_modules=[
            "q_proj", "k_proj", "v_proj", "o_proj",
            "gate_proj", "up_proj", "down_proj"
        ],
        bias="none",
    )

    model = get_peft_model(model, lora_config)
    return model, lora_config

training_args = TrainingArguments(
    output_dir="./output/qwen-lora",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    warmup_ratio=0.1,
    learning_rate=2e-4,
    lr_scheduler_type="cosine",
    logging_steps=10,
    save_strategy="epoch",
    eval_strategy="epoch",
    bf16=True,
    gradient_checkpointing=True,
    optim="adamw_torch",
    max_grad_norm=1.0,
)

trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    packing=True,               # 序列打包提升效率
    max_seq_length=2048,
)

2.3 QLoRA（显存优化）

from transformers import BitsAndBytesConfig

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype="bfloat16",
    bnb_4bit_use_double_quant=True,
)

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto",
)

三、评估流程

class ModelEvaluator:
    def __init__(self, model_path, test_data):
        self.model_path = model_path
        self.test_data = test_data

    def evaluate(self) -> dict:
        results = {
            "loss": self._eval_loss(),
            "bleu": self._eval_bleu(),
            "rouge": self._eval_rouge(),
            "human_like": self._eval_human_like(),
            "safety": self._eval_safety(),
            "latency_p50": self._eval_latency(),
        }
        return results

    def _eval_safety(self) -> float:
        """安全评估：检测有害输出比例"""
        harmful_count = 0
        for sample in self.test_data:
            output = self._generate(sample["input"])
            if self._is_harmful(output):
                harmful_count += 1
        return 1.0 - harmful_count / len(self.test_data)

    def _is_harmful(self, text: str) -> bool:
        harmful_patterns = [
            r"如何(制造|获取).*(武器|毒品)",
            r"(自杀|自残)的方法",
            r"歧视.*(种族|性别|宗教)",
        ]
        return any(re.search(p, text) for p in harmful_patterns)

四、版本管理

import mlflow

class ModelRegistry:
    def __init__(self, tracking_uri="http://mlflow:5000"):
        mlflow.set_tracking_uri(tracking_uri)

    def register_model(self, model_path, name, metrics, tags=None):
        with mlflow.start_run():
            mlflow.log_metrics(metrics)
            mlflow.log_artifacts(model_path)
            mlflow.register_model(
                f"runs:/{mlflow.active_run().info.run_id}/model",
                name,
                tags=tags or {}
            )

    def get_version(self, name, stage="Production"):
        client = mlflow.tracking.MlflowClient()
        versions = client.get_latest_versions(name, stages=[stage])
        return versions[0] if versions else None

五、灰度发布与 A/B 测试

class CanaryDeployer:
    """灰度发布：逐步增加新模型流量比例"""

    def __init__(self, old_model: str, new_model: str):
        self.old_model = old_model
        self.new_model = new_model
        self.traffic_split = 0.0  # 新模型流量比例
        self.metrics = {"old": [], "new": []}

    def should_use_new(self) -> bool:
        import random
        return random.random() < self.traffic_split

    def canary_stages(self):
        """分阶段灰度"""
        stages = [
            {"split": 0.05, "duration": "1h", "check": "error_rate < 1%"},
            {"split": 0.20, "duration": "6h", "check": "error_rate < 1%, latency_p99 < 10s"},
            {"split": 0.50, "duration": "24h", "check": "all_metrics_stable"},
            {"split": 1.00, "duration": "∞", "check": "promoted"},
        ]
        return stages

    def evaluate_and_promote(self):
        """评估指标决定是否推进"""
        new_error_rate = self._calc_error_rate("new")
        old_error_rate = self._calc_error_rate("old")
        new_latency = self._calc_p99("new")
        old_latency = self._calc_p99("old")

        if new_error_rate > old_error_rate * 1.5:
            self._rollback()
            return "ROLLBACK: error rate too high"
        if new_latency > old_latency * 1.3:
            self._rollback()
            return "ROLLBACK: latency regression"
        return "PROMOTE: metrics OK"

六、回滚机制

class RollbackManager:
    def __init__(self, registry: ModelRegistry):
        self.registry = registry

    def rollback(self, model_name: str, reason: str):
        """回滚到上一个 Production 版本"""
        client = mlflow.tracking.MlflowClient()
        versions = client.search_model_versions(
            f"name='{model_name}'", order_by=["version_number DESC"]
        )
        prod_versions = [v for v in versions if v.current_stage == "Production"]
        archived = [v for v in versions if v.current_stage == "Archived"]

        if len(prod_versions) >= 1 and archived:
            # 当前 prod 版本归档，上一个 archived 版本恢复
            client.transition_model_version_stage(
                name=model_name,
                version=prod_versions[0].version,
                stage="Archived",
            )
            client.transition_model_version_stage(
                name=model_name,
                version=archived[0].version,
                stage="Production",
            )
            logger.info(f"Rolled back {model_name}: {reason}")
            return True
        return False

总结

LLM 微调 MLOps 流水线的核心环节：数据质量决定上限，LoRA/QLoRA 平衡效果与成本，评估必须覆盖质量+安全+性能三维度，灰度发布配合自动回滚是生产安全的最后防线。建议使用 MLflow 统一管理模型版本，从训练到部署全链路可追溯。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。