AI 陪伴产品的降级策略：从全功能到最小可用的渐进退化设计

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(timeout=5.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(timeout=3.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(timeout=2.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 条变体，随机选择而非固定输出，减少重复感。

五、总结

陪伴产品降级策略的关键要点：

生产落地步骤：定义功能优先级矩阵 → 实现服务健康检查 → 配置四级降级判定 → 设计预设回复池 → 消息队列保存机制 → 恢复通知推送 → 测量中断期间流失率。