第03章 人工评估：系统性收集人类判断
第03章 人工评估：系统性收集人类判断
“当机器无法判断好坏时，人类仍然是最终的裁判。” —— 评估工程实践
LLM-as-Judge 很强大，但对于需要高精度的评估（涉及专业判断、用户偏好、行业知识），人工评估仍然不可替代。本章讲的不是"让人随便看一看"，而是如何系统性地设计和执行高质量的人工评估。
3.1 人工评估的设计原则
# ====================================================
# 人工评估系统设计
# ====================================================
"""
人工评估的常见失败原因：
1. 标注任务设计不清晰 → 标注员不知道怎么判断
修复：提供详细的评分手册 + 示例
2. 标注员选择不当 → 标注员不具备判断所需的知识
修复：明确标注员资质要求，必要时培训
3. 标注员内部一致性低 → 同一任务不同时间打了不同的分
修复：加入重复案例，监控自一致性
4. 标注员间一致性低 → 不同人打了不同的分
修复：培训期的校准练习，定期对齐会议
5. 顺序效应 → 看了一批高质量的案例后，对下一批的评分更苛刻
修复：随机化案例顺序，定期重置基准
"""
import asyncio
import random
import statistics
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from openai import AsyncOpenAI
import json
import time
client = AsyncOpenAI()
@dataclass
class AnnotationTask:
"""一个标注任务"""
task_id: str
input_text: str
output_text: str
is_calibration: bool = False # 是否是校准用的已知分案例
gold_score: Optional[float] = None # 校准案例的标准分数
metadata: Dict = field(default_factory=dict)
@dataclass
class Annotation:
"""一条标注记录"""
task_id: str
annotator_id: str
scores: Dict[str, float] # dimension → score
comments: str = ""
duration_seconds: float = 0.0
timestamp: float = field(default_factory=time.time)
class AnnotationQualityMonitor:
"""
标注质量监控器
监控：
1. 标注员自一致性（同一个案例不同时间的分是否一致）
2. 标注员间一致性（不同标注员对同一案例的分是否一致）
3. 校准准确性（标注员在校准案例上的表现）
4. 异常检测（答题时间异常短，可能是随机填写）
"""
def __init__(self, calibration_threshold: float = 0.8):
self.calibration_threshold = calibration_threshold
self.annotations: List[Annotation] = []
def add_annotation(self, annotation: Annotation):
self.annotations.append(annotation)
def calculate_inter_annotator_agreement(
self,
task_id: str,
dimension: str
) -> Optional[float]:
"""
计算特定任务和维度上的标注员间一致性
使用 Cohen's Kappa 的简化版（相关系数）
"""
# 找出对这个任务打了分的所有标注员
task_annotations = [
a for a in self.annotations
if a.task_id == task_id and dimension in a.scores
]
if len(task_annotations) < 2:
return None
scores = [a.scores[dimension] for a in task_annotations]
if len(scores) < 2:
return None
# 简化：用分数的标准差衡量一致性（标准差越小，一致性越高）
# 真实场景应该用 Cohen's Kappa 或 Krippendorff's Alpha
std_dev = statistics.stdev(scores) if len(scores) > 1 else 0
max_possible_std = 2.0 # 假设分数范围0-5
agreement = 1.0 - (std_dev / max_possible_std)
return max(0.0, agreement)
def check_annotator_reliability(
self,
annotator_id: str,
calibration_tasks: List[AnnotationTask]
) -> Dict:
"""
评估标注员的可靠性：
- 在校准案例上的准确率
- 平均评分时间（太快可能在随机填写）
"""
annotator_annotations = [
a for a in self.annotations
if a.annotator_id == annotator_id
]
if not annotator_annotations:
return {"status": "no_data"}
# 检查校准案例准确率
calibration_scores = []
for task in calibration_tasks:
if not task.is_calibration or task.gold_score is None:
continue
matching = next(
(a for a in annotator_annotations if a.task_id == task.task_id),
None
)
if matching:
# 用第一个维度的分数和标准答案比较
first_score = list(matching.scores.values())[0] if matching.scores else None
if first_score is not None:
diff = abs(first_score - task.gold_score)
calibration_scores.append(1.0 - min(diff / 4.0, 1.0)) # 归一化
calibration_accuracy = (
statistics.mean(calibration_scores)
if calibration_scores else None
)
# 检查评分时间
avg_duration = statistics.mean([a.duration_seconds for a in annotator_annotations])
suspiciously_fast = avg_duration < 10 # 10秒以内完成一个任务很可疑
# 检查分数分布（是否只给一个分数，如全给5分）
all_scores_flat = [
s for a in annotator_annotations
for s in a.scores.values()
]
score_variance = statistics.variance(all_scores_flat) if len(all_scores_flat) > 1 else 0
low_variance = score_variance < 0.1 # 几乎没有差异化
return {
"annotator_id": annotator_id,
"total_annotations": len(annotator_annotations),
"calibration_accuracy": calibration_accuracy,
"avg_duration_seconds": avg_duration,
"reliability_flags": {
"too_fast": suspiciously_fast,
"low_variance": low_variance,
"calibration_fail": (
calibration_accuracy is not None and
calibration_accuracy < self.calibration_threshold
),
}
}
def generate_report(self) -> str:
if not self.annotations:
return "暂无标注数据"
annotators = list(set(a.annotator_id for a in self.annotations))
total = len(self.annotations)
report = f"标注质量报告\n"
report += f"总标注数: {total}，标注员数: {len(annotators)}\n\n"
avg_duration = statistics.mean([a.duration_seconds for a in self.annotations])
report += f"平均标注时间: {avg_duration:.1f}秒/条\n"
return report
# 演示人工评估流程
async def simulate_human_evaluation():
print("=== 人工评估流程演示 ===\n")
monitor = AnnotationQualityMonitor()
# 模拟3个标注员对同一个案例打分
task = AnnotationTask(
task_id="t001",
input_text="分析电动汽车市场的竞争格局",
output_text="特斯拉领先，比亚迪快速追赶，传统车企转型中...",
)
# 模拟标注员打分（真实情况下是人工界面采集）
simulated_annotations = [
Annotation("t001", "ann_001", {"准确性": 4.0, "完整性": 3.0}, duration_seconds=45),
Annotation("t001", "ann_002", {"准确性": 4.0, "完整性": 4.0}, duration_seconds=52),
Annotation("t001", "ann_003", {"准确性": 3.0, "完整性": 3.0}, duration_seconds=38),
]
for ann in simulated_annotations:
monitor.add_annotation(ann)
# 计算一致性
agreement_accuracy = monitor.calculate_inter_annotator_agreement("t001", "准确性")
agreement_completeness = monitor.calculate_inter_annotator_agreement("t001", "完整性")
print(f"案例 t001 标注员间一致性:")
print(f" 准确性维度: {agreement_accuracy:.2f}" if agreement_accuracy else " 准确性维度: 数据不足")
print(f" 完整性维度: {agreement_completeness:.2f}" if agreement_completeness else " 完整性维度: 数据不足")
# 0.8以上认为一致性良好
if agreement_accuracy and agreement_accuracy >= 0.8:
print("\n✅ 标注员间一致性良好，可以信任这批标注数据")
else:
print("\n⚠️ 标注员间分歧较大，建议增加讨论对齐，或使用平均值")
print(monitor.generate_report())
asyncio.run(simulate_human_evaluation())
3.2 小规模高效人工评估策略
# ====================================================
# 高效人工评估：没有标注预算时的策略
# ====================================================
"""
现实情况：大多数 AI Agent 项目没有专业标注预算
解决策略：
1. 开发者自评 + 校准
- 风险：主观偏见
- 对策：使用盲评（隐藏版本信息），延迟评估（隔天再看）
2. 用户反馈转化
- 利用现有用户反馈（点赞/踩、重新生成率）
- 主动设计简单的1-3分评价界面
3. 聚焦精评
- 不是每条输出都评，而是重点评估：
a. 最坏的案例（找到问题）
b. 最好的案例（建立基准）
c. 随机采样（代表性样本）
4. 精简评估维度
- 从5-8个维度压缩到1-2个最关键的维度
- 甚至只做偏好评估（A 和 B 哪个更好？）
"""
class LeanEvaluationPlan:
"""
低成本高效评估计划
专为资源有限的场景设计
"""
def __init__(self, time_budget_hours: float, evaluators: int):
self.time_budget_hours = time_budget_hours
self.evaluators = evaluators
# 假设每条评估需要3分钟
self.total_capacity = int(time_budget_hours * 60 / 3 * evaluators)
def generate_sampling_plan(
self,
total_outputs: int,
high_risk_count: int = 0
) -> Dict:
"""
生成采样计划：在预算内最大化覆盖
"""
budget = self.total_capacity
# 优先级1：高风险案例（100%评估）
high_risk_eval = min(high_risk_count, int(budget * 0.3))
budget -= high_risk_eval
# 优先级2：随机采样
random_sample = min(
max(50, int(total_outputs * 0.1)), # 至少50条，或10%
budget
)
return {
"total_capacity": self.total_capacity,
"high_risk_cases": high_risk_eval,
"random_sample": random_sample,
"total_to_evaluate": high_risk_eval + random_sample,
"coverage_rate": (high_risk_eval + random_sample) / total_outputs,
"recommendation": (
"样本量充足，结果可信" if random_sample >= 100
else "样本量偏小，增加评估员或延长时间"
)
}
def suggest_quick_eval_format(self) -> str:
"""
根据预算推荐最简化的评估格式
"""
if self.total_capacity < 50:
return "偏好评估（A vs B 哪个更好？）— 最快，每条只需30秒"
elif self.total_capacity < 200:
return "1-3分整体评分 — 简单，每条约1分钟"
else:
return "多维度评分（2-3个维度）— 细致，每条约3分钟"
# 演示低成本评估计划
plan = LeanEvaluationPlan(time_budget_hours=4, evaluators=2)
sampling = plan.generate_sampling_plan(total_outputs=500, high_risk_count=20)
print("=== 低成本评估计划 ===\n")
print(f"时间预算: 4小时，2名评估员")
print(f"总评估能力: {sampling['total_capacity']} 条")
print(f"\n采样分配:")
print(f" 高风险案例: {sampling['high_risk_cases']} 条")
print(f" 随机采样: {sampling['random_sample']} 条")
print(f" 总计: {sampling['total_to_evaluate']} 条")
print(f" 覆盖率: {sampling['coverage_rate']:.1%}")
print(f"\n建议评估格式: {plan.suggest_quick_eval_format()}")
print(f"评估建议: {sampling['recommendation']}")
本章小结
-
人工评估的质量取决于设计，而不是执行：如果标注任务设计不清晰，增加标注员只会增加更多的噪声。在开始收集数据之前，先把评估协议、评分量表和示例准备好。
-
标注员间一致性是评估数据可信度的核心指标：如果两个人对同一个输出的评分差异超过一个等级，说明评估标准不清晰，或标注员需要更多培训。目标是一致性（相关系数）> 0.8。
-
校准案例是发现低质量标注员的最有效工具：在任务集中随机混入"已知答案"的案例，可以快速识别随机作答或理解有偏差的标注员，而不需要昂贵的全数据校验。
-
资源有限时，聚焦精评而不是广评：与其粗略评估1000条，不如精细评估100条（覆盖高风险案例 + 随机样本）。评估质量比数量更重要。
-
偏好评估（A vs B）通常比绝对打分更可靠：人类更擅长做比较判断，而不是给出绝对分数。如果预算有限，让人判断"哪个更好"比让人打1-5分通常更一致。
# 核心行动：建立你的首批人工评估数据
def create_annotation_batch(agent_outputs: list, sample_size: int = 30) -> list:
"""
从一批输出中创建标注任务：
- 10条随机采样（代表性）
- 10条最短输出（可能质量差）
- 10条最长输出（复杂案例）
"""
sorted_by_len = sorted(agent_outputs, key=lambda x: len(x.get("output", "")))
tasks = (
random.sample(agent_outputs, min(10, len(agent_outputs))) +
sorted_by_len[:10] +
sorted_by_len[-10:]
)
return tasks[:sample_size] # 去重后不超过 sample_size
本章提示词模板
【模板1：标注任务设计提示词】
我需要设计一个人工评估任务，评估我的 AI Agent 的输出质量：
Agent 类型：{agent_type}
评估维度：{dimensions}（如：准确性、完整性、可读性）
评估人：{evaluator_profile}（如：没有专业背景的内部员工）
每条任务的预期时间：{time_per_task}分钟
请设计：
1. 评估界面应该展示什么信息（输入？输出？参考答案？）
2. 每个维度的评分量表（1-5分，每分的明确定义）
3. 3-5个校准案例（含标准答案和评分理由）
4. 标注员培训材料的核心内容（一页纸）
5. 如何处理争议案例（标注员分歧较大时）
【模板2：人工评估结果分析提示词】
我完成了一轮人工评估，有以下结果：
评估样本量：{sample_size}
各维度平均分：{scores_by_dimension}
标注员间一致性：{agreement_scores}
校准案例准确率：{calibration_accuracy}
评分分布：{score_distribution}（如：1分占10%，2分占15%...）
请帮我分析：
1. 这批评估数据的可信度如何？（根据一致性和校准准确率）
2. 哪个维度的问题最突出？
3. 分数分布是否存在异常（如评分过于集中）？
4. 根据评估结果，最优先需要改进的是什么？
5. 下一轮评估应该增加哪类案例？