RAG 常见故障排查：幻觉、漏检、延迟的根因分析

故障报告 → 症状分类 → 根因定位 → 修复方案 → 验证
              ↓
    ┌─────────┼─────────┐
    ↓         ↓         ↓
  检索层     生成层    系统层
  漏检       幻觉      延迟
  误检       偏题      超时
  排序错     不完整    成本高

class HallucinationDiagnostic:
    def diagnose(self, question: str, answer: str, contexts: list):
        # 1. 拆分答案为原子论断
        claims = self._decompose_claims(answer)
        
        # 2. 逐条验证
        results = []
        for claim in claims:
            evidence = self._find_evidence(claim, contexts)
            results.append({
                "claim": claim,
                "has_evidence": evidence["score"] > 0.5,
                "evidence_text": evidence["text"],
                "score": evidence["score"]
            })
        
        hallucinated = [r for r in results if not r["has_evidence"]]
        
        return {
            "total_claims": len(claims),
            "hallucinated_claims": len(hallucinated),
            "hallucination_rate": len(hallucinated) / len(claims),
            "details": results,
            "likely_causes": self._identify_causes(hallucinated, contexts)
        }
    
    def _identify_causes(self, hallucinated, contexts):
        causes = []
        
        # 原因1：上下文不足
        if len(contexts) < 3:
            causes.append({
                "cause": "context_insufficient",
                "description": "检索到的上下文数量太少",
                "fix": "增加 top_k 参数或扩大检索范围"
            })
        
        # 原因2：上下文不相关
        if contexts and all(not self._is_relevant(c) for c in contexts):
            causes.append({
                "cause": "context_irrelevant",
                "description": "检索到的上下文与问题不相关",
                "fix": "检查 Embedding 模型质量或分块策略"
            })
        
        # 原因3：LLM 忽略上下文
        if contexts and any(self._is_relevant(c) for c in contexts):
            causes.append({
                "cause": "llm_ignoring_context",
                "description": "上下文包含相关信息但 LLM 未使用",
                "fix": "优化 Prompt，强调'仅基于上下文回答'"
            })
        
        return causes

# 修复前（容易幻觉）
BAD_PROMPT = "根据以下信息回答问题：{context}\n问题：{question}"

# 修复后（抑制幻觉）
GOOD_PROMPT = """请严格基于以下参考信息回答问题。

规则：
1. 只使用参考信息中的内容
2. 如果参考信息不足以回答，请说"根据现有信息无法回答"
3. 每个事实陈述后用 [1], [2] 标注来源
4. 不要编造任何信息

参考信息：
{context}

问题：{question}

回答："""

class RetrievalDiagnostic:
    def diagnose(self, question: str, retrieved: list, ground_truth_docs: list):
        report = {}
        
        # 1. 召回率分析
        retrieved_ids = {r.id for r in retrieved}
        gt_ids = {d.id for d in ground_truth_docs}
        
        report["recall"] = len(retrieved_ids & gt_ids) / len(gt_ids)
        report["missed_docs"] = list(gt_ids - retrieved_ids)
        
        # 2. 分析漏检原因
        for missed_id in report["missed_docs"]:
            missed_doc = self._get_doc(missed_id)
            root_cause = self._analyze_miss(question, missed_doc)
            report.setdefault("miss_reasons", []).append(root_cause)
        
        return report
    
    def _analyze_miss(self, question: str, missed_doc: Document):
        # 原因A: Embedding 相似度过低
        q_emb = self.embedder.encode(question)
        d_emb = self.embedder.encode(missed_doc.content)
        similarity = cosine_similarity(q_emb, d_emb)
        
        if similarity < 0.5:
            return {
                "doc_id": missed_doc.id,
                "cause": "embedding_similarity_low",
                "similarity": similarity,
                "fix": "考虑使用 HyDE 或 Query 重写"
            }
        
        # 原因B: 分块导致信息碎片化
        if len(missed_doc.content) < 50:
            return {
                "doc_id": missed_doc.id,
                "cause": "chunk_too_short",
                "fix": "调整分块策略，增大 chunk_size"
            }
        
        # 原因C: 元数据过滤排除了该文档
        return {
            "doc_id": missed_doc.id,
            "cause": "metadata_filter_excluded",
            "similarity": similarity,
            "fix": "检查元数据过滤条件"
        }

模式1: 语义鸿沟
  Query: "怎么提高模型准确率"
  文档: "模型精度优化方法"  ← 同义词但 Embedding 可能不够近
  修复: 加入 Query 重写或同义词扩展

模式2: 多跳信息分散
  Query: "A和B的区别"
  文档: 分别有A和B的描述，但没有对比性文档
  修复: 使用多查询策略或 Agentic RAG

模式3: 长尾知识
  Query: 某罕见概念
  文档: 存在但 Embedding 训练时未见
  修复: 加入关键词检索做补充

模式4: 版本过期
  Query: 最新信息
  文档: 旧版本未更新
  修复: 定期增量更新 + 时间戳过滤

class LatencyProfiler:
    def profile(self, query: str):
        timings = {}
        
        # 1. Query Embedding
        t0 = time.time()
        q_emb = self.embedder.encode(query)
        timings["embedding"] = time.time() - t0
        
        # 2. 向量检索
        t0 = time.time()
        results = self.vector_store.search(q_emb, top_k=50)
        timings["vector_search"] = time.time() - t0
        
        # 3. 重排序
        t0 = time.time()
        reranked = self.reranker.rerank(query, results, top_k=5)
        timings["rerank"] = time.time() - t0
        
        # 4. Prompt 构建
        t0 = time.time()
        prompt = self._build_prompt(query, reranked)
        timings["prompt_build"] = time.time() - t0
        
        # 5. LLM 生成
        t0 = time.time()
        answer = self.llm.generate(prompt)
        timings["llm_generate"] = time.time() - t0
        
        timings["total"] = sum(timings.values())
        
        return timings

class OptimizedRAG:
    def __init__(self):
        self.semantic_cache = SemanticCache(
            threshold=0.95,
            ttl=3600
        )
        self.query_classifier = QueryClassifier()
    
    def query(self, question: str):
        # 0. 语义缓存检查（~1ms）
        cached = self.semantic_cache.get(question)
        if cached:
            return cached, "cache_hit"
        
        # 1. 查询分类：简单 vs 复杂
        complexity = self.query_classifier.predict(question)
        
        if complexity < 0.3:
            # 快速通道：跳过重排序
            results = self._fast_query(question)
        else:
            # 标准通道：完整流程
            results = self._full_query(question)
        
        # 写入缓存
        self.semantic_cache.set(question, results)
        
        return results
    
    def _fast_query(self, question):
        """快速通道：向量检索 → LLM（跳过重排序）"""
        q_emb = self.embedder.encode(question)
        results = self.vector_store.search(q_emb, top_k=5)
        return self.llm.generate(question, context=results, stream=True)

def diagnose_off_topic(question, answer, contexts):
    # 检查问题覆盖
    question_aspects = extract_question_aspects(question)
    answer_coverage = check_coverage(answer, question_aspects)
    
    # 检查检索相关性
    retrieval_relevance = assess_relevance(question, contexts)
    
    return {
        "question_aspects": question_aspects,
        "covered_aspects": answer_coverage["covered"],
        "missed_aspects": answer_coverage["missed"],
        "retrieval_relevance": retrieval_relevance,
        "root_cause": identify_cause(answer_coverage, retrieval_relevance)
    }

□ 检索阶段
  □ 检索结果是否相关？ → 查看原始检索结果
  □ top_k 是否足够？ → 检查召回率
  □ Embedding 模型是否合适？ → 测试语义相似度
  □ 分块是否合理？ → 检查块边界
  □ 是否需要 Query 重写？ → 检查 Query 与文档的语义鸿沟

□ 生成阶段
  □ Prompt 是否清晰约束？ → 检查 Prompt 模板
  □ 上下文是否足够？ → 检查上下文长度
  □ temperature 是否过高？ → 降到 0-0.3
  □ 模型能力是否足够？ → 换更强模型测试

□ 系统阶段
  □ 延迟在哪个环节？ → Profile 各阶段耗时
  □ 缓存是否生效？ → 检查缓存命中率
  □ 是否有资源瓶颈？ → 检查 GPU/CPU/内存