企业官网建设流程全解析

RAG 回答异常怎么查：串起切片、检索与模型调用

RAG 回答异常时，先把切片、检索结果、上下文组装和模型调用串起来。每段保存脱敏摘要与耗时，就能看出内容在哪一步偏离，而不是把错误都归给模型。

向量检索与上下文长度的常见边界

搭建 RAG 系统时，开发者很容易沉迷于 Baseline 演示的顺利表现，从而忽视了生产环境下的复杂边缘条件。常见问题通常来自两个相互叠加的环节：

第一，切片粒度与向量相似度阈值配置失当。分块（Chunking）策略过于粗暴，导致上下文缺乏关键语义关联。相似度匹配算法将高噪音的文档片段强行送入上下文窗口，直接引发了大模型的理解偏差。

第二，缺乏 Prompt Token 溢出截断保护。当向量数据库返回了 5 个长文本片段后，编排逻辑直接将它们拼接进 Prompt 中，超出了上游模型 API 的最大输入限制。而上游 API 没有返回明确的错误码，直接进入无限挂起状态，直至网关层抛出 HTTP 504。

+-------------------------------------------------------------------------+ | RAG 线上常见故障诱因与现象对照 | +-------------------------------------------------------------------------+ | 故障环节 | 根因分析 | 前端外部表现 | +----------------+-------------------------------+------------------------+ | 向量化阶段 | Embedding 模型离线或超时 | 检索为空，回答与背景无关 | | 检索匹配阶段 | Top-K 过大且 Threshold 设得太低 | 产生严重幻觉，答非所问 | | 上下文组装 | 未计算 Token 边界致 Prompt 溢出| 服务挂起，返回 504 超时 | | 模型推理阶段 | API 速率受限 (429 Rate Limit) | 界面长时间停滞或报错500 | +----------------+-------------------------------+------------------------+

线上故障定位的核心证据链拓扑

证据链可以从入口分配Trace-ID，再记录 Chunk 得分、Token 用量和各阶段耗时。日志应脱敏，并设置采样与留存期限；Trace 只能帮助还原调用链，不能替代责任判断。

Python 可部署的 RAG 链路追踪与全链路容错断路器

以下代码展示 RAG API 编排器的几个边界：用 Trace ID 关联日志、重排 Top-K Chunk、限制 Token 输入，并在外部 API 失败时触发熔断。每条分支仍需用超时、空结果和依赖故障测试验证。

import uuid import time import logging from typing import List, Dict, Any, Optional logging.basicConfig(level=logging.INFO, format="[%(asctime)s][%(levelname)s][Trace: %(threadName)s] %(message)s") logger = logging.getLogger("RAGPipeline") class VectorStoreUnavailableError(Exception): """向量数据库不可用自定义异常""" pass class LLMAPIError(Exception): """上游 LLM 调用异常""" pass class SafeRAGPipeline: def __init__(self, similarity_threshold: float = 0.72, max_prompt_tokens: int = 1500): self.similarity_threshold = similarity_threshold self.max_prompt_tokens = max_prompt_tokens def mock_vector_search(self, query: str, trace_id: str) -> List[Dict[str, Any]]: """模拟向量数据库检索，带异常保护""" logger.info(f"[{trace_id}] 开始向量数据库检索, Query: '{query}'") # 模拟检索耗时 time.sleep(0.15) # 模拟边缘场景：查询触发特殊条件抛出超时异常 if "timeout_trigger" in query: raise VectorStoreUnavailableError("Vector DB index search timed out (Connection limit reached)") return [ {"chunk_id": "c101", "text": "爷爷在1975年参加了铁路线建设，负责日常养路与调度工作。", "score": 0.89}, {"chunk_id": "c102", "text": "修筑铁路期间，工人们常常在篝火旁哼唱地方民歌。", "score": 0.75}, {"chunk_id": "c103", "text": "现代电力机车采用双向供电系统，额定电压为25kV。", "score": 0.42} # 低相关度噪音 ] def truncate_context_by_tokens(self, chunks: List[Dict[str, Any]], max_tokens: int, trace_id: str) -> str: """根据 Token 估算安全限制裁剪上下文""" accumulated_text = "" current_tokens = 0 for c in chunks: # 粗略计算：1 汉字 ≈ 1.5 Tokens estimated_tokens = int(len(c["text"]) * 1.5) if current_tokens + estimated_tokens > max_tokens: logger.warning(f"[{trace_id}] 上下文触发安全截断! 已忽略后续 Chunk: {c['chunk_id']}") break accumulated_text += f"\n[参考文档]: {c['text']}" current_tokens += estimated_tokens return accumulated_text def call_llm_with_circuit_breaker(self, prompt: str, trace_id: str) -> str: """带熔断机制的 LLM 调用""" logger.info(f"[{trace_id}] 请求上游 LLM, Prompt 字符数: {len(prompt)}") # 模拟 API 失败率机制 if len(prompt) > 4000: raise LLMAPIError("HTTP 400: Context length exceeded threshold limit") time.sleep(0.4) return "基于记录，您的爷爷曾在1975年参与铁路线养护工作，那些岁月留下了踏实而温暖的足迹。" def process_query(self, user_query: str) -> Dict[str, Any]: trace_id = f"tr-{uuid.uuid4().hex[:8]}" start_timestamp = time.time() evidence_chain = {"trace_id": trace_id, "query": user_query, "steps": []} try: # 1. 向量检索阶段 raw_chunks = self.mock_vector_search(user_query, trace_id) evidence_chain["steps"].append({"step": "VECTOR_SEARCH", "raw_chunks_count": len(raw_chunks)}) # 2. 过滤低相关度 Chunk filtered_chunks = [c for c in raw_chunks if c["score"] >= self.similarity_threshold] logger.info(f"[{trace_id}] 过滤后有效 Chunk 数: {len(filtered_chunks)}") evidence_chain["steps"].append({"step": "SCORE_FILTER", "valid_chunks": len(filtered_chunks)}) if not filtered_chunks: logger.warning(f"[{trace_id}] 未找到高可信度匹配项，降级为温情预置提示") return { "trace_id": trace_id, "answer": "抱歉，关于这段回忆暂未检索到确切的文字记载。但或许您可以重新描述一下那个温暖的故事？", "status": "FALLBACK_NO_MATCH" } # 3. 组装并裁剪上下文 safe_context = self.truncate_context_by_tokens(filtered_chunks, self.max_prompt_tokens, trace_id) final_prompt = f"上下文资料：{safe_context}\n\n问题：{user_query}\n请用亲切温暖的语气回答。" # 4. LLM 生成 answer = self.call_llm_with_circuit_breaker(final_prompt, trace_id) total_duration = round((time.time() - start_timestamp) * 1000, 2) evidence_chain["steps"].append({"step": "LLM_SUCCESS", "duration_ms": total_duration}) return { "trace_id": trace_id, "answer": answer, "status": "SUCCESS", "evidence": evidence_chain } except VectorStoreUnavailableError as ve: logger.error(f"[{trace_id}] 向量库服务不可用: {str(ve)}") return { "trace_id": trace_id, "answer": "回忆档案库正处于定期维护中，请稍后再试。", "status": "DEGRADED_VECTOR_ERROR" } except LLMAPIError as le: logger.error(f"[{trace_id}] 大模型 API 异常: {str(le)}") return { "trace_id": trace_id, "answer": "智能助手思绪有些紊乱，稍等片刻它就会恢复正常。", "status": "DEGRADED_LLM_ERROR" } except Exception as e: logger.critical(f"[{trace_id}] 未知运行时崩溃: {str(e)}", exc_info=True) return { "trace_id": trace_id, "answer": "系统遇到了未预期的扰动。", "status": "SYSTEM_CRITICAL_FAILURE" } # 验证故障捕获逻辑 if __name__ == "__main__": pipeline = SafeRAGPipeline(similarity_threshold=0.7) print("\n=== 测试场景 A: 正常检索流程 ===") res_a = pipeline.process_query("讲讲爷爷修铁路的事") print(f"回答结果: {res_a['answer']}\n状态: {res_a['status']}") print("\n=== 测试场景 B: 向量库检索超时降级 ===") res_b = pipeline.process_query("timeout_trigger 相关的历史") print(f"回答结果: {res_b['answer']}\n状态: {res_b['status']}")

用证据链支持故障排查

大模型应用除了正常路径，还要覆盖空检索、模型超时和上下文超限。每条失败路径都应有可观察状态和明确回退。

Trace ID 应贯穿检索与模型调用，相似度阈值和超时则从评测集与服务目标中得出。日志能还原输入、候选片段和调用状态后，RAG 问题才有稳定的排查入口。