RAG与微调

RAG 重排序指南：Cohere Rerank vs bge-reranker vs Cross-Encoder 引言 RAG 系统的标准流程是：向量检索 Top-K → 直接喂给 LLM。但向量检索（双塔模型）的弱点是精度有限：它擅长快速召回大量相关文档，但不擅长精细区分「真正相关」和「看起来相关」。 重排序（Reranking）是解决这一问题的关键环节：用一个更强大的模型对检索结果重新打分排序，将最相关的文档排到前面。 Query → 向量检索 Top-50 → 重排序 → Top-5 → LLM 生成 本文深入对比三种主流重排序方案。 1. 为什么需要重排序？ 1.1 双塔 vs 交叉编码器 特性 双塔模型（Bi-Encoder） 交叉编码器（Cross-Encoder） 架构 Query 和 Doc 独立编码 Query 和 Doc 拼接后联合编码 交互 无（仅在最后做余弦相似度） 全程（Attention 层交互） 精度 中 高 速度 极快（可预计算索引） 慢（每对 Q-D 需独立前向） 用途 初筛召回 精排 1.2 RAG 中的两阶段检索 # 两阶段检索流程 def two_stage_retrieve( query: str, vector_index, # 双塔向量索引 reranker, # 交叉编码器重排序 first_stage_k: int = 50, # 初筛数量 final_k: int = 5, # 最终数量 ) -> list: # Stage 1: 向量检索（快，召回多） candidates = vector_index.search(query, top_k=first_stage_k) # Stage 2: 重排序（精，筛少） pairs = [(query, doc["content"]) for doc in candidates] scores = reranker.predict(pairs) # 排序并取 Top-K ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in ranked[:final_k]] 1.3 效果提升 指标 无重排序 有重排序 提升 Top-1 准确率 62.3% 78.1% +15.8% Top-5 召回率 81.2% 91.5% +10.3% MRR 0.681 0.832 +0.151 2. 方案一：Cohere Rerank API 2.1 概述 Cohere Rerank 是托管的商业重排序 API，基于自家训练的 rerank 模型，支持多语言。 ...

持续预训练实践：领域大模型的训练方法论 引言 通用大模型（GPT-4、LLaMA-3）在医疗、法律、金融等垂直领域常出现专业术语理解偏差、领域知识幻觉、行业惯例缺失等问题。全量预训练成本过高，而 SFT（指令微调）又不足以注入深层领域知识。 持续预训练（Continual Pre-Training, CPT）是两者之间的平衡点：在通用基座上用领域语料继续预训练，让模型深层理解领域知识，同时保留通用能力。 1. CPT 在训练体系中的定位 预训练（PT）→ 持续预训练（CPT）→ 指令微调（SFT）→ 偏好对齐（DPO/RLHF） ↑ 本文重点 阶段 数据量 数据形式 学习目标 参数更新 PT 万亿 tokens 纯文本 Next token prediction 全量 CPT 百亿~千亿 tokens 纯文本（领域） Next token prediction 全量或部分 SFT 百万~千万条 指令-响应对 指令跟随 全量或 PEFT DPO 数万~数十万条 偏好对 偏好优化 全量或 PEFT 1.1 CPT vs SFT：本质区别 维度 CPT SFT 数据 无标注领域文本 人工标注指令对 目标 学习领域知识分布 学习指令跟随格式 深度 改变模型内部知识表征 调整输出行为 数据量 10B-100B tokens 1M-10M 条 灾难性遗忘 严重（需缓解） 较轻 2. 领域数据构建 2.1 数据来源与采集 from typing import List, Dict import json import os class DomainDataBuilder: """领域预训练数据构建器""" def __init__(self, domain: str): self.domain = domain # "medical", "legal", "financial", etc. def collect_sources(self) -> Dict[str, List[str]]: """定义数据来源""" sources = { "medical": { "textbooks": ["医学教科书 PDF", "诊疗指南"], "papers": ["PubMed 论文", "中文医学期刊"], "records": ["脱敏病历", "医学百科"], "guidelines": ["临床路径", "用药指南"], "qa": ["医学问答", "丁香园问答"], }, "legal": { "textbooks": ["法学教材", "法律释义"], "papers": ["法学期刊论文"], "records": ["裁判文书", "法律案例"], "guidelines": ["司法解释", "部门规章"], "qa": ["法律咨询问答"], }, "financial": { "textbooks": ["金融学教材", "CFA教材"], "papers": ["金融研究论文"], "reports": ["年报", "券商研报"], "guidelines": ["会计准则", "监管文件"], "qa": ["金融问答", "投资者教育"], }, } return sources.get(self.domain, {}) def estimate_tokens(self, file_sizes_mb: List[float]) -> int: """估算 token 数（中文约 1.5 字/token，英文约 0.75 词/token）""" total_chars = sum(size * 1024 * 1024 * 3 for size in file_sizes_mb) # 假设中文3字节 return int(total_chars / 1.5) # 粗略估算 2.2 数据清洗管道 import re from typing import Optional class DataCleaner: """预训练数据清洗""" @staticmethod def clean_text(text: str) -> Optional[str]: """单文档清洗""" # 1. 去除 HTML 标签 text = re.sub(r'<[^>]+>', '', text) # 2. 去除 URL text = re.sub(r'https?://\S+', '', text) # 3. 去除多余空白 text = re.sub(r'\s+', ' ', text).strip() # 4. 去除乱码（非可打印字符） text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]', '', text) # 5. 去重检测（文档级） if len(text) < 100: # 太短，丢弃 return None # 6. 质量过滤：字母/汉字占比 alpha_ratio = len(re.findall(r'[a-zA-Z\u4e00-\u9fff]', text)) / len(text) if alpha_ratio < 0.5: # 非文字内容太多 return None return text @staticmethod def deduplicate(texts: List[str], threshold: float = 0.8) -> List[str]: """MinHash 去重""" from datasketch import MinHash, MinHashLSH lsh = MinHashLSH(threshold=threshold, num_perm=128) unique_texts = [] for i, text in enumerate(texts): m = MinHash(num_perm=128) # 按 5-gram 分词 words = [text[j:j+5] for j in range(0, len(text)-4, 3)] for word in words: m.update(word.encode('utf-8')) # 检查是否重复 result = lsh.query(m) if not result: lsh.insert(str(i), m) unique_texts.append(text) return unique_texts 2.3 数据混合比例 def build_mixed_dataset( domain_data: List[str], # 领域文本 general_data: List[str], # 通用文本（如预训练数据子集） domain_ratio: float = 0.7, # 领域数据占比 ) -> List[str]: """ 混合领域数据和通用数据，防止灾难性遗忘 """ import random n_domain = int(len(domain_data) * domain_ratio / (1 - domain_ratio)) n_domain = min(n_domain, len(domain_data)) sampled_domain = random.sample(domain_data, n_domain) sampled_general = random.sample(general_data, len(domain_data) - n_domain + len(domain_data)) # 交错排列（避免连续大量同类型数据） mixed = [] domain_iter = iter(sampled_domain) general_iter = iter(sampled_general) while True: try: mixed.append(next(domain_iter)) except StopIteration: pass try: mixed.append(next(general_iter)) except StopIteration: pass if not (domain_iter or general_iter): break return mixed # 推荐混合比例 ratio_guide = { "medical": {"domain": 0.6, "general": 0.3, "code": 0.1}, # 医疗 "legal": {"domain": 0.65, "general": 0.25, "code": 0.10}, # 法律 "financial": {"domain": 0.55, "general": 0.35, "code": 0.10}, # 金融 "code": {"domain": 0.7, "general": 0.2, "code": 0.10}, # 代码 "general_domain": {"domain": 0.5, "general": 0.4, "code": 0.1}, # 通用领域 } 3. 训练策略 3.1 学习率调度 CPT 的学习率通常远低于预训练阶段，但高于 SFT： ...

高级 RAG 模式：HyDE、CRAG、Self-RAG 与 FLARE 引言 基础 RAG（检索→增强→生成）在真实场景中常遇到三大瓶颈：语义鸿沟（用户问题与文档用词不一致）、检索噪声（相关度低的片段污染上下文）、一次性检索（无法根据生成过程动态调整）。高级 RAG 模式针对这些问题提出了系统性解法。 本文深入解析四种最具代表性的高级 RAG 模式，给出可运行的 Python 实现，并在最后提供综合对比。 1. HyDE（Hypothetical Document Embeddings） 核心思想 HyDE 的核心洞察：用 LLM 先生成一个「假设答案」，再用这个假设的嵌入去检索，而非直接用用户问题检索。 原因是：假设答案在语义空间中比原始问题更接近真实文档，尤其当用户问题很短或措辞与知识库不一致时。 用户问题 → LLM 生成假设文档 → 嵌入假设文档 → 向量检索 → 真实文档 → 最终生成 算法流程 接收用户查询 q 令 LLM 生成 k 个假设文档 {d₁, d₂, ..., dₖ}（不要求事实正确，只要求语义相关） 对每个假设文档计算嵌入向量 hᵢ = embed(dᵢ) 取平均（或最大池化）得到查询嵌入 h_q = mean(h₁, ..., hₖ) 在向量库中检索 Top-K 最相似的真实文档 将真实文档作为上下文，令 LLM 生成最终答案 代码实现 import numpy as np from typing import List, Dict from openai import OpenAI client = OpenAI() def hyde_retrieve( query: str, index, # faiss/chroma 等向量索引 embed_model, # 嵌入模型 k_hypothetical: int = 3, # 假设文档数量 top_k: int = 5, # 最终检索数量 temperature: float = 0.7, ) -> List[Dict]: """ HyDE 检索：生成假设文档 → 嵌入 → 检索真实文档 """ # Step 1: 生成假设文档 hypothetical_docs = [] for _ in range(k_hypothetical): resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个知识丰富的助手。请根据用户问题，撰写一段可能回答该问题的文档片段。不需要事实完全准确，重点是覆盖相关概念和关键词。"}, {"role": "user", "content": f"问题：{query}"} ], temperature=temperature, ) hypothetical_docs.append(resp.choices[0].message.content) # Step 2: 嵌入假设文档 hyp_embeddings = embed_model.encode(hypothetical_docs) # shape: (k, dim) query_embedding = hyp_embeddings.mean(axis=0, keepdims=True) # 平均池化 # Step 3: 检索真实文档 distances, indices = index.search(query_embedding.astype(np.float32), top_k) results = [{"index": idx, "score": dist} for idx, dist in zip(indices[0], distances[0])] return results # 单文档 HyDE（k=1 的简化版） def hyde_single(query: str, index, embed_model, top_k: int = 5): """单假设文档版本，速度更快""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "写一个可能回答问题的段落，覆盖相关关键词。"}, {"role": "user", "content": query} ], temperature=0.7, ) hypo_doc = resp.choices[0].message.content q_emb = embed_model.encode([hypo_doc]) distances, indices = index.search(q_emb.astype(np.float32), top_k) return indices[0].tolist(), distances[0].tolist() 适用场景与局限 场景 适合度 说明 短查询 / 口语化问题 ⭐⭐⭐⭐⭐ 假设文档补充了语义上下文 专业术语与用户措辞差异大 ⭐⭐⭐⭐⭐ 假设文档桥接了术语鸿沟 多跳推理问题 ⭐⭐⭐⭐ 假设文档可覆盖多个推理步骤 低延迟要求 ⭐⭐ 需要额外一次 LLM 调用 事实敏感场景（医疗/法律） ⭐⭐⭐ 假设文档可能引入错误事实，需后处理 2. CRAG（Corrective RAG） 核心思想 CRAG 引入检索质量评估器，在检索后对文档相关性进行打分，动态决定： ...

为什么需要对齐 大模型预训练后学会了"续写文本"，但不会"按要求回答"。对齐（Alignment）让模型学会按人类偏好行事——有帮助、诚实、无害。 预训练模型："如何制作蛋糕？" → 续写："如何制作饼干？"（续写模式） 对齐后模型："如何制作蛋糕？" → 回答："制作蛋糕的步骤如下：1..."（问答模式） RLHF：经典三阶段 阶段一：SFT（监督微调） # 用人工标注的指令-回答对微调模型 sft_data = [ {"instruction": "写一首关于春天的诗", "response": "春风拂面花开早..."}, {"instruction": "解释什么是量子纠缠", "response": "量子纠缠是指..."}, ] # 标准的下一个 Token 预测训练 def sft_loss(model, batch): input_ids = batch["input_ids"] # instruction + response labels = batch["labels"] # 只对 response 部分计算 loss outputs = model(input_ids, labels=labels) return outputs.loss 阶段二：奖励模型（RM） 训练一个模型来预测人类偏好分数： class RewardModel(nn.Module): def __init__(self, base_model): super().__init__() self.backbone = base_model # 通常用 SFT 模型初始化 self.reward_head = nn.Linear(hidden_size, 1) def forward(self, input_ids): hidden = self.backbone(input_ids).last_hidden_state[:, -1, :] reward = self.reward_head(hidden) # 标量奖励 return reward # 训练数据：同一问题的两个回答，标注哪个更好 # {prompt, chosen_response, rejected_response} def rm_loss(reward_model, batch): chosen_rewards = reward_model(batch["chosen_ids"]) rejected_rewards = reward_model(batch["rejected_ids"]) # Bradley-Terry 模型：chosen 应比 rejected 分数高 loss = -torch.log(torch.sigmoid(chosen_rewards - rejected_rewards)) return loss.mean() 阶段三：PPO（近端策略优化） 用奖励模型的分数作为信号，通过 RL 优化 SFT 模型： ...

GraphRAG 是什么 GraphRAG 是微软研究院于 2024 年提出的 RAG 增强架构。传统 Vector RAG 基于向量相似度检索文档片段，难以回答需要跨文档推理、全局性总结的问题。GraphRAG 通过构建知识图谱，利用实体关系和社区结构来增强检索和生成能力。 核心问题：Vector RAG 擅长"找相关文档"，但不擅长"综合多文档的全局信息"。 问题示例： Vector RAG 擅长："公司2024年Q3营收是多少？"（单文档事实） Vector RAG 失败："总结整个行业的主要参与者及其关系"（全局综合） GraphRAG 擅长：以上两者都能处理 架构全貌 ┌──────────────────────────────────────────────────────┐ │ GraphRAG Pipeline │ │ │ │ ┌─────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ 文档分块 │──▶│ 实体抽取 │──▶│ 关系图构建 │ │ │ └─────────┘ └──────────┘ └────────┬─────────┘ │ │ │ │ │ ┌─────────▼──────────┐ │ │ │ 社区检测 (Leiden) │ │ │ └─────────┬──────────┘ │ │ │ │ │ ┌────────────────┼────────────┐ │ │ ▼ ▼ │ │ │ ┌──────────────┐ ┌────────────────┐ │ │ │ │ 社区摘要生成 │ │ 向量索引 │ │ │ │ └──────┬───────┘ └───────┬────────┘ │ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────┐ ┌────────────────┐ │ │ │ │ Global Search│ │ Local Search │ │ │ │ └──────────────┘ └────────────────┘ │ │ └──────────────────────────────────────────────────────┘ 核心步骤详解 1. 文档分块 与标准 RAG 一样，先对源文档分块： ...

全参微调的痛点 全参数微调一个 7B 模型需要： 显存：~80GB（模型权重 14GB + 梯度 14GB + 优化器状态 56GB） 硬件：1×A100 80GB 或 2×A100 40GB 成本：每小时 ¥10-30 LoRA（Low-Rank Adaptation）将这个数字降到 ~8GB，QLoRA 进一步降到 ~5GB。 LoRA 原理：低秩分解 核心数学 LoRA 假设模型微调时的权重更新 ΔW 是低秩的。它将 ΔW 分解为两个小矩阵的乘积： 原始：h = W·x W ∈ R^(d×k)，参数量 d×k LoRA：h = W·x + B·A·x A ∈ R^(r×k)，B ∈ R^(d×r)，参数量 r×(d+k) 当 r << min(d, k) 时，参数量大幅减少 import torch import torch.nn as nn class LoRALayer(nn.Module): def __init__(self, original_layer, rank=8, alpha=16): super().__init__() self.original = original_layer # 冻结的原始权重 self.rank = rank self.alpha = alpha self.scaling = alpha / rank d_out, d_in = original_layer.weight.shape # 低秩矩阵 A 和 B self.lora_A = nn.Parameter(torch.zeros(rank, d_in)) self.lora_B = nn.Parameter(torch.zeros(d_out, rank)) # A 用 Kaiming 初始化，B 用零初始化 nn.init.kaiming_uniform_(self.lora_A, a=5**0.5) # B 初始为 0，所以训练开始时 ΔW = 0，不改变原模型行为 # 冻结原始权重 for param in self.original.parameters(): param.requires_grad = False def forward(self, x): original_output = self.original(x) lora_output = (x @ self.lora_A.T) @ self.lora_B * self.scaling return original_output + lora_output 参数量对比 以 7B 模型为例（隐藏层 4096）： ...

核心差异：外部知识 vs 内化知识 RAG 和微调本质上是两种不同的知识注入方式。RAG 通过检索外部知识库为模型提供上下文，模型在推理时"查阅资料"回答问题；微调则通过梯度更新将知识"内化"到模型参数中。 # RAG：推理时检索 def rag_inference(query): docs = vector_store.search(query, k=5) prompt = f"基于以下文档回答：\n{docs}\n\n问题：{query}" return llm.generate(prompt) # 微调：知识已内化在参数中 def finetuned_inference(query): return finetuned_llm.generate(query) # 无需外部检索 RAG 的核心优势 1. 实时性 RAG 的知识库可以实时更新。当文档变更时，只需更新向量数据库，无需重新训练模型： # 知识更新：只需重新嵌入文档 new_doc = load_document("updated_policy.pdf") embedding = embed_model.encode(new_doc) vector_store.upsert(embedding, metadata={"source": "policy_v2"}) # 微调更新知识：需要重新训练 # 数据准备 → 训练 → 评估 → 部署（数小时到数天） 2. 可追溯性 RAG 的每个回答都能追溯到具体文档来源，这对企业合规和审计至关重要： response = rag_pipeline.query("公司的退款政策是什么？") print(response.answer) # "退款政策规定..." print(response.sources) # [{"doc": "policy.pdf", "page": 3, "score": 0.92}] 3. 低成本 维度 RAG 微调 初始成本 向量数据库 + 嵌入模型 GPU 训练 + 数据标注 更新成本 重新嵌入变更文档 重新训练（全量或增量） 推理成本 增加检索延迟 + Token 消耗 与基座模型持平 人力成本 文档维护 数据工程 + 训练调优 微调的核心优势 1. 风格与格式控制 微调能改变模型的输出风格、语气和格式，RAG 做不到这点： ...

为什么分块决定 RAG 质量 分块（Chunking）是 RAG 管道中被低估的环节。分块质量直接决定三件事： 检索精度：块太大，检索引入噪声；块太小，丢失上下文 生成质量：LLM 收到的上下文是否完整连贯 Token 成本：块大小直接影响每次推理的 Token 消耗 文档 → [分块] → 嵌入 → 向量库 → 检索 → 生成 ↑ 这一步决定了后面所有环节的上限 策略一：固定大小分块 原理 按固定 Token/字符数切割，最简单但最粗暴。 from langchain.text_splitter import CharacterTextSplitter def fixed_chunk(text, chunk_size=500, chunk_overlap=50): splitter = CharacterTextSplitter( separator="\n", chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len ) return splitter.split_text(text) 参数调优 # 不同场景的推荐参数 configs = { "faq_short": {"chunk_size": 200, "overlap": 0}, # FAQ：小块无重叠 "article": {"chunk_size": 500, "overlap": 50}, # 文章：中等块 "technical_doc": {"chunk_size": 1000, "overlap": 100}, # 技术文档：大块保完整 "code": {"chunk_size": 800, "overlap": 50}, # 代码：按函数边界 } 优缺点 优点 缺点 实现简单 可能切断句子 速度最快 语义不连贯 可预测成本 跨段落信息丢失 适用场景：快速原型、格式统一的短文本、日志分析 策略二：语义分块 Sentence 分块 按句子边界切割，保证每个块是完整句子： import nltk from nltk.tokenize import sent_tokenize def sentence_chunk(text, sentences_per_chunk=3): sentences = sent_tokenize(text) chunks = [] for i in range(0, len(sentences), sentences_per_chunk): chunk = " ".join(sentences[i:i+sentences_per_chunk]) chunks.append(chunk) return chunks 语义相似度分块 更高级的方法：计算相邻句子的嵌入相似度，在语义断裂处分块： ...

两种范式 随着模型上下文窗口从 4K 扩展到 1M+ token，出现了一个根本性问题： RAG 方案：检索相关片段，只放入少量上下文 长上下文方案：把所有文档直接塞进上下文窗口 两种方案各有优劣，选择哪种取决于具体场景。 长上下文模型现状 模型 上下文长度 约等于 定价倍数(vs 4K) GPT-4o 128K ~300页文档 2x Claude 3.5 Sonnet 200K ~500页文档 2x Gemini 1.5 Pro 1M ~2500页文档 1.5x Gemini 1.5 Flash 1M ~2500页文档 0.5x GPT-4 Turbo 128K ~300页文档 2x RAG 优势 1. 成本优势 场景: 10万页文档库，每次查询需要其中 5 页 RAG: 检索 5 页 (~2K token) → 输入 2K token → 成本: $0.006 长上下文: 全部塞入 (~250K token) → 输入 250K token → 成本: $0.75 RAG 成本仅为长上下文的 1/125。 ...

对齐技术全景 预训练模型 (Base) ↓ SFT (Supervised Fine-Tuning) ← 监督微调，学习指令遵循 ↓ DPO / RLHF ← 偏好对齐，学习人类价值观 ↓ 对齐模型 (Chat/Instruct) 阶段 目标 数据 方法 SFT 学会跟随指令 (指令, 回答) 对 监督学习 RLHF 对齐人类偏好 (prompt, chosen, rejected) 强化学习 DPO 对齐人类偏好 同上 直接优化，无需 RL SFT：监督微调 数据格式 Alpaca 格式： { "instruction": "将以下句子翻译为英文", "input": "今天天气真好", "output": "The weather is really nice today." } ShareGPT 格式（多轮对话）： { "conversations": [ {"from": "human", "value": "你好"}, {"from": "gpt", "value": "你好！有什么可以帮你的？"}, {"from": "human", "value": "解释一下量子计算"}, {"from": "gpt", "value": "量子计算利用量子力学原理..."} ] } 数据构造 import json import random class SFTDataBuilder: def __init__(self, llm_call): self.llm = llm_call def generate_alpaca_style(self, seed_topics: list[str], n_samples: int = 1000): """用 LLM 自动生成 SFT 数据""" dataset = [] for topic in seed_topics: prompt = f"""生成 10 条多样化的指令-回答对。 主题：{topic} 要求： 1. 指令类型多样（问答、翻译、摘要、代码、推理等） 2. 难度从简单到复杂 3. 回答准确、自然 输出 JSON 数组格式： [{{"instruction": "", "input": "", "output": ""}}]""" result = self.llm(prompt) try: items = json.loads(result) dataset.extend(items) except json.JSONDecodeError: continue # 去重 + 打乱 seen = set() unique = [] for item in dataset: key = item["instruction"] + item.get("input", "") if key not in seen: seen.add(key) unique.append(item) random.shuffle(unique) return unique[:n_samples] def to_sharegpt(self, alpaca_data: list[dict]) -> list[dict]: """Alpaca → ShareGPT 格式转换""" converted = [] for item in alpaca_data: instruction = item["instruction"] if item.get("input"): instruction = f"{instruction}\n\n输入: {item['input']}" converted.append({ "conversations": [ {"from": "human", "value": instruction}, {"from": "gpt", "value": item["output"]}, ] }) return converted 训练配置 from transformers import ( AutoModelForCausalLM, AutoTokenizer, TrainingArguments, Trainer, ) from datasets import Dataset from peft import LoraConfig, get_peft_model, TaskType import torch class SFTTrainer: def __init__(self, model_path: str, use_lora: bool = True): self.tokenizer = AutoTokenizer.from_pretrained(model_path) self.tokenizer.pad_token = self.tokenizer.eos_token self.model = AutoModelForCausalLM.from_pretrained( model_path, torch_dtype=torch.bfloat16, device_map="auto", ) if use_lora: self._setup_lora() def _setup_lora(self): """配置 LoRA：只训练少量参数""" lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=16, # LoRA rank lora_alpha=32, # 缩放因子 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj", ], ) self.model = get_peft_model(self.model, lora_config) self.model.print_trainable_parameters() # 输出: trainable params: 13M || all params: 7B || trainable%: 0.19% def format_prompt(self, instruction: str, input_text: str = "", output: str = "") -> str: """格式化为 ChatML 模板""" user_msg = instruction if input_text: user_msg += f"\n\n{input_text}" prompt = f"<|im_start|>user\n{user_msg}<|im_end|>\n" if output: prompt += f"<|im_start|>assistant\n{output}<|im_end|>" return prompt def train(self, data: list[dict], learning_rate: float = 2e-5, num_epochs: int = 3, batch_size: int = 4, max_length: int = 1024): # 格式化 texts = [ self.format_prompt( d["instruction"], d.get("input", ""), d["output"] ) for d in data ] dataset = Dataset.from_dict({"text": texts}) def tokenize(examples): tokenized = self.tokenizer( examples["text"], truncation=True, max_length=max_length, padding="max_length", ) tokenized["labels"] = tokenized["input_ids"].copy() return tokenized dataset = dataset.map(tokenize, batched=True) training_args = TrainingArguments( output_dir="./sft_output", num_train_epochs=num_epochs, per_device_train_batch_size=batch_size, gradient_accumulation_steps=4, learning_rate=learning_rate, warmup_ratio=0.03, lr_scheduler_type="cosine", logging_steps=10, save_strategy="epoch", bf16=True, optim="adamw_torch", ) trainer = Trainer( model=self.model, args=training_args, train_dataset=dataset, ) trainer.train() LoRA 超参数选择 参数 推荐值 说明 r (rank) 8-64 越大能力越强，但过拟合风险 lora_alpha 2×r 经验值，平衡缩放 target_modules all linear 全覆盖效果最好 dropout 0.05 防过拟合 learning_rate 1e-4 ~ 5e-5 LoRA 可用较大学习率 DPO：直接偏好优化 DPO (Rafailov et al., 2023) 是 RLHF 的简化替代——无需训练奖励模型，无需强化学习： ...