Most LLM applications ship and stagnate. This lesson shows you how to instrument your app to capture user feedback, structure preference pairs for DPO and RLHF workflows, and build a complete pipeline that turns raw signals into training data — so your model actually gets better over time.
Your LLM application shipped. Users are interacting with it daily. And somewhere in the gap between what the model produces and what users actually want, you're losing trust, quality, and retention — you just can't see it yet.
Most teams treat LLM deployment as a finish line. In reality, it's the starting gun for a continuous improvement loop. The models that get better over time — the ones that feel increasingly "right" to users — are backed by systematic feedback collection, structured preference data, and disciplined pipelines for turning that signal into model or prompt improvements. Teams that skip this end up with stale prompts, drift they can't diagnose, and no ground truth to evaluate against.
By the end of this lesson, you'll have built a complete feedback collection and human preference dataset pipeline: from capturing raw signals at the application layer, through structuring preference pairs, to storing and versioning the data in a format ready for fine-tuning or evaluation. You'll understand how to instrument your application, design a preference annotation interface, and maintain dataset quality over time.
What you'll learn:
You should be comfortable with Python and have working knowledge of building LLM-powered applications (API calls, prompt templating, basic RAG). Familiarity with SQL or a document store is helpful. You don't need to have done RLHF or DPO training before — we'll set up the data pipeline that feeds those processes, not the training itself.
Before we instrument anything, it's worth understanding why preference data is structurally different from standard labeled datasets — because that difference shapes every design decision downstream.
A standard training dataset has inputs and outputs: a question and its correct answer, a document and its summary label. Human preference data doesn't assert that any single response is correct. Instead, it captures relative quality: given prompt P, response A is preferred over response B by a human rater. This pairwise structure is what makes the data useful for techniques like Reinforcement Learning from Human Feedback (RLHF) and Direct Preference Optimization (DPO).
The data looks like this:
{
"prompt": "Explain the difference between precision and recall to a non-technical stakeholder.",
"chosen": "Precision is about whether what we flagged was actually a problem. Recall is about whether we caught all the problems. High precision means fewer false alarms. High recall means fewer things slipping through.",
"rejected": "Precision is TP/(TP+FP) and recall is TP/(TP+FN) where TP is true positives, FP is false positives, and FN is false negatives."
}
The "chosen" response isn't necessarily perfect. It's just better than "rejected" for this prompt in this context. That's a crucial distinction — your pipeline doesn't need to collect gold-standard outputs. It needs to collect honest comparative judgments.
This also means your feedback pipeline has two separate concerns:
Most teams conflate these and end up with data that's hard to use. We'll keep them separate.
The architecture has three layers: the instrumentation layer inside your application, the ingestion layer that receives and stores raw signals, and the structuring layer that transforms raw signals into preference pairs.
Here's the high-level flow:
LLM Application
│
├── Captures: completions, metadata, user context
├── Exposes: feedback UI (thumbs, ratings, edits)
│
▼
Feedback Ingestion API
│
├── Validates and stores raw events
│
▼
Preference Dataset Builder
│
├── Pairs candidates (same prompt, different responses)
├── Routes to annotation queue
│
▼
Preference Store (versioned)
│
└── Feeds: evaluations, fine-tuning, prompt analysis
Let's build each layer in turn.
The first thing to instrument is your completion logging. Every response your application generates needs a persistent record that can be referenced later when feedback arrives.
Here's a practical completion logger you can drop into an existing FastAPI or Flask application:
import uuid
import time
import json
from dataclasses import dataclass, asdict, field
from typing import Optional
import boto3 # or swap for your storage client
import hashlib
@dataclass
class CompletionRecord:
completion_id: str
session_id: str
user_id: Optional[str]
prompt: str
prompt_version: str # track which prompt template was active
model: str # e.g., "gpt-4o", "claude-3-5-sonnet"
response: str
latency_ms: int
input_tokens: int
output_tokens: int
temperature: float
application_context: dict # feature flags, RAG sources used, etc.
timestamp: float = field(default_factory=time.time)
def prompt_hash(self) -> str:
"""Canonical hash of the prompt for grouping variants."""
return hashlib.sha256(self.prompt.encode()).hexdigest()[:16]
class CompletionLogger:
def __init__(self, storage_backend):
self.storage = storage_backend
def log(self, record: CompletionRecord) -> str:
record_dict = asdict(record)
record_dict["prompt_hash"] = record.prompt_hash()
self.storage.put(
key=f"completions/{record.timestamp:.0f}/{record.completion_id}",
value=json.dumps(record_dict)
)
return record.completion_id
def get(self, completion_id: str) -> Optional[dict]:
return self.storage.get(completion_id)
def log_completion(
prompt: str,
response: str,
model: str,
session_id: str,
prompt_version: str,
latency_ms: int,
token_counts: dict,
temperature: float = 0.7,
user_id: Optional[str] = None,
context: dict = None
) -> str:
record = CompletionRecord(
completion_id=str(uuid.uuid4()),
session_id=session_id,
user_id=user_id,
prompt=prompt,
prompt_version=prompt_version,
model=model,
response=response,
latency_ms=latency_ms,
input_tokens=token_counts.get("input", 0),
output_tokens=token_counts.get("output", 0),
temperature=temperature,
application_context=context or {}
)
return logger.log(record)
The prompt_version field is non-negotiable. When you're doing analysis six weeks later, you need to know whether quality differences across completions were due to model changes, prompt changes, or just natural variance. Track it from day one.
The application_context dictionary is where you capture anything situational: which RAG chunks were retrieved, which user persona is active, whether the user is on a free or paid tier. This becomes essential for slicing your preference data later.
Now instrument your UI feedback mechanisms. The most common are thumbs up/down, star ratings, and free-text corrections. Here's the API endpoint and event schema:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from typing import Literal, Optional
import time
import uuid
app = FastAPI()
class FeedbackEvent(BaseModel):
completion_id: str
session_id: str
user_id: Optional[str] = None
feedback_type: Literal["thumbs", "rating", "edit", "flag"]
# For thumbs feedback
thumbs_value: Optional[Literal["up", "down"]] = None
# For star rating (1-5)
rating: Optional[int] = None
# For edit feedback — user rewrote the response
edited_response: Optional[str] = None
edit_comment: Optional[str] = None
# For flagging problematic content
flag_reason: Optional[Literal["factually_wrong", "harmful", "off_topic", "too_verbose", "too_brief", "other"]] = None
# Optional free text always
comment: Optional[str] = None
@validator("rating")
def rating_in_range(cls, v):
if v is not None and not (1 <= v <= 5):
raise ValueError("Rating must be between 1 and 5")
return v
class FeedbackStore:
"""Simplified — swap for your actual database."""
def __init__(self, db_connection):
self.db = db_connection
def insert(self, event: FeedbackEvent, feedback_id: str):
self.db.execute("""
INSERT INTO feedback_events
(feedback_id, completion_id, session_id, user_id,
feedback_type, thumbs_value, rating, edited_response,
edit_comment, flag_reason, comment, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
feedback_id,
event.completion_id,
event.session_id,
event.user_id,
event.feedback_type,
event.thumbs_value,
event.rating,
event.edited_response,
event.edit_comment,
event.flag_reason,
event.comment,
time.time()
))
@app.post("/feedback")
async def collect_feedback(event: FeedbackEvent):
feedback_id = str(uuid.uuid4())
# Verify the completion exists before accepting feedback
completion = completion_logger.get(event.completion_id)
if not completion:
raise HTTPException(status_code=404, detail="Completion not found")
feedback_store.insert(event, feedback_id)
return {"feedback_id": feedback_id, "status": "recorded"}
Explicit thumbs-up signals are gold, but users don't always click them. Implicit signals — copy events, session abandonment, regeneration requests — are often more abundant and sometimes more honest.
class ImplicitSignalEvent(BaseModel):
completion_id: str
session_id: str
signal_type: Literal[
"copy_to_clipboard", # strong positive
"regenerate_requested", # negative
"continued_conversation", # positive — user built on the response
"session_abandoned", # negative if shortly after response
"response_shared", # strong positive
"scroll_depth", # how far into a long response user read
]
signal_metadata: dict = {} # e.g., {"scroll_percent": 0.3, "time_to_abandon_seconds": 4}
@app.post("/implicit-signal")
async def collect_implicit_signal(event: ImplicitSignalEvent):
# Store with lower weight than explicit signals
signal_id = str(uuid.uuid4())
implicit_store.insert(event, signal_id, weight=0.3)
return {"signal_id": signal_id}
Weight your signals carefully. A user copying a response to clipboard is a genuinely strong positive signal. A session being abandoned is ambiguous — the user might have gotten what they needed. Assign weights in your schema now so you can tune them later without restructuring your data.
Raw feedback events are not preference pairs. To make the data usable for fine-tuning and evaluation, you need to construct pairwise comparisons. This is where most teams either don't think hard enough — and produce noisy, unusable pairs — or overthink it and never ship.
The core idea: for any given prompt (or prompt hash), if you have multiple responses with different feedback signals, you can construct pairs where the higher-feedback response is "chosen" and the lower-feedback response is "rejected."
from dataclasses import dataclass
from typing import Optional
import json
@dataclass
class PreferencePair:
pair_id: str
prompt: str
prompt_hash: str
prompt_version: str
chosen: str
chosen_completion_id: str
chosen_feedback_signal: str # what made us pick this as chosen
chosen_signal_score: float
rejected: str
rejected_completion_id: str
rejected_feedback_signal: str
rejected_signal_score: float
construction_method: str # "implicit", "explicit", "annotator"
quality_tier: str # "high", "medium", "low"
metadata: dict
created_at: float
class PreferencePairConstructor:
# Explicit thumbs-up/down carries the most weight
SIGNAL_WEIGHTS = {
"thumbs_up": 1.0,
"thumbs_down": -1.0,
"rating_5": 0.9,
"rating_4": 0.6,
"rating_3": 0.1,
"rating_2": -0.5,
"rating_1": -0.9,
"edit_submitted": -0.7, # user had to fix it
"copy_to_clipboard": 0.5,
"regenerate_requested": -0.6,
"continued_conversation": 0.4,
"session_abandoned_early": -0.3,
}
def score_completion(self, completion_id: str) -> tuple[float, list[str]]:
"""
Aggregate all feedback signals for a completion into a single score.
Returns (score, list_of_signals_used).
"""
events = self.feedback_store.get_events_for_completion(completion_id)
implicit = self.implicit_store.get_signals_for_completion(completion_id)
total_score = 0.0
signals_used = []
for event in events:
if event.feedback_type == "thumbs":
key = f"thumbs_{event.thumbs_value}"
elif event.feedback_type == "rating":
key = f"rating_{event.rating}"
elif event.feedback_type == "edit":
key = "edit_submitted"
else:
continue
weight = self.SIGNAL_WEIGHTS.get(key, 0.0)
total_score += weight
signals_used.append(key)
for signal in implicit:
weight = self.SIGNAL_WEIGHTS.get(signal.signal_type, 0.0) * 0.3
total_score += weight
signals_used.append(f"implicit:{signal.signal_type}")
return total_score, signals_used
def construct_pairs_for_prompt(self, prompt_hash: str) -> list[PreferencePair]:
"""
Given a prompt hash, find all completions for that prompt,
score them, and construct valid preference pairs.
"""
completions = self.completion_store.get_by_prompt_hash(prompt_hash)
if len(completions) < 2:
return [] # Need at least two responses to compare
scored = []
for completion in completions:
score, signals = self.score_completion(completion["completion_id"])
scored.append((score, signals, completion))
# Sort descending by score
scored.sort(key=lambda x: x[0], reverse=True)
pairs = []
# Pair highest with lowest, second-highest with second-lowest, etc.
# Only create pairs where there's a meaningful score gap
top = scored[:len(scored)//2]
bottom = scored[len(scored)//2:]
for (high_score, high_signals, high_comp), (low_score, low_signals, low_comp) in zip(top, reversed(bottom)):
score_gap = high_score - low_score
if score_gap < 0.5:
# Gap too small — this pair would introduce noise
quality_tier = "low"
elif score_gap < 1.2:
quality_tier = "medium"
else:
quality_tier = "high"
pair = PreferencePair(
pair_id=str(uuid.uuid4()),
prompt=high_comp["prompt"],
prompt_hash=prompt_hash,
prompt_version=high_comp["prompt_version"],
chosen=high_comp["response"],
chosen_completion_id=high_comp["completion_id"],
chosen_feedback_signal=", ".join(high_signals),
chosen_signal_score=high_score,
rejected=low_comp["response"],
rejected_completion_id=low_comp["completion_id"],
rejected_feedback_signal=", ".join(low_signals),
rejected_signal_score=low_score,
construction_method="implicit" if all("implicit" in s for s in high_signals) else "explicit",
quality_tier=quality_tier,
metadata={
"model_chosen": high_comp["model"],
"model_rejected": low_comp["model"],
"prompt_version": high_comp["prompt_version"],
},
created_at=time.time()
)
pairs.append(pair)
return pairs
Don't pair responses across different prompt versions. If your prompt template changed between two completions, any quality difference could be due to the prompt, not the model behavior. Your
prompt_versionfield prevents this contamination — filter strictly.
Implicit signals are abundant but noisy. For your highest-stakes preference pairs — cases where you'll use data for fine-tuning — you want human judgment. Let's build a lightweight annotation queue and interface.
from enum import Enum
class AnnotationStatus(str, Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETE = "complete"
DISPUTED = "disputed"
SKIPPED = "skipped"
class AnnotationQueue:
"""
Manages assignment of preference pairs to human annotators.
Supports redundant annotation (same pair to multiple annotators)
for high-quality pairs.
"""
def __init__(self, db, target_annotations_per_pair: int = 2):
self.db = db
self.target_n = target_annotations_per_pair
def enqueue(self, pair: PreferencePair, priority: str = "normal"):
self.db.execute("""
INSERT INTO annotation_queue
(pair_id, priority, annotations_needed, annotations_received, status, created_at)
VALUES (?, ?, ?, 0, 'pending', ?)
""", (pair.pair_id, priority, self.target_n, time.time()))
def get_next_for_annotator(self, annotator_id: str) -> Optional[dict]:
"""
Fetch the next pair for an annotator, skipping pairs they've already annotated.
Prioritize 'high' priority items and items closest to completion.
"""
result = self.db.execute("""
SELECT q.pair_id, p.*
FROM annotation_queue q
JOIN preference_pairs p ON q.pair_id = p.pair_id
WHERE q.status = 'pending'
AND q.pair_id NOT IN (
SELECT pair_id FROM annotations WHERE annotator_id = ?
)
ORDER BY
CASE q.priority WHEN 'high' THEN 0 WHEN 'normal' THEN 1 ELSE 2 END,
q.annotations_received DESC
LIMIT 1
""", (annotator_id,)).fetchone()
return dict(result) if result else None
def submit_annotation(
self,
pair_id: str,
annotator_id: str,
chosen_id: str, # annotator picks which completion_id is better
confidence: int, # 1-3 scale: 1=uncertain, 2=confident, 3=very confident
reasoning: Optional[str] = None,
quality_issues: Optional[list] = None
):
annotation_id = str(uuid.uuid4())
self.db.execute("""
INSERT INTO annotations
(annotation_id, pair_id, annotator_id, chosen_completion_id,
confidence, reasoning, quality_issues, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
annotation_id, pair_id, annotator_id, chosen_id,
confidence, reasoning, json.dumps(quality_issues or []), time.time()
))
# Update queue progress
self.db.execute("""
UPDATE annotation_queue
SET annotations_received = annotations_received + 1,
status = CASE
WHEN annotations_received + 1 >= annotations_needed THEN 'complete'
ELSE status
END
WHERE pair_id = ?
""", (pair_id,))
# Check for inter-annotator agreement
self._check_agreement(pair_id)
def _check_agreement(self, pair_id: str):
"""Flag pairs where annotators disagree for review."""
annotations = self.db.execute("""
SELECT chosen_completion_id, COUNT(*) as cnt
FROM annotations WHERE pair_id = ?
GROUP BY chosen_completion_id
""", (pair_id,)).fetchall()
if len(annotations) > 1:
# Disagreement — mark as disputed
self.db.execute("""
UPDATE annotation_queue SET status = 'disputed' WHERE pair_id = ?
""", (pair_id,))
For the annotation UI itself, you can use Streamlit for internal tooling without building a full frontend. Here's the backend logic that feeds it:
import streamlit as st
def render_annotation_interface(annotator_id: str, queue: AnnotationQueue):
st.title("Response Quality Annotation")
pair_data = queue.get_next_for_annotator(annotator_id)
if not pair_data:
st.success("Queue empty — great work!")
return
st.markdown("### Prompt")
st.text_area("", value=pair_data["prompt"], height=150, disabled=True, key="prompt_display")
st.markdown("---")
st.markdown("**Which response is better for this prompt?**")
col1, col2 = st.columns(2)
# Randomize which response appears left/right to avoid position bias
import random
responses = [
("A", pair_data["chosen_completion_id"], pair_data["chosen"]),
("B", pair_data["rejected_completion_id"], pair_data["rejected"])
]
random.shuffle(responses)
with col1:
st.markdown(f"**Response {responses[0][0]}**")
st.text_area("", value=responses[0][2], height=300, disabled=True, key="resp_left")
with col2:
st.markdown(f"**Response {responses[1][0]}**")
st.text_area("", value=responses[1][2], height=300, disabled=True, key="resp_right")
chosen_label = st.radio(
"Better response:",
options=[responses[0][0], responses[1][0], "Neither / Skip"],
horizontal=True
)
confidence = st.slider("How confident are you?", 1, 3, 2,
help="1=uncertain, 2=confident, 3=very confident")
reasoning = st.text_area("Optional: Why did you choose this response?", height=80)
quality_issues = st.multiselect(
"Any quality issues with either response?",
["Factual errors", "Hallucinated content", "Inappropriate tone",
"Too verbose", "Incomplete", "Harmful content"]
)
if st.button("Submit", type="primary"):
if chosen_label == "Neither / Skip":
queue.skip(pair_data["pair_id"], annotator_id)
st.rerun()
else:
idx = 0 if chosen_label == responses[0][0] else 1
chosen_completion_id = responses[idx][1]
queue.submit_annotation(
pair_id=pair_data["pair_id"],
annotator_id=annotator_id,
chosen_id=chosen_completion_id,
confidence=confidence,
reasoning=reasoning or None,
quality_issues=quality_issues
)
st.rerun()
Always randomize left/right presentation. Studies on human annotation consistently show position bias — annotators slightly favor responses presented first or on the left. Randomize per-session and record which position each response appeared in so you can check for this in your data quality analysis.
Your final dataset needs to be versioned, filterable, and exportable in the formats that fine-tuning frameworks expect. Here's a schema and export layer that handles both.
-- Core preference pairs table
CREATE TABLE preference_pairs (
pair_id TEXT PRIMARY KEY,
prompt TEXT NOT NULL,
prompt_hash TEXT NOT NULL,
prompt_version TEXT NOT NULL,
chosen TEXT NOT NULL,
chosen_completion_id TEXT NOT NULL,
chosen_signal_score REAL,
rejected TEXT NOT NULL,
rejected_completion_id TEXT NOT NULL,
rejected_signal_score REAL,
construction_method TEXT NOT NULL, -- 'implicit', 'explicit', 'annotator'
quality_tier TEXT NOT NULL, -- 'high', 'medium', 'low'
dataset_version TEXT, -- assigned when exported
is_validated BOOLEAN DEFAULT FALSE,
is_excluded BOOLEAN DEFAULT FALSE,
exclusion_reason TEXT,
metadata JSON,
created_at REAL NOT NULL
);
-- Human annotation results (post-reconciliation)
CREATE TABLE validated_preferences (
validation_id TEXT PRIMARY KEY,
pair_id TEXT REFERENCES preference_pairs(pair_id),
final_chosen_completion_id TEXT NOT NULL,
annotator_agreement_rate REAL, -- e.g., 1.0 if all agreed
total_annotations INT,
avg_confidence REAL,
validation_method TEXT, -- 'consensus', 'expert_review', 'auto'
created_at REAL NOT NULL
);
-- Dataset versions for reproducibility
CREATE TABLE dataset_versions (
version_id TEXT PRIMARY KEY,
version_tag TEXT UNIQUE NOT NULL, -- e.g., 'v1.2.0-customer-support'
pair_count INT,
quality_filter TEXT, -- JSON: filters applied
created_at REAL NOT NULL,
notes TEXT
);
CREATE INDEX idx_pairs_prompt_hash ON preference_pairs(prompt_hash);
CREATE INDEX idx_pairs_quality ON preference_pairs(quality_tier, is_excluded);
CREATE INDEX idx_pairs_version ON preference_pairs(dataset_version);
Different fine-tuning frameworks want different formats. Here's an exporter that handles the most common ones:
import json
from pathlib import Path
from datasets import Dataset # HuggingFace datasets
import pandas as pd
class PreferenceDatasetExporter:
def __init__(self, db):
self.db = db
def query_pairs(
self,
quality_tiers: list = ["high", "medium"],
construction_methods: list = ["annotator", "explicit"],
prompt_versions: list = None,
min_signal_gap: float = 0.5,
exclude_disputed: bool = True,
limit: int = None
) -> list[dict]:
query = """
SELECT p.*, v.final_chosen_completion_id, v.annotator_agreement_rate
FROM preference_pairs p
LEFT JOIN validated_preferences v ON p.pair_id = v.pair_id
WHERE p.is_excluded = FALSE
AND p.quality_tier IN ({quality_placeholders})
AND p.construction_method IN ({method_placeholders})
AND (p.chosen_signal_score - p.rejected_signal_score) >= ?
""".format(
quality_placeholders=",".join("?" * len(quality_tiers)),
method_placeholders=",".join("?" * len(construction_methods))
)
params = quality_tiers + construction_methods + [min_signal_gap]
if prompt_versions:
query += f" AND p.prompt_version IN ({','.join('?' * len(prompt_versions))})"
params += prompt_versions
if exclude_disputed:
query += """
AND p.pair_id NOT IN (
SELECT pair_id FROM annotation_queue WHERE status = 'disputed'
)
"""
if limit:
query += f" LIMIT {limit}"
return [dict(row) for row in self.db.execute(query, params).fetchall()]
def to_dpo_format(self, pairs: list[dict]) -> list[dict]:
"""
Export in TRL/DPO format compatible with HuggingFace TRL library.
"""
records = []
for pair in pairs:
# If human-validated, use the human's choice; otherwise use signal-based
if pair.get("final_chosen_completion_id"):
if pair["final_chosen_completion_id"] == pair["chosen_completion_id"]:
chosen, rejected = pair["chosen"], pair["rejected"]
else:
chosen, rejected = pair["rejected"], pair["chosen"]
else:
chosen, rejected = pair["chosen"], pair["rejected"]
records.append({
"prompt": pair["prompt"],
"chosen": chosen,
"rejected": rejected,
})
return records
def to_openai_messages_format(self, pairs: list[dict]) -> list[dict]:
"""
Export in OpenAI fine-tuning messages format.
Only uses 'chosen' — treats preference data as SFT data.
Useful when you want to fine-tune on preferred responses only.
"""
records = []
for pair in pairs:
chosen = pair["chosen"]
if pair.get("final_chosen_completion_id") and \
pair["final_chosen_completion_id"] == pair["rejected_completion_id"]:
chosen = pair["rejected"]
records.append({
"messages": [
{"role": "user", "content": pair["prompt"]},
{"role": "assistant", "content": chosen}
]
})
return records
def export(
self,
output_path: str,
format: str = "dpo",
version_tag: str = None,
**query_kwargs
) -> str:
pairs = self.query_pairs(**query_kwargs)
if format == "dpo":
records = self.to_dpo_format(pairs)
elif format == "openai_sft":
records = self.to_openai_messages_format(pairs)
else:
raise ValueError(f"Unknown format: {format}")
path = Path(output_path)
path.mkdir(parents=True, exist_ok=True)
# Always write JSONL
output_file = path / f"preferences_{version_tag or 'latest'}.jsonl"
with open(output_file, "w") as f:
for record in records:
f.write(json.dumps(record) + "\n")
# Write version manifest
manifest = {
"version": version_tag,
"format": format,
"pair_count": len(records),
"query_params": query_kwargs,
"created_at": time.time()
}
with open(path / "manifest.json", "w") as f:
json.dump(manifest, f, indent=2)
print(f"Exported {len(records)} pairs to {output_file}")
return str(output_file)
# Usage example — export high-quality DPO pairs for the v2 prompt
exporter = PreferenceDatasetExporter(db)
exporter.export(
output_path="./datasets/customer_support/",
format="dpo",
version_tag="v1.0.0",
quality_tiers=["high"],
construction_methods=["annotator"],
min_signal_gap=1.0
)
Collecting data without acting on it is expensive bookkeeping. Here's how to actually use what you've built.
Your preference dataset doubles as an evaluation benchmark. Once you have 200+ validated pairs, you can run your current model against them automatically:
def evaluate_against_preferences(
model_client,
preference_pairs: list[dict],
judge_model: str = "gpt-4o"
) -> dict:
"""
Use an LLM judge to assess whether the current model produces
responses that would be preferred over the 'rejected' baseline.
"""
wins, losses, ties = 0, 0, 0
for pair in preference_pairs:
# Generate a fresh response with the current model
current_response = model_client.complete(pair["prompt"])
# Ask judge to compare current response vs the 'rejected' baseline
judge_prompt = f"""
You are evaluating two responses to the same prompt.
Prompt: {pair["prompt"]}
Response A: {current_response}
Response B: {pair["rejected"]}
Which response is better? Answer with exactly one of: A, B, or TIE.
Reason briefly.
"""
judgment = model_client.complete(judge_prompt, model=judge_model)
if "Response A" in judgment or judgment.strip().startswith("A"):
wins += 1
elif "Response B" in judgment or judgment.strip().startswith("B"):
losses += 1
else:
ties += 1
total = len(preference_pairs)
return {
"win_rate": wins / total,
"loss_rate": losses / total,
"tie_rate": ties / total,
"total_evaluated": total
}
This gives you a win rate against your historical "rejected" baseline — a number you can track over time as you improve your prompts or fine-tune your model.
Track your win rate against a frozen baseline. If you keep updating what "rejected" means, you lose the ability to compare across time. Pick a representative sample of your worst historical responses and freeze it as a standing benchmark.
Build a minimal version of this pipeline end-to-end using a local SQLite database and a fictional customer support application.
Scenario: You're building a customer support assistant for a SaaS product. Users ask questions like "How do I export my data to CSV?" and your LLM generates answers. Some users give thumbs up/down.
Tasks:
Create the SQLite schema for completion_records, feedback_events, and preference_pairs.
Write a script that seeds the database with 20 synthetic completions across 5 distinct prompts. Vary the responses in quality (some are verbose and accurate, some are brief and accurate, some are inaccurate). Assign plausible feedback events (thumbs up/down) to each completion.
Use the PreferencePairConstructor to generate pairs for each prompt hash. Print each pair with its quality tier.
Export the resulting pairs in DPO format to a .jsonl file. Open it and verify it looks correct.
Inspect your output: which prompt produced the clearest preference pairs? Which ones were too close to call (low quality tier)? What would you need to do to resolve those ambiguous ones?
Stretch goal: Modify the annotation queue to detect when the same user submits feedback twice on the same completion (a sign of accidental duplicate clicks) and deduplicate automatically.
Pairing responses from different prompt versions. If your prompt template changes between two completions, any preference signal you extract is confounded. You can't tell if users preferred response B because the model was better or because the new prompt was clearer. Always filter pairs to share the same prompt_version.
Treating all thumbs-down equally. A user who thumbs-down a response immediately after reading it is different from one who scrolls to the bottom and then thumbs-down. Capture response interaction timing when you can, and use it to weight the signal.
Not randomizing annotator response order. If response A always appears on the left, your annotators will develop a systematic preference for "left" responses. This poisons your dataset silently. Randomize and record the presentation order in your annotation schema.
Building pairs with too little signal gap. If completion A has a score of 0.2 and completion B has a score of 0.0, the gap is so small that noise dominates. Use a min_signal_gap threshold — start conservatively at 0.8 and relax it only when you need more data volume.
Not versioning your dataset. Fine-tuning is expensive. If you can't reproduce the exact dataset that produced a given model checkpoint, you can't debug regressions. Write your manifest file every time you export. Store it with the model artifact.
Letting disputed pairs silently enter your training data. Inter-annotator disagreement is a signal that the pair is genuinely ambiguous. Ambiguous pairs add noise to fine-tuning. Route disputed pairs to a senior reviewer or exclude them — don't let them slip in automatically.
Collecting feedback only from power users. If you have a feedback button that requires clicking through to a response detail page, only your most engaged users will use it. This biases your dataset toward expert-level preferences. Instrument in-line feedback at the response level, visible to all users, to capture a more representative distribution.
You've built a complete preference data pipeline: instrumented completion logging, multi-signal feedback collection, preference pair construction with quality scoring, a human annotation queue with inter-annotator agreement checking, versioned dataset storage, and export to DPO-compatible format.
The pipeline you built follows a deliberate architecture — each layer has a single responsibility, and data moves from raw signal to structured preference pair through explicit transformation steps. This makes it auditable, debuggable, and extensible as your needs grow.
Here's where to go next:
Improve your reward modeling. Your score_completion function is a simple weighted sum. For a production system, you can train a small reward model on your validated preference pairs that scores completions more accurately than a hand-crafted formula.
Set up scheduled exports and drift alerts. Once your pipeline is running, schedule weekly exports and track your preference data statistics over time: number of new pairs per week, ratio of explicit to implicit signal, annotation agreement rate. Sharp drops in any of these usually indicate a bug or a product change you need to investigate.
Connect to your prompt engineering workflow. When you analyze which prompts generate the most "rejected" completions, you've identified your highest-priority prompt improvement targets. Build a dashboard that surfaces this — it becomes the agenda for your weekly prompt review.
Explore DPO fine-tuning. With a high-quality JSONL dataset in hand, the HuggingFace TRL library makes it straightforward to run a DPO fine-tuning job on an open-source base model. Your next step is learning how to set up that training pipeline and evaluate the resulting checkpoint against your frozen evaluation benchmark.
The feedback loop is built. Now feed it.