章节导语

通用大模型很强大，但针对你的场景可能不够好用。微调（Fine-tuning）就是让通用模型变成领域专家的技术。

比如通用GPT不懂你公司的术语，但微调后的模型就能准确理解。微调不是训练新模型，而是让现有模型”适配”你的场景。

本文系统讲解微调的完整流程，从数据准备到模型训练，再到效果评估，每步都有实战代码。

一、前置说明

1.1 学习路径

1.2 读者需要的基础

• Python基础：能写训练脚本
• 深度学习：知道什么是神经网络、梯度下降
• LLM基础：用过GPT、知道Embedding

1.3 学习目标

学完本文，你将能够：

• 理解什么时候该微调，什么时候不该
• 掌握微调数据的准备和处理
• 使用OpenAI/Hugging Face进行微调
• 评估微调效果

二、微调 vs Prompt工程

2.1 什么时候用微调

图1：微调 vs RAG vs Prompt

不是所有场景都需要微调。先问自己：Prompt能否解决？

2.2 微调的典型应用

• 客服机器人：适配品牌语气、FAQ
• 代码助手：学会特定代码库的风格
• 医疗助手：理解医学术语、病历格式
• 法律助手：理解法律条文、合同格式
• 特定语言风格：口语化、年轻化、专业化

2.3 微调的代价

• 成本高：训练需要GPU资源
• 周期长：数据准备+训练+评估
• 不够灵活：知识更新需要重新训练
• 可能过拟合：训练数据不够会导致泛化能力差

三、环境配置

3.1 OpenAI微调

# OpenAI微调依赖
pip install openai tqdmHv

# 准备数据格式转换
pip install pandas jsonlines

3.2 开源模型微调（Hugging Face）

# Hugging Face微调依赖
pip install transformers datasets accelerate peft

# qlora相关（高效微调）
pip install bitsandbytes trl rouge-score

3.3 硬件要求

四、数据准备

4.1 数据格式

OpenAI微调需要JSONL格式：

图2：微调数据格式

{"messages": [
  {"role": "system", "content": "你是一个客服助手"},
  {"role": "user", "content": "怎么退货？"},
  {"role": "assistant", "content": "请提供订单号，我来帮您查询退货流程..."}
]}

{"messages": [
  {"role": "system", "content": "你是一个客服助手"},
  {"role": "user", "content": "什么时候发货？"},
  {"role": "assistant", "content": "正常情况下24小时内发货..."}
]}

4.2 数据准备代码

import json
from typing import List, Dict

class FineTuningDataPreparer:
    """微调数据准备器
    
    将原始数据转换为微调格式
    """
    
    def __init__(self, system_prompt: str = ""):
        self.system_prompt = system_prompt
    
    def create_conversation(self, messages: List[Dict]) -> Dict:
        """创建对话格式
        
        参数:
            messages: 消息列表，格式：[{"role": "user", "content": "..."}]
        返回:
            微调格式的对话
        """
        conversation = {"messages": []}
        
        # 添加系统提示词
        if self.system_prompt:
            conversation["messages"].append({
                "role": "system",
                "content": self.system_prompt
            })
        
        # 添加对话
        for msg in messages:
            conversation["messages"].append({
                "role": msg["role"],
                "content": msg["content"]
            })
        
        return conversation
    
    def prepare_from_qa(self, qa_pairs: List[Dict], output_file: str):
        """从问答对准备数据
        
        qa_pairs格式: [{"question": "...", "answer": "..."}]
        """
        print(f"📝 准备 {len(qa_pairs)} 个问答对...")
        
        with open(output_file, 'w', encoding='utf-8') as f:
            for i, qa in enumerate(qa_pairs):
                conversation = self.create_conversation([
                    {"role": "user", "content": qa["question"]},
                    {"role": "assistant", "content": qa["answer"]}
                ])
                
                f.write(json.dumps(conversation, ensure_ascii=False) + '\n')
                
                if (i + 1) % 100 == 0:
                    print(f"   已处理 {i+1}/{len(qa_pairs)}")
        
        print(f"✅ 数据已保存到 {output_file}")
    
    def prepare_from_csv(self, csv_file: str, output_file: str, 
                        question_col: str, answer_col: str):
        """从CSV文件准备数据"""
        import pandas as pd
        
        print(f"📂 读取CSV: {csv_file}")
        df = pd.read_csv(csv_file)
        
        qa_pairs = []
        for _, row in df.iterrows():
            qa_pairs.append({
                "question": str(row[question_col]),
                "answer": str(row[answer_col])
            })
        
        self.prepare_from_qa(qa_pairs, output_file)
    
    def validate_data(self, file_path: str) -> Dict:
        """验证数据质量"""
        print(f"🔍 验证数据: {file_path}")
        
        total = 0
        valid = 0
        errors = []
        
        with open(file_path, 'r', encoding='utf-8') as f:
            for i, line in enumerate(f, 1):
                total += 1
                try:
                    data = json.loads(line.strip())
                    
                    # 检查必要字段
                    if "messages" not in data:
                        errors.append(f"行{i}: 缺少messages字段")
                        continue
                    
                    msgs = data["messages"]
                    
                    # 必须有user和assistant消息
                    roles = [m.get("role") for m in msgs]
                    if "user" not in roles or "assistant" not in roles:
                        errors.append(f"行{i}: 缺少user或assistant消息")
                        continue
                    
                    valid += 1
                    
                except json.JSONDecodeError as e:
                    errors.append(f"行{i}: JSON解析错误 - {e}")
        
        print(f"   总数: {total}, 有效: {valid}, 错误: {len(errors)}")
        
        if errors:
            print("   错误示例:")
            for err in errors[:5]:
                print(f"     - {err}")
        
        return {
            "total": total,
            "valid": valid,
            "errors": errors[:10]  # 只返回前10个错误
        }

# 使用
if __name__ == "__main__":
    preparer = FineTuningDataPreparer(
        system_prompt="你是一个专业的客服助手，礼貌、专业地回答用户问题。"
    )
    
    # 示例问答数据
    qa_pairs = [
        {"question": "你们支持退货吗？", "answer": "支持，7天内可以申请退货，请登录账号进入'我的订单'操作。"},
        {"question": "退货需要运费吗？", "answer": "质量问题我们承担运费，非质量问题退货运费由用户承担。"},
        {"question": "退款多久到账？", "answer": "退款会在15个工作日内原路返回您的支付账户。"},
    ]
    
    # 生成训练数据
    preparer.prepare_from_qa(qa_pairs, "training_data.jsonl")
    
    # 验证
    preparer.validate_data("training_data.jsonl")

4.3 数据质量检查清单

• 数量：通常100-1000条有效样本
• 多样性：避免重复句式
• 准确性：答案必须正确
• 格式：JSON格式正确
• 长度：单条不超过模型上下文窗口

五、OpenAI微调实战

5.1 上传数据

from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class OpenAIFineTuner:
    """OpenAI微调管理器"""
    
    def __init__(self):
        self.client = client
    
    def upload_data(self, file_path: str) -> str:
        """上传微调数据
        
        返回:
            file_id
        """
        print(f"📤 上传数据: {file_path}")
        
        with open(file_path, "rb") as f:
            response = self.client.files.create(
                file=f,
                purpose="fine-tune"
            )
        
        file_id = response.id
        print(f"✅ 上传成功: {file_id}")
        return file_id
    
    def create_fine_tune_job(self, file_id: str, model: str = "gpt-3.5-turbo") -> str:
        """创建微调任务
        
        参数:
            file_id: 上传文件的ID
            model: 基础模型
        返回:
            job_id
        """
        print(f"🚀 创建微调任务: {model}")
        
        job = self.client.fine_tuning.jobs.create(
            training_file=file_id,
            model=model
        )
        
        job_id = job.id
        print(f"✅ 任务创建成功: {job_id}")
        return job_id
    
    def get_job_status(self, job_id: str) -> Dict:
        """获取任务状态"""
        job = self.client.fine_tuning.jobs.get(job_id)
        
        status = job.status
        progress = job.progress
        
        print(f"📊 状态: {status}, 进度: {progress}%")
        
        return {
            "id": job.id,
            "status": status,
            "progress": progress,
            "model": job.fine_tuned_model,
            "error": job.error
        }
    
    def list_jobs(self) -> List[Dict]:
        """列出所有微调任务"""
        jobs = self.client.fine_tuning.jobs.list(limit=10)
        
        print(f"\n📋 微调任务列表:")
        for job in jobs.data:
            print(f"  {job.id} | {job.status} | {job.model}")
        
        return [{"id": j.id, "status": j.status} for j in jobs.data]
    
    def cancel_job(self, job_id: str):
        """取消任务"""
        self.client.fine_tuning.jobs.cancel(job_id)
        print(f"✅ 任务已取消: {job_id}")

# 使用
if __name__ == "__main__":
    tuner = OpenAIFineTuner()
    
    # 1. 上传数据
    file_id = tuner.upload_data("training_data.jsonl")
    
    # 2. 创建微调任务
    job_id = tuner.create_fine_tune_job(file_id, model="gpt-3.5-turbo")
    
    # 3. 查看状态
    for i in range(5):
        status = tuner.get_job_status(job_id)
        if status["status"] in ["completed", "failed", "cancelled"]:
            print(f"\n🎉 最终状态: {status['status']}")
            if status["model"]:
                print(f"   模型: {status['model']}")
            break
        import time
        time.sleep(30)

5.2 使用微调后的模型

# 使用微调后的模型
response = client.chat.completions.create(
    model="ft:gpt-3.5-turbo:your_org:custom_suffix",  # 微调后的模型名
    messages=[
        {"role": "system", "content": "你是一个专业的客服助手"},
        {"role": "user", "content": "怎么退货？"}
    ]
)

print(response.choices[0].message.content)

5.3 成本估算

六、Hugging Face微调（LoRA）

6.1 LoRA原理

图3：LoRA高效微调原理

LoRA（Low-Rank Adaptation）只训练少量参数，大幅降低显存需求。

核心思想：不改变原始权重W，而是添加低秩分解的增量ΔW=BA。

6.2 LoRA微调代码

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from datasets import load_dataset

class LoRAFineTuner:
    """LoRA微调器
    
    使用Hugging Face PEFT库进行高效微调
    """
    
    def __init__(self, model_name: str = "meta-llama/Llama-2-7b-hf"):
        self.model_name = model_name
        self.model = None
        self.tokenizer = None
    
    def load_model(self):
        """加载模型"""
        print(f"📦 加载模型: {self.model_name}")
        
        # 加载tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        self.tokenizer.pad_token = self.tokenizer.eos_token
        
        # 加载模型（4bit量化，减少显存）
        self.model = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            load_in_4bit=True,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        
        # 准备训练
        self.model = prepare_model_for_kbit_training(self.model)
        print("✅ 模型加载完成")
    
    def apply_lora(self, r: int = 8, lora_alpha: int = 16, target_modules: list = None):
        """应用LoRA
        
        参数:
            r: LoRA秩，越大效果越好但更慢
            lora_alpha: 缩放因子
            target_modules: 要应用LoRA的模块
        """
        if target_modules is None:
            target_modules = ["q_proj", "v_proj", "k_proj", "o_proj"]
        
        lora_config = LoraConfig(
            r=r,
            lora_alpha=lora_alpha,
            target_modules=target_modules,
            lora_dropout=0.05,
            bias="none",
            task_type="CAUSAL_LM"
        )
        
        self.model = get_peft_model(self.model, lora_config)
        
        # 打印可训练参数
        self.model.print_trainable_parameters()
    
    def prepare_dataset(self, file_path: str):
        """准备数据集"""
        def format_prompt(examples):
            """格式化样本"""
            prompts = []
            for instruction, output in zip(examples["instruction"], examples["output"]):
                prompt = f"""### Instruction:
{instruction}

### Response:
{output}

### End"""
                prompts.append(prompt)
            return {"text": prompts}
        
        # 加载数据集
        dataset = load_dataset("json", data_files=file_path, split="train")
        
        # 格式化
        dataset = dataset.map(
            format_prompt,
            batched=True,
            remove_columns=dataset.column_names
        )
        
        # Tokenize
        def tokenize(example):
            result = self.tokenizer(
                example["text"],
                truncation=True,
                max_length=512,
                padding="max_length"
            )
            result["labels"] = result["input_ids"].copy()
            return result
        
        dataset = dataset.map(tokenize, batched=True)
        
        # 划分训练/验证集
        split_dataset = dataset.train_test_split(test_size=0.1)
        
        print(f"📊 训练集: {len(split_dataset['train'])} 样本")
        print(f"📊 验证集: {len(split_dataset['test'])} 样本")
        
        return split_dataset
    
    def train(self, dataset, output_dir: str = "./lora_model", epochs: int = 3):
        """训练"""
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=epochs,
            per_device_train_batch_size=4,
            gradient_accumulation_steps=4,
            learning_rate=2e-4,
            warmup_ratio=0.03,
            lr_scheduler_type="cosine",
            logging_steps=10,
            save_steps=100,
            eval_steps=100,
            evaluation_strategy="steps",
            fp16=True,
            optim="paged_adamw_8bit"
        )
        
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=dataset["train"],
            eval_dataset=dataset["test"],
        )
        
        print("🚀 开始训练...")
        trainer.train()
        print("✅ 训练完成")
        
        # 保存模型
        self.model.save_pretrained(output_dir)
        self.tokenizer.save_pretrained(output_dir)
        print(f"💾 模型已保存到: {output_dir}")

# 使用
if __name__ == "__main__":
    tuner = LoRAFineTuner("meta-llama/Llama-2-7b-hf")
    
    tuner.load_model()
    tuner.apply_lora(r=8, lora_alpha=16)
    
    # 准备数据（需要先准备好JSON数据）
    # dataset = tuner.prepare_dataset("training_data.json")
    # tuner.train(dataset, epochs=3)

七、效果评估

7.1 评估指标

图4：微调效果评估

7.2 自动化评估代码

from rouge_score import rouge_scorer
from sklearn.metrics import accuracy_score
import json

class FineTuningEvaluator:
    """微调效果评估器"""
    
    def __init__(self):
        self.rouge_scorer = rouge_scorer.RougeScorer(['rouge1', 'rougeL'], use_stemmer=True)
    
    def evaluate_rouge(self, predictions: list, references: list) -> dict:
        """评估ROUGE分数
        
        参数:
            predictions: 模型生成的答案列表
            references: 标准答案列表
        """
        rouge1_scores = []
        rougeL_scores = []
        
        for pred, ref in zip(predictions, references):
            scores = self.rouge_scorer.score(ref, pred)
            rouge1_scores.append(scores['rouge1'].fmeasure)
            rougeL_scores.append(scores['rougeL'].fmeasure)
        
        return {
            'rouge1': sum(rouge1_scores) / len(rouge1_scores),
            'rougeL': sum(rougeL_scores) / len(rougeL_scores)
        }
    
    def evaluate_exact_match(self, predictions: list, references: list) -> float:
        """评估精确匹配率"""
        return accuracy_score(references, predictions)
    
    def evaluate_quality(self, model_response: str, reference: str) -> dict:
        """评估回答质量（多维度）"""
        scores = {
            'accuracy': 0.0,
            'fluency': 0.0,
            'relevance': 0.0,
            'overall': 0.0
        }
        
        # 简单的自动化评分（实际应该用LLM Judge）
        if model_response and len(model_response) > 10:
            scores['fluency'] = min(1.0, len(model_response) / 100)
        
        # 与参考答案的重叠度
        if reference:
            overlap = len(set(model_response) & set(reference))
            scores['accuracy'] = overlap / max(len(set(reference)), 1)
            scores['relevance'] = overlap / max(len(set(model_response)), 1)
        
        scores['overall'] = (scores['accuracy'] + scores['fluency'] + scores['relevance']) / 3
        
        return scores
    
    def evaluate(self, model_responses: list, references: list) -> dict:
        """综合评估"""
        print("🔍 开始评估...")
        
        # ROUGE分数
        rouge_scores = self.evaluate_rouge(model_responses, references)
        print(f"   ROUGE-1: {rouge_scores['rouge1']:.4f}")
        print(f"   ROUGE-L: {rouge_scores['rougeL']:.4f}")
        
        # 精确匹配
        exact_match = self.evaluate_exact_match(model_responses, references)
        print(f"   Exact Match: {exact_match:.4f}")
        
        # 质量评估
        quality_scores = []
        for pred, ref in zip(model_responses, references):
            score = self.evaluate_quality(pred, ref)
            quality_scores.append(score)
        
        avg_quality = {
            'accuracy': sum(s['accuracy'] for s in quality_scores) / len(quality_scores),
            'fluency': sum(s['fluency'] for s in quality_scores) / len(quality_scores),
            'relevance': sum(s['relevance'] for s in quality_scores) / len(quality_scores),
            'overall': sum(s['overall'] for s in quality_scores) / len(quality_scores)
        }
        
        print(f"   质量-准确性: {avg_quality['accuracy']:.4f}")
        print(f"   质量-流畅性: {avg_quality['fluency']:.4f}")
        print(f"   质量-相关性: {avg_quality['relevance']:.4f}")
        print(f"   质量-总体: {avg_quality['overall']:.4f}")
        
        return {
            'rouge': rouge_scores,
            'exact_match': exact_match,
            'quality': avg_quality
        }

# 使用
if __name__ == "__main__":
    evaluator = FineTuningEvaluator()
    
    # 测试数据
    responses = [
        "您好，退货请联系客服申请。",
        "我们支持7天无理由退货。",
        "退款会在15个工作日内到账。"
    ]
    
    references = [
        "退货请联系客服申请退货流程。",
        "支持7天无理由退货。",
        "退款15个工作日内到账。"
    ]
    
    results = evaluator.evaluate(responses, references)
    
    print("\n📊 最终评估结果：")
    print(f"   ROUGE-L: {results['rouge']['rougeL']:.4f}")
    print(f"   质量总分: {results['quality']['overall']:.4f}")

八、实战：客服机器人微调

8.1 完整流程

"""
客服机器人微调完整实战

流程：
1. 准备训练数据
2. 数据验证
3. 创建微调任务
4. 监控训练
5. 评估效果
6. 部署使用
"""

import json
from openai import OpenAI
import os

class CustomerServiceFineTuner:
    """客服机器人微调系统"""
    
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.fine_tuned_model = None
    
    def prepare_training_data(self, qa_data: list, output_file: str):
        """准备训练数据"""
        system_prompt = """你是一个专业、友好的客服助手。
规则：
1. 礼貌问候客户
2. 清晰准确地回答问题
3. 如无法解决，引导客户联系人工
4. 不确定时诚实说明"""
        
        with open(output_file, 'w') as f:
            for qa in qa_data:
                conversation = {
                    "messages": [
                        {"role": "system", "content": system_prompt},
                        {"role": "user", "content": qa["question"]},
                        {"role": "assistant", "content": qa["answer"]}
                    ]
                }
                f.write(json.dumps(conversation, ensure_ascii=False) + '\n')
        
        print(f"✅ 生成了 {len(qa_data)} 条训练数据")
    
    def fine_tune(self, training_file: str) -> str:
        """执行微调"""
        # 1. 上传文件
        print("📤 上传训练数据...")
        with open(training_file, 'rb') as f:
            file = self.client.files.create(
                file=f,
                purpose="fine-tune"
            )
        file_id = file.id
        
        # 2. 创建任务
        print("🚀 创建微调任务...")
        job = self.client.fine_tuning.jobs.create(
            training_file=file_id,
            model="gpt-3.5-turbo",
            hyperparameters={
                "n_epochs": 3
            }
        )
        
        return job.id
    
    def check_status(self, job_id: str) -> dict:
        """检查训练状态"""
        job = self.client.fine_tuning.jobs.get(job_id)
        
        return {
            "status": job.status,
            "progress": job.progress,
            "model": job.fine_tuned_model
        }
    
    def use_model(self, user_message: str) -> str:
        """使用微调后的模型"""
        if not self.fine_tuned_model:
            return "请先完成微调"
        
        response = self.client.chat.completions.create(
            model=self.fine_tuned_model,
            messages=[
                {"role": "user", "content": user_message}
            ]
        )
        
        return response.choices[0].message.content

# 使用
if __name__ == "__main__":
    tuner = CustomerServiceFineTuner()
    
    # 1. 准备数据
    qa_data = [
        {"question": "怎么退货？", "answer": "您好！退货流程如下：\n1. 登录账号进入'我的订单'\n2. 选择需要退货的订单\n3. 点击'申请退货'并填写原因\n4. 提交后等待审核\n如有疑问随时联系我～"},
        {"question": "支持七天无理由退货吗？", "answer": "支持的！自收到商品之日起7天内，在商品完好不影响二次销售的情况下，可以申请无理由退货。"},
        {"question": "退货需要运费吗？", "answer": "关于运费：\n• 7天内非质量问题退货，买家承担寄回运费\n• 质量问题或发错货，我们承担来回运费\n• 请保留快递单据以便核实"},
        # ... 可以添加更多数据
    ]
    
    tuner.prepare_training_data(qa_data, "cs_training.jsonl")
    
    # 2. 微调（需要等待）
    # job_id = tuner.fine_tune("cs_training.jsonl")
    
    # 3. 使用
    # response = tuner.use_model("我要退货")
    # print(response)

九、避坑指南

9.1 数据相关

• 数据不足：少于100条效果不明显
• 数据偏差：某个类别过多会导致模型偏科
• 标签错误：错误答案会让模型学会错误行为
• 长度不一：悬殊的长度差异会影响训练

9.2 训练相关

• 学习率过高：Loss震荡，无法收敛
• 训练过长：过拟合，泛化能力下降
• batch太小：训练不稳定

9.3 评估相关

• 只看Loss：Loss低不等于效果好
• 只用自动评估：必须有人工评估验证
• 不看泛化：测试集表现好不等于真实场景好

十、总结与练习

10.1 要点回顾

• 何时微调：Prompt和RAG无法解决时
• 数据准备：格式正确、数量足够、质量高
• OpenAI微调：简单、托管、付费
• LoRA微调：开源、高效、低显存
• 效果评估：自动+人工多维度

10.2 选型指南

10.3 延伸阅读

• OpenAI微调文档：官方指南
• Hugging Face PEFT：PEFT库
• LoRA论文：LoRA: Low-Rank Adaptation

10.4 课后练习

基础题：准备100条问答数据，执行一次OpenAI微调。

进阶题：对比LoRA不同秩（r=4/8/16）的效果差异。

挑战题：实现一个增量微调系统，支持追加训练数据而不丢失之前学到的能力。