AI 用户满意度指标体系：从 CSAT 到 AI-NPS

AI 用户满意度指标体系
├── 基础指标（继承自传统软件）
│   ├── CSAT-AI（会话级满意度）
│   ├── NPS-AI（推荐意愿）
│   └── CES-AI（任务完成努力度）
├── AI 特有指标
│   ├── 信任度（Trust Score）
│   ├── 可控性感知（Controllability）
│   ├── 透明度感知（Transparency）
│   └── 惊喜指数（Delight Index）
└── 行为指标（隐式测量）
    ├── 重试率（Retry Rate）
    ├── 人工求助率（Escalation Rate）
    ├── 会话深度（Conversation Depth）
    └── 回访留存（Return Retention）

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CSATQuestion:
    question_id: str
    question: str
    scale: int = 5  # 1-5 或 1-10
    labels: dict = field(default_factory=dict)

# CSAT-AI 问卷设计
CSAT_AI_QUESTIONS = [
    CSATQuestion(
        question_id="overall",
        question="整体来看，本次与 AI 的对话是否满足您的需求？",
        scale=5,
        labels={1: "非常不满意", 2: "不满意", 3: "一般", 4: "满意", 5: "非常满意"}
    ),
    CSATQuestion(
        question_id="accuracy",
        question="AI 提供的信息是否准确可靠？",
        scale=5,
        labels={1: "完全不可靠", 5: "非常可靠"}
    ),
    CSATQuestion(
        question_id="understanding",
        question="AI 是否准确理解了您的意图？",
        scale=5,
        labels={1: "完全没理解", 5: "完全理解"}
    ),
    CSATQuestion(
        question_id="completeness",
        question="AI 的回答是否完整地解答了您的问题？",
        scale=5,
        labels={1: "非常不完整", 5: "非常完整"}
    ),
    CSATQuestion(
        question_id="tone",
        question="AI 的表达方式是否令您感到舒适？",
        scale=5,
        labels={1: "非常不适", 5: "非常舒适"}
    ),
]

class CSATAI:
    """CSAT-AI 计算器"""
    
    def __init__(self, questions: list[CSATQuestion] = None):
        self.questions = questions or CSAT_AI_QUESTIONS

    def compute(self, responses: list[dict]) -> dict:
        """
        responses: [{"question_id": str, "score": int, "user_id": str, "session_id": str}]
        """
        from collections import defaultdict
        import numpy as np

        by_question = defaultdict(list)
        by_user = defaultdict(list)
        by_session = defaultdict(list)

        for r in responses:
            by_question[r["question_id"]].append(r["score"])
            by_user[r["user_id"]].append(r["score"])
            by_session[r["session_id"]].append(r["score"])

        results = {}
        # 各维度得分
        results["dimensions"] = {
            q_id: {
                "mean": np.mean(scores),
                "std": np.std(scores),
                "count": len(scores),
                "pct_satisfied": np.mean([1 if s >= 4 else 0 for s in scores]),
                "pct_dissatisfied": np.mean([1 if s <= 2 else 0 for s in scores]),
            }
            for q_id, scores in by_question.items()
        }
        # 整体 CSAT（满意比例 = 4-5 分占比）
        all_scores = [r["score"] for r in responses]
        results["overall"] = {
            "csat_score": np.mean([1 if s >= 4 else 0 for s in all_scores]) * 100,
            "mean_score": np.mean(all_scores),
            "response_count": len(all_scores),
        }
        return results

class CSATSampling:
    """CSAT 采样策略"""
    
    STRATEGIES = {
        "random": "随机采样（5-10% 的会话）",
        "long_conversation": "长对话优先（>10轮）",
        "negative_signals": "负面信号触发（用户修改/重试）",
        "first_time": "新用户首次对话",
        "milestone": "里程碑触发（第10次/第50次对话）",
    }

    def __init__(self, strategy: str = "random", rate: float = 0.08):
        self.strategy = strategy
        self.rate = rate

    def should_survey(self, session: dict) -> bool:
        if self.strategy == "random":
            import random
            return random.random() < self.rate
        elif self.strategy == "negative_signals":
            return self._has_negative_signals(session)
        elif self.strategy == "long_conversation":
            return session.get("turn_count", 0) > 10
        elif self.strategy == "first_time":
            return session.get("is_first_session", False)
        elif self.strategy == "milestone":
            return session.get("session_count", 0) in [1, 10, 50, 100]
        return False

    def _has_negative_signals(self, session: dict) -> bool:
        signals = [
            session.get("retry_count", 0) >= 2,  # 用户重试了2次以上
            session.get("message_edits", 0) >= 3,  # 用户编辑消息3次以上
            session.get("contains_negative_words", False),  # 包含负面词汇
            session.get("session_duration_ms", 0) > 300000,  # 会话超过5分钟仍未完成
        ]
        return any(signals)

@dataclass
class AINPSSurvey:
    # 标准 NPS 问题
    nps_question: str = "您有多大可能向朋友或同事推荐本产品？"
    nps_scale: int = 11  # 0-10

    # AI 扩展问题
    trust_question: str = "在多大程度上您信任 AI 的回答足以用于重要决策？"
    trust_scale: int = 11

    dependency_question: str = "如果本产品明天消失，您会有多困扰？"
    dependency_scale: int = 11

    # 开放性问题
    open_questions: list = field(default_factory=lambda: [
        "AI 做得最好的方面是什么？",
        "AI 最需要改进的方面是什么？",
        "在什么场景下您不敢使用 AI 的回答？",
    ])


class AINPSCalculator:
    def __init__(self):
        self.survey = AINPSSurvey()

    def compute_nps(self, scores: list[int]) -> dict:
        """计算 NPS"""
        promoters = sum(1 for s in scores if s >= 9)
        passives = sum(1 for s in scores if 7 <= s <= 8)
        detractors = sum(1 for s in scores if s <= 6)
        total = len(scores)
        
        nps = (promoters - detractors) / total * 100 if total > 0 else 0
        return {
            "nps": round(nps, 1),
            "promoters_pct": round(promoters / total * 100, 1) if total else 0,
            "passives_pct": round(passives / total * 100, 1) if total else 0,
            "detractors_pct": round(detractors / total * 100, 1) if total else 0,
            "total_responses": total,
            "mean_score": round(sum(scores) / total, 2) if total else 0,
        }

    def compute_trust_score(self, scores: list[int]) -> dict:
        """计算信任度"""
        import numpy as np
        return {
            "mean_trust": round(np.mean(scores), 2),
            "high_trust_pct": round(np.mean([1 if s >= 8 else 0 for s in scores]) * 100, 1),
            "low_trust_pct": round(np.mean([1 if s <= 3 else 0 for s in scores]) * 100, 1),
        }

    def full_report(self, responses: list[dict]) -> dict:
        nps_scores = [r["nps_score"] for r in responses]
        trust_scores = [r["trust_score"] for r in responses]
        dep_scores = [r["dependency_score"] for r in responses]
        
        return {
            "nps": self.compute_nps(nps_scores),
            "trust": self.compute_trust_score(trust_scores),
            "dependency": self.compute_trust_score(dep_scores),  # 复用计算逻辑
            "trust_nps_gap": self._trust_nps_gap(nps_scores, trust_scores),
        }

    def _trust_nps_gap(self, nps_scores: list[int], trust_scores: list[int]) -> dict:
        """信任度与推荐意愿的差距分析"""
        import numpy as np
        gap = np.mean(trust_scores) - np.mean(nps_scores)
        return {
            "gap": round(gap, 2),
            "interpretation": (
                "信任度低于推荐度：用户愿意推荐但不信任——口碑风险"
                if gap < -1 else
                "信任度与推荐度一致"
                if abs(gap) <= 1 else
                "信任度高于推荐度：用户信任但不推荐——可能有使用门槛"
            )
        }

class TrustMeter:
    """AI 信任度多维测量"""
    
    DIMENSIONS = {
        "competence": "能力信任：AI 是否具备解决问题的能力？",
        "reliability": "可靠信任：AI 的回答是否前后一致且可预测？",
        "benevolence": "善意信任：AI 是否以用户利益为优先？",
        "transparency": "透明信任：AI 是否清楚说明其能力和局限？",
    }

    def measure(self, user_responses: dict) -> dict:
        """
        user_responses: {"competence": 1-7, "reliability": 1-7, ...}
        """
        import numpy as np
        scores = {dim: user_responses.get(dim, 4) for dim in self.DIMENSIONS}
        overall = np.mean(list(scores.values()))
        return {
            "overall_trust": round(overall, 2),
            "dimensions": scores,
            "trust_level": self._trust_level(overall),
            "weakest_dimension": min(scores, key=scores.get),
        }

    def _trust_level(self, score: float) -> str:
        if score >= 6: return "高度信任"
        elif score >= 4.5: return "中等信任"
        elif score >= 3: return "低信任"
        else: return "不信任"

class ControllabilityMeter:
    """可控性感知测量"""
    
    QUESTIONS = [
        ("steering", "我能引导 AI 按照我想要的方向工作"),
        ("correction", "当 AI 理解错误时，我能有效地纠正它"),
        ("override", "我能让 AI 忽略它的默认行为"),
        ("feedback_speed", "我的反馈能快速影响 AI 的后续行为"),
        ("predictability", "我能预测 AI 对我的输入会做出什么反应"),
    ]

    def measure(self, responses: dict) -> dict:
        scores = {q[0]: responses.get(q[0], 3) for q in self.QUESTIONS}
        import numpy as np
        return {
            "overall": round(np.mean(list(scores.values())), 2),
            "dimensions": scores,
            "lowest": min(scores, key=scores.get),
        }

class DelightMeter:
    """惊喜指数测量"""
    
    # 隐式信号
    IMPLICIT_SIGNALS = {
        "copy_action": {"weight": 0.15, "desc": "用户复制了 AI 的回答"},
        "share_action": {"weight": 0.30, "desc": "用户分享了 AI 的回答"},
        "bookmark": {"weight": 0.20, "desc": "用户收藏了对话"},
        "positive_emoji": {"weight": 0.10, "desc": "用户发送了正面 emoji"},
        "follow_up_deep": {"weight": 0.15, "desc": "用户基于 AI 回答深入追问"},
        "no_edit_copy": {"weight": 0.10, "desc": "用户直接复制使用，无修改"},
    }

    # 显式信号
    EXPLICIT_QUESTION = "AI 的回答是否超出了您的预期？"

    def measure(self, session: dict, explicit_score: int = None) -> dict:
        # 隐式评分
        implicit_score = 0
        triggered = []
        for signal, config in self.IMPLICIT_SIGNALS.items():
            if session.get(signal, False):
                implicit_score += config["weight"]
                triggered.append(signal)

        # 显式评分（如有）
        result = {
            "implicit_score": round(implicit_score, 3),
            "triggered_signals": triggered,
        }
        if explicit_score is not None:
            result["explicit_score"] = explicit_score
            result["delight_index"] = round(
                (implicit_score * 0.4 + (explicit_score / 10) * 0.6), 3
            )
        else:
            result["delight_index"] = implicit_score

        return result

class BehavioralMetrics:
    """行为指标计算器"""
    
    def compute_all(self, sessions: list[dict]) -> dict:
        return {
            "retry_rate": self._retry_rate(sessions),
            "escalation_rate": self._escalation_rate(sessions),
            "conversation_depth": self._conversation_depth(sessions),
            "return_retention": self._return_retention(sessions),
            "abandonment_rate": self._abandonment_rate(sessions),
            "rephrase_rate": self._rephrase_rate(sessions),
        }

    def _retry_rate(self, sessions: list[dict]) -> dict:
        """重试率：用户对 AI 回答不满意而重新提问的比例"""
        total_turns = sum(s.get("turn_count", 0) for s in sessions)
        retry_turns = sum(s.get("retry_count", 0) for s in sessions)
        rate = retry_turns / total_turns if total_turns > 0 else 0
        return {
            "rate": round(rate, 4),
            "total_retries": retry_turns,
            "total_turns": total_turns,
            "interpretation": "健康" if rate < 0.1 else ("需关注" if rate < 0.2 else "严重")
        }

    def _escalation_rate(self, sessions: list[dict]) -> dict:
        """人工求助率：用户从 AI 转向人工客服的比例"""
        ai_sessions = len(sessions)
        escalated = sum(1 for s in sessions if s.get("escalated_to_human", False))
        rate = escalated / ai_sessions if ai_sessions > 0 else 0
        return {
            "rate": round(rate, 4),
            "escalated_count": escalated,
            "total_sessions": ai_sessions,
        }

    def _conversation_depth(self, sessions: list[dict]) -> dict:
        """会话深度：平均对话轮次"""
        import numpy as np
        depths = [s.get("turn_count", 0) for s in sessions]
        return {
            "mean": round(np.mean(depths), 2) if depths else 0,
            "median": round(np.median(depths), 2) if depths else 0,
            "p75": round(np.percentile(depths, 75), 2) if depths else 0,
        }

    def _return_retention(self, sessions: list[dict]) -> dict:
        """回访留存：用户在7天内再次使用 AI 的比例"""
        users = {}
        for s in sessions:
            uid = s.get("user_id")
            if uid not in users:
                users[uid] = []
            users[uid].append(s.get("timestamp"))
        
        retained = 0
        total = 0
        for uid, timestamps in users.items():
            if len(timestamps) >= 2:
                timestamps.sort()
                first = timestamps[0]
                # 检查7天内是否有第二次使用
                for t in timestamps[1:]:
                    if (t - first).days <= 7:
                        retained += 1
                        break
            total += 1
        
        return {
            "d7_retention": round(retained / total, 4) if total > 0 else 0,
            "total_users": total,
            "retained_users": retained,
        }

    def _abandonment_rate(self, sessions: list[dict]) -> dict:
        """放弃率：用户在 AI 回答后直接离开，无后续交互"""
        abandoned = sum(1 for s in sessions if s.get("abandoned", False))
        total = len(sessions)
        return {
            "rate": round(abandoned / total, 4) if total > 0 else 0,
            "count": abandoned,
        }

    def _rephrase_rate(self, sessions: list[dict]) -> dict:
        """重述率：用户重新组织语言再次提问的比例"""
        total_first_turns = 0
        rephrased = 0
        for s in sessions:
            turns = s.get("turns", [])
            for i, turn in enumerate(turns):
                if i == 0 or turn.get("is_retry", False):
                    continue
                prev = turns[i-1] if i > 0 else None
                if prev and self._is_rephrase(prev.get("user_message", ""), 
                                               turn.get("user_message", "")):
                    rephrased += 1
                total_first_turns += 1
        return {
            "rate": round(rephrased / total_first_turns, 4) if total_first_turns > 0 else 0,
            "count": rephrased,
        }

    def _is_rephrase(self, msg1: str, msg2: str) -> bool:
        """简单判断是否为重述（语义相似度高）"""
        from difflib import SequenceMatcher
        ratio = SequenceMatcher(None, msg1.lower(), msg2.lower()).ratio()
        return 0.4 < ratio < 0.8  # 相似但不完全相同

class SatisfactionDashboard:
    """满意度指标仪表盘"""
    
    def generate(self, data: dict) -> dict:
        csat = data.get("csat", {})
        nps = data.get("nps", {})
        behavioral = data.get("behavioral", {})
        trust = data.get("trust", {})
        
        # 健康度评分
        health = self._health_score(csat, nps, behavioral, trust)
        
        return {
            "summary": {
                "health_score": health["score"],
                "health_level": health["level"],
                "key_metrics": {
                    "CSAT": csat.get("overall", {}).get("csat_score", "N/A"),
                    "NPS": nps.get("nps", {}).get("nps", "N/A"),
                    "Trust Score": trust.get("overall_trust", "N/A"),
                    "Retry Rate": behavioral.get("retry_rate", {}).get("rate", "N/A"),
                    "D7 Retention": behavioral.get("return_retention", {}).get("d7_retention", "N/A"),
                },
            },
            "trends": self._compute_trends(data),
            "alerts": self._generate_alerts(csat, nps, behavioral, trust),
        }

    def _health_score(self, csat, nps, behavioral, trust) -> dict:
        score = 0
        # CSAT (25%)
        csat_score = csat.get("overall", {}).get("csat_score", 0)
        score += min(csat_score / 100, 1) * 25
        # NPS (25%)
        nps_score = nps.get("nps", {}).get("nps", 0)
        score += max(0, min((nps_score + 100) / 200, 1)) * 25
        # 信任度 (25%)
        trust_score = trust.get("overall_trust", 0)
        score += min(trust_score / 7, 1) * 25
        # 行为指标 (25%)
        retry = behavioral.get("retry_rate", {}).get("rate", 0.5)
        retention = behavioral.get("return_retention", {}).get("d7_retention", 0)
        score += (1 - retry) * 12.5 + retention * 12.5
        
        score = round(score, 1)
        level = "优秀" if score >= 80 else ("良好" if score >= 60 else ("需关注" if score >= 40 else "严重"))
        return {"score": score, "level": level}

    def _generate_alerts(self, csat, nps, behavioral, trust) -> list:
        alerts = []
        if csat.get("overall", {}).get("csat_score", 100) < 60:
            alerts.append({"level": "critical", "metric": "CSAT", "msg": "CSAT 低于 60，用户满意度严重不足"})
        if nps.get("nps", {}).get("nps", 0) < 0:
            alerts.append({"level": "critical", "metric": "NPS", "msg": "NPS 为负数，贬损者超过推荐者"})
        if behavioral.get("retry_rate", {}).get("rate", 0) > 0.2:
            alerts.append({"level": "warning", "metric": "Retry Rate", "msg": "重试率超过 20%，AI 回答质量可能存在问题"})
        if trust.get("overall_trust", 7) < 3.5:
            alerts.append({"level": "warning", "metric": "Trust", "msg": "用户信任度偏低，可能影响产品留存"})
        return alerts