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

修复方案：

原因解决方案
查询太短添加 Query 扩展/改写
Embedding 模型差换用更强的 Embedding 模型
分块太大/太小调整 chunk_size（推荐 256-512 token）
向量库数据少补充知识库内容
相似度阈值过高降低 similarity_threshold

问题2: 检索到但生成时幻觉

async def debug_hallucination(query, retrieved_docs, response):
    """排查幻觉根因"""
    # 检查1: 检索文档是否包含答案
    has_answer = await llm.judge(
        f"以下文档是否包含回答此问题的信息？\n"
        f"问题：{query}\n文档：{docs_to_text(retrieved_docs)}\n"
        f"回答 yes 或 no"
    )

    # 检查2: 回答是否能在文档中找到支撑
    support_check = await llm.judge(
        f"以下回答的每个论断是否能在文档中找到支撑？\n"
        f"回答：{response}\n文档：{docs_to_text(retrieved_docs)}\n"
        f"输出 JSON：{{"supported": ["论断1"], "unsupported": ["论断2"]}}"
    )

    # 检查3: Prompt 是否明确约束了"基于文档回答"
    prompt_check = "根据" in assembled_prompt or "基于" in assembled_prompt

    return {
        "docs_contain_answer": has_answer,
        "unsupported_claims": support_check.get("unsupported", []),
        "prompt_has_grounding": prompt_check,
    }

问题3: 延迟过高

async def debug_latency(trace: RAGTrace):
    """分析延迟瓶颈"""
    breakdown = {
        "embedding_ms": trace.embedding_latency_ms,
        "retrieval_ms": trace.retrieval_latency_ms,
        "reranking_ms": trace.reranking_latency_ms,
        "prompt_assembly_ms": trace.prompt_latency_ms,
        "generation_ms": trace.generation_latency_ms,
        "total_ms": trace.total_latency_ms,
    }

    # 找到最大瓶颈
    bottleneck = max(breakdown, key=breakdown.get)
    breakdown["bottleneck"] = bottleneck

    # 建议优化
    suggestions = {
        "embedding_ms": "考虑本地 Embedding 模型或缓存 embedding",
        "retrieval_ms": "减少 top_k、优化索引、或用 ANN 替代精确搜索",
        "reranking_ms": "减少 rerank 文档数或换用更快的 reranker",
        "generation_ms": "减少 context 长度、换用更快模型、或启用流式",
    }
    breakdown["suggestion"] = suggestions.get(bottleneck, "检查整体架构")
    return breakdown

日志设计

结构化日志

import structlog

logger = structlog.get_logger()

async def rag_with_logging(query):
    log = logger.bind(query=query, trace_id=uuid.uuid4().hex)

    # 检索阶段
    log.info("retrieval_start")
    results = await vector_db.search(embed(query), top_k=10)
    log.info("retrieval_done",
             result_count=len(results),
             top_score=results[0].score if results else 0,
             latency_ms=elapsed)

    # 过滤阶段
    filtered = [r for r in results if r.score > 0.5]
    log.info("filter_done",
             before=len(results), after=len(filtered),
             threshold=0.5)

    # 生成阶段
    log.info("generation_start",
             context_docs=len(filtered),
             model="gpt-4o")
    response = await llm.complete(query, context=filtered)
    log.info("generation_done",
             tokens_in=count_tokens(query),
             tokens_out=count_tokens(response),
             latency_ms=elapsed)

    return response

调试 Checklist

□ 查询 Embedding 是否正常生成（维度、范数）
□ 向量库中是否有足够数据（总文档数 > 1000）
□ 检索 top_k 是否合理（5-20 之间）
□ 相似度阈值是否过高（建议 0.3-0.7）
□ 文档分块大小是否合理（256-512 token）
□ Prompt 是否明确约束"基于文档回答"
□ Prompt 是否要求"不确定时说不知道"
□ 模型 temperature 是否过高（RAG 建议 0-0.3）
□ 是否存在检索到但未传入生成的文档
□ 是否有重排（reranking）来提升 top 结果质量
□ 缓存 key 是否包含影响结果的变量
□ 是否监控了回答质量（用户反馈/LLM judge）

总结

RAG 调试的核心是"全链路追踪"——从查询 embedding、向量检索、重排过滤到 prompt 组装、模型生成，每一步都要有可观测的 trace。常见问题集中在三类：检索不到（embedding/chunking/数据量）、检索到了但生成不好（prompt/模型/上下文窗口）、延迟高（top_k 太大/context 太长）。建立结构化日志和调试工具链，让问题定位从"猜"变成"看"。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。