失败兜底 — AI 交互模式实验室

设计思路

解决的问题：AI 产品依赖外部 API——可能超时、限流、返回格式错误。如果这些错误直接暴露给用户（白屏、红字报错、堆栈信息），用户会觉得产品不可靠，甚至怀疑"是不是我的操作把系统搞崩了"。

设计决策：5 层防御体系——网络重试（指数退避）、JSON 解析回退（3 层）、流式失败转非流式、安全中间件兜底、错误信息脱敏。引用预防错误和帮助用户识别、诊断并恢复（尼尔森第九原则）。

生活类比：餐厅上错菜了——服务员不会说"厨房数据解析错误"或者把炒糊的菜直接端给你。他们会说"不好意思，我帮您换一份"。不管后厨出了什么问题，你作为顾客只应该看到一道正常的菜。这就是兜底——把系统内部的错误翻译成用户能理解、能行动的信息。

预防错误帮助识别与恢复优雅降级系统状态可见性

交互 Demo

切换 Tab 查看三种失败场景的兜底处理方式

模拟网络中断场景：API 调用失败后自动重试，用户只看到最终成功或友好提示

蓝色 = 信息，黄色 = 警告，红色 = 错误，绿色 = 恢复成功，紫色 = 处理步骤。点击 [运行演示] 开始。

实现方式

retry_with_backoff — 指数退避重试

网络失败时按 1s、2s、4s 逐步增加等待时间重试，最多 2 次。避免雪崩式重试加重服务器负担。

parse_json_response — 三层 JSON 回退

第 1 层：直接 json.loads()。第 2 层：去掉 Markdown fences (```json ... ```)。第 3 层：括号匹配提取完整 JSON 对象。三层都失败才报错。

流式失败切非流式 fallback

SSE 流解析失败时，自动切换为非流式请求重试，确保至少能拿到结果。同时 sanitize_error() 脱敏错误信息，防止 API Key 等敏感信息泄露到前端。

源码参考

src/backend/services/utils.py

async def retry_with_backoff(fn, max_retries=2, base_delay=1.0):
    """Exponential backoff: 1s -> 2s -> fail"""
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except Exception as e:
            if attempt == max_retries: raise
            delay = base_delay * (2 ** attempt)
            logger.warning(f"Retry {attempt+1}/{max_retries} in {delay}s: {e}")
            await asyncio.sleep(delay)

def parse_json_response(raw: str):
    """3-layer JSON parsing fallback."""
    # Layer 1: Direct parse
    try: return json.loads(raw)
    except JSONDecodeError: pass

    # Layer 2: Strip markdown fences
    cleaned = re.sub(r'^```(?:json)?\s*|\s*```$', '', raw.strip(), flags=re.M)
    try: return json.loads(cleaned)
    except JSONDecodeError: pass

    # Layer 3: Bracket matching
    match = re.search(r'\{.*\}', raw, re.DOTALL)
    if match:
        try: return json.loads(match.group())
        except JSONDecodeError: pass

    raise ValueError("All 3 JSON parsing layers failed")

def sanitize_error(msg: str) -> str:
    """Strip API keys, tokens, and truncate long messages."""
    msg = re.sub(r'sk-[a-zA-Z0-9]+', '[REDACTED]', msg)
    msg = re.sub(r'Bearer\s+[a-zA-Z0-9\-_]+', 'Bearer [REDACTED]', msg)
    return msg[:300] if len(msg) > 300 else msg