import os
from typing import Dict, List, Optional
from loguru import logger
from data_juicer.ops.mapper.annotation.annotation_mapper import (
LabelStudioAnnotationMapper,
)
from ...base_op import OPERATORS
[docs]
@OPERATORS.register_module("human_preference_annotation_mapper")
class HumanPreferenceAnnotationMapper(LabelStudioAnnotationMapper):
"""Operator for human preference annotation using Label Studio."""
DEFAULT_LABEL_CONFIG = """
<View className="root">
<Style>
.root {
box-sizing: border-box;
margin: 0;
padding: 0;
font-family: 'Roboto',
sans-serif;
line-height: 1.6;
background-color: #f0f0f0;
}
.container {
margin: 0 auto;
padding: 20px;
background-color: #ffffff;
border-radius: 5px;
box-shadow: 0 4px 8px 0 rgba(0, 0, 0, 0.1), 0 6px 20px 0 rgba(0, 0, 0, 0.1);
}
.prompt {
padding: 20px;
background-color: #0084ff;
color: #ffffff;
border-radius: 5px;
margin-bottom: 20px;
box-shadow: 0 2px 4px 0 rgba(0, 0, 0, 0.1), 0 3px 10px 0 rgba(0, 0, 0, 0.1);
}
.answers {
display: flex;
justify-content: space-between;
flex-wrap: wrap;
gap: 20px;
}
.answer-box {
flex-basis: 49%;
padding: 20px;
background-color: rgba(44, 62, 80, 0.9);
color: #ffffff;
border-radius: 5px;
box-shadow: 0 2px 4px 0 rgba(0, 0, 0, 0.1), 0 3px 10px 0 rgba(0, 0, 0, 0.1);
}
.answer-box p {
word-wrap: break-word;
}
.answer-box:hover {
background-color: rgba(52, 73, 94, 0.9);
cursor: pointer;
transition: all 0.3s ease;
}
.lsf-richtext__line:hover {
background: unset;
}
.answer-box .lsf-object {
padding: 20px
}
</Style>
<View className="container">
<View className="prompt">
<Text name="prompt" value="$prompt" />
</View>
<View className="answers">
<Pairwise name="comparison" toName="answer1,answer2"
selectionStyle="background-color: #27ae60; box-shadow: 0 4px 8px 0 rgba(0, 0, 0, 0.2), 0 6px 20px 0 rgba(0, 0, 0, 0.2); border: 2px solid #2ecc71; cursor: pointer; transition: all 0.3s ease;"
leftChoiceValue="answer1" rightChoiceValue="answer2" />
<View className="answer-box">
<Text name="answer1" value="$answer1" />
</View>
<View className="answer-box">
<Text name="answer2" value="$answer2" />
</View>
</View>
</View>
</View>
""" # noqa: E501
[docs]
def __init__(
self,
label_config_file: str = None,
answer1_key: str = "answer1",
answer2_key: str = "answer2",
prompt_key: str = "prompt",
chosen_key: str = "chosen",
rejected_key: str = "rejected",
**kwargs,
):
"""Initialize the human preference annotation operator."""
# Store our class-specific attributes
self.answer1_key = answer1_key
self.answer2_key = answer2_key
self.prompt_key = prompt_key
self.chosen_key = chosen_key
self.rejected_key = rejected_key
# Ensure text_key is set to prompt_key if not explicitly provided
if "text_key" not in kwargs:
kwargs["text_key"] = prompt_key
# Prepare the label_config parameter
if label_config_file and os.path.exists(label_config_file):
with open(label_config_file, "r") as f:
kwargs["label_config"] = f.read().strip()
logger.info(f"Loaded label config from {label_config_file}")
else:
kwargs["label_config"] = self.DEFAULT_LABEL_CONFIG.strip()
logger.info("Using default UI config for human preference annotation")
# Initialize the parent class with remaining kwargs
super().__init__(**kwargs)
def _format_task(self, samples: List[Dict]) -> Dict:
"""Format samples as a Label Studio task for human preference.
Args:
samples: List of samples to include in the task
Returns:
Dict: Formatted task data
"""
# For human preference, we need a special format
if len(samples) != 1:
logger.warning("Human preference requires exactly one sample per task")
sample = samples[0]
task = {"data": {}}
# Add the prompt/question
if self.prompt_key in sample:
task["data"]["prompt"] = sample[self.prompt_key]
else:
logger.warning(f"Sample missing required field '{self.prompt_key}'")
task["data"]["prompt"] = "No prompt provided"
# Add the answer options
if self.answer1_key in sample:
task["data"]["answer1"] = sample[self.answer1_key]
else:
logger.warning(f"Sample missing required field '{self.answer1_key}'")
task["data"]["answer1"] = "No answer 1 provided"
if self.answer2_key in sample:
task["data"]["answer2"] = sample[self.answer2_key]
else:
logger.warning(f"Sample missing required field '{self.answer2_key}'")
task["data"]["answer2"] = "No answer 2 provided"
# Add any other metadata as string values only
for key, value in sample.items():
if key not in [self.prompt_key, self.answer1_key, self.answer2_key]:
if isinstance(value, (str, int, float, bool)) or value is None:
# Convert to string to ensure compatibility
task["data"][f"meta:{key}"] = str(value) if value is not None else ""
# Log the task for debugging
logger.debug(f"Formatted task: {task}")
return task
def _get_task_annotation(self, task_id: int) -> Optional[Dict]:
"""Get annotation for a task if available"""
annotation = super()._get_task_annotation(task_id)
# Process the annotation if available
if annotation and "result" in annotation:
# Extract the preference information
for item in annotation["result"]:
if item.get("type") == "pairwise":
# Get the selected option (from_id or to_id)
selected = item.get("value", {}).get("selected")
if selected:
# Add the preference to the annotation
annotation["preference"] = selected
return annotation
def _process_annotation_result(self, annotation: Dict, sample: Dict) -> Dict:
"""Process human preference annotation result and update the sample
Args:
annotation: The annotation result from Label Studio
sample: The original sample that was annotated
Returns:
Dict: The updated sample with preference results
"""
# Extract the preference information
logger.debug(f"Processing annotation result: {annotation}")
all_keys = f"{self.answer1_key}{self.answer2_key}"
preference = None
for item in annotation["result"]:
if item.get("type") == "pairwise":
# Get the selected option
selected = item.get("value", {}).get("selected")
if selected:
# Map 'left'/'right' to 'answer1'/'answer2'
if selected == "left":
preference = self.answer1_key
elif selected == "right":
preference = self.answer2_key
else:
# In case it's already 'answer1'/'answer2'
preference = selected
break
# Store the preference result directly in the sample
chosen = preference if preference else "Unanswered"
rejected = all_keys.replace(preference, "") if preference else "Unanswered"
sample[self.chosen_key] = sample[chosen]
sample[self.rejected_key] = sample[rejected]
logger.debug(f"Updated sample: {sample}")
return sample