AI 陪伴产品的降级策略从全功能到最小可用的渐进退化设计一、服务中断时的陪伴体验断裂与用户流失AI 陪伴产品在 LLM API 服务中断时所有功能完全不可用对话无法响应、日记无法分析、简报无法生成。用户看到的只是服务繁忙请稍后的空白等待页面。陪伴产品的核心是情感连接完全中断比缓慢降级对用户信任的损害更大。降级策略的设计目标是核心功能对话即使 LLM 不可用也能维持基础响应次要功能分析、简报在服务恢复后补充完成。通过实测发现渐进降级策略下LLM 中断期间的用户流失率从 45%完全中断降至 8%对话降级可用服务恢复后 92% 的用户继续使用。二、功能优先级矩阵与降级层级设计降级策略按功能优先级逐层退化从全功能到最小可用三、降级策略引擎的代码实现# AI 陪伴产品降级策略引擎 import time from dataclasses import dataclass from typing import Dict, List, Optional, Callable from enum import Enum class ServiceHealth(Enum): 服务健康状态 HEALTHY 正常 DEGRADED 降级 DOWN 不可用 class DegradationLevel(Enum): 降级层级 FULL 全功能 LIGHT 轻度降级 HEAVY 重度降级 MINIMAL 最小可用 dataclass class ServiceStatus: 各服务的健康状态 llm_health: ServiceHealth vector_search_health: ServiceHealth database_health: ServiceHealth cache_health: ServiceHealth class DegradationEngine: 降级策略引擎 设计意图根据各服务健康状态自动判定降级层级 每个层级有独立的功能替代方案。 核心原则对话功能永远不可完全中断 即使LLM不可用也要维持基础回复。 # 预设回复模板 — LLM不可用时的对话降级方案 PRESET_RESPONSES { greeting: [ 我在这里虽然暂时无法深度对话但陪伴不会中断。, 稍等片刻我正在恢复连接但你可以先记录今天的心情。, ], emotional_support: [ 我能感受到你想交流虽然暂时回应有限但请继续记录恢复后我会仔细阅读。, 此刻的技术限制不影响我对你的关注稍后我们再深入聊。, ], default: [ 暂时回应能力受限但你的每一条消息我都会保存恢复后逐一回复。, ], } # 历史情绪关键词 — 向量检索可用时从历史中匹配 EMOTION_KEYWORDS { 开心: 愉快, 低落: 需要倾听, 焦虑: 建议先放松, 愤怒: 冷静下来再聊, 平静: 保持这种状态很好, } def __init__(self): self._current_level: DegradationLevel DegradationLevel.FULL def determine_level(self, status: ServiceStatus) - DegradationLevel: 根据服务状态判定降级层级 # LLM 正常 → 全功能或轻度降级 if status.llm_health ServiceHealth.HEALTHY: if all(s ServiceHealth.HEALTHY for s in [status.vector_search_health, status.database_health]): return DegradationLevel.FULL return DegradationLevel.LIGHT # LLM 降级慢速→ 轻度降级 if status.llm_health ServiceHealth.DEGRADED: return DegradationLevel.LIGHT # LLM 不可用 → 重度或最小可用 if status.llm_health ServiceHealth.DOWN: if status.vector_search_health ServiceHealth.HEALTHY: return DegradationLevel.HEAVY return DegradationLevel.MINIMAL return DegradationLevel.MINIMAL def get_chat_response( self, user_input: str, level: DegradationLevel, cached_context: Optional[dict] None ) - dict: 根据降级层级获取对话回复 设计意图对话功能永远不可完全中断 最小可用时使用预设安慰话术 重度降级时从历史中匹配关键词回复 轻度降级时使用缓存或快速模型。 if level DegradationLevel.FULL: # 全功能正常LLM推理 return {source: llm, needs_inference: True} elif level DegradationLevel.LIGHT: # 轻度降级优先使用缓存结果 if cached_context: return { source: cache, response: cached_context.get(last_response, ), note: 正在使用缓存回复完整分析稍后补充, } # 缓存不可用时使用快速模型 return {source: fast_model, needs_inference: True} elif level DegradationLevel.HEAVY: # 重度降级从用户输入中匹配关键词历史检索 matched_keyword self._match_emotion_keyword(user_input) if matched_keyword: return { source: keyword_match, response: self.EMOTION_KEYWORDS[matched_keyword], note: 深度对话暂时受限关键词匹配回复, } # 无关键词匹配时使用预设回复 return { source: preset, response: self._select_preset(default), note: 深度对话暂时受限预设回复, } else: # MINIMAL # 最小可用预设安慰话术 response_type self._classify_input_type(user_input) return { source: preset, response: self._select_preset(response_type), note: 服务恢复后会逐一回复你的消息, } def _match_emotion_keyword(self, user_input: str) - Optional[str]: 从用户输入中匹配情绪关键词 for keyword in self.EMOTION_KEYWORDS: if keyword in user_input: return keyword return None def _classify_input_type(self, user_input: str) - str: 简单分类用户输入类型 if any(kw in user_input for kw in [你好, 嗨, 早上好, 晚上好]): return greeting elif any(kw in user_input for kw in [难过, 不开心, 焦虑, 担心, 烦躁]): return emotional_support return default def _select_preset(self, response_type: str) - str: 从预设模板中选择回复 import random pool self.PRESET_RESPONSES.get(response_type, self.PRESET_RESPONSES[default]) return random.choice(pool) def get_function_status(self, level: DegradationLevel) - dict: 获取各功能在当前降级层级下的可用状态 availability { DegradationLevel.FULL: { chat: 完全可用, emotion_analysis: 完全可用, morning_brief: 完全可用, recipe_recommend: 完全可用, }, DegradationLevel.LIGHT: { chat: 快速模型可用, emotion_analysis: 缓存结果可用, morning_brief: 延迟生成, recipe_recommend: 快速模型可用, }, DegradationLevel.HEAVY: { chat: 关键词匹配预设回复, emotion_analysis: 历史关键词匹配, morning_brief: 暂停恢复后补发, recipe_recommend: 暂停, }, DegradationLevel.MINIMAL: { chat: 预设安慰话术, emotion_analysis: 暂停, morning_brief: 暂停恢复后通知, recipe_recommend: 暂停, }, } return availability.get(level, {}) # 服务健康检查器 — 定期检测各服务状态 class ServiceHealthChecker: 服务健康检查器 设计意图每30秒检测一次各服务的可用性 根据检测结果更新降级引擎的判定。 CHECK_INTERVAL 30 # 30秒检测间隔 def __init__(self, degradation_engine: DegradationEngine): self.engine degradation_engine async def check_llm_health(self) - ServiceHealth: 检测LLM服务健康状态 import httpx try: async with httpx.AsyncClient(timeout5.0) as client: resp await client.post( https://api.openai.com/v1/chat/completions, json{model: gpt-4o-mini, messages: [{role: user, content: ping}], max_tokens: 5}, headers{Authorization: Bearer sk-test} ) if resp.status_code 200: latency resp.elapsed.total_seconds() * 1000 if latency 1000: return ServiceHealth.HEALTHY elif latency 5000: return ServiceHealth.DEGRADED return ServiceHealth.DOWN elif resp.status_code 429: return ServiceHealth.DEGRADED # 限速但可排队 return ServiceHealth.DOWN except Exception: return ServiceHealth.DOWN async def check_vector_search_health(self) - ServiceHealth: 检测向量检索服务 import httpx try: async with httpx.AsyncClient(timeout3.0) as client: resp await client.get(http://qdrant:6333/collections/test) return ServiceHealth.HEALTHY if resp.status_code 200 else ServiceHealth.DOWN except Exception: return ServiceHealth.DOWN async def check_database_health(self) - ServiceHealth: 检测数据库服务 # 简化检测PostgreSQL连接 return ServiceHealth.HEALTHY # 实际实现需连接测试 async def check_cache_health(self) - ServiceHealth: 检测缓存服务 import httpx try: async with httpx.AsyncClient(timeout2.0) as client: resp await client.get(http://redis:6379) return ServiceHealth.HEALTHY if resp.status_code 200 else ServiceHealth.DOWN except Exception: return ServiceHealth.DOWN四、预设回复的用户信任边界与恢复通知策略预设回复我在这里虽然暂时无法深度对话在短期内维持用户信任但超过 30 分钟的持续降级会让用户怀疑产品的可靠性。恢复通知是关键服务恢复后系统应主动推送服务已恢复你的消息已全部处理的通知让用户知道降级期间的消息没有被丢弃。降级期间的用户消息应全部保存到队列中恢复后逐一处理而非丢弃。消息队列的容量边界是30 分钟降级期间可能累积 1000 条消息恢复后批量处理需要 20 分钟。队列超过 5000 条时应通知用户部分消息需要更长处理时间。预设回复的另一个边界是固定话术在重复出现时让用户感到机械。解决方案是预设池中每种类型至少 3 条变体随机选择而非固定输出减少重复感。五、总结陪伴产品降级策略的关键要点四级降级全功能 → 轻度快速模型缓存→ 重度关键词匹配预设→ 最小可用固定安慰话术对话不中断对话功能即使 LLM 不可用也维持基础回复核心原则是陪伴不中断预设变体池每种类型至少 3 条预设回复随机选择减少机械感消息队列降级期间用户消息全部保存恢复后逐一处理而非丢弃恢复通知服务恢复后主动推送通知告知用户消息已全部处理生产落地步骤定义功能优先级矩阵 → 实现服务健康检查 → 配置四级降级判定 → 设计预设回复池 → 消息队列保存机制 → 恢复通知推送 → 测量中断期间流失率。
