RAG

幻觉：AGI 之路的隐形地雷 当智能体开始"说话"，它就不可避免地会"说错话"。幻觉（Hallucination）——即模型生成看似合理但与事实不符的内容——是 AGI 领域最棘手的安全问题之一。在聊天场景中，幻觉可能只是一次尴尬的回答；但在智能体场景中，幻觉可能导致错误的工具调用、误导性的决策建议，甚至危险的实际操作。 本文将从幻觉的根源出发，系统性地介绍检测、缓解和修复策略，帮助你构建更可信赖的智能体系统。 一、幻觉的根源：为什么会"胡说八道" 1.1 训练数据层面 大模型的幻觉根源深植于其训练过程中： 训练数据的噪声：互联网数据包含大量错误信息、过时信息和矛盾信息。模型在学习时，不可避免地吸收了这些噪声。当它"回忆"某个事实时，可能提取的是错误版本。 知识截止日期：模型的训练数据有截止日期，但世界在不断变化。模型可能自信地回答一个已经过时的事实——比如一个已经废弃的 API 方法。 长尾知识稀疏：对于常见话题，训练数据丰富，模型表现可靠。但在长尾领域（冷门技术、小众领域），训练数据稀疏，模型更容易通过"编造"来填补空白。 1.2 推理机制层面 自回归生成的本质缺陷：模型逐 token 生成文本，每一步都基于前面的内容。一旦早期生成了一个错误的前提，后续内容会围绕这个错误前提展开，形成"幻觉级联"。 缺乏知识边界感知：模型不知道自己"不知道什么"。它没有可靠机制来判断某个问题是否超出了自己的知识范围，因此倾向于对任何问题都给出回答，而不是说"我不知道"。 概率而非检索：模型生成的是"最可能"的答案，而非"最准确"的答案。在需要精确事实的场景中，这种差异就表现为幻觉。 1.3 智能体场景的特殊幻觉 在智能体场景中，幻觉有了新的表现形式： 工具参数幻觉：模型编造不存在的参数值，或混淆不同工具的参数格式 工具结果误解：工具返回了正确结果，但模型在解读时加入了自己的"想象" 调用链幻觉：模型声称调用了某个工具但实际没有，或编造工具调用的返回结果 上下文遗忘幻觉：在长对话中，模型"忘记"了早期确认的事实，生成与之前矛盾的内容 二、幻觉检测：从规则到模型 2.1 基于规则的检测 最基础的检测层是规则匹配： import re from typing import List, Tuple class RuleBasedHallucinationDetector: def __init__(self): self.rules = [ # 数字一致性检查 self._check_number_consistency, # 实体引用检查 self._check_entity_reference, # 日期合理性检查 self._check_date_validity, # 自相矛盾检查 self._check_self_contradiction, ] def _check_number_consistency(self, claim: str, sources: List[str]) -> List[Tuple[str, float]]: issues = [] # 提取声明中的所有数字 claim_numbers = set(re.findall(r'\d+\.?\d*', claim)) for source in sources: source_numbers = set(re.findall(r'\d+\.?\d*', source)) # 检查声明中的数字是否在来源中出现 for num in claim_numbers: if num not in source_numbers: issues.append(( f"数字 '{num}' 在来源中未找到", 0.7 # 置信度 )) return issues def _check_entity_reference(self, claim: str, entities: dict) -> List[Tuple[str, float]]: issues = [] # 检查声明中提到的实体是否存在于已知实体库 claim_entities = self._extract_entities(claim) for entity in claim_entities: if entity not in entities: issues.append(( f"实体 '{entity}' 不在已知实体库中", 0.6 )) return issues def _check_date_validity(self, claim: str) -> List[Tuple[str, float]]: issues = [] dates = re.findall(r'\d{4}年\d{1,2}月\d{1,2}日', claim) for date_str in dates: # 检查日期是否合理（如未来的日期、不可能的日期） try: from datetime import datetime d = datetime.strptime(date_str, '%Y年%m月%d日') if d.year < 1900 or d.year > 2100: issues.append((f"日期 '{date_str}' 不合理", 0.8)) except ValueError: issues.append((f"日期 '{date_str}' 格式无效", 0.9)) return issues def _check_self_contradiction(self, claim: str, history: List[str]) -> List[Tuple[str, float]]: issues = [] # 简化的矛盾检测：检查是否与历史声明中的数字/日期矛盾 for prev in history: prev_numbers = set(re.findall(r'\d+\.?\d*', prev)) claim_numbers = set(re.findall(r'\d+\.?\d*', claim)) # 如果同一话题但数字不同，可能存在矛盾 overlap = self._semantic_overlap(prev, claim) if overlap > 0.5: if prev_numbers != claim_numbers and prev_numbers and claim_numbers: issues.append(( f"与历史声明可能矛盾: 之前说 {prev_numbers}, 现在说 {claim_numbers}", overlap * 0.8 )) return issues 2.2 基于模型的检测 更高级的检测方案使用专门的模型来判断幻觉： ...

前言 Embedding 模型是 RAG（检索增强生成）系统和语义搜索的基石。一个优秀的 Embedding 模型可以让你的向量数据库"更聪明"地理解语义，大幅提升下游 LLM 的回答质量。2026 年，Embedding 模型领域迎来了多语言、长文本和多粒度的重要进展。本文将全面对比四款主流 Embedding 模型，帮助你为 RAG 系统选择最合适的向量引擎。 一、为什么 Embedding 选型如此重要？ 1.1 Embedding 在 RAG 中的角色 用户查询 → [Embedding 模型] → 查询向量 ↓ 知识库文档 → [Embedding 模型] → 文档向量 → 向量数据库 ↓ 向量相似度检索 → Top-K 结果 ↓ [LLM] → 生成回答 Embedding 模型决定了检索的质量上限。 即使 LLM 再强大，如果检索阶段召回的文档不相关，最终回答也会大打折扣。这就是所谓的"垃圾进，垃圾出"（Garbage In, Garbage Out）。 1.2 好的 Embedding 模型的标准 标准 说明 重要性 语义区分度 能区分近义但不同的概念 ⭐⭐⭐⭐⭐ 多语言一致性 同一概念在不同语言下向量接近 ⭐⭐⭐⭐ 长文本处理 支持段落/篇章级 Embedding ⭐⭐⭐⭐ 多粒度 同时支持短查询和长文档 ⭐⭐⭐⭐ 推理速度 编码延迟低 ⭐⭐⭐⭐ 维度适中 存储和检索效率高 ⭐⭐⭐ 开箱即用 无需复杂微调 ⭐⭐⭐ 二、参评模型概览 模型 开发商 参数量 向量维度 最大输入 多语言 开源 BGE-M3 智源研究院 568M 1024 8192 ✅ 100+ 语言 ✅ E5-Mistral-7B Microsoft 7B 4096 32768 ✅ 100+ 语言 ✅ OpenAI text-embedding-3-large OpenAI 未知 3072 (可截断) 8191 ✅ ❌ Cohere embed-v4 Cohere 未知 1536 512 ✅ 100+ 语言 ❌ GTE-Qwen2-7B 阿里云 7B 3584 32768 ✅ ✅ stella-v5-1.5B PDLN 1.5B 1536 16384 ✅ ✅ 三、基准测试对比 3.1 MTEB 排行榜（主流基准） MTEB（Massive Text Embedding Benchmark）是 Embedding 领域最权威的综合评测，覆盖 8 类任务、58 个数据集。 ...

GraphRAG 解析：知识图谱增强的检索增强生成 引言 传统 RAG 基于向量相似度检索文档块，存在三个结构性缺陷： 多跳推理弱：答案需要关联多个文档中的实体关系时，向量检索难以覆盖 全局视角缺失：只返回局部相关片段，无法回答「整个文档集的主要主题是什么」 实体消歧差：同名实体或指代关系容易混淆 GraphRAG（微软 2024 年提出）通过知识图谱 + 层级社区摘要解决这些问题，在全局性问题上显著优于传统 RAG。 1. GraphRAG 架构总览 原始文档 ↓ [1] 文本分块 ↓ [2] 实体 & 关系抽取（LLM） ↓ [3] 知识图谱构建（实体节点 + 关系边） ↓ [4] 社区检测（Leiden 算法） ↓ [5] 层级社区摘要（LLM） ↓ [6] 检索：局部检索（实体子图） + 全局检索（社区摘要） ↓ [7] 生成 1.1 与传统 RAG 的关键差异 维度 传统 RAG GraphRAG 检索单元 文档块（文本片段） 实体、关系、社区摘要 索引结构 向量索引 图结构 + 向量索引 多跳推理 ❌ 依赖单次检索 ✅ 图遍历天然支持 全局问题 ❌ 只看局部片段 ✅ 社区摘要提供全局视角 构建成本 低（嵌入即可） 高（需 LLM 抽取实体） 查询延迟 低（向量检索） 中高（图检索+摘要） 2. 实体与关系抽取 2.1 LLM 驱动的信息抽取 GraphRAG 使用 LLM 从文本块中抽取实体和关系： ...

引言 RAG（Retrieval-Augmented Generation）是当前最流行的 LLM 应用架构之一。然而，从 Demo 到生产之间横亘着巨大的鸿沟。本文基于多个 RAG 生产项目的实战经验，总结 12 个最常见、最致命的坑，并给出经过验证的解决方案。 坑 1：文档分块策略不当 问题 天真地按固定长度分块（如每 512 字符），导致： 语义被截断（一个完整的段落从中间切断） 关键信息分散在多个块中，检索时只命中一部分 表格和列表被拆碎，失去结构信息 解决方案：语义分块 + 重叠窗口 from dataclasses import dataclass from typing import List import re @dataclass class Chunk: text: str metadata: dict token_count: int = 0 class SemanticChunker: """基于语义边界的智能分块器""" def __init__( self, target_size: int = 400, # 目标块大小（tokens） min_size: int = 100, # 最小块大小 max_size: int = 600, # 最大块大小 overlap: int = 50, # 重叠区间 ): self.target_size = target_size self.min_size = min_size self.max_size = max_size self.overlap = overlap def chunk_document(self, text: str, source: str = "") -> List[Chunk]: """分块主流程""" # Step 1: 按结构边界切分 sections = self._split_by_structure(text) # Step 2: 对每个 section 按 paragraph 切分 paragraphs = [] for section in sections: paragraphs.extend(self._split_by_paragraph(section)) # Step 3: 合并过小的段落，拆分过大的段落 chunks = self._merge_and_split(paragraphs) # Step 4: 添加重叠 chunks = self._add_overlap(chunks) # Step 5: 附加元数据 return [ Chunk( text=c, metadata={"source": source, "chunk_index": i, "total_chunks": len(chunks)}, token_count=len(c) // 2, # 粗略估算 ) for i, c in enumerate(chunks) ] def _split_by_structure(self, text: str) -> List[str]: """按标题、分隔符等结构边界切分""" # 按 Markdown 标题切分 pattern = r'(?=^#{1,6}\s)' sections = re.split(pattern, text, flags=re.MULTILINE) return [s.strip() for s in sections if s.strip()] def _split_by_paragraph(self, text: str) -> List[str]: """按段落（双换行）切分""" paras = text.split("\n\n") return [p.strip() for p in paras if p.strip()] def _merge_and_split(self, paragraphs: List[str]) -> List[str]: """合并过小段落，拆分过大段落""" chunks = [] buffer = [] buffer_size = 0 for para in paragraphs: para_size = len(para) // 2 # 粗略 token 估算 if buffer_size + para_size > self.max_size and buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if para_size > self.max_size: # 单段落过长，按句子切分 if buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 sentences = re.split(r'(?<=[。！？.!?])\s+', para) sent_buffer = [] sent_size = 0 for sent in sentences: if sent_size + len(sent) // 2 > self.max_size and sent_buffer: chunks.append(" ".join(sent_buffer)) sent_buffer = [] sent_size = 0 sent_buffer.append(sent) sent_size += len(sent) // 2 if sent_buffer: chunks.append(" ".join(sent_buffer)) else: buffer.append(para) buffer_size += para_size if buffer_size >= self.target_size: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if buffer: chunks.append("\n\n".join(buffer)) return chunks def _add_overlap(self, chunks: List[str]) -> List[str]: """为相邻块添加重叠""" if self.overlap <= 0 or len(chunks) <= 1: return chunks result = [chunks[0]] for i in range(1, len(chunks)): prev_text = chunks[i - 1] overlap_text = prev_text[-self.overlap * 2:] # 粗略取后半段 result.append(overlap_text + " " + chunks[i]) return result # 使用示例 chunker = SemanticChunker(target_size=400, overlap=50) document = open("knowledge_base/product_manual.md").read() chunks = chunker.chunk_document(document, source="product_manual.md") print(f"分块完成: {len(chunks)} 个块") 分块策略对比 策略 优点 缺点 适用场景 固定长度 实现简单 语义截断 不推荐 按段落 保持语义 块大小不均 短文档 语义分块 语义完整 实现复杂 通用推荐 按文档结构 保持层级 需要结构化输入 Markdown/HTML 递归分块 灵活适配 可控性差 混合内容 坑 2：Embedding 模型与 LLM 不匹配 问题 用 OpenAI 的 text-embedding-3-large 做向量，但生成用的是 Claude 模型。两者对语义的理解不同，可能导致检索到的内容并非生成模型"认为"最相关的。 ...

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 模型，支持多语言。 ...

高级 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 引入检索质量评估器，在检索后对文档相关性进行打分，动态决定： ...

引言 “我们应该微调还是用 RAG？"——这是 2024-2026 年 AI 工程师被问得最多的问题。两者都能让 LLM 适配特定领域，但适用场景截然不同。本文不给标准答案，而是给一个决策框架。 核心差异：先理解本质 微调（Fine-tuning） 微调是改变模型内部权重，让模型"学会"新的知识或行为模式。 原始模型 → 注入领域数据 → 新权重 → 直接推理 RAG（检索增强生成） RAG是不改变模型，在推理时"查阅"外部知识库。 用户问题 → 检索知识库 → 拼接到 Prompt → 模型生成答案 本质对比 维度 微调 RAG 知识存储位置 模型权重中 外部数据库中 知识更新方式 重新训练 更新数据库 推理时依赖 无外部依赖 需要检索系统 “记忆"方式 隐式（权重） 显式（文本） 幻觉风险 较高（知识编码不精确） 较低（有原文参考） 知识溯源 无法溯源 可追溯到来源文档 决策框架：7 个关键问题 问题 1：知识更新频率多高？ 更新频率 │ ├─ 每日/每周更新 ──────→ RAG ✅ │ （产品目录、新闻、工单） │ ├─ 每月/每季度更新 ─────→ RAG 或混合 │ （技术文档、政策） │ └─ 几乎不变 ───────────→ 微调 ✅ （行业术语、品牌语调） 原理：微调每次更新知识都需要重新训练（数小时到数天），RAG 只需更新数据库（秒级）。 ...

LlamaIndex 的演进路线 LlamaIndex 起家于 RAG（检索增强生成），最初的定位是"连接你的数据与 LLM"。但随着 LLM 能力增强，单纯检索已经不够——用户需要 Agent 来动态决定何时检索、如何检索、检索后做什么。 版本阶段 核心能力 定位 0.9 之前 Index + Query Engine RAG 框架 0.10+ LlamaHub + Workflows 数据连接 + 工作流 当前 Data Agent + AgentWorkflow RAG + Agent 融合 0.10 是分水岭——LlamaIndex 重构了整个包结构，引入了 Workflow（事件驱动工作流）和 Data Agent，正式从 RAG 框架进化为 Agent 框架。 Data Agent：带检索能力的智能体 LlamaIndex 的 Data Agent = LLM + Tools + 检索能力： from llama_index.core.agent import FunctionAgent from llama_index.llms.openai import OpenAI from llama_index.core.tools import FunctionTool, QueryEngineTool llm = OpenAI(model="gpt-4o") # 工具1：网络搜索 def search_web(query: str) -> str: """搜索互联网获取最新信息""" from llama_index.tools.tavilyi import TavilyToolSpec return TavilyToolSpec().search(query) web_tool = FunctionTool.from_defaults(fn=search_web) # 工具2：RAG 查询（复用现有索引） from llama_index.core import VectorStoreIndex index = VectorStoreIndex.from_documents(documents) query_engine = index.as_query_engine(similarity_top_k=5) rag_tool = QueryEngineTool.from_defaults( query_engine=query_engine, name="knowledge_base", description="查询内部知识库，包含产品文档和技术规范" ) # 创建 Data Agent agent = FunctionAgent( tools=[web_tool, rag_tool], llm=llm, system_prompt="你是一个研究助手。先查内部知识库，找不到再搜网络。" ) response = await agent.run("我们的API限流策略是什么？和业界最佳实践比如何？") # Agent会: 1. 先用rag_tool查内部 2. 再用web_tool搜业界实践 3. 综合回答 这是 LlamaIndex 的差异化优势：RAG 不是外部组件，而是 Agent 的原生能力。在 LangChain 里你要手动把 Retriever 包装成 Tool，在 LlamaIndex 里这是内建的。 ...

核心差异：外部知识 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 全链路问题分类 RAG 系统由检索和生成两大环节组成，问题往往难以定位是检索还是生成的锅。先看常见问题清单： 问题现象 可能根因 所在环节 回答"不知道" 检索未命中相关文档 检索 回答包含错误事实 检索到无关文档 / 模型幻觉 检索+生成 回答缺少关键信息 文档分块不当 / 检索 top_k 太小 检索 回答自相矛盾 检索到冲突文档 / 模型推理错误 检索+生成 回答过于笼统 Prompt 未约束细节 / 检索文档质量差 生成+数据 延迟过高 检索过多文档 / 模型 token 过多 全链路 调试工具链 1. 请求追踪中间件 import json, time, uuid from dataclasses import dataclass, asdict @dataclass class RAGTrace: trace_id: str query: str # 检索阶段 query_embedded: list = None retrieval_raw: list = None # 原始检索结果 retrieval_reranked: list = None # 重排后结果 retrieval_filtered: list = None # 过滤后结果 retrieval_latency_ms: float = 0 # 生成阶段 prompt_assembled: str = None model_used: str = None generation_latency_ms: float = 0 # 结果 response: str = None input_tokens: int = 0 output_tokens: int = 0 class RAGTracer: def __init__(self, sink=None): self.sink = sink # elasticsearch / file / stdout def start(self, query) -> RAGTrace: return RAGTrace(trace_id=str(uuid.uuid4()), query=query) async def end(self, trace: RAGTrace): if self.sink: await self.sink.write(json.dumps(asdict(trace), default=str)) # 使用方式 tracer = RAGTracer() async def rag_pipeline(query): trace = tracer.start(query) # 检索 t0 = time.time() results = await vector_db.search(embed(query), top_k=10) trace.retrieval_raw = [{"id": r.id, "score": r.score, "text": r.text[:200]} for r in results] trace.retrieval_latency_ms = (time.time() - t0) * 1000 # 生成 t0 = time.time() response = await llm.complete(query, context=results[:5]) trace.generation_latency_ms = (time.time() - t0) * 1000 trace.response = response await tracer.end(trace) return response 2. 检索质量分析器 class RetrievalAnalyzer: """分析检索结果的质量""" def analyze(self, query, results, ground_truth_ids=None): report = { "query": query, "total_results": len(results), "score_distribution": self._score_stats(results), "score_gap": self._score_gap(results), "potential_issues": [], } # 检查1: 结果太少 if len(results) < 3: report["potential_issues"].append("too_few_results") # 检查2: 分数过低 avg_score = sum(r["score"] for r in results) / len(results) if avg_score < 0.5: report["potential_issues"].append("low_similarity_scores") # 检查3: 分数无区分度 if report["score_gap"] < 0.05: report["potential_issues"].append("no_score_discrimination") # 检查4: 有标注数据时计算 Recall@K if ground_truth_ids: hit = sum(1 for r in results if r["id"] in ground_truth_ids) report["recall_at_k"] = hit / len(ground_truth_ids) if report["recall_at_k"] < 0.7: report["potential_issues"].append("low_recall") return report def _score_stats(self, results): scores = [r["score"] for r in results] return { "min": min(scores), "max": max(scores), "mean": sum(scores) / len(scores), "std": (sum((s - sum(scores)/len(scores))**2 for s in scores) / len(scores)) ** 0.5, } def _score_gap(self, results): """最高分与第二高分的差距""" if len(results) < 2: return 0 sorted_scores = sorted([r["score"] for r in results], reverse=True) return sorted_scores[0] - sorted_scores[1] 常见问题排障 问题1: 空检索 / 低相关度 async def debug_empty_retrieval(query, embed_model, vector_db): """排查检索为空或相关度低的根因""" report = {} # Step 1: 检查 embedding 是否正常 embedding = await embed_model.embed(query) if not embedding or len(embedding) == 0: return {"error": "embedding_failed", "query": query} # Step 2: 检查向量维度是否匹配 db_dim = await vector_db.dimension() if len(embedding) != db_dim: return {"error": "dimension_mismatch", "query_dim": len(embedding), "db_dim": db_dim} # Step 3: 尝试不同 top_k results_variants = {} for k in [5, 10, 20, 50]: results = await vector_db.search(embedding, top_k=k) results_variants[f"top_{k}"] = { "count": len(results), "top_score": results[0]["score"] if results else 0, "avg_score": (sum(r["score"] for r in results) / len(results)) if results else 0, } # Step 4: 检查查询是否有拼写错误/过于简短 report["query_analysis"] = { "length": len(query), "word_count": len(query.split()), "has_special_chars": any(c in query for c in "!@#$%^&*()"), "embedding_norm": sum(x**2 for x in embedding) ** 0.5, } report["retrieval_variants"] = results_variants return report 修复方案： ...