第二章：Embedding 管线与召回质量保障

开篇语

Embedding 管线是 RAG 系统的"语义理解入口"——文本进去，向量出来，质量直接决定下游召回的生死。这一章，我们不谈"跑通 Demo"，而是深入生产线：如何选模型、如何微调、如何量化召回质量、如何让向量索引在百万级更新下仍保持一致性。

---

2.1 Embedding 模型选型与微调：何时用通用模型，何时必须微调

2.1.1 通用模型的能力边界

核心矛盾：通用模型在海量数据上训练，覆盖广，但垂直领域语义理解差。

典型失败案例：

场景	通用模型表现	原因
法律合同检索	召回率 < 60%	"当事人"被理解为普通人，而非法律主体
医疗问诊	混淆"症状"和"疾病"	医学术语向量空间分布不均
金融研报	无法区分"买入"和"增持"	专业术语粒度不够细

决策树：

2.1.2 微调的触发条件（量化指标）

不要"感觉效果不好"就微调，用数据说话：

必须微调的硬指标：

1. 召回率（Recall@10）< 70%

   • 测算方法：人工标注 200 条查询-相关文档对，计算召回率
   • 例子：200 条查询，前 10 个结果中平均只有 140 个相关文档 → Recall@10 = 70%

2. MRR < 0.5

   • 第一个相关结果平均排名 > 2
   • 用户需要翻页才能找到答案，体验差

3. 领域术语相似度误差 > 30%

   • 例子："期货合约"和"期权合约"的余弦相似度 > 0.8（应该 < 0.5）

不需要微调的情况：

• 通用场景（客服 FAQ、新闻检索）
• 数据量太少（< 1000 条，微调容易过拟合）
• 预算有限（GPU 成本高，ROI 不合算）

2.1.3 数学原理：为什么微调能提升召回率？

对比学习的目标函数（详细版）：

$${\mathcal{L}}_{\text{contrastive}}=-\log \frac{\exp (\text{sim}(x_i,x_i^{+})/\tau )}{{\sum }_{j=1}^N\exp (\text{sim}(x_i,x_{i,j})/\tau )}$$

其中：

• $x_i$：查询文本
• $x_i^{+}$：相关文档（正样本）
• $x_{i,j}$：不相关文档（负样本）
• $\tau$：温度参数（通常 0.05-0.1）

微调的本质：
通过调整模型参数 $\theta$，让 $\text{sim}(x_i,x_i^{+})$ 变大，$\text{sim}(x_i,x_{i,j})$ 变小。

关键洞察：

• 通用模型在通用数据上优化，对垂直领域数据不敏感
• 微调在垂直数据上继续优化，让模型"记住"领域知识

2.1.4 实际工作中的 Gotchas

Gotcha 1：盲目微调导致通用能力退化

现象：微调法律模型后，通用问答效果下降 20%。

原因：灾难性遗忘（Catastrophic Forgetting）—— 微调时只给领域数据，模型忘了通用知识。

解决方案：

• 混合训练：70% 领域数据 + 30% 通用数据
• 小学习率：2e-5 或更低
• LoRA 微调：只训练 0.1% 的参数，保留大部分预训练权重

Gotcha 2：负样本采样不当

现象：微调后，模型把所有文档都判定为"相关"。

原因：负样本太简单（随机采样的文档显然不相关），模型学不到细粒度区分。

解决方案：

• 困难负样本挖掘（Hard Negative Mining）：用当前模型检索，取排名 10-100 的结果作为负样本
• 对抗性负样本：人工构造"看似相关但实际不相关"的样本

Gotcha 3：忽略模型的最大序列长度

现象：用 512 token 的模型编码长文档，结果后面的内容被截断。

原因：超出长度的 token 被直接丢弃。

解决方案：

• 换长文本模型（bge-m3 支持 8192 token）
• 切片编码（Chunking）：将长文档切成 512 token 的片段
• 摘要 + 全文编码：用 LLM 生成摘要，分别编码

---

2.2 实战：用 LoRA 微调 Embedding 模型提升垂直领域召回率

2.2.1 为什么选 LoRA？

全参数微调的问题：

• 模型太大（bge-large-zh 有 326M 参数）
• 显存需求高（A100 80GB 才能微调）
• 训练慢（10 万条数据需要 3 天）

LoRA 的优势：

• 只训练低秩矩阵（0.1% 的参数）
• 显存需求降低 90%（A10 就能微调）
• 训练速度提升 3-5 倍
• 可以随时切换回原始模型（不破坏预训练权重）

LoRA 原理（简化版）：

在原始权重矩阵 $W$ 旁边加两个小矩阵 $A$ 和 $B$：

$$W^{\prime }=W+\Delta W=W+BA$$

其中：

• $W\in {\mathbb{R}}^{d\times d}$
• $B\in {\mathbb{R}}^{d\times r}$
• $A\in {\mathbb{R}}^{r\times d}$
• $r\ll d$（秩，通常 8-64）

训练时只更新 $A$ 和 $B$，冻结 $W$。

2.2.2 数据准备（法律领域示例）

数据格式（每行一个 JSON）：

{
  "query": "劳动合同到期不续签怎么赔偿",
  "positive": "根据《劳动合同法》第四十六条，劳动合同期满终止的，用人单位应当支付经济补偿...",
  "negative": "劳动合同签订时应当注意以下事项..."
}

数据量建议：

• 最少：3000 条（1000 条训练 + 1000 条验证 + 1000 条测试）
• 推荐：10000 条
• 上限：50000 条（再多边际收益递减）

数据质量检查清单：

• 每条 query 确实有对应的 positive 文档
• negative 文档确实不相关（不能只是随机采样）
• 覆盖主要业务场景（不能只有一种问题类型）
• 人工抽查 200 条，准确率 > 95%

2.2.3 LoRA 微调代码实现

from typing import List, Optional, Dict, Any, Tuple
from pydantic import BaseModel, Field, validator
from sentence_transformers import SentenceTransformer, InputExample, losses
from sentence_transformers.trainer import SentenceTransformerTrainer
from sentence_transformers.training_args import SentenceTransformerTrainingArguments
from peft import LoraConfig, get_peft_model, TaskType
import torch
from torch.utils.data import DataLoader
import json
from pathlib import Path
import logging
from dataclasses import dataclass
from datetime import datetime

logger = logging.getLogger(__name__)

# ==================== 配置模型 ====================

class LoRAFineTuningConfig(BaseModel):
    """LoRA 微调配置"""
    model_name: str = Field(..., description="基础模型名称")
    train_data_path: str = Field(..., description="训练数据路径")
    eval_data_path: Optional[str] = Field(None, description="评估数据路径")
    output_dir: str = Field(..., description="输出目录")
    
    # LoRA 参数
    lora_rank: int = Field(8, description="LoRA 秩（越小参数越少，越大表达能力越强）")
    lora_alpha: int = Field(32, description="LoRA 缩放因子（通常 2*rank 或 4*rank）")
    lora_dropout: float = Field(0.1, description="LoRA dropout（防止过拟合）")
    
    # 训练参数
    num_epochs: int = Field(3, description="训练轮数")
    batch_size: int = Field(16, description="批大小（LoRA 可以用更大的批）")
    learning_rate: float = Field(3e-4, description="学习率（LoRA 可以用更大的 LR）")
    warmup_ratio: float = Field(0.1, description="预热比例")
    max_seq_length: int = Field(512, description="最大序列长度")
    
    # 数据参数
    use_hard_negatives: bool = Field(True, description="是否使用困难负样本")

class LoRAFineTuningService:
    """LoRA 微调服务"""
    
    def __init__(self, config: LoRAFineTuningConfig):
        self.config = config
        
        # 加载基础模型
        logger.info(f"加载基础模型: {config.model_name}")
        self.model = SentenceTransformer(config.model_name)
        self.model.max_seq_length = config.max_seq_length
        
        # 应用 LoRA
        self._apply_lora()
        
    def _apply_lora(self):
        """应用 LoRA 到模型"""
        # 定义 LoRA 配置
        lora_config = LoraConfig(
            task_type=TaskType.FEATURE_EXTRACTION,  # Embedding 模型用这个任务类型
            r=self.config.lora_rank,  # 秩
            lora_alpha=self.config.lora_alpha,  # 缩放因子
            lora_dropout=self.config.lora_dropout,  # dropout
            target_modules=["query", "value"],  # 对哪些模块应用 LoRA（Transformer 的 Q/V 矩阵）
        )
        
        # 应用 LoRA
        self.model = get_peft_model(self.model, lora_config)
        
        # 打印可训练参数占比
        trainable_params = sum(p.numel() for p in self.model.parameters() if p.requires_grad)
        total_params = sum(p.numel() for p in self.model.parameters())
        logger.info(f"可训练参数: {trainable_params} ({trainable_params/total_params*100:.2f}%)")
        
    def load_training_data(self, data_path: str) -> List[InputExample]:
        """加载训练数据"""
        examples = []
        data_file = Path(data_path)
        
        if not data_file.exists():
            raise FileNotFoundError(f"训练数据文件不存在: {data_path}")
        
        with open(data_file, 'r', encoding='utf-8') as f:
            for line in f:
                if not line.strip():
                    continue
                data = json.loads(line)
                query = data['query']
                positive = data['positive']
                
                # 构建训练样本
                if 'negative' in data and data['negative']:
                    # 有三元组的情况
                    examples.append(InputExample(
                        texts=[query, positive, data['negative']]
                    ))
                else:
                    # 只有查询-正样本的情况
                    examples.append(InputExample(
                        texts=[query, positive]
                    ))
        
        logger.info(f"加载了 {len(examples)} 条训练样本")
        return examples
    
    def mine_hard_negatives(self, train_examples: List[InputExample]) -> List[InputExample]:
        """挖掘困难负样本"""
        if not self.config.use_hard_negatives:
            return train_examples
        
        logger.info("正在挖掘困难负样本...")
        
        # 用当前模型编码所有查询
        queries = [ex.texts[0] for ex in train_examples]
        query_embeddings = self.model.encode(queries, convert_to_tensor=True)
        
        # 用当前模型编码所有正样本
        positives = [ex.texts[1] for ex in train_examples]
        positive_embeddings = self.model.encode(positives, convert_to_tensor=True)
        
        # 计算相似度矩阵
        similarity_matrix = torch.matmul(query_embeddings, positive_embeddings.T)
        
        # 对每个查询，取相似度排名 10-100 的作为困难负样本
        hard_negatives_examples = []
        for i, ex in enumerate(train_examples):
            # 取第 i 行（查询 i 与所有正样本的相似度）
            similarities = similarity_matrix[i]
            
            # 排序（降序）
            sorted_indices = torch.argsort(similarities, descending=True)
            
            # 取排名 10-100 的作为困难负样本
            hard_indices = sorted_indices[10:100]
            
            if len(hard_indices) > 0:
                # 随机选一个困难负样本
                hard_idx = hard_indices[torch.randperm(len(hard_indices))[0]]
                hard_negative = positives[hard_idx]
                
                # 替换原来的负样本（如果有）
                if len(ex.texts) > 2:
                    ex.texts[2] = hard_negative
                else:
                    ex.texts.append(hard_negative)
            
            hard_negatives_examples.append(ex)
        
        logger.info(f"挖掘了 {len(hard_negatives_examples)} 条困难负样本")
        return hard_negatives_examples
    
    def create_trainer(self, train_examples: List[InputExample], 
                       eval_examples: Optional[List[InputExample]] = None):
        """创建训练器"""
        
        # 选择损失函数（对比学习损失，适合检索任务）
        train_loss = losses.MultipleNegativesRankingLoss(self.model)
        
        # 训练参数
        training_args = SentenceTransformerTrainingArguments(
            output_dir=self.config.output_dir,
            num_train_epochs=self.config.num_epochs,
            per_device_train_batch_size=self.config.batch_size,
            learning_rate=self.config.learning_rate,
            warmup_ratio=self.config.warmup_ratio,
            fp16=torch.cuda.is_available(),  # 混合精度训练
            save_steps=1000,
            save_total_limit=3,
            logging_steps=100,
            evaluation_strategy="steps" if eval_examples else "no",
            eval_steps=500 if eval_examples else None,
            save_safetensors=True,  # 保存为 safetensors 格式（更安全）
        )
        
        # 创建训练器
        trainer = SentenceTransformerTrainer(
            model=self.model,
            args=training_args,
            train_dataset=train_examples,
            eval_dataset=eval_examples,
            loss=train_loss,
        )
        
        return trainer
    
    async def fine_tune(self) -> Dict[str, Any]:
        """执行 LoRA 微调"""
        logger.info(f"开始 LoRA 微调: {self.config.model_name}")
        
        # 加载数据
        train_examples = self.load_training_data(self.config.train_data_path)
        eval_examples = None
        if self.config.eval_data_path:
            eval_examples = self.load_training_data(self.config.eval_data_path)
        
        # 挖掘困难负样本（可选）
        if self.config.use_hard_negatives:
            train_examples = self.mine_hard_negatives(train_examples)
        
        # 创建训练器
        trainer = self.create_trainer(train_examples, eval_examples)
        
        # 执行训练
        trainer.train()
        
        # 保存 LoRA 权重（只保存 LoRA 参数，不保存整个模型）
        self.model.save_pretrained(self.config.output_dir)
        
        # 也保存完整模型（方便部署）
        merged_model = self.model.merge_and_unload()  # 合并 LoRA 权重到原始模型
        merged_model.save(self.config.output_dir + "_merged")
        
        logger.info(f"LoRA 权重已保存到: {self.config.output_dir}")
        logger.info(f"合并后的模型已保存到: {self.config.output_dir}_merged")
        
        # 返回训练信息
        return {
            "lora_weights_path": self.config.output_dir,
            "merged_model_path": self.config.output_dir + "_merged",
            "base_model": self.config.model_name,
            "train_samples": len(train_examples),
            "lora_rank": self.config.lora_rank,
            "trainable_params_ratio": self._get_trainable_params_ratio()
        }
    
    def _get_trainable_params_ratio(self) -> float:
        """计算可训练参数占比"""
        trainable_params = sum(p.numel() for p in self.model.parameters() if p.requires_grad)
        total_params = sum(p.numel() for p in self.model.parameters())
        return trainable_params / total_params

# ==================== 使用示例 ====================

async def main():
    config = LoRAFineTuningConfig(
        model_name="BAAI/bge-large-zh-v1.5",
        train_data_path="./data/legal_train.jsonl",
        eval_data_path="./data/legal_eval.jsonl",
        output_dir="./models/bge-large-zh-legal-lora",
        lora_rank=16,  # 秩 16，平衡参数数量和表达能力
        lora_alpha=32,  # 缩放因子 32
        num_epochs=3,
        batch_size=16,
        learning_rate=3e-4,  # LoRA 可以用更大的学习率
        use_hard_negatives=True
    )
    
    service = LoRAFineTuningService(config)
    result = await service.fine_tune()
    print(f"LoRA 微调完成: {result}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

代码关键点：

1. LoRA 配置：

   • r=16：秩 16，只训练 0.1% 的参数
   • lora_alpha=32：缩放因子，通常设为 2*r 或 4*r
   • target_modules=["query", "value"]：对 Transformer 的 Q/V 矩阵应用 LoRA

2. 困难负样本挖掘：

   • 用当前模型检索，取排名 10-100 的结果作为负样本
   • 比随机采样难，能提升模型细粒度区分能力

3. 保存策略：

   • 保存 LoRA 权重（小文件，方便版本管理）
   • 保存合并后的模型（方便部署，不需要 PEFT 库）

2.2.4 微调效果评估

评估指标：

from sentence_transformers import util
import torch

async def evaluate_fine_tuned_model(model_path: str, test_data_path: str):
    """评估微调后的模型"""
    
    # 加载微调后的模型
    model = SentenceTransformer(model_path)
    
    # 加载测试数据
    queries = []
    positives = []
    with open(test_data_path, 'r', encoding='utf-8') as f:
        for line in f:
            data = json.loads(line)
            queries.append(data['query'])
            positives.append(data['positive'])
    
    # 编码
    query_embeddings = model.encode(queries, convert_to_tensor=True)
    doc_embeddings = model.encode(positives, convert_to_tensor=True)
    
    # 计算余弦相似度
    cos_scores = util.cos_sim(query_embeddings, doc_embeddings)
    
    # 计算 Recall@10
    recall_at_10 = 0
    for i, scores in enumerate(cos_scores):
        sorted_indices = torch.argsort(scores, descending=True)
        if i in sorted_indices[:10]:  # 检查正确答案是否在前 10
            recall_at_10 += 1
    
    recall_at_10 /= len(queries)
    
    # 计算 MRR
    mrr = 0
    for i, scores in enumerate(cos_scores):
        sorted_indices = torch.argsort(scores, descending=True)
        rank = (sorted_indices == i).nonzero(as_tuple=True)[0].item()
        mrr += 1.0 / (rank + 1)
    
    mrr /= len(queries)
    
    print(f"Recall@10: {recall_at_10:.4f}")
    print(f"MRR: {mrr:.4f}")
    
    return {"recall_at_10": recall_at_10, "mrr": mrr}

真实案例效果：

模型	Recall@10	MRR	训练时间	GPU 成本
通用 bge-large-zh	0.65	0.42	-	-
LoRA 微调后	0.89	0.71	6 小时	¥120（A10）
全参数微调	0.91	0.73	3 天	¥3600（A100）

结论：LoRA 微调效果接近全参数微调，但成本低 30 倍。

2.2.5 实际工作中的 Gotchas

Gotcha 1：LoRA 秩选择不当

现象：秩太小（r=4），微调后效果提升不明显；秩太大（r=64），显存溢出。

解决方案：

• 从 r=16 开始实验
• 监控验证集损失，不再下降时停止
• 显存不够时减小 r 或增大 lora_alpha

Gotcha 2：忘记合并 LoRA 权重

现象：部署时发现需要加载 PEFT 库，但生产环境不支持。

解决方案：

• 部署前用 model.merge_and_unload() 合并权重
• 保存合并后的模型（不需要 PEFT 库就能加载）

Gotcha 3：微调后模型在某些查询上效果变差

现象：大部分查询效果提升，但少数查询效果下降。

原因：训练数据覆盖不均，模型在某些子领域过拟合。

解决方案：

• 在训练集中增加多样性（覆盖所有子领域）
• 用 A/B 测试监控各子领域的效果

---

2.3 召回质量评估体系搭建：从人工标注到自动化回归测试

2.3.1 为什么需要召回质量评估？

问题：你上线了一个 RAG 系统，用户抱怨"搜不到想要的结果"，但你不知道是模型问题、数据问题还是检索策略问题。

解决思路：建立量化的召回质量评估体系，持续监控和改进。

2.3.2 评估体系架构

2.3.3 人工标注规范

标注工具推荐：

• LabelStudio（开源，支持自定义标注流程）
• Prodigy（付费，AI 辅助标注）

标注任务设计：

1. 查询-文档相关性标注（3 档标注）：

   • 相关（2 分）：文档能完整回答查询
   • 部分相关（1 分）：文档包含部分答案
   • 不相关（0 分）：文档与查询无关

2. 困难样本标注：

   • 对模型检索错误的结果，人工标注"为什么错了"
   • 常见原因：术语理解错误、上下文缺失、文档切片不当

标注质量控制：

• 每个样本由 3 个人标注，取多数投票
• 计算标注一致性（Cohen’s Kappa），目标 > 0.7
• 定期抽查，发现标注质量下降时重新培训标注员

2.3.4 自动化评估流水线实现

from typing import List, Dict, Any, Optional, Tuple
from pydantic import BaseModel, Field
from enum import Enum
import json
from pathlib import Path
import logging
from dataclasses import dataclass
from datetime import datetime
import asyncio

logger = logging.getLogger(__name__)

# ==================== 数据模型 ====================

class RelevanceLabel(str, Enum):
    """相关性标注"""
    RELEVANT = "relevant"  # 相关（2 分）
    PARTIALLY_RELEVANT = "partially_relevant"  # 部分相关（1 分）
    NOT_RELEVANT = "not_relevant"  # 不相关（0 分）

class TestSample(BaseModel):
    """测试样本"""
    query: str = Field(..., description="查询文本")
    relevant_docs: List[str] = Field(..., description="相关文档 ID 列表")
    partially_relevant_docs: List[str] = Field(default_factory=list, description="部分相关文档 ID 列表")
    metadata: Dict[str, Any] = Field(default_factory=dict, description="元数据（场景、难度等）")

class RetrievalResult(BaseModel):
    """检索结果"""
    query: str = Field(..., description="查询文本")
    retrieved_docs: List[Tuple[str, float]] = Field(..., description="检索到的文档 ID 和相似度分数")
    latency_ms: int = Field(..., description="检索耗时（毫秒）")

class EvaluationReport(BaseModel):
    """评估报告"""
    test_set_name: str = Field(..., description="测试集名称")
    model_name: str = Field(..., description="模型名称")
    timestamp: datetime = Field(default_factory=datetime.now, description="评估时间")
    
    # 核心指标
    recall_at_10: float = Field(..., description="Recall@10")
    recall_at_50: float = Field(..., description="Recall@50")
    mrr: float = Field(..., description="MRR")
    ndcg_at_10: float = Field(..., description="NDCG@10")
    
    # 分场景指标
    scenario_metrics: Dict[str, Dict[str, float]] = Field(default_factory=dict, description="分场景指标")
    
    # 失败案例分析
    failure_cases: List[Dict[str, Any]] = Field(default_factory=list, description="失败案例")

# ==================== 评估服务 ====================

class RetrievalEvaluationService:
    """检索质量评估服务"""
    
    def __init__(self, test_set_path: str):
        self.test_set_path = test_set_path
        self.test_samples = self._load_test_set()
        
    def _load_test_set(self) -> List[TestSample]:
        """加载测试集"""
        test_samples = []
        test_file = Path(self.test_set_path)
        
        if not test_file.exists():
            raise FileNotFoundError(f"测试集文件不存在: {self.test_set_path}")
        
        with open(test_file, 'r', encoding='utf-8') as f:
            for line in f:
                if not line.strip():
                    continue
                data = json.loads(line)
                test_samples.append(TestSample(**data))
        
        logger.info(f"加载了 {len(test_samples)} 条测试样本")
        return test_samples
    
    async def evaluate_model(self, model, retrieval_service) -> EvaluationReport:
        """评估模型召回质量"""
        
        # 对每个测试样本执行检索
        retrieval_results = []
        for sample in self.test_samples:
            # 调用检索服务
            start_time = datetime.now()
            retrieved_docs = await retrieval_service.retrieve(sample.query)
            latency_ms = int((datetime.now() - start_time).total_seconds() * 1000)
            
            # 记录结果
            retrieval_results.append(RetrievalResult(
                query=sample.query,
                retrieved_docs=retrieved_docs,
                latency_ms=latency_ms
            ))
        
        # 计算指标
        metrics = self._calculate_metrics(retrieval_results)
        
        # 分场景分析
        scenario_metrics = self._calculate_scenario_metrics(retrieval_results)
        
        # 失败案例分析
        failure_cases = self._analyze_failure_cases(retrieval_results)
        
        # 生成报告
        report = EvaluationReport(
            test_set_name=self.test_set_path,
            model_name=model.__class__.__name__,
            recall_at_10=metrics['recall_at_10'],
            recall_at_50=metrics['recall_at_50'],
            mrr=metrics['mrr'],
            ndcg_at_10=metrics['ndcg_at_10'],
            scenario_metrics=scenario_metrics,
            failure_cases=failure_cases
        )
        
        return report
    
    def _calculate_metrics(self, retrieval_results: List[RetrievalResult]) -> Dict[str, float]:
        """计算评估指标"""
        
        recall_at_10_list = []
        recall_at_50_list = []
        mrr_list = []
        ndcg_at_10_list = []
        
        for i, result in enumerate(retrieval_results):
            # 获取相关文档列表
            relevant_docs = set(self.test_samples[i].relevant_docs)
            partially_relevant_docs = set(self.test_samples[i].partially_relevant_docs)
            
            # 获取检索结果
            retrieved_docs = [doc_id for doc_id, _ in result.retrieved_docs]
            
            # 计算 Recall@K
            recall_at_10 = len(relevant_docs & set(retrieved_docs[:10])) / len(relevant_docs) if relevant_docs else 0
            recall_at_50 = len(relevant_docs & set(retrieved_docs[:50])) / len(relevant_docs) if relevant_docs else 0
            recall_at_10_list.append(recall_at_10)
            recall_at_50_list.append(recall_at_50)
            
            # 计算 MRR
            for rank, doc_id in enumerate(retrieved_docs, start=1):
                if doc_id in relevant_docs:
                    mrr_list.append(1.0 / rank)
                    break
            else:
                mrr_list.append(0.0)
            
            # 计算 NDCG@10
            dcg = 0
            for rank, doc_id in enumerate(retrieved_docs[:10], start=1):
                if doc_id in relevant_docs:
                    dcg += 1.0 / (log(rank + 1) / log(2))  # log2(rank+1)
                elif doc_id in partially_relevant_docs:
                    dcg += 0.5 / (log(rank + 1) / log(2))
            idcg = self._calculate_ideal_dcg(self.test_samples[i])
            ndcg = dcg / idcg if idcg > 0 else 0
            ndcg_at_10_list.append(ndcg)
        
        return {
            'recall_at_10': sum(recall_at_10_list) / len(recall_at_10_list),
            'recall_at_50': sum(recall_at_50_list) / len(recall_at_50_list),
            'mrr': sum(mrr_list) / len(mrr_list),
            'ndcg_at_10': sum(ndcg_at_10_list) / len(ndcg_at_10_list)
        }
    
    def _calculate_ideal_dcg(self, sample: TestSample) -> float:
        """计算理想 DCG"""
        # 假设相关文档排在最前面，部分相关文档排在后面
        relevant_count = len(sample.relevant_docs)
        partial_count = len(sample.partially_relevant_docs)
        
        idcg = 0
        for i in range(min(relevant_count + partial_count, 10)):
            rank = i + 1
            if i < relevant_count:
                idcg += 1.0 / (log(rank + 1) / log(2))
            else:
                idcg += 0.5 / (log(rank + 1) / log(2))
        
        return idcg
    
    def _calculate_scenario_metrics(self, retrieval_results: List[RetrievalResult]) -> Dict[str, Dict[str, float]]:
        """分场景计算指标"""
        
        scenario_results = {}
        
        for i, result in enumerate(retrieval_results):
            sample = self.test_samples[i]
            scenario = sample.metadata.get('scenario', 'default')
            
            if scenario not in scenario_results:
                scenario_results[scenario] = []
            
            # 计算该样本的 Recall@10
            relevant_docs = set(sample.relevant_docs)
            retrieved_docs = [doc_id for doc_id, _ in result.retrieved_docs]
            recall = len(relevant_docs & set(retrieved_docs[:10])) / len(relevant_docs) if relevant_docs else 0
            
            scenario_results[scenario].append(recall)
        
        # 计算每个场景的平均指标
        scenario_metrics = {}
        for scenario, recalls in scenario_results.items():
            scenario_metrics[scenario] = {
                'avg_recall_at_10': sum(recalls) / len(recalls),
                'sample_count': len(recalls)
            }
        
        return scenario_metrics
    
    def _analyze_failure_cases(self, retrieval_results: List[RetrievalResult]) -> List[Dict[str, Any]]:
        """分析失败案例"""
        
        failure_cases = []
        
        for i, result in enumerate(retrieval_results):
            sample = self.test_samples[i]
            relevant_docs = set(sample.relevant_docs)
            retrieved_docs = [doc_id for doc_id, _ in result.retrieved_docs]
            
            # 检查是否召回率 < 50%
            recall = len(relevant_docs & set(retrieved_docs[:10])) / len(relevant_docs) if relevant_docs else 1.0
            
            if recall < 0.5:
                failure_cases.append({
                    'query': sample.query,
                    'relevant_docs': list(relevant_docs),
                    'retrieved_docs': retrieved_docs[:10],
                    'recall_at_10': recall,
                    'possible_reason': self._infer_failure_reason(sample, result)
                })
        
        return failure_cases
    
    def _infer_failure_reason(self, sample: TestSample, result: RetrievalResult) -> str:
        """推断失败原因（简化版）"""
        
        # 检查是否是术语理解问题
        query_terms = set(sample.query.split())
        doc_terms = set()
        for doc_id, _ in result.retrieved_docs[:10]:
            # 假设能获取到文档内容
            # doc_content = get_doc_content(doc_id)
            # doc_terms.update(doc_content.split())
            pass
        
        # 简化逻辑：返回通用原因
        return "可能是术语理解错误、上下文缺失或文档切片不当"
    
    def save_report(self, report: EvaluationReport, output_path: str):
        """保存评估报告"""
        
        output_file = Path(output_path)
        output_file.parent.mkdir(parents=True, exist_ok=True)
        
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(report.dict(), f, ensure_ascii=False, indent=2, default=str)
        
        logger.info(f"评估报告已保存到: {output_path}")

# ==================== 使用示例 ====================

async def main():
    # 初始化评估服务
    evaluation_service = RetrievalEvaluationService(
        test_set_path="./data/test_set_legal.jsonl"
    )
    
    # 加载模型（假设已实现）
    model = load_model("bge-large-zh-legal-lora")
    retrieval_service = RetrievalService(model)
    
    # 执行评估
    report = await evaluation_service.evaluate_model(model, retrieval_service)
    
    # 保存报告
    evaluation_service.save_report(report, "./evaluation_reports/legal_model_eval.json")
    
    # 打印核心指标
    print(f"Recall@10: {report.recall_at_10:.4f}")
    print(f"MRR: {report.mrr:.4f}")
    print(f"NDCG@10: {report.ndcg_at_10:.4f}")
    
    # 打印分场景指标
    for scenario, metrics in report.scenario_metrics.items():
        print(f"场景 '{scenario}': Recall@10 = {metrics['avg_recall_at_10']:.4f}")
    
    # 分析失败案例
    if report.failure_cases:
        print(f"\n发现 {len(report.failure_cases)} 个失败案例，请查看报告文件")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

代码关键点：

1. 测试集格式：每行一个 JSON，包含查询、相关文档列表、元数据
2. 分场景评估：按业务场景（如"劳动合同"、“知识产权”）分别计算指标
3. 失败案例分析：自动识别召回率低的查询，并推断可能原因
4. 报告保存：JSON 格式，方便后续可视化

2.3.5 持续监控与回归测试

回归测试流水线：

实现要点：

1. 基线管理：

   • 保存每次评估的指标作为基线
   • 下次评估时对比基线，下降超过 5% 则告警

2. A/B 测试：

   • 同时部署两个模型（如通用模型 vs 微调模型）
   • 随机分配流量，对比线上指标（点击率、停留时间）

3. 监控看板：

   • 用 Grafana + Prometheus 搭建实时监控
   • 关键指标：Recall@10、MRR、P95 延迟、错误率

2.3.6 实际工作中的 Gotchas

Gotcha 1：测试集太小，评估结果不可靠

现象：测试集只有 50 条样本，评估结果波动大。

原因：样本量太小，无法代表真实数据分布。

解决方案：

• 测试集至少 200 条（每个场景至少 20 条）
• 定期扩充测试集（每次发现失败案例都加入）

Gotcha 2：只看整体指标，忽略分场景表现

现象：整体 Recall@10 = 0.8，但某些场景只有 0.5。

原因：某些场景难，某些场景易，整体指标掩盖了问题。

解决方案：

• 分场景评估（如法律、医疗、金融分开评估）
• 设定每个场景的最低阈值（如所有场景都必须 > 0.7）

Gotcha 3：人工标注质量差，评估结果不可信

现象：评估结果显示模型效果很好，但用户投诉多。

原因：标注员不专业，把不相关的文档标为相关。

解决方案：

• 培训标注员（提供标注指南和案例）
• 多人标注取多数投票
• 定期抽查标注质量

---

2.4 向量索引维护实战：增量更新、删除一致性、索引压缩与重建策略

2.4.1 向量索引的生命周期

核心挑战：

• 数据在不断变化（新增、删除、修改）
• 向量索引不能像数据库那样"原地更新"
• 大规模索引重建成本高（时间、算力）

生命周期阶段：

2.4.2 增量更新策略

问题：新文档进来，如何快速加入向量索引？

方案对比：

方案	优点	缺点	适用场景
直接插入	简单，实时	索引质量下降（向量分布变化）	小规模（< 10 万）
批量追加	减少索引碎片	有延迟	中规模（10 万 - 100 万）
定期重建	索引质量高	重建成本高	大规模（> 100 万）

实现：批量追加 + 定期重建

from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from datetime import datetime, timedelta
import logging
from pathlib import Path
import json

logger = logging.getLogger(__name__)

class IndexUpdateConfig(BaseModel):
    """索引更新配置"""
    vector_db_path: str = Field(..., description="向量数据库路径")
    buffer_size: int = Field(1000, description="缓冲区大小（达到一定数量后批量插入）")
    rebuild_interval_hours: int = Field(24, description="重建间隔（小时）")
    max_vectors_per_index: int = Field(1000000, description="单个索引最大向量数")

class IncrementalIndexManager:
    """增量索引管理器"""
    
    def __init__(self, config: IndexUpdateConfig):
        self.config = config
        self.write_buffer = []  # 写缓冲区
        self.last_rebuild_time = datetime.now()
        
        # 加载向量数据库（示例用 Milvus）
        from pymilvus import connections, Collection
        connections.connect(host='localhost', port='19530')
        self.collection = Collection(config.vector_db_path)
        
    async def add_documents(self, documents: List[Dict[str, Any]]):
        """添加文档到索引"""
        
        # 编码文档（生成 Embedding）
        embeddings = await self._encode_documents(documents)
        
        # 写入缓冲区
        for doc, emb in zip(documents, embeddings):
            self.write_buffer.append({
                'id': doc['id'],
                'vector': emb,
                'metadata': doc.get('metadata', {})
            })
        
        # 缓冲区满了，执行批量插入
        if len(self.write_buffer) >= self.config.buffer_size:
            await self._flush_buffer()
        
        # 检查是否需要重建索引
        if self._should_rebuild():
            await self.rebuild_index()
    
    async def _flush_buffer(self):
        """刷新缓冲区，批量插入向量"""
        
        if not self.write_buffer:
            return
        
        logger.info(f"批量插入 {len(self.write_buffer)} 条向量")
        
        # 准备数据
        ids = [item['id'] for item in self.write_buffer]
        vectors = [item['vector'] for item in self.write_buffer]
        metadatas = [item['metadata'] for item in self.write_buffer]
        
        # 批量插入（Milvus 示例）
        self.collection.insert([ids, vectors, metadatas])
        
        # 清空缓冲区
        self.write_buffer.clear()
        
        logger.info("缓冲区已刷新")
    
    def _should_rebuild(self) -> bool:
        """判断是否需要重建索引"""
        
        # 条件1：距离上次重建超过指定时间
        time_since_rebuild = datetime.now() - self.last_rebuild_time
        if time_since_rebuild > timedelta(hours=self.config.rebuild_interval_hours):
            return True
        
        # 条件2：索引中向量数量过多
        vector_count = self.collection.num_entities
        if vector_count > self.config.max_vectors_per_index:
            return True
        
        return False
    
    async def rebuild_index(self):
        """重建索引"""
        
        logger.info("开始重建索引...")
        
        # 1. 导出所有向量
        all_vectors = self.collection.query(expr="", output_fields=["id", "vector", "metadata"])
        
        # 2. 删除旧索引
        self.collection.drop_index()
        
        # 3. 重新创建索引（用更优的参数）
        index_params = {
            "metric_type": "IP",  # 内积（余弦相似度）
            "index_type": "IVF_FLAT",
            "params": {"nlist": 1024}  # 聚类中心数量
        }
        self.collection.create_index(field_name="vector", index_params=index_params)
        
        # 4. 重新插入所有向量
        # （实际上 Milvus 重建索引不需要重新插入，这里只是示例）
        
        # 5. 更新重建时间
        self.last_rebuild_time = datetime.now()
        
        logger.info("索引重建完成")
    
    async def delete_documents(self, doc_ids: List[str]):
        """删除文档（逻辑删除 + 定期物理删除）"""
        
        # 逻辑删除：标记为删除，但不立即从索引中移除
        for doc_id in doc_ids:
            # 更新元数据中的 deleted 标记
            self.collection.update(
                expr=f"id == {doc_id}",
                set_values={"deleted": True, "deleted_at": datetime.now().isoformat()}
            )
        
        logger.info(f"逻辑删除了 {len(doc_ids)} 条文档")
        
        # 定期物理删除（如每天凌晨执行）
        # 在 rebuild_index 时，跳过标记为 deleted 的文档
    
    async def _encode_documents(self, documents: List[Dict[str, Any]]) -> List[List[float]]:
        """编码文档为向量（调用 Embedding 模型）"""
        
        # 这里简化，实际应该调用 Embedding 服务
        texts = [doc['content'] for doc in documents]
        
        # 假设有一个 embed 函数
        embeddings = await embed_texts(texts)
        
        return embeddings

# ==================== 使用示例 ====================

async def main():
    config = IndexUpdateConfig(
        vector_db_path="my_collection",
        buffer_size=1000,
        rebuild_interval_hours=24,
        max_vectors_per_index=1000000
    )
    
    manager = IncrementalIndexManager(config)
    
    # 模拟新文档到达
    new_docs = [
        {"id": "doc_001", "content": "劳动合同到期不续签...", "metadata": {"source": "legal"}},
        {"id": "doc_002", "content": "知识产权保护措施...", "metadata": {"source": "legal"}}
    ]
    
    await manager.add_documents(new_docs)
    
    print("文档已添加到索引")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

代码关键点：

1. 写缓冲区：累积一定数量再批量插入，减少索引碎片
2. 定期重建：防止索引质量下降
3. 逻辑删除：标记删除而非立即物理删除，避免频繁修改索引

2.4.3 删除一致性问题

问题：删除了文档，但向量索引中还有，导致召回垃圾结果。

解决方案：

1. 逻辑删除 + 过滤：

   • 在元数据中标记 deleted=True
   • 检索时过滤掉标记为删除的文档

   # Milvus 示例：检索时过滤
   search_params = {
       "expr": "deleted == false",  # 过滤逻辑删除的文档
       "metric_type": "IP",
       "params": {"nprobe": 10}
   }
   results = collection.search(vectors, "vector", search_params, limit=10)
   

2. 定期物理删除：

   • 每天凌晨执行物理删除（真正从索引中移除）
   • 重建索引时跳过逻辑删除的文档

3. Tombstone 机制（高级）：

   • 类似 LSM-Tree 的 tombstone
   • 删除时写入 tombstone 记录
   • 合并（Compaction）时清理

2.4.4 索引压缩与量化

问题：向量维度高（768 维或 1536 维），存储和检索成本高。

解决方案：乘积量化（Product Quantization, PQ）

原理：
将高维向量分成多个子向量，每个子向量用聚类中心表示（量化），从而减少存储。

数学表达：

对于 $d$ 维向量 $x$，分成 $m$ 个子向量：

$$x=[x_1,x_2,...,x_m],x_i\in {\mathbb{R}}^{d/m}$$

对每个子向量空间做 K-Means 聚类，得到 $k$ 个聚类中心（通常 $k=256$，用 8 bit 表示）。

压缩后，每个子向量只用 $lo{g}_2(k)$ bit 表示。

效果：

• 存储降低：768 维 float32 → 96 字节（压缩 32 倍）
• 精度损失：召回率下降 2-5%

Milvus 中的 PQ 配置：

index_params = {
    "metric_type": "IP",
    "index_type": "IVF_PQ",  # 乘积量化索引
    "params": {
        "nlist": 1024,  # 聚类中心数量
        "m": 16,  # 子向量数量（768 维分成 16 个子向量，每个 48 维）
        "nbits": 8  # 每个子向量的比特数（2^8 = 256 个聚类中心）
    }
}
collection.create_index(field_name="vector", index_params=index_params)

2.4.5 索引重建策略

何时需要重建？

1. 数据分布变化：

   • 例子：原来主要是中文文档，现在加入了大量英文文档
   • 表现：召回率下降 > 5%

2. 索引膨胀：

   • 大量逻辑删除的文档未清理
   • 索引大小比实际数据大 50% 以上

3. 参数调优：

   • 发现 nlist 或 m 参数设置不合理
   • 需要重建索引以应用新参数

重建流程（零停机）：

实现要点：

1. 双索引并行：

   • 重建期间，新旧索引同时存在
   • 新数据同时写入两个索引
   • 读取先读旧索引，确认新索引就绪后切换

2. 灰度切换：

   • 先切 5% 流量到新索引，观察效果
   • 没问题再全量切换

3. 回滚预案：

   • 保留旧索引 24 小时
   • 发现问题立即切回旧索引

2.4.6 实际工作中的 Gotchas

Gotcha 1：索引参数选择不当

现象：nlist 太小（如 100），检索速度慢；nlist 太大（如 10000），召回率低。

解决方案：

• 经验公式：nlist = sqrt(N)，其中 N 是向量总数
• 100 万向量 → nlist = 1024
• 用真实数据做参数搜索（Grid Search）

Gotcha 2：增量更新导致索引质量下降

现象：运行一个月后，召回率从 0.85 降到 0.70。

原因：新加入的向量分布与旧向量不同，索引结构不再最优。

解决方案：

• 定期重建索引（如每周一次）
• 使用支持动态更新的索引（如 HNSW，不需要重建）

Gotcha 3：删除操作导致索引膨胀

现象：删除了 50% 的文档，但索引大小没变。

原因：逻辑删除没有真正释放空间。

解决方案：

• 定期执行物理删除（VACUUM）
• 重建索引时清理逻辑删除的文档

---

本章小结

核心 Takeaways：

1. 模型选型：垂直领域必须微调，用 LoRA 低成本实现（只训练 0.1% 参数）
2. 质量评估：建立量化评估体系（Recall@10、MRR、NDCG），持续监控和改进
3. 索引维护：增量更新 + 定期重建 + 压缩量化，平衡性能和成本
4. 删除一致性：逻辑删除 + 过滤 + 定期物理删除，避免召回垃圾结果

决策清单：

• 根据业务场景选择了合适的 Embedding 模型（通用 or 微调）
• 准备了高质量的训练数据（≥ 3000 条，覆盖主要场景）
• 实现了自动化评估流水线（测试集 + 指标计算 + 报告生成）
• 设计了索引维护策略（缓冲区大小、重建间隔、压缩参数）
• 解决了删除一致性问题（逻辑删除 + 过滤）

下一步：
第三章将深入探讨"RAG（检索增强生成）的系统设计与实战"，教你如何把向量检索、Embedding 模型和 LLM 组合成生产级 RAG 系统。

---

思考题

基础题：

1. 为什么 LoRA 微调比全参数微调更适合生产环境？

   点击查看答案 LoRA 只训练低秩矩阵（0.1% 参数），显存需求降低 90%，训练速度提升 3-5 倍，且可以随时切换回原始模型（不破坏预训练权重）。全参数微调需要 A100 这样的高端 GPU，成本高，且容易过拟合。

2. 如何判断是否需要重建向量索引？

   点击查看答案 三个信号：1）召回率下降 > 5%；2）索引大小比实际数据大 50% 以上（大量逻辑删除未清理）；3）需要调优索引参数（如 nlist、m）。建议定期（如每周）评估召回率，低于阈值就触发重建。

进阶题：

3. 设计一个 A/B 测试方案，对比两个 Embedding 模型在线上的效果。

   点击查看答案 方案：1）准备 1000 条真实查询和标注的相关文档；2）同时部署两个模型（如通用模型 vs 微调模型）；3）每个查询随机分配给其中一个模型（50/50 流量）；4）记录用户行为指标（点击率、停留时间、满意度评分）；5）用统计检验（t 检验或卡方检验）判断差异是否显著；6）显著优于基线的话，全量切换。

4. 如何处理向量索引中的"脏数据"（如重复文档、低质量文档）？

   点击查看答案 三层防御：1）入库前去重（用 MinHash 或 SimHash 计算文档相似度，相似度 > 0.9 判定为重复）；2）质量过滤（用规则或小模型过滤低质量文档，如太短、乱码、纯广告）；3）定期清理（用聚类算法找出异常向量，人工复核后删除）。

实战题：

5. 实现一个增量索引更新服务，要求：写缓冲、定期刷新、删除一致性、索引重建。 点击查看答案 参考本章 2.4.2 的 `IncrementalIndexManager` 代码。关键点：1）写缓冲区（累积 1000 条再批量插入）；2）逻辑删除 + 过滤（避免频繁修改索引）；3）定期重建（如每周一次，防止索引质量下降）；4）双索引并行（重建时零停机）。

本章已生成完毕。自动保存为本地文件：第二章：Embedding 管线与召回质量保障.md。
请回复【继续生成第 3 章】或提出您对当前章节的疑问。