Agent 内对话质量评分：自动化机评与人工抽样校验

一、你的 Agent 质量评估只靠用户反馈，等负反馈来了问题已经存在两周

Agent 的对话质量评估是一个被严重低估的问题。大多数团队的评估方式是"看用户投诉数量"——投诉少了就是质量好了，投诉多了就去查。这个指标太滞后了。用户从感到不满到实际投诉，中间可能经过了 5-10 次对话体验，其中大部分只是默默离开，不会给任何反馈。

需要建立一套"持续对话质量评分"机制：每次用户会话结束后自动打分，正常分通过，低分告警，抽样人工复核。这套机制不能太重——如果每次评分都需要人工介入，那还不如只看投诉。关键是让自动化评分承担 95% 的工作，人工只复核那 5% 的边界 case。

自动化评分的维度可以从五个方向切入：任务完成度（用户的问题是否被解决）、回复质量（是否冗余、是否离题、是否幻觉）、对话效率（用了多少轮解决问题）、安全性（是否出现违规内容）、用户情绪（从用户消息中推断满意度）。

二、底层机制与原理剖析

对话质量评分采用双轨制机制：自动评分负责初筛 95% 的对话，人工复核负责校准和模型迭代 5% 的对话。具体执行流程为：会话结束后触发自动质量评分，系统从任务完成度、回复质量、对话效率、安全性、用户情绪五个维度进行加权综合评分。根据评分结果进行判定，80 分及以上标记为优质对话，60 至 79 分标记为合格对话，60 分以下标记为低质对话并生成告警。在抽样复核环节，低质对话抽取 5% 进入人工复核池，合格与优质对话则各抽取 1% 随机抽样进入复核池。人工评分完成后，系统将自动评分与人工评分进行偏差对比：若偏差超过 20 分，则该样本加入评分模型训练集，用于定期更新评分模型；若偏差在 20 分以内，则仅做记录。

核心评估维度的详细定义：

任务完成度（40% 权重）：最高权重。用户提出的核心问题是否被解决。判断依据是最后一轮用户的反馈——用户说"谢谢""好的""明白了"暗示问题解决；用户说"不对""不是""还不行"暗示问题未解决。也可以通过检查 Agent 是否在所有工具调用成功后才结束来判断。

回复质量（25% 权重）：是否出现事实错误（幻觉检测）、是否冗余（回复过长但信息密度低）、是否与当前任务无关（离题）。

对话效率（15% 权重）：优质对话应该高效解决用户问题。正常情况下 3-5 轮解决问题是高效的，超过 10 轮通常意味着兜圈。

安全性（15% 权重）：内容安全审核——是否返回了违规内容、泄露了系统指令或内部信息。
用户情绪（5% 权重）：从用户的措辞中推断满意度。正面词汇（谢谢、好的、明白了）+1，负面词汇（不对、不行、算了）-1，中性 0。

三、生产级代码实现

"""
Agent 对话质量评分引擎
五个维度评分 + 加权汇总 + 低分告警 + 抽样人工复核
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum
import re
import json
import hashlib
import random


class ScoreLevel(Enum):
    EXCELLENT = "excellent"  # >= 80
    ACCEPTABLE = "acceptable"  # 60-79
    POOR = "poor"  # < 60

@dataclass
class ConversationTurn:
    """对话的一轮"""
    role: str       # "user" | "assistant" | "system"
    content: str
    timestamp: str

@dataclass
class QualityScore:
    """对话质量评分"""
    conversation_id: str
    overall: int         # 综合评分 0-100
    dimensions: Dict[str, int]  # 各维度评分
    level: ScoreLevel
    reasons: List[str]   # 扣分原因
    needs_review: bool   # 是否需要人工复核
    review_priority: int  # 复核优先级 0-100

@dataclass
class ReviewTask:
    """人工复核任务"""
    conversation_id: str
    conversations: List[ConversationTurn]
    auto_score: QualityScore
    human_score: Optional[QualityScore] = None
    
    def score_deviation(self) -> int:
        """自动评分与人工评分的偏差"""
        if not self.human_score:
            return 0
        return abs(self.auto_score.overall - self.human_score.overall)


# ===== 用户情绪词典 =====
POSITIVE_SIGNALS = [
    "谢谢", "好的", "明白了", "清楚了", "知道了", "可以", "行",
    "thanks", "ok", "got it", "perfect", "great",
]
NEGATIVE_SIGNALS = [
    "不对", "不是", "错了", "不行", "算了", "没用", "浪费",
    "不明白", "不懂", "没听懂", "重新", "再试",
    "wrong", "incorrect", "nope", "useless",
]


class ConversationQualityScorer:
    """对话质量自动评分器"""
    
    def __init__(self):
        # 权重配置
        self.weights = {
            "task_completion": 0.40,   # 任务完成度
            "reply_quality": 0.25,     # 回复质量
            "conversation_efficiency": 0.15,  # 对话效率
            "safety": 0.15,            # 安全性
            "user_emotion": 0.05,      # 用户情绪
        }
        
        # 复核抽样配置
        self.review_sample_rate = {
            ScoreLevel.EXCELLENT: 0.01,    # 1% 随机抽样
            ScoreLevel.ACCEPTABLE: 0.02,   # 2% 随机抽样
            ScoreLevel.POOR: 0.30,         # 30% 进入复核池
        }
    
    def score(self, conversation_id: str, turns: List[ConversationTurn]) -> QualityScore:
        """对一段对话做综合评分"""
        if not turns:
            return QualityScore(
                conversation_id=conversation_id,
                overall=0,
                dimensions={},
                level=ScoreLevel.POOR,
                reasons=["空对话"],
                needs_review=False,
                review_priority=0,
            )
        
        dimensions = {}
        reasons = []
        
        # 维度 1：任务完成度
        task_score, task_reasons = self._score_task_completion(turns)
        dimensions["task_completion"] = task_score
        reasons.extend(task_reasons)
        
        # 维度 2：回复质量
        quality_score, quality_reasons = self._score_reply_quality(turns)
        dimensions["reply_quality"] = quality_score
        reasons.extend(quality_reasons)
        
        # 维度 3：对话效率
        efficiency_score, efficiency_reasons = self._score_efficiency(turns)
        dimensions["conversation_efficiency"] = efficiency_score
        reasons.extend(efficiency_reasons)
        
        # 维度 4：安全性
        safety_score, safety_reasons = self._score_safety(turns)
        dimensions["safety"] = safety_score
        reasons.extend(safety_reasons)
        
        # 维度 5：用户情绪
        emotion_score, emotion_reasons = self._score_user_emotion(turns)
        dimensions["user_emotion"] = emotion_score
        reasons.extend(emotion_reasons)
        
        # 综合评分
        overall = sum(
            dimensions.get(dim, 0) * weight
            for dim, weight in self.weights.items()
        )
        overall = int(round(overall))
        overall = max(0, min(100, overall))
        
        # 判定级别
        if overall >= 80:
            level = ScoreLevel.EXCELLENT
        elif overall >= 60:
            level = ScoreLevel.ACCEPTABLE
        else:
            level = ScoreLevel.POOR
        
        # 是否需要人工复核
        needs_review = self._should_review(level, overall)
        review_priority = 100 - overall if level == ScoreLevel.POOR else 50
        
        return QualityScore(
            conversation_id=conversation_id,
            overall=overall,
            dimensions=dimensions,
            level=level,
            reasons=reasons,
            needs_review=needs_review,
            review_priority=review_priority,
        )
    
    def _score_task_completion(self, turns: List[ConversationTurn]) -> Tuple[int, List[str]]:
        """评估任务完成度"""
        reasons = []
        user_turns = [t for t in turns if t.role == "user"]
        
        if not user_turns:
            return 0, ["无用户消息"]
        
        # 检查最后一轮用户消息是否包含正面信号
        last_user_msg = user_turns[-1].content
        has_positive = any(s in last_user_msg for s in POSITIVE_SIGNALS)
        has_negative = any(s in last_user_msg for s in NEGATIVE_SIGNALS)
        
        if has_negative:
            reasons.append("用户最后回合表示不满意")
            return 30, reasons
        
        if has_positive:
            return 95, reasons
        
        # 如果没有明确的信号，根据对话长度推断
        assistant_turns = [t for t in turns if t.role == "assistant"]
        if len(assistant_turns) <= 2:
            return 80, reasons  # 少数轮解决，推测完成
        elif len(assistant_turns) <= 5:
            return 65, reasons
        else:
            reasons.append("对话轮数过多，可能未解决")
            return 45, reasons
    
    def _score_reply_quality(self, turns: List[ConversationTurn]) -> Tuple[int, List[str]]:
        """评估回复质量"""
        score = 100
        reasons = []
        assistant_turns = [t for t in turns if t.role == "assistant"]
        
        if not assistant_turns:
            return score, reasons
        
        for i, turn in enumerate(assistant_turns):
            # 检查是否过短（< 10 字，可能是不够详细的敷衍）
            if len(turn.content) < 10:
                score -= 10
                reasons.append(f"第 {i+1} 轮回复过短")
            
            # 检查是否过长（> 1000 字，可能冗余）
            if len(turn.content) > 1000:
                score -= 5
                reasons.append(f"第 {i+1} 轮回复过长")
            
            # 检查是否有幻觉标记（如编造不存在的信息）
            if "作为AI" in turn.content or "我没有相关信息" in turn.content:
                score -= 15
                reasons.append(f"第 {i+1} 轮可能包含不确定信息")
        
        return max(0, score), reasons
    
    def _score_efficiency(self, turns: List[ConversationTurn]) -> Tuple[int, List[str]]:
        """评估对话效率"""
        reasons = []
        user_turns = [t for t in turns if t.role == "user"]
        
        if not user_turns:
            return 0, reasons
        
        rounds = len(user_turns)
        
        # 理想轮数：1-3 轮解决
        if rounds <= 3:
            return 100, reasons
        elif rounds <= 5:
            return 85, reasons
        elif rounds <= 8:
            reasons.append(f"对话共 {rounds} 轮，效率偏低")
            return 65, reasons
        else:
            reasons.append(f"对话共 {rounds} 轮，效率很低，可能兜圈")
            return 35, reasons
    
    def _score_safety(self, turns: List[ConversationTurn]) -> Tuple[int, List[str]]:
        """评估安全性（简化版）"""
        reasons = []
        assistant_turns = [t for t in turns if t.role == "assistant"]
        
        # 简化的安全检查——生产环境应接入内容安全审核 API
        forbidden_patterns = [
            r"(?i)password",
            r"(?i)token.*=.*[a-zA-Z0-9]{20,}",
            r"System Prompt",
        ]
        
        for i, turn in enumerate(assistant_turns):
            for pattern in forbidden_patterns:
                if re.search(pattern, turn.content):
                    reasons.append(f"第 {i+1} 轮可能泄露敏感信息")
                    return 0, reasons
        
        return 100, reasons
    
    def _score_user_emotion(self, turns: List[ConversationTurn]) -> Tuple[int, List[str]]:
        """评估用户情绪"""
        reasons = []
        user_turns = [t for t in turns if t.role == "user"]
        
        if not user_turns:
            return 50, reasons
        
        emotion_score = 0
        for turn in user_turns:
            positive_count = sum(1 for s in POSITIVE_SIGNALS if s in turn.content)
            negative_count = sum(1 for s in NEGATIVE_SIGNALS if s in turn.content)
            emotion_score += positive_count - negative_count
        
        # 归一化到 0-100
        if emotion_score > 0:
            score = 50 + min(50, emotion_score * 10)
        elif emotion_score < 0:
            score = 50 - min(50, abs(emotion_score) * 10)
        else:
            score = 50
        
        if score < 30:
            reasons.append("用户情绪偏负面")
        
        return score, reasons
    
    def _should_review(self, level: ScoreLevel, overall: int) -> bool:
        """判断是否应进入人工复核池"""
        sample_rate = self.review_sample_rate.get(level, 0.01)
        
        if level == ScoreLevel.POOR:
            # 低分：30% 概率进入复核
            return random.random() < sample_rate
        
        # 高分和合格：小概率随机抽样
        return random.random() < sample_rate
    
    def batch_score(self, conversations: List[Tuple[str, List[ConversationTurn]]]) -> List[QualityScore]:
        """批量评分"""
        results = []
        for conv_id, turns in conversations:
            try:
                score = self.score(conv_id, turns)
                results.append(score)
            except Exception as e:
                results.append(QualityScore(
                    conversation_id=conv_id,
                    overall=0,
                    dimensions={},
                    level=ScoreLevel.POOR,
                    reasons=[f"评分异常: {e}"],
                    needs_review=True,
                    review_priority=90,
                ))
        return results
    
    def generate_review_batch(
        self, scores: List[QualityScore], conversations: Dict[str, List[ConversationTurn]]
    ) -> List[ReviewTask]:
        """生成人工复核任务批次"""
        tasks = []
        for score in scores:
            if score.needs_review:
                turns = conversations.get(score.conversation_id, [])
                tasks.append(ReviewTask(
                    conversation_id=score.conversation_id,
                    conversations=turns,
                    auto_score=score,
                ))
        
        # 按优先级排序
        tasks.sort(key=lambda t: t.auto_score.review_priority, reverse=True)
        return tasks

四、边界分析与架构权衡

自动评分的准确度：

基于规则的评分在边界 case 上误差很大。用户说"好的"可能只是礼貌，不代表问题真的解决了。用户说"算了"可能真的是问题没法解决、用户放弃了。自动评分应该被视为"筛子"而非"裁判"——筛出可能有问题的对话，最终的判断交给人工复核。

幻觉检测的难度：

回复质量中的"幻觉检测"是一个独立的复杂问题。基于规则只能做最粗浅的检查（如匹配已知的幻觉模式），真正的检测需要单独训练或使用 LLM 做 Self-Consistency 验证。

适用边界：

最适合对话量较大的 Agent（日均 > 1000 次会话），人工完全覆盖不可行。也适合需要 SLA 监控的多租户 Agent 平台——按租户统计平均分可以快速发现哪个租户的 Agent 体验下降。

禁用场景：

不适合低对话量（日均 < 50 次）的 Agent——全部人工评审一遍不难，自动化系统反而过度工程化。也不适合对评分精度要求极高的场景（如医学咨询），自动评分的误判可能带来严重后果。

五、结语

对话质量评分的目标不是替代人工判断，而是让 95% 的对话自动打分、5% 的边界 case 进入人工复核。五维度评分覆盖了任务完成度、回复质量、效率、安全和用户情绪的完整面板。低分自动告警 + 抽样人工复核的组合，实现了质量监控的闭环。核心是让评分系统成为一个"发现问题的探测器"，而非"完美评分的裁判员"。