排障

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 修复方案： ...