data-labeling

数据标注

此技能可帮助 AI Agent 为机器学习项目设计并执行数据标注工作流。它涵盖使用 Label Studio 等工具进行人工标注、使用模型辅助预标注进行半自动标注、优先选择信息量最大的样本的主动学习循环，以及使用标注函数进行程序化弱监督。Agent 负责设计标签模式、编写标注员指南、通过标注员间一致性进行质量控制，并将数据导出为适用于机器学习的格式。

工作流

1. 定义标注模式和指南： 设计标签分类体系——分类任务的类别、NER 的实体类型、目标检测的边界框类别，或语义分割的片段标签。为每个标签编写清晰的标注员指南，并提供正例和反例，涵盖边界情况和模糊场景。

2. 设置标注环境： 使用标注工具（Label Studio、Labelbox 或 Prodigy）配置标签模式，导入原始数据，并设置具有相应权限的用户帐户。定义与任务类型相匹配的标注界面模板——文本分类、跨度标注、图像边界框或多轮对话标注。

3. 使用模型预测进行预标注： 使用现有模型或启发式规则为数据集生成初步标签。标注员随后审核并修正这些预测，而不是从头开始标注，这样可将标注时间缩短 40-60%。对于已有不错基线模型的任务，这尤其有价值。

4. 执行带质量控制的标注： 为标注员分配标注任务，并内置冗余机制——安排 2-3 名标注员标注相同项目，以衡量标注员间一致性（Cohen's kappa 或 Fleiss' kappa）。将一致性较低的项目标记出来，交由高级标注员审核。根据嵌入任务队列的金标准数据集跟踪标注员的准确率。

5. 开展主动学习迭代： 创建初始标注集后，训练模型并使用不确定性采样或委员会查询，选择信息量最大的未标注样本，以供下一轮标注。这样可以最大限度地提高每个已标注样本带来的模型改进，在标注预算有限时至关重要。

6. 导出并验证： 按照训练流水线要求的格式（JSONL、COCO、CoNLL、CSV）导出已标注数据。运行验证检查，确保标签一致、没有遗漏标注，并确认类别分布符合要求。记录标注流程和数据集统计信息，以确保可复现性。

支持的技术

• 标注工具： Label Studio、Labelbox、Prodigy（spaCy）、Amazon SageMaker Ground Truth、CVAT
• 弱监督： Snorkel、Flyingsquid、Skweak
• 主动学习： modAL、ALiPy、Prodigy 主动学习配方
• 一致性指标： Cohen's kappa、Fleiss' kappa、Krippendorff's alpha
• 导出格式： COCO JSON、Pascal VOC XML、CoNLL、JSONL、Hugging Face Datasets

用法

向 Agent 提供原始数据集、任务类型（分类、NER、目标检测等）和标签类别。还可选择指定偏好的标注工具和质量要求（最低标注员间一致性）。Agent 将配置标注环境、设置质量控制并管理标注工作流。

示例

示例 1：用于文本分类的 Label Studio 流水线

Label Studio 标注界面配置（config.xml）：

<View>
  <Header value="Classify the customer review sentiment:" />
  <Text name="text" value="$text" />
  <Choices name="sentiment" toName="text" choice="single-column" showInline="true">
    <Choice value="positive" />
    <Choice value="negative" />
    <Choice value="neutral" />
  </Choices>
  <Textarea name="notes" toName="text" placeholder="Optional: explain ambiguous cases"
            maxSubmissions="1" editable="true" />
</View>

用于设置项目并导入数据的 Python 脚本：

from label_studio_sdk import Client

ls = Client(url="http://localhost:8080", api_key="your-api-key")

project = ls.start_project(
    title="Customer Review Sentiment",
    label_config=open("config.xml").read(),
    description="Label customer reviews as positive, negative, or neutral.",
)

# Import tasks from a CSV file
import csv
tasks = []
with open("reviews.csv") as f:
    for row in csv.DictReader(f):
        tasks.append({"data": {"text": row["review_text"]}, "meta": {"source_id": row["id"]}})

project.import_tasks(tasks)

# Configure inter-annotator overlap: each task gets 2 annotators
project.set_params(maximum_annotations=2, overlap_cohort_percentage=100)
print(f"Created project with {len(tasks)} tasks, 2 annotators per task")

# After annotation, export results
annotations = project.export_tasks(export_type="JSON")
# Compute agreement
from sklearn.metrics import cohen_kappa_score
labels_a1 = [a["annotations"][0]["result"][0]["value"]["choices"][0] for a in annotations if len(a["annotations"]) >= 2]
labels_a2 = [a["annotations"][1]["result"][0]["value"]["choices"][0] for a in annotations if len(a["annotations"]) >= 2]
print(f"Cohen's kappa: {cohen_kappa_score(labels_a1, labels_a2):.3f}")

示例 2：使用 Snorkel 标注函数进行弱监督

import pandas as pd
import numpy as np
from snorkel.labeling import labeling_function, PandasLFApplier, LFAnalysis
from snorkel.labeling.model import LabelModel

SPAM = 1
HAM = 0
ABSTAIN = -1

df = pd.DataFrame({
    "text": [
        "Congratulations! You've won a free iPhone!", "Meeting at 3pm tomorrow",
        "URGENT: claim your prize now!!!", "Can you review the Q3 report?",
        "Buy cheap meds online fast", "Lunch plans for Thursday?",
        "Click here for a free vacation", "Project deadline is next Friday",
    ]
})

@labeling_function()
def lf_contains_free(x):
    return SPAM if "free" in x.text.lower() else ABSTAIN

@labeling_function()
def lf_contains_urgent(x):
    return SPAM if "urgent" in x.text.lower() else ABSTAIN

@labeling_function()
def lf_contains_click(x):
    return SPAM if "click" in x.text.lower() else ABSTAIN

@labeling_function()
def lf_excessive_punctuation(x):
    return SPAM if x.text.count("!") >= 3 else ABSTAIN

@labeling_function()
def lf_contains_meeting(x):
    return HAM if any(w in x.text.lower() for w in ["meeting", "project", "report", "deadline"]) else ABSTAIN

@labeling_function()
def lf_short_and_casual(x):
    return HAM if len(x.text.split()) < 8 and "?" in x.text else ABSTAIN

lfs = [lf_contains_free, lf_contains_urgent, lf_contains_click,
       lf_excessive_punctuation, lf_contains_meeting, lf_short_and_casual]

applier = PandasLFApplier(lfs=lfs)
L_train = applier.apply(df=df)

print(LFAnalysis(L=L_train, lfs=lfs).lf_summary())

# Train the label model to combine noisy labeling functions
label_model = LabelModel(cardinality=2, verbose=True)
label_model.fit(L_train=L_train, n_epochs=500, log_freq=100, seed=42)

# Get probabilistic labels
probs = label_model.predict_proba(L=L_train)
df["label"] = label_model.predict(L=L_train)
df["confidence"] = np.max(probs, axis=1)

# Filter out low-confidence samples for manual review
confident = df[df["confidence"] > 0.8]
needs_review = df[df["confidence"] <= 0.8]
print(f"Confidently labeled: {len(confident)}, needs manual review: {len(needs_review)}")

最佳实践

• 编写详细的标注指南，为每个标签至少提供 3 个正例和 3 个负例，并涵盖边界情况。标注过程中出现模糊情况时，及时更新指南。
• 在标注队列中嵌入金标准条目（占任务的 5-10%），并附上已知的正确标签，以持续监控标注人员的质量，及时发现疲劳或困惑。
• 使用 Cohen's kappa（2 名标注人员）或 Fleiss' kappa（3 名及以上）衡量标注人员间的一致性。若 kappa 低于 0.6，则在继续之前需要修订指南或标签体系。
• 标注预算有限时使用主动学习——通过将标注精力集中在最不确定或最有信息量的样本上，以减少 30-50% 的已标注样本量，同时达到相同的模型性能。
• 对已标注数据集进行版本管理，并记录清晰的元数据（标注人员 ID、时间戳、指南版本），以便追溯标签对应的具体标注活动并复现结果。
• 先开展一轮试点，标注 50-100 个样本，再扩大规模。利用试点校准指南、估算标注速度，并尽早发现标签体系的问题。

边界情况

• 标注人员分歧较大： 当 kappa 低于 0.4 时，任务定义可能存在歧义。可将有问题的标签拆分为更具体的子标签、在指南中补充更多示例，或组织标注人员开展校准会议，以统一理解。
• 原始数据中的类别严重不平衡： 如果目标类别占比很低（< 5%），随机抽样将只能得到少量正例。可使用关键词筛选、基于模型的预选或分层抽样，为标注队列补充可能的正例。
• 弱监督中的标注函数冲突： 多个标注函数对同一样本给出不同结果时，标签模型可能会生成置信度较低的预测。可编写更多更具体的标注函数以提高覆盖率，或将冲突样本转交人工标注。
• 大批次标注导致标注人员疲劳： 连续标注 2-3 小时后，质量会下降。可将工作拆分为每批 100-200 个条目的多个时段，随机排列任务顺序，并根据金标准跟踪每个时段的准确率，以发现质量下滑。
• 项目进行中演变标签体系： 如果标注开始后新增标签类别，就必须重新审核所有已标注数据，以纳入新类别。应使用带版本的标签体系和重新标注队列，而不是追溯修改现有标注。