RAG 问答对生成：用 LLM 自动构建检索评测集的工程化实践 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_factorydict) 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_clientNone, 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_exceptionsTrue) 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[段落 {i1}] {chunk} for i, chunk in enumerate(selected) ) for attempt in range(max_retries): try: result await self._call_llm(template.format(contextcontext)) 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_asciiFalse) 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(fLLM 返回格式无法解析: {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( questionquestion, answeranswer, source_chunkssource_chunks, difficultydifficulty, question_typeq_type, anchor_textstr(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, RAGRetrieval-Augmented Generation是一种融合信息检索和文本生成的 AI 架构。 * 5, 向量数据库如 Milvus 和 Qdrant 专门用于存储和检索高维向量数据。 * 5, 异步编程是 Python 中处理并发的重要范式asyncio 库提供了事件循环和协程支持。 * 5, ] generator QAGenerator(similarity_threshold0.85) pairs await generator.generate( test_docs, target_count20, easy_ratio0.4, medium_ratio0.35, hard_ratio0.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, encodingutf-8) as f: json.dump(output, f, ensure_asciiFalse, indent2) if __name__ __main__: asyncio.run(demo())四、边界分析与架构权衡LLM 生成偏差LLM 倾向于生成它擅长回答的问题。如果你的 LLM 特别擅长代码相关的问题它生成的评测集会偏重代码问题导致评测失真。需要通过 Prompt 明确限定领域和主题分布来纠正。锚定验证的精度简单的关键词重叠检测会把语义相关但事实无关的句子错判为锚点。如果需要高精度验证建议用 NLI自然语言推理模型判断答案和源文档的逻辑关系。评测集的领域泛化性如果知识库文档本身覆盖范围很窄生成的评测集也只会覆盖这个窄领域。这对于垂直领域 RAG 系统是可以接受的但如果你希望评测集有跨领域泛化性需要引入外部文档采样。成本控制生成 1000 个问答对可能需要调用 LLM 2000 次含重试。建议使用性价比高的模型如 GPT-4o-mini做批量生成只在锚定验证失败时用更强的模型重试。五、总结LLM 自动构建 RAG 评测集的关键不是能不能生成而是生成的质量和可控性。分层难度设计保证评测集能区分不同检索能力的差距锚定验证保证答案有据可查去重和多样性检查保证评测集的广度。把这套流程融入 CI/CD每次文档更新后自动生成新评测集RAG 系统的质量就能持续得到量化保证。