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

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 进入人工复核。五维度评分覆盖了任务完成度、回复质量、效率、安全和用户情绪的完整面板。低分自动告警 + 抽样人工复核的组合，实现了质量监控的闭环。核心是让评分系统成为一个"发现问题的探测器"，而非"完美评分的裁判员"。