当前位置：首页 > news >正文

RAG 问答对生成：用 LLM 自动构建检索评测集的工程化实践

news 2026/7/11 19:39:14

RAG 问答对生成：用 LLM 自动构建检索评测集的工程化实践

一、深度引言与场景痛点

RAG 系统上线后，你总要回答一个问题："这个系统的检索质量到底怎么样？"回答这个问题的前提是你有一个评测集——一组（问题, 标准答案）对。但问题来了：评测集从哪里来？

如果你有真实用户的查询日志，当然可以直接用。但很多 RAG 项目在上线前就需要做评测，或者业务领域太专业、日志样本不够。人工标注呢？100 个问答对可能就得花一整天，而且标注一致性很难保证——同一个文档，不同标注员会提出不同的问题，侧重点也不同。

这就是用 LLM 自动构建评测集的用武之地。给定你的知识库文档，让 LLM 基于文档内容自动生成多样化的问答对。但这里有一个容易掉坑的地方：LLM 生成的问答对如果太简单（答案就在文档的第一句话），就不能区分出检索质量的差异。好的评测集应该覆盖多文档检索、需要推理的、需要跨段落信息的等多层次难度。

二、底层机制与原理深度剖析

用 LLM 构建 RAG 评测集的核心流程：

文档采样：从知识库中按比例采样文档，保证评测集覆盖不同主题和文档类型。这里的关键是分层采样——短文档和长文档、结构化文档和自由文本都要覆盖到。

难度分层生成：通过 Prompt 工程控制 LLM 生成不同难度的问答对。简单难度：答案在同一段落中明确出现；中等难度：答案需要跨段落综合信息；困难难度：答案跨文档，需要推理和多文档比较。

答案锚定验证：LLM 生成的答案必须能在源文档中找到确切的文本锚点。如果生成的答案在文档中找不到依据，说明 LLM 自行发挥了——这是不能接受的。

多样性保证：通过指令约束和温度参数控制，确保生成的问答对在问题类型（What / How / Why / Compare）、主题分布、语言风格上有多样性。

flowchart TB subgraph "文档准备" KB["知识库文档"] --> SAMPLE["分层采样\n按主题、长度分层"] SAMPLE --> CHUNKS["文档分块\n生成上下文窗口"] end subgraph "LLM 生成 — 三轮迭代" CHUNKS --> P1["第一轮 Prompt\n生成简单问答\n(答案在段落内)"] P1 --> Q1["简单问答集\n(40%)"] CHUNKS --> P2["第二轮 Prompt\n生成中等问答\n(跨段落综合)"] P2 --> Q2["中等问答集\n(35%)"] CHUNKS --> P3["第三轮 Prompt\n生成困难问答\n(跨文档 + 推理)"] P3 --> Q3["困难问答集\n(25%)"] end subgraph "质量校验" Q1 --> VAL["锚定验证\n答案在源文档中可追溯？"] Q2 --> VAL Q3 --> VAL VAL -->|"通过"| DEDUP["去重检测\n语义相似度 < 阈值"] VAL -->|"不通过"| RETRY["重试（max 3次）"] DEDUP --> DIVERSITY["多样性检查\n问题类型/主题分布"] DIVERSITY --> FINAL["最终评测集"] end style KB fill:#4A90D9,color:#fff style P1 fill:#5CB85C,color:#fff style P2 fill:#5CB85C,color:#fff style P3 fill:#5CB85C,color:#fff style VAL fill:#E8A838,color:#fff style FINAL fill:#D9534F,color:#fff

三、生产级代码实现

下面是一套完整的问答对生成和校验 pipeline，支持难度分层和答案锚定验证。

import asyncio import json import logging import random from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass class QAPair: """问答对数据模型。""" question: str answer: str source_chunks: list[str] # 源文档块 difficulty: str # easy / medium / hard question_type: str # what / how / why / compare anchor_text: str = "" # 答案在源文档中的锚点文本 metadata: dict = field(default_factory=dict) class QAGenerator: """自动构建 RAG 评测集的问答对生成器。 特性： - 分层难度生成（简单/中等/困难） - 答案锚定验证 - 多样性保障 - 去重检测 """ # Prompt 模板 PROMPT_TEMPLATES = { "easy": """你是一个 RAG 评测集构建助手。请基于以下文档段落，生成一个简单难度的问答对。 要求： 1. 问题的答案必须**明确且完整地**出现在提供的段落中，不能有任何推理成分 2. 问题类型可以是 What/Who/When/Where 3. 答案必须在 3 句话以内 4. 在答案中引用原文的具体句子 文档段落： {context} 请严格按以下 JSON 格式输出： ```json {{"question": "...", "answer": "...", "anchor": "原文中支持答案的句子"}} ```""", "medium": """你是一个 RAG 评测集构建助手。请基于以下多个文档段落，生成一个中等难度的问答对。 要求： 1. 问题需要综合**至少两个不同的段落**的信息才能回答 2. 问题类型可以是 How/Why/Explain 3. 答案需要包含逻辑推理或信息归纳 4. 明确指出答案分别来自哪些段落 文档段落： {context} 请严格按以下 JSON 格式输出： ```json {{"question": "...", "answer": "...", "anchors": ["段落1的锚点", "段落2的锚点"]}} ```""", "hard": """你是一个 RAG 评测集构建助手。请基于以下多个文档段落，生成一个困难难度的问答对。 要求： 1. 问题需要**跨文档比较、分析或推断**，不能仅靠检索一个段落回答 2. 问题可能涉及对比评估、因果分析、多角度论证 3. 答案需要 5-10 句话，展示完整的推理过程 4. 明确指出答案依赖的各个锚点 文档段落： {context} 请严格按以下 JSON 格式输出： ```json {{"question": "...", "answer": "...", "reasoning": "推理过程", "anchors": ["锚点1", "锚点2", "锚点3"]}} ```""", } def __init__(self, llm_client=None, similarity_threshold: float = 0.85): self.llm = llm_client self.threshold = similarity_threshold self._generated_questions: set[str] = set() async def generate( self, documents: list[str], target_count: int = 100, easy_ratio: float = 0.4, medium_ratio: float = 0.35, hard_ratio: float = 0.25, ) -> list[QAPair]: """构建完整的评测集。 Args: documents: 知识库文档列表 target_count: 目标问答对数量 easy_ratio: 简单题占比 medium_ratio: 中等题占比 hard_ratio: 困难题占比 """ if not documents: raise ValueError("文档列表不能为空") # 1. 分块和分层采样 chunks = self._chunk_documents(documents) logger.info("文档分块完成，共 %d 个块", len(chunks)) # 2. 按比例生成不同难度的问答对 easy_count = int(target_count * easy_ratio) medium_count = int(target_count * medium_ratio) hard_count = target_count - easy_count - medium_count all_pairs: list[QAPair] = [] # 并行生成 tasks = [ self._generate_batch(chunks, "easy", easy_count), self._generate_batch(chunks, "medium", medium_count), self._generate_batch(chunks, "hard", hard_count), ] results = await asyncio.gather(*tasks, return_exceptions=True) for i, result in enumerate(results): if isinstance(result, Exception): logger.error("批次生成失败: %s", result) else: all_pairs.extend(result) # 3. 质量校验 validated = await self._validate_all(all_pairs) deduped = self._deduplicate(validated) # 4. 统计报告 self._print_stats(deduped, target_count) return deduped async def _generate_batch( self, chunks: list[str], difficulty: str, count: int, ) -> list[QAPair]: """生成一批指定位难度的问答对。""" pairs: list[QAPair] = [] template = self.PROMPT_TEMPLATES[difficulty] max_retries = 3 for _ in range(count): # 根据难度选择不同数量的上下文块 if difficulty == "easy": n_chunks = 1 elif difficulty == "medium": n_chunks = random.randint(2, 4) else: n_chunks = random.randint(3, 6) selected = random.sample( chunks, min(n_chunks, len(chunks)), ) context = "\n\n---\n\n".join( f"[段落 {i+1}] {chunk}" for i, chunk in enumerate(selected) ) for attempt in range(max_retries): try: result = await self._call_llm(template.format(context=context)) pair = self._parse_llm_response(result, selected, difficulty) # 锚定验证 if self._verify_anchor(pair): pairs.append(pair) break else: logger.warning( "锚定验证失败，重试 (%d/%d)", attempt + 1, max_retries, ) except Exception as e: logger.error("生成失败 (%d/%d): %s", attempt + 1, max_retries, e) logger.info("%s 难度生成完成: %d/%d", difficulty, len(pairs), count) return pairs async def _call_llm(self, prompt: str) -> str: """调用 LLM 生成回答。替换为你的 LLM 客户端。""" # ===== 实际调用替换 ===== # response = await self.llm.ainvoke(prompt) # return response.content await asyncio.sleep(0.5) # 模拟延迟 return json.dumps({ "question": f"基于文档内容的测试问题（关于{prompt[:30]}...）", "answer": "测试答案", "anchor": "原文锚点文本", }, ensure_ascii=False) def _parse_llm_response( self, response: str, source_chunks: list[str], difficulty: str, ) -> QAPair: """解析 LLM 返回的 JSON，提取问答对。""" # 提取 JSON（可能在 markdown 代码块中） import re json_match = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', response, re.DOTALL) if json_match: json_str = json_match.group(1) else: json_str = response try: data = json.loads(json_str) except json.JSONDecodeError as e: # 尝试修复常见格式问题 logger.warning("JSON 解析失败，尝试修复: %s", e) json_str = json_str.replace("'", '"') try: data = json.loads(json_str) except json.JSONDecodeError: raise ValueError(f"LLM 返回格式无法解析: {response[:200]}") question = data.get("question", "") answer = data.get("answer", "") anchor = data.get("anchor") or ( data.get("anchors", [""])[0] if data.get("anchors") else "" ) if not question or not answer: raise ValueError("问答对缺少 question 或 answer 字段") # 确定问题类型 question_lower = question.lower() if any(kw in question_lower for kw in ["什么是", "what", "谁", "who"]): q_type = "what" elif any(kw in question_lower for kw in ["如何", "怎么", "how"]): q_type = "how" elif any(kw in question_lower for kw in ["为什么", "why"]): q_type = "why" elif any(kw in question_lower for kw in ["对比", "比较", "compare", "vs"]): q_type = "compare" else: q_type = "what" return QAPair( question=question, answer=answer, source_chunks=source_chunks, difficulty=difficulty, question_type=q_type, anchor_text=str(anchor), ) def _verify_anchor(self, pair: QAPair) -> bool: """验证答案锚点：答案的关键信息必须在源文档中可追溯。""" if not pair.anchor_text: return False anchor_lower = pair.anchor_text.lower().strip() for chunk in pair.source_chunks: chunk_lower = chunk.lower() # 模糊匹配：锚点文本的核心部分必须在 chunk 中出现 if len(anchor_lower) > 10 and anchor_lower[:30] in chunk_lower: return True # 关键词重叠检测 anchor_words = set(anchor_lower.split()) chunk_words = set(chunk_lower.split()) overlap = anchor_words & chunk_words if len(overlap) >= min(3, len(anchor_words) * 0.5): return True logger.debug("锚点验证失败: anchor=「%s」", pair.anchor_text[:80]) return False async def _validate_all(self, pairs: list[QAPair]) -> list[QAPair]: """批量校验所有问答对的质量。""" valid = [] for pair in pairs: # 1. 长度检查 if len(pair.question) < 5 or len(pair.answer) < 5: logger.warning("问答对过短，丢弃: %s", pair.question[:30]) continue # 2. 锚定验证（已在生成时做，这里是二次保证） if not self._verify_anchor(pair): logger.warning("问答对锚定验证失败: %s", pair.question[:30]) continue # 3. 问题质量检查 if pair.question in self._generated_questions: logger.warning("重复问题: %s", pair.question[:30]) continue self._generated_questions.add(pair.question) valid.append(pair) return valid def _deduplicate(self, pairs: list[QAPair]) -> list[QAPair]: """基于语义相似度的去重。""" # 简化版：基于问题文本的编辑距离或 Jaccard 相似度 from difflib import SequenceMatcher unique: list[QAPair] = [] for pair in pairs: is_duplicate = False for existing in unique: similarity = SequenceMatcher( None, pair.question, existing.question, ).ratio() if similarity > self.threshold: is_duplicate = True break if not is_duplicate: unique.append(pair) logger.info("去重: %d → %d", len(pairs), len(unique)) return unique @staticmethod def _chunk_documents(documents: list[str], chunk_size: int = 500) -> list[str]: """将长文档切分为适合 LLM 上下文窗口的块。""" chunks = [] for doc in documents: for i in range(0, len(doc), chunk_size): chunk = doc[i:i + chunk_size] if chunk.strip(): chunks.append(chunk) return chunks @staticmethod def _print_stats(pairs: list[QAPair], target: int): """打印评测集统计报告。""" difficulty_counts = {} type_counts = {} for p in pairs: difficulty_counts[p.difficulty] = difficulty_counts.get(p.difficulty, 0) + 1 type_counts[p.question_type] = type_counts.get(p.question_type, 0) + 1 logger.info("=" * 40) logger.info("评测集生成完成") logger.info("目标数量: %d, 实际数量: %d", target, len(pairs)) logger.info("难度分布: %s", difficulty_counts) logger.info("问题类型分布: %s", type_counts) logger.info("=" * 40) async def demo(): test_docs = [ "Python 是一种解释型、面向对象的高级编程语言。它由 Guido van Rossum 创建，首次发布于 1991 年。" * 5, "RAG（Retrieval-Augmented Generation）是一种融合信息检索和文本生成的 AI 架构。" * 5, "向量数据库如 Milvus 和 Qdrant 专门用于存储和检索高维向量数据。" * 5, "异步编程是 Python 中处理并发的重要范式，asyncio 库提供了事件循环和协程支持。" * 5, ] generator = QAGenerator(similarity_threshold=0.85) pairs = await generator.generate( test_docs, target_count=20, easy_ratio=0.4, medium_ratio=0.35, hard_ratio=0.25, ) # 保存到文件 output = [{ "question": p.question, "answer": p.answer, "difficulty": p.difficulty, "type": p.question_type, "anchor": p.anchor_text[:100], } for p in pairs] with open("rag_eval_set.json", "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) if __name__ == "__main__": asyncio.run(demo())