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 的简化替代——无需训练奖励模型，无需强化学习： ...

LoRA 原理：低秩分解 全量微调更新所有参数：ΔW 与 W 同尺寸。LoRA 的核心洞察：微调时的权重更新是低秩的——可以用两个小矩阵的乘积近似。 全量微调: W' = W + ΔW 参数量 d×k LoRA: W' = W + B×A 参数量 r×(d+k), r << min(d,k) A: 高斯初始化, B: 零初始化 7B 模型的 4096×4096 权重矩阵：全量微调 16.7M 参数，LoRA(r=8) 仅 65K 参数，减少 99.6%。 超参数选择 参数 推荐值 影响 r (rank) 8-64 越大表达能力越强 alpha 通常等于r 控制更新幅度 target_modules q_proj,v_proj / 全线性层 影响效果和显存 dropout 0.05-0.1 防止过拟合 r 的经验法则：r=8 简单任务（格式调整）；r=16 中等任务（领域适配）；r=32 复杂任务（推理能力）。 QLoRA：再省显存 QLoRA = Quantization + LoRA：基础模型量化到 4-bit 存储，仅在 LoRA 参数上用 16-bit 训练。 ...

RAG 流水线全景 一个生产级 RAG 系统远不止"Embedding + 向量检索 + LLM 生成"这么简单。完整的优化链路： 用户查询 → 查询改写 → 多路召回 → 重排序 → 上下文压缩 → LLM 生成 → 流式输出 ↑ ↓ 缓存 引用标注 每个环节都有优化空间。下面逐一拆解。 1. 分块策略 分块决定了文档被切分成什么粒度的片段，直接影响检索精度。 固定长度分块 最简单的策略，按固定 token 数切分： from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=64, separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""] ) chunks = splitter.split_text(long_document) chunk_overlap 很关键：64-128 的重叠可以避免句子被截断导致语义丢失。但重叠太大会增加存储和检索冗余。 语义分块 按语义完整性切分，而非固定长度： from langchain.text_splitter import SemanticChunker from langchain.embeddings import HuggingFaceEmbeddings embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5") splitter = SemanticChunker( embeddings, breakpoint_threshold_type="percentile", # 或 "standard_deviation" breakpoint_threshold_amount=95 ) chunks = splitter.split_text(long_document) 语义分块在文档结构复杂时效果更好，但计算成本高（每句话都要算 Embedding）。 结构感知分块 利用文档结构（Markdown 标题、HTML 标签）分块： ...

RAG 评估的特殊性 RAG 系统由检索和生成两个阶段组成。传统评估只看最终回答质量，但 RAG 的问题可能出在检索阶段——检索到的文档是错的，再好的生成模型也无法给出正确回答。因此 RAG 评估必须是全链路的。 评估维度 用户问题 → [检索器] → 检索文档 → [生成器] → 最终回答 ↑ ↑ 检索质量评估 生成质量评估 检索质量指标 指标 含义 Context Precision 检索文档中相关部分的比例 Context Recall 答案所需信息被检索到的比例 Hit Rate 至少检索到一个相关文档的比例 MRR 相关文档的排名倒数均值 生成质量指标 指标 含义 Faithfulness 回答是否忠于检索文档（无幻觉） Answer Relevance 回答与问题的相关程度 Answer Correctness 回答与标准答案的匹配度 Hallucination Rate 无法从文档推导的陈述比例 RAGAS 框架 RAGAS 是最流行的 RAG 评估框架，支持无参考评估（不需要标准答案）。 from ragas import evaluate from ragas.metrics import ( faithfulness, answer_relevancy, context_precision, context_recall, ) from datasets import Dataset eval_data = Dataset.from_dict({ "question": ["什么是量子计算？", "如何防止SQL注入？"], "answer": [ "量子计算利用量子力学原理进行计算，使用量子比特...", "防止SQL注入的方法包括参数化查询、输入验证...", ], "contexts": [ ["量子计算利用量子力学原理...", "量子比特可同时处于0和1的叠加态..."], ["SQL注入通过嵌入恶意SQL语句实现...", "参数化查询是最有效方法..."], ], "ground_truth": [ "量子计算利用量子叠加和纠缠原理进行并行计算。", "使用参数化查询、输入验证、ORM来防止SQL注入。", ], }) results = evaluate(eval_data, metrics=[ faithfulness, answer_relevancy, context_precision, context_recall, ]) # {'faithfulness': 0.85, 'answer_relevancy': 0.92, # 'context_precision': 0.78, 'context_recall': 0.88} Faithfulness 详解 衡量回答中每个陈述是否能从检索文档推导： ...

核心问题 “我应该用 Prompt 工程还是微调？“答案是：取决于瓶颈在哪。 模型理解能力不足 → 微调 上下文信息不足 → RAG / Prompt 工程 输出格式不稳定 → Prompt 工程（或少量微调） 领域知识缺失 → RAG（而非微调） Prompt 工程的能力边界 能做什么 # 1. 角色设定和行为控制 SYSTEM_PROMPT = "你是专业法律顾问，仅基于提供的条文回答" # 2. 输出格式控制 FORMAT_PROMPT = '输出JSON: {"risk_level":"high|medium|low", "factors":[]}' # 3. 少样本学习 FEW_SHOT = "示例：\n输入：订单没到\n输出：{\"intent\":\"logistics\"}\n现在处理：{input}" # 4. 思维链推理 COT = "请一步步思考：首先分析...然后...最后..." 天花板 # 解决不了的问题： # 1. 风格深度定制 — 需要大量token示例，成本高且不稳定 style_prompt = "请模仿以下风格（附10000字示例）..." # token成本爆炸 # 2. 领域术语理解 — Prompt过长，模型注意力分散 medical_prompt = "医学缩写对照表（500个）..." # 效果差 # 3. 复杂工具调用模式 — 规则太多时遵循率下降 tool_prompt = "调用工具的50条注意事项..." # 遵循率低 微调的适用场景 场景一：风格和格式定制 # 微调数据：客服风格定制 training_data = [ {"messages": [ {"role": "system", "content": "你是XX品牌客服"}, {"role": "user", "content": "你们的东西太贵了"}, {"role": "assistant", "content": "理解您的感受～我们坚持品质路线，现有新人9折优惠😊"}, ]}, # 500-1000条这样的数据 ] # 微调后：无需长Prompt，模型自然使用品牌语气 场景二：降低推理成本 # 微调前：GPT-4o + 2000 token system prompt → $0.017/请求 # 微调后：GPT-4o-mini fine-tuned → $0.0015/请求 # 成本降低91%，回本周期约0.5个月 # 蒸馏：用GPT-4生成数据训练小模型 distillation_data = [] for input_text in training_inputs: gpt4_output = await gpt4.generate(input_text, system_prompt=LONG_PROMPT) distillation_data.append({"input": input_text, "output": gpt4_output}) 成本对比 一次性成本 项目 Prompt 工程 微调 数据标注 $0 $500-$5000 开发时间 数小时-数天 1-2 周 训练计算 $0 $10-$100（LoRA） 运行时成本 每天10000次请求对比：Prompt工程(GPT-4o+2000token)月$5100 vs 微调(GPT-4o-mini-ft)月$472，回本约13天。 ...

RAG 是 Agent 最实用的知识增强方案。本文从架构设计到代码实现，完整拆解生产级 RAG 系统的每个环节。

RAG 和 Fine-tuning 不是竞争关系，而是互补关系。本文用决策树帮你选对方案。

RAG 系统的心脏是向量数据库。本文实测 5 大向量数据库，帮你选出最适合的那个。