眾包標註的品質控制

annotation_task_name: "Sentiment Annotation with Checks"
 
surveyflow:
  on: true
  order:
    - survey_instructions
    - annotation
    - survey_attention_check
    - annotation
    - survey_completion

# In your surveyflow survey definitions
survey_attention_check:
  - question: "To confirm you're paying attention, please select 'Strongly Agree'."
    type: radio
    options:
      - Strongly Disagree
      - Disagree
      - Neutral
      - Agree
      - Strongly Agree

annotation_task_name: "Multi-Annotator Sentiment Task"
 
data_files:
  - path: data.json
    list_as_text: false
    sampling: random
 
# Control how many annotators see each item through assignment logic
# This is typically managed through your annotator assignment system

from sklearn.metrics import cohen_kappa_score
 
# After collecting annotations
annotator1_labels = ["Positive", "Negative", "Positive", ...]
annotator2_labels = ["Positive", "Negative", "Neutral", ...]
 
kappa = cohen_kappa_score(annotator1_labels, annotator2_labels)
print(f"Cohen's Kappa: {kappa:.3f}")

from statsmodels.stats.inter_rater import fleiss_kappa
import numpy as np
 
# Build a matrix of label counts per item
# Each row is an item, each column is a label category
ratings_matrix = np.array([
    [3, 0, 0],  # Item 1: 3 Positive, 0 Negative, 0 Neutral
    [2, 1, 0],  # Item 2: 2 Positive, 1 Negative, 0 Neutral
    [0, 0, 3],  # Item 3: 0 Positive, 0 Negative, 3 Neutral
    ...
])
 
kappa = fleiss_kappa(ratings_matrix)
print(f"Fleiss' Kappa: {kappa:.3f}")

[
  {
    "id": "gold_001",
    "text": "I absolutely love this product! Best purchase ever!",
    "is_gold": true,
    "gold_label": "Positive"
  },
  {
    "id": "gold_002",
    "text": "This is terrible. Complete waste of money. Worst experience.",
    "is_gold": true,
    "gold_label": "Negative"
  },
  {
    "id": "regular_001",
    "text": "The product arrived on time and works as expected.",
    "is_gold": false
  }
]

import json
 
def calculate_gold_accuracy(annotations_file, gold_labels):
    with open(annotations_file) as f:
        annotations = json.load(f)
 
    annotator_scores = {}
 
    for item_id, item_annotations in annotations.items():
        if item_id in gold_labels:
            expected = gold_labels[item_id]
            for annotator, label in item_annotations.items():
                if annotator not in annotator_scores:
                    annotator_scores[annotator] = {'correct': 0, 'total': 0}
                annotator_scores[annotator]['total'] += 1
                if label == expected:
                    annotator_scores[annotator]['correct'] += 1
 
    for annotator, scores in annotator_scores.items():
        accuracy = scores['correct'] / scores['total']
        print(f"{annotator}: {accuracy:.1%} gold accuracy")
 
    return annotator_scores

import json
from statistics import mean, stdev
 
def analyze_timing(annotations_file):
    with open(annotations_file) as f:
        data = json.load(f)
 
    times = []
    for item in data.values():
        if 'time_spent' in item:
            times.append(item['time_spent'])
 
    avg_time = mean(times)
    std_time = stdev(times)
 
    # Flag annotations that are too fast (< 2 std below mean)
    threshold = max(avg_time - 2 * std_time, 2)  # At least 2 seconds
 
    flagged = [t for t in times if t < threshold]
    print(f"Average time: {avg_time:.1f}s")
    print(f"Flagged as too fast: {len(flagged)} items")

def quality_check_annotations(annotations_file):
    with open(annotations_file) as f:
        data = json.load(f)
 
    issues = []
 
    for annotator_id, items in group_by_annotator(data).items():
        labels = [item['label'] for item in items]
 
        # Check for single-label bias (always selecting same option)
        unique_labels = set(labels)
        if len(unique_labels) == 1 and len(labels) > 10:
            issues.append(f"{annotator_id}: Only used label '{labels[0]}'")
 
        # Check for position bias (always selecting first option)
        # Requires knowing option order in your schema
 
        # Check for very fast submissions
        times = [item.get('time_spent', 0) for item in items]
        avg_time = sum(times) / len(times) if times else 0
        if avg_time < 3:
            issues.append(f"{annotator_id}: Average time only {avg_time:.1f}s")
 
    return issues