Human Feedback Tooling: Label Studio, Argilla, and LangSmith

# Code Fragment 29.12.2: Label Studio preference ranking project setup
# Install: pip install label-studio label-studio-sdk

from label_studio_sdk import Client

# Connect to a running Label Studio instance
ls_client = Client(url="http://localhost:8080", api_key="your-api-key")

# XML template for pairwise preference ranking of LLM responses
preference_template = """
<View>
 <Style>
 .prompt-box { background: #f0f4f8; padding: 12px; border-radius: 8px; }
 .response-pair { display: flex; gap: 16px; margin-top: 12px; }
 .response-card { flex: 1; padding: 12px; border: 1px solid #ddd; border-radius: 8px; }
 </Style>

 <View className="prompt-box">
 <Header value="User Prompt" />
 <Text name="prompt" value="$prompt" />
 </View>

 <View className="response-pair">
 <View className="response-card">
 <Header value="Response A" />
 <Text name="response_a" value="$response_a" />
 </View>
 <View className="response-card">
 <Header value="Response B" />
 <Text name="response_b" value="$response_b" />
 </View>
 </View>

 <Header value="Which response is better?" />
 <Choices name="preference" toName="prompt" choice="single-radio">
 <Choice value="response_a_much_better" />
 <Choice value="response_a_slightly_better" />
 <Choice value="tie" />
 <Choice value="response_b_slightly_better" />
 <Choice value="response_b_much_better" />
 </Choices>

 <Header value="Rate on specific dimensions (optional)" />
 <Rating name="helpfulness" toName="prompt" maxRating="5" />
 <Rating name="safety" toName="prompt" maxRating="5" />

 <TextArea name="rationale" toName="prompt"
 placeholder="Briefly explain your preference..."
 rows="3" />
</View>
"""

# Create the project
project = ls_client.start_project(
 title="LLM Preference Ranking v2",
 label_config=preference_template,
 description="Pairwise comparison of model responses for RLHF training",
)

# Import tasks with pre-generated response pairs
tasks = [
 {
 "data": {
 "prompt": "Explain quantum entanglement to a high school student.",
 "response_a": "Quantum entanglement is when two particles...",
 "response_b": "Imagine you have two magic coins...",
 },
 "predictions": [{
 "model_version": "gpt-4o-preliminary",
 "result": [{"from_name": "preference", "to_name": "prompt",
 "type": "choices", "value": {"choices": ["response_b_slightly_better"]}}],
 "score": 0.73, # Model confidence for pre-annotation
 }],
 },
 # ... additional tasks
]

project.import_tasks(tasks)
print(f"Created project {project.id} with {len(tasks)} tasks")

# Code Fragment 29.12.2: Argilla dataset curation with feedback collection
# Install: pip install argilla

import argilla as rg

# Connect to Argilla server (or Hugging Face Spaces deployment)
client = rg.Argilla(api_url="http://localhost:6900", api_key="argilla.apikey")

# Define the annotation schema for preference feedback
settings = rg.Settings(
 fields=[
 rg.TextField(name="instruction", title="User Instruction"),
 rg.TextField(name="response_a", title="Response A", use_markdown=True),
 rg.TextField(name="response_b", title="Response B", use_markdown=True),
 ],
 questions=[
 rg.LabelQuestion(
 name="preference",
 title="Which response is better?",
 labels={
 "a_much_better": "A is much better",
 "a_slightly_better": "A is slightly better",
 "tie": "About the same",
 "b_slightly_better": "B is slightly better",
 "b_much_better": "B is much better",
 },
 ),
 rg.RatingQuestion(
 name="helpfulness",
 title="Rate the helpfulness of the chosen response",
 values=[1, 2, 3, 4, 5],
 ),
 rg.TextQuestion(
 name="rationale",
 title="Why did you choose this response?",
 required=False,
 ),
 ],
 metadata=[
 rg.TermsMetadataProperty(name="model_a", title="Model A"),
 rg.TermsMetadataProperty(name="model_b", title="Model B"),
 rg.FloatMetadataProperty(name="prompt_length", title="Prompt Length"),
 ],
 guidelines=(
 "Compare both responses to the user instruction. Choose the response "
 "that is more helpful, accurate, and safe. When responses are close in "
 "quality, prefer the one that is more concise and direct."
 ),
)

# Create the dataset
dataset = rg.Dataset(name="preference-pairs-v3", settings=settings, workspace="default")
dataset.create()

# Add records from a Hugging Face dataset or custom source
records = [
 rg.Record(
 fields={
 "instruction": "Write a haiku about machine learning.",
 "response_a": "Data flows like streams\nWeights adjust in silence deep\nPatterns start to speak",
 "response_b": "Neurons fire and learn\nGradient descent finds the way\nLoss function drops low",
 },
 metadata={"model_a": "llama-3.1-8b", "model_b": "mistral-7b", "prompt_length": 42.0},
 ),
 # ... additional records
]

dataset.records.log(records)

# After annotation, export to Hugging Face datasets format
annotated = dataset.records.to_list()
print(f"Collected {len(annotated)} annotated records")

# Push curated dataset to Hugging Face Hub
dataset.records.to_datasets().push_to_hub("my-org/preference-pairs-v3")

# Code Fragment 29.12.5: LangSmith trace evaluation pipeline
# Install: pip install langsmith langchain-openai

import os
from langsmith import Client
from langsmith.evaluation import evaluate
from langsmith.schemas import Example, Run

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

client = Client()

# Step 1: Create an evaluation dataset from production traces
# (Traces are collected automatically when tracing is enabled)
dataset = client.create_dataset(
 dataset_name="customer-support-eval-v2",
 description="Curated customer support interactions for evaluation",
)

# Add examples (typically selected from production traces)
examples = [
 {
 "inputs": {"question": "How do I reset my password?"},
 "outputs": {"answer": "Go to Settings > Security > Reset Password..."},
 "metadata": {"category": "account", "difficulty": "easy"},
 },
 {
 "inputs": {"question": "Why was I charged twice for my subscription?"},
 "outputs": {"answer": "I can see the duplicate charge. Let me process a refund..."},
 "metadata": {"category": "billing", "difficulty": "medium"},
 },
]

for ex in examples:
 client.create_example(
 inputs=ex["inputs"],
 outputs=ex["outputs"],
 metadata=ex["metadata"],
 dataset_id=dataset.id,
 )

# Step 2: Define custom evaluators for human-in-the-loop assessment
def correctness_evaluator(run: Run, example: Example) -> dict:
 """Automated correctness check as a first pass."""
 prediction = run.outputs.get("answer", "")
 reference = example.outputs.get("answer", "")
 # Simple overlap heuristic; real implementations use LLM-as-judge
 overlap = len(set(prediction.split()) & set(reference.split()))
 total = max(len(set(reference.split())), 1)
 score = overlap / total
 return {"key": "correctness", "score": score}

def helpfulness_evaluator(run: Run, example: Example) -> dict:
 """Flag examples needing human review based on confidence."""
 prediction = run.outputs.get("answer", "")
 # Low-confidence heuristic: short responses or hedging language
 hedging = any(w in prediction.lower() for w in ["maybe", "i think", "not sure", "possibly"])
 short = len(prediction.split()) < 15
 needs_review = hedging or short
 return {
 "key": "helpfulness_auto",
 "score": 0.5 if needs_review else 0.8,
 "comment": "Flagged for human review" if needs_review else "Auto-passed",
 }

# Step 3: Run the target application against the evaluation dataset
def my_support_bot(inputs: dict) -> dict:
 """The application under evaluation."""
 from langchain_openai import ChatOpenAI
 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 response = llm.invoke(inputs["question"])
 return {"answer": response.content}

# Execute evaluation experiment
results = evaluate(
 my_support_bot,
 data="customer-support-eval-v2",
 evaluators=[correctness_evaluator, helpfulness_evaluator],
 experiment_prefix="support-bot-v2.1",
 max_concurrency=4,
)

# Step 4: Create annotation queue for human review of flagged examples
# (Done through LangSmith UI: select runs with low auto-scores,
# add to annotation queue, assign to reviewers)
print(f"Evaluation complete: {results.experiment_name}")
print("Review flagged examples in the LangSmith annotation queue UI.")

# Code Fragment 29.12.4: Converting annotation exports to RLHF training format

import json
from collections import Counter
from datasets import Dataset

def argilla_to_preference_pairs(argilla_records: list[dict]) -> list[dict]:
 """Convert Argilla annotation exports to chosen/rejected pairs.

 Filters out ties and low-agreement examples.
 """
 preference_pairs = []

 for record in argilla_records:
 instruction = record["fields"]["instruction"]
 response_a = record["fields"]["response_a"]
 response_b = record["fields"]["response_b"]

 # Collect all annotator responses for this record
 responses = record.get("responses", [])
 if not responses:
 continue

 # Aggregate preferences across annotators
 pref_counts = Counter()
 for resp in responses:
 pref = resp["values"]["preference"]["value"]
 pref_counts[pref] += 1

 # Map 5-point scale to binary preference
 a_votes = pref_counts.get("a_much_better", 0) + pref_counts.get("a_slightly_better", 0)
 b_votes = pref_counts.get("b_much_better", 0) + pref_counts.get("b_slightly_better", 0)
 tie_votes = pref_counts.get("tie", 0)
 total = a_votes + b_votes + tie_votes

 # Skip ties and low-agreement examples
 if total == 0 or tie_votes > (a_votes + b_votes):
 continue

 # Require at least 60% agreement for binary preference
 majority = max(a_votes, b_votes)
 if majority / total < 0.6:
 continue

 if a_votes > b_votes:
 chosen, rejected = response_a, response_b
 else:
 chosen, rejected = response_b, response_a

 preference_pairs.append({
 "prompt": instruction,
 "chosen": chosen,
 "rejected": rejected,
 "agreement_ratio": majority / total,
 "num_annotators": total,
 })

 return preference_pairs

def format_for_dpo_training(pairs: list[dict]) -> Dataset:
 """Format preference pairs for DPO/RLHF training with TRL."""

 formatted = []
 for pair in pairs:
 formatted.append({
 "prompt": pair["prompt"],
 "chosen": [
 {"role": "user", "content": pair["prompt"]},
 {"role": "assistant", "content": pair["chosen"]},
 ],
 "rejected": [
 {"role": "user", "content": pair["prompt"]},
 {"role": "assistant", "content": pair["rejected"]},
 ],
 })

 dataset = Dataset.from_list(formatted)
 return dataset

# Example usage
raw_export = json.loads(open("argilla_export.json").read())
pairs = argilla_to_preference_pairs(raw_export)
print(f"Extracted {len(pairs)} preference pairs from {len(raw_export)} records")
print(f"Mean agreement: {sum(p['agreement_ratio'] for p in pairs) / len(pairs):.2f}")

dpo_dataset = format_for_dpo_training(pairs)
dpo_dataset.push_to_hub("my-org/preference-data-v3")

# Code Fragment 29.12.3: Computing inter-annotator agreement metrics
import numpy as np
from sklearn.metrics import cohen_kappa_score

def compute_pairwise_kappa(annotations: dict[str, list[int]]) -> dict:
 """Compute pairwise Cohen's kappa for all annotator pairs.

 Args:
 annotations: mapping of annotator_id to list of labels
 (one label per item, in the same order)

 Returns:
 Dictionary with pairwise kappa scores and overall statistics.
 """
 annotator_ids = list(annotations.keys())
 n_annotators = len(annotator_ids)
 kappa_scores = {}

 for i in range(n_annotators):
 for j in range(i + 1, n_annotators):
 id_i, id_j = annotator_ids[i], annotator_ids[j]
 labels_i = annotations[id_i]
 labels_j = annotations[id_j]

 # Filter to items where both annotators provided labels
 paired = [
 (a, b) for a, b in zip(labels_i, labels_j)
 if a is not None and b is not None
 ]
 if len(paired) < 10:
 continue

 a_labels, b_labels = zip(*paired)
 kappa = cohen_kappa_score(a_labels, b_labels, weights="linear")
 kappa_scores[f"{id_i}_vs_{id_j}"] = {
 "kappa": round(kappa, 3),
 "n_items": len(paired),
 "agreement_pct": round(
 sum(1 for a, b in paired if a == b) / len(paired) * 100, 1
 ),
 }

 # Compute summary statistics
 all_kappas = [v["kappa"] for v in kappa_scores.values()]
 summary = {
 "pairwise_scores": kappa_scores,
 "mean_kappa": round(np.mean(all_kappas), 3) if all_kappas else None,
 "min_kappa": round(np.min(all_kappas), 3) if all_kappas else None,
 "max_kappa": round(np.max(all_kappas), 3) if all_kappas else None,
 "interpretation": interpret_kappa(np.mean(all_kappas)) if all_kappas else "insufficient data",
 }

 return summary

def interpret_kappa(kappa: float) -> str:
 """Interpret kappa using Landis and Koch (1977) scale."""
 if kappa < 0.0:
 return "poor (less than chance)"
 elif kappa < 0.20:
 return "slight agreement"
 elif kappa < 0.40:
 return "fair agreement"
 elif kappa < 0.60:
 return "moderate agreement"
 elif kappa < 0.80:
 return "substantial agreement"
 else:
 return "almost perfect agreement"

# Example: three annotators labeling 100 preference comparisons
# Labels: 0 = A much better, 1 = A slightly better, 2 = tie,
# 3 = B slightly better, 4 = B much better
np.random.seed(42)
annotations = {
 "annotator_1": list(np.random.choice([0, 1, 2, 3, 4], size=100, p=[0.15, 0.25, 0.15, 0.30, 0.15])),
 "annotator_2": list(np.random.choice([0, 1, 2, 3, 4], size=100, p=[0.15, 0.25, 0.15, 0.30, 0.15])),
 "annotator_3": list(np.random.choice([0, 1, 2, 3, 4], size=100, p=[0.15, 0.25, 0.15, 0.30, 0.15])),
}

results = compute_pairwise_kappa(annotations)
print(f"Mean kappa: {results['mean_kappa']}")
print(f"Interpretation: {results['interpretation']}")
for pair, score in results["pairwise_scores"].items():
 print(f" {pair}: kappa={score['kappa']}, agreement={score['agreement_pct']}%")