Annotation Workflow Guide

Overview

This guide walks you through setting up and running a complete data labeling workflow — from raw unlabeled data to model-ready training labels.

The Labeling Lifecycle

┌─────────────┐    ┌──────────────┐    ┌───────────────┐    ┌──────────┐
│ Raw Data    │ →  │ Task Queue   │ →  │ Annotation    │ →  │ Consensus│
│ (unlabeled) │    │ (prioritised)│    │ (N annotators)│    │ Resolution│
└─────────────┘    └──────────────┘    └───────────────┘    └──────────┘
                                                                  │
┌─────────────┐    ┌──────────────┐    ┌───────────────┐         │
│ Training    │ ←  │ Export       │ ←  │ Quality Audit │ ←───────┘
│ Dataset     │    │ (approved)   │    │ & Adjudication│
└─────────────┘    └──────────────┘    └───────────────┘

Step-by-Step Setup

1. Define Your Schema

Before anything else, decide what labels you need. Open configs/schema_config.yaml and define:

• Categories: What are the possible labels?
• Descriptions: How should annotators interpret each category?
• Examples: What are clear-cut examples of each?
• Hierarchy: Do labels have parent-child relationships?

from src.schema import LabelSchema, Category, LabelType

schema = LabelSchema(
    name="my_project",
    version="1.0",
    label_type=LabelType.CLASSIFICATION,
    categories=[
        Category(
            name="positive",
            description="Expresses satisfaction or approval",
            examples=["Great product!", "Highly recommend"],
            counter_examples=["It's okay (→ neutral)"],
        ),
        # ... more categories
    ],
)

Key decisions:

• Start with fewer categories. You can always split later.
• Write descriptions as if explaining to a new hire who's never seen your data.
• Include counter-examples for every common mistake.

2. Configure the Workflow

Edit configs/labeling_config.yaml. The critical settings:

3. Prepare Your Data

Create Task objects from your unlabeled data:

from src.workflow import Task, WorkflowEngine

tasks = [
    Task(
        task_id=f"item_{i:04d}",
        data={"text": row["text"]},
        difficulty="normal",  # or "easy" / "hard"
    )
    for i, row in enumerate(your_data)
]

engine = WorkflowEngine(redundancy=3, min_agreement=0.67)
engine.add_tasks(tasks)

4. Create Batches and Assign

annotator_ids = ["ann_alice", "ann_bob", "ann_charlie"]

batch = engine.create_batch(
    batch_id="batch_001",
    annotator_ids=annotator_ids,
    strategy="round_robin",
    max_tasks=50,
)

5. Collect Labels

As annotators work (through your UI or the included templates/labeling_interface.html):

engine.submit_label(
    assignment_id="asgn_abc123",
    label="positive",
    duration_sec=4.2,
)

6. Resolve Consensus

Once all annotators have completed their assignments:

results = engine.resolve_all()

for result in results:
    if result.needs_adjudication:
        print(f"  {result.task_id} needs expert review")
    else:
        print(f"  {result.task_id} → '{result.final_label}'")

7. Adjudicate Disagreements

Tasks that can't reach consensus go to a senior annotator:

for task in engine.tasks_needing_adjudication:
    # Show the task + all submitted labels to the expert
    engine.adjudicate(task.task_id, "neutral", adjudicator_id="expert_01")

8. Run Quality Audit

from src.quality_audit import generate_audit_report, Annotation

# Convert engine's internal data to Annotation objects for auditing
annotations = [...]  # build from engine._assignments
report = generate_audit_report(annotations)
print(report)

9. Export Training Data

approved_data = engine.export_approved()
# → [{"task_id": ..., "data": {...}, "label": ..., "metadata": {...}}, ...]

Active Learning Loop

For maximum labeling efficiency, use the active learning module:

┌─────────────────────────────────────────────────┐
│                                                 │
│   ┌──────────┐    ┌───────────┐    ┌────────┐  │
└─→ │ Train    │ →  │ Score     │ →  │ Select │ ─┘
    │ Model    │    │ Unlabeled │    │ Top-K  │
    └──────────┘    └───────────┘    └────────┘
                                          │
                                          ↓
                                    ┌──────────┐
                                    │ Label    │
                                    │ Selected │
                                    └──────────┘

from src.active_learning import entropy_sampling

# After training your model on the current labeled set:
probs = model.predict_proba(X_unlabeled)
result = entropy_sampling(probs.tolist(), budget=100)

# Queue the selected samples for labeling
selected_tasks = [all_tasks[i] for i in result.indices]
engine.add_tasks(selected_tasks)

Handling Schema Changes

When you need to add/rename/merge categories mid-project:

from src.schema import SchemaMigration, MigrationStep, MigrationAction

migration = SchemaMigration(
    from_version="1.0",
    to_version="2.0",
    steps=[
        MigrationStep(
            action=MigrationAction.ADD_CATEGORY,
            params={"name": "neutral"},
            reason="Annotators struggled with ambiguous reviews",
        ),
    ],
)

# Apply to existing labels
migrated_labels, n_changed = migration.apply_to_labels(existing_labels)

Common Pitfalls

Metrics to Track

1. Inter-annotator agreement (Cohen's κ / Fleiss' κ): target ≥ 0.6

2. Adjudication rate: should decrease over time as guidelines improve

3. Label velocity: items labeled per annotator per hour

4. Active learning efficiency: accuracy gain per labeled sample vs. random

5. Distribution drift: chi-squared p-value between weekly batches

Part of ML Engineer Toolkit