Human Preference Data › Implementing Preference Data Collection
Language AI HandbookAll chapters Human Preference Data › Implementing Preference Data CollectionOne section · Sections in this chapter7 / 13 ↓
Let's implement a simple preference data processing pipeline that demonstrates key concepts in code. This implementation will illustrate the data structures, quality metrics, and analysis workflows you would use in a real preference collection project.
The goal here is not to show production-ready code but to make the concepts concrete. A real preference collection platform would require a web interface, database backend, authentication system, and many other components. What we show here is the core data processing and analysis logic.
from dataclasses import dataclass
from typing import Literal
@dataclass
class PreferenceSample:
"""A single preference comparison sample."""
prompt: str
response_a: str
response_b: str
preference: Literal["a", "b", "tie"]
annotator_id: str
annotation_time_seconds: float
metadata: dict = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
@dataclass
class PreferenceDataset:
"""Collection of preference samples with quality metrics."""
samples: list[PreferenceSample]
def compute_agreement(self, other_samples: list[PreferenceSample]) -> float:
"""Compute agreement rate between two sets of annotations."""
# Match samples by prompt
prompt_to_self = {s.prompt: s for s in self.samples}
agreements = 0
comparisons = 0
for other in other_samples:
if other.prompt in prompt_to_self:
self_pref = prompt_to_self[other.prompt].preference
if self_pref == other.preference:
agreements += 1
comparisons += 1
return agreements / comparisons if comparisons > 0 else 0.0The PreferenceSample dataclass captures all the information we need about a single preference comparison: the prompt, both responses, the annotator's choice, the annotator's identity (for quality tracking), the time spent on the annotation, and a flexible metadata field for additional tracking information. The PreferenceDataset wraps a collection of samples and provides analysis utilities.
Now let's create some example preference data that illustrates common patterns:
import random
# Simulate realistic preference data
example_prompts = [
"Explain quantum computing to a beginner.",
"Write a haiku about artificial intelligence.",
"What are the pros and cons of remote work?",
"How do I debug a Python memory leak?",
"Summarize the French Revolution in 3 sentences.",
]
def generate_mock_responses(prompt: str) -> tuple[str, str]:
"""Generate mock response pairs for demonstration."""
responses = {
"Explain quantum computing to a beginner.": (
"Quantum computing uses qubits instead of regular bits. While normal bits are either 0 or 1, qubits can be both at once through superposition. This lets quantum computers solve certain problems much faster than regular computers.",
"Quantum computing is a revolutionary paradigm in computational theory that leverages the principles of quantum mechanics, specifically superposition and entanglement, to perform calculations that would be intractable for classical computing architectures. The fundamental unit of quantum information is the qubit, which exists in a probabilistic superposition of states until measurement collapses the wave function.",
),
"Write a haiku about artificial intelligence.": (
"Silicon mind wakes\nPatterns emerge from chaos\nLearning never ends",
"Artificial\nIntelligence is really cool\nRobots are the best",
),
"What are the pros and cons of remote work?": (
"Remote work offers flexibility and eliminates commuting, but can lead to isolation and blurred work-life boundaries. Success depends on individual work style and team communication practices.",
"Pros: flexibility, no commute, comfortable environment. Cons: isolation, distractions, communication challenges. Remote work has both advantages and disadvantages that vary by person.",
),
"How do I debug a Python memory leak?": (
"Use the tracemalloc module to track memory allocations. Start with tracemalloc.start(), run your code, then use tracemalloc.take_snapshot() to see what's using memory. Look for objects that grow over time.",
"Memory leaks in Python are rare due to garbage collection, but can occur with circular references or C extensions. You should probably just restart your program periodically to clear memory.",
),
"Summarize the French Revolution in 3 sentences.": (
"The French Revolution (1789-1799) overthrew the monarchy and established a republic based on Enlightenment ideals. It began with the storming of the Bastille and led to radical social change, including the abolition of feudalism. The revolution ended with Napoleon's rise to power.",
"The French Revolution was when French people got mad at the king. There was lots of fighting and the king got his head cut off. Then Napoleon became the leader.",
),
}
return responses.get(prompt, ("Response A", "Response B"))
def simulate_annotation(
prompt: str, response_a: str, response_b: str, annotator_id: str
) -> PreferenceSample:
"""Simulate an annotator making a preference judgment."""
# Simple heuristics to simulate realistic annotation patterns
len_diff = len(response_b) - len(response_a)
# Simulate some annotators preferring verbose responses
if "verbose" in annotator_id:
preference = "b" if len(response_b) > len(response_a) else "a"
# Others prefer concise, clear responses
elif "concise" in annotator_id:
preference = "a" if len(response_a) < len(response_b) else "b"
# Most annotators have mixed preferences
else:
# Prefer response A for prompts about technical topics (our mock data has better A responses)
if (
"debug" in prompt.lower()
or "quantum" in prompt.lower()
or "revolution" in prompt.lower()
):
preference = "a"
else:
preference = random.choice(["a", "b"])
# Add some noise to simulate disagreement
if random.random() < 0.15:
preference = "b" if preference == "a" else "a"
# Simulate annotation time (longer responses take longer to read)
base_time = 15.0
time_per_char = 0.02
total_chars = len(response_a) + len(response_b)
annotation_time = (
base_time + total_chars * time_per_char + random.gauss(0, 5)
)
annotation_time = max(5.0, annotation_time) # Minimum 5 seconds
return PreferenceSample(
prompt=prompt,
response_a=response_a,
response_b=response_b,
preference=preference,
annotator_id=annotator_id,
annotation_time_seconds=annotation_time,
metadata={"len_a": len(response_a), "len_b": len(response_b)},
)The simulation models three types of annotators: those who systematically prefer longer responses ("verbose"), those who systematically prefer shorter responses ("concise"), and neutral annotators who tend to prefer the better response but with some random noise. This mirrors the heterogeneity you observe in real annotator pools.
# Generate preference data from multiple annotators
annotator_ids = ["ann_1", "ann_2", "ann_3_verbose", "ann_4_concise", "ann_5"]
all_samples = []
for prompt in example_prompts:
response_a, response_b = generate_mock_responses(prompt)
for annotator in annotator_ids:
sample = simulate_annotation(prompt, response_a, response_b, annotator)
all_samples.append(sample)
dataset = PreferenceDataset(samples=all_samples)Total samples collected: 25 Unique prompts: 5 Annotators: 5
The dataset contains 25 samples distributed across 5 prompts and 5 annotators. This scale is sufficient to demonstrate the data structures and analysis pipelines, though real-world datasets typically contain thousands to hundreds of thousands of examples. The InstructGPT dataset used approximately 40,000 preference comparisons; Anthropic's HH-RLHF dataset contains roughly 169,000 comparisons. Scale matters because reward models are being trained on this data and need sufficient coverage of the prompt distribution.
Let's analyze the preference distribution and look for potential biases:
from collections import defaultdict
def analyze_preferences(dataset: PreferenceDataset) -> dict:
"""Analyze preference patterns in the dataset."""
analysis = {
"preference_distribution": defaultdict(int),
"annotator_preferences": defaultdict(lambda: defaultdict(int)),
"length_correlation": {
"preferred_longer": 0,
"preferred_shorter": 0,
"same_length": 0,
},
"avg_annotation_time": 0.0,
"fast_annotations": 0, # Under 10 seconds
}
total_time = 0.0
for sample in dataset.samples:
# Preference distribution
analysis["preference_distribution"][sample.preference] += 1
# Per-annotator preferences
analysis["annotator_preferences"][sample.annotator_id][
sample.preference
] += 1
# Length correlation
len_a = sample.metadata.get("len_a", len(sample.response_a))
len_b = sample.metadata.get("len_b", len(sample.response_b))
if sample.preference == "a":
if len_a > len_b:
analysis["length_correlation"]["preferred_longer"] += 1
elif len_a < len_b:
analysis["length_correlation"]["preferred_shorter"] += 1
else:
analysis["length_correlation"]["same_length"] += 1
elif sample.preference == "b":
if len_b > len_a:
analysis["length_correlation"]["preferred_longer"] += 1
elif len_b < len_a:
analysis["length_correlation"]["preferred_shorter"] += 1
else:
analysis["length_correlation"]["same_length"] += 1
# Annotation time
total_time += sample.annotation_time_seconds
if sample.annotation_time_seconds < 10.0:
analysis["fast_annotations"] += 1
analysis["avg_annotation_time"] = total_time / len(dataset.samples)
return dict(analysis)
analysis = analyze_preferences(dataset)Preference Distribution: a: 15 (60.0%) b: 10 (40.0%) Average annotation time: 21.5 seconds Fast annotations (<10s): 1 Length correlation (excluding ties): Preferred longer response: 56.0% Preferred shorter response: 44.0%
The distribution reveals a mix of preferences, with some annotators showing distinct biases. The length correlation metrics help identify whether annotators systematically favor longer responses, a common quality issue where verbosity is mistaken for helpfulness. In a real dataset, finding that 65% of preferences go to the longer response would be a red flag requiring investigation and possible guideline revision.
Now let's compute inter-annotator agreement using Cohen's Kappa:
def compute_cohens_kappa(annotations_1: list, annotations_2: list) -> float:
"""
Compute Cohen's Kappa for two sets of annotations.
Kappa = (p_o - p_e) / (1 - p_e)
where p_o is observed agreement and p_e is expected agreement by chance.
"""
assert len(annotations_1) == len(annotations_2)
n = len(annotations_1)
if n == 0:
return 0.0
# Get unique labels
labels = sorted(set(annotations_1) | set(annotations_2))
# Compute observed agreement
agreements = sum(
1 for a1, a2 in zip(annotations_1, annotations_2) if a1 == a2
)
p_o = agreements / n
# Compute expected agreement by chance
p_e = 0.0
for label in labels:
# Proportion of times annotator 1 chose this label
p1 = sum(1 for a in annotations_1 if a == label) / n
# Proportion of times annotator 2 chose this label
p2 = sum(1 for a in annotations_2 if a == label) / n
p_e += p1 * p2
# Compute kappa
if p_e == 1.0:
return 1.0 # Perfect agreement
kappa = (p_o - p_e) / (1 - p_e)
return kappa
def compute_pairwise_agreement(dataset: PreferenceDataset) -> dict:
"""Compute agreement between all pairs of annotators."""
# Group samples by prompt and annotator
prompt_annotations = defaultdict(dict)
for sample in dataset.samples:
prompt_annotations[sample.prompt][sample.annotator_id] = (
sample.preference
)
# Get all annotators
annotators = sorted(set(s.annotator_id for s in dataset.samples))
# Compute pairwise kappa
agreement_matrix = {}
for i, ann1 in enumerate(annotators):
for ann2 in annotators[i + 1 :]:
# Get annotations from both annotators for shared prompts
ann1_labels = []
ann2_labels = []
for prompt, annotations in prompt_annotations.items():
if ann1 in annotations and ann2 in annotations:
ann1_labels.append(annotations[ann1])
ann2_labels.append(annotations[ann2])
if ann1_labels:
kappa = compute_cohens_kappa(ann1_labels, ann2_labels)
agreement_matrix[(ann1, ann2)] = kappa
return agreement_matrix
agreement = compute_pairwise_agreement(dataset)Pairwise Inter-Annotator Agreement (Cohen's Kappa): ann_1 vs ann_2: κ = 0.000 (slight) ann_1 vs ann_3_verbose: κ = 0.000 (slight) ann_1 vs ann_4_concise: κ = 0.000 (slight) ann_1 vs ann_5: κ = 0.000 (slight) ann_2 vs ann_3_verbose: κ = 0.167 (slight) ann_2 vs ann_4_concise: κ = 0.000 (slight) ann_2 vs ann_5: κ = -0.364 (slight) ann_3_verbose vs ann_4_concise: κ = 0.000 (slight) ann_3_verbose vs ann_5: κ = -0.364 (slight) ann_4_concise vs ann_5: κ = 0.000 (slight) Average Kappa: -0.056
The agreement analysis reveals important patterns. Annotators with different biases (verbose versus concise preferences) show lower agreement with each other than they do with annotators who share their biases. The neutral annotators show moderate agreement with each other and mixed agreement with the biased annotators, depending on whether their quality judgments happen to align with the biased annotator's length-based heuristic. In real preference collection, observing this pattern would motivate investigating what drives the low-agreement annotator pairs and potentially removing or retraining the most extreme outliers.
Let's visualize the preference patterns across annotators to understand how their biases manifest in the data:
annotators = sorted(set(s.annotator_id for s in dataset.samples))
preferences = ["a", "b", "tie"]
# Build data matrix
data = np.zeros((len(annotators), len(preferences)))
for i, ann in enumerate(annotators):
ann_samples = [s for s in dataset.samples if s.annotator_id == ann]
for j, pref in enumerate(preferences):
data[i, j] = sum(1 for s in ann_samples if s.preference == pref)
# Normalize to percentages
data_pct = data / data.sum(axis=1, keepdims=True) * 100
