RAG 服务的优雅降级当向量库或 LLM 不可用时如何保持基本可用性一、深度引言与场景痛点用户满怀期待地输入问题等了三秒钟等来一个红色错误提示或者一个空白页面。后端日志里可能写着一行ConnectionRefusedError: milvus:19530但这行日志用户看不到用户只知道这个系统坯了。RAG 服务依赖的外部组件很多向量数据库Milvus/Pinecone/Redis Stack、LLM 服务OpenAI/本地模型、Embedding 服务、重排序服务。任一个组件出问题整个链路就断了。更糟的是这些组件的故障模式各不相同——向量库可能是连接池耗尽LLM 可能是 Rate LimitEmbedding 服务可能是超时。不能用一个统一的异常处理应对所有情况。优雅降级的核心思想是当外部依赖不可用时不退回到出错而是退回到一个质量稍低但仍然能产出有意义答案的模式。用户可能注意到回复不如平时精准但至少不用面对冰冷的错误页面。二、底层机制与原理深度剖析降级不是要么全功能要么全挂而是沿着一个质量降低、可用性保持的阶梯逐级下沉。L0 全功能检索正常的向量检索 LLM 推理链路。在入口处设置健康检查探针每 10 秒对向量库和 LLM 做一次轻量 ping。健康检查不是简单的 TCP 连接检测而是发一条真实的查询如test ping验证完整的读写链路。L1 关键词降级当向量库不可用时退回到关键词检索。方案有两种一是保留一份基于 Whoosh 或 Tantivy 的本地全文索引二是直接对知识库做 in-memory 的 BM25 匹配。关键词检索的精度比向量检索低但不需要外部向量库的依赖。LLM 依然可用的话把关键词结果和问题一起发给 LLM 做答案生成。L1B 轻量模型降级当 LLM 因为 Rate Limit 或超时不可用时不是放弃 LLM而是切换到更便宜的模型。比如 GPT-4 不可用就切 GPT-4o-mini或者切到本地部署的 Qwen-7B。回复质量下降但格式和逻辑框架还在。L2 缓存降级当 LLM 完全不可用时退回到查缓存。把历史上高频问题及其回答缓存下来可以用 Redis 或本地 SQLite。当前问题来了先查缓存——命中直接返回带一个可能不是最新信息的免责声明未命中再进入 L3。L3 静态降级最后的兜底。返回一份预设的 FAQ 列表或搜索引导页告诉用户系统正在维护中您可以尝试以下常用问题或联系人工客服。这一步的关键是让用户知道发生了什么以及他们可以做什么——而不是扔一个500 Internal Server Error。三、生产级代码实现import asyncio import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Awaitable import aiohttp logger logging.getLogger(__name__) class ServiceStatus(str, Enum): HEALTHY healthy DEGRADED degraded UNHEALTHY unhealthy dataclass class CircuitBreaker: 断路器保护系统不被持续故障的外部服务拖垮 name: str failure_threshold: int 3 recovery_timeout: float 30.0 # 半开后等待多久尝试恢复 half_open_max: int 1 # 半开状态允许的试探请求数 _failure_count: int field(default0, initFalse) _last_failure_time: float field(default0, initFalse) _status: ServiceStatus field(defaultServiceStatus.HEALTHY, initFalse) _half_open_requests: int field(default0, initFalse) property def status(self) - ServiceStatus: 检查是否应该从 UNHEALTHY 转为 HALF_OPEN if self._status ServiceStatus.UNHEALTHY: if time.monotonic() - self._last_failure_time self.recovery_timeout: self._status ServiceStatus.HEALTHY # 允许试探 self._half_open_requests 0 return self._status def record_success(self) - None: 记录成功重置故障计数 self._failure_count 0 self._status ServiceStatus.HEALTHY def record_failure(self) - None: 记录失败可能触发熔断 self._failure_count 1 self._last_failure_time time.monotonic() if self._failure_count self.failure_threshold: self._status ServiceStatus.UNHEALTHY logger.error(Circuit breaker [%s] OPENED after %d failures, self.name, self._failure_count) def allow_request(self) - bool: 判断当前是否允许发送请求 return self.status ! ServiceStatus.UNHEALTHY class GracefulDegradationEngine: RAG 服务优雅降级引擎 def __init__( self, vector_search_fn: Callable[[str], Awaitable[list[dict[str, Any]]]], keyword_search_fn: Callable[[str], Awaitable[list[dict[str, Any]]]], llm_generate_fn: Callable[[str, list[dict[str, Any]]], Awaitable[str]], faq_cache: dict[str, str] | None None, ) - None: self._vector_search vector_search_fn self._keyword_search keyword_search_fn self._llm_generate llm_generate_fn self._faq_cache faq_cache or {} # 每个外部依赖独立的断路器 self._vector_breaker CircuitBreaker(namevector_db, failure_threshold3) self._llm_breaker CircuitBreaker(namellm_service, failure_threshold5) # 响应缓存 self._response_cache: dict[str, str] {} async def query(self, user_query: str) - dict[str, Any]: 带降级的主查询入口 docs: list[dict[str, Any]] [] degradation_level 0 # ---- 检索阶段 ---- if self._vector_breaker.allow_request(): try: docs await self._vector_search(user_query) self._vector_breaker.record_success() except (aiohttp.ClientError, asyncio.TimeoutError, ConnectionError) as e: logger.warning(Vector search failed, falling back: %s, e) self._vector_breaker.record_failure() degradation_level 1 else: degradation_level 1 # L1 降级关键词检索 if degradation_level 1 and not docs: try: docs await self._keyword_search(user_query) logger.info(Falling back to keyword search, got %d results, len(docs)) except Exception as e: logger.warning(Keyword search also failed: %s, e) degradation_level 2 # ---- LLM 生成阶段 ---- content if self._llm_breaker.allow_request(): try: content await self._llm_generate(user_query, docs) self._llm_breaker.record_success() except (aiohttp.ClientError, asyncio.TimeoutError, ConnectionError) as e: logger.warning(LLM generation failed, falling back: %s, e) self._llm_breaker.record_failure() degradation_level max(degradation_level, 2) else: degradation_level max(degradation_level, 2) # L2 降级缓存 / FAQ if degradation_level 2 and not content: content self._lookup_cache_or_faq(user_query) if content: content \n\n[提示当前为缓存回复信息可能不是最新] degradation_level 3 if not content else degradation_level # L3 降级静态兜底 if not content: content ( 系统正在维护升级中暂时无法为您提供完整服务。\n\n 您可以尝试以下操作\n 1. 浏览我们的帮助中心https://help.example.com\n 2. 联系人工客服获取即时帮助\n 3. 稍后重试系统预计在 5 分钟内恢复 ) degradation_level 3 return { content: content, degradation_level: degradation_level, docs_used: len(docs), } def _lookup_cache_or_faq(self, query: str) - str: 在响应缓存和 FAQ 中查找 # 精确缓存命中 if query in self._response_cache: return self._response_cache[query] # FAQ 模糊匹配 for faq_q, faq_a in self._faq_cache.items(): if _simple_similarity(query, faq_q) 0.6: return faq_a return def _simple_similarity(a: str, b: str) - float: 简单的字符级相似度 set_a, set_b set(a), set(b) if not set_a or not set_b: return 0.0 return len(set_a set_b) / len(set_a | set_b) async def main() - None: # 模拟外部依赖 async def mock_vector_search(q: str) - list[dict[str, Any]]: # 模拟向量库故障 raise aiohttp.ClientError(Milvus connection refused) async def mock_keyword_search(q: str) - list[dict[str, Any]]: return [{content: 关键词检索结果相关的文档片段...}] async def mock_llm(q: str, docs: list[dict[str, Any]]) - str: return f基于 {len(docs)} 条文档生成的回复 engine GracefulDegradationEngine( vector_search_fnmock_vector_search, keyword_search_fnmock_keyword_search, llm_generatemock_llm, faq_cache{如何重置密码: 请访问设置页面...}, ) result await engine.query(如何重置密码) print(f降级等级: L{result[degradation_level]}) print(f回复: {result[content]}) if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())断路器的设计要点每个外部依赖有独立的失败计数和恢复策略。向量库的故障阈值是 3连接失败要快速熔断LLM 的故障阈值是 5超时比连接失败更常见容忍度稍高。半开状态是恢复的关键——断路器打开后不是一直关上而是等recovery_timeout后允许少量试探请求通过。试探成功则闭合断路器恢复正常失败则继续打开。这个机制避免了雪崩效应——一旦断路器触发后续请求直接走降级路径不再尝试连故障服务给下游恢复的时间。四、边界分析与架构权衡降级不是无代价的。L1 关键词检索的结果质量远不如向量检索如果用户习惯了高精度的语义匹配突然收到关键词匹配的粗糙结果会感到明显的质量落差。建议在降级回复的前面加一个明确的提示如当前检索模式为简化版管理用户预期。缓存降级还有一个时效性风险。FAQ 缓存可能包含过时信息。如果缓存的是退货政策2025版但政策已经更新到 2026 版用户会收到错误信息。解决方法是给缓存答案打上时间戳和来源文档版本号降级返回时告知用户信息的生成时间。断路器的阈值设置也需要权衡。阈值太低比如 1 次失败就熔断网络抖动就会触发降级频繁切换影响用户体验阈值太高比如 10 次故障发生时大量请求堆积超时放大下游压力。生产环境的经验是连接拒绝 2-3 次熔断超时 5 次熔断从低阈值开始上线根据误报率逐步调高。五、总结RAG 服务的优雅降级不是出了问题再想办法而是在系统设计时就把部分组件不可用作为正常状态来对待。多层降级金字塔向量检索→关键词检索→缓存/FAQ→静态兜底让系统在任意组件故障时都能输出有意义的内容。断路器模式保护系统不被故障拖垮半开状态实现自动恢复。关键是降级后的回复质量不如正常模式但用户至少得到一个答案而不是一个错误。这就是优雅的意义——即使摔倒了也要确保着地的姿势不难看。
