:::tip 🎮 Interactive Playground Visualize this concept: Try the Human Evaluation Process demo on the EngineersOfAI Playground - no code required. :::
Feedback Collection for LLM Systems
The Feedback Black Hole
It was month four of production. The AI writing assistant had 18,000 active users, 600,000 responses in the database, and the engineering team was proud of the numbers: P99 latency at 2.1 seconds, error rate at 0.04%, availability at 99.97%. The operations dashboard was green in every direction.
Then the quarterly NPS survey landed: 21. Below average for a SaaS product. The open-ended responses were cryptic - "sometimes great, sometimes totally wrong" and "I don't always trust it anymore" and "I love the idea but I can't rely on it." Nobody could identify what was wrong. The team had no feedback data. Not because users hadn't experienced problems - because nobody had built a system to capture those experiences.
The five-star rating modal that appeared every third session? Users closed it in under two seconds. The "Report an issue" button buried three levels deep in settings? Zero clicks in four months. The team had built a sophisticated AI system and a complete feedback void simultaneously. They were flying blind at altitude.
This is the paradox of feedback collection in AI products: the best AI experiences feel seamless, and seamless experiences generate no feedback signal. The system works - users don't feel the need to comment. But when it fails, the failure is subtle. The AI gives a confident-sounding wrong answer. The user is unsure whether to trust it. They quietly stop using the feature. By the time you see churn data, you are months behind the actual problem.
Feedback collection is a product design problem as much as a data engineering problem. The goal is to capture high-quality signal about model quality with the minimum friction that still yields sufficient volume and diversity to be actionable. Get the design wrong, and you either collect too little (users don't engage) or the wrong kind (biased, unrepresentative). Get it right, and you build a closed loop where every bad response makes the next generation of the model better.
This lesson covers the full stack: signal taxonomy, UI patterns, implicit signal collection, data schemas, bias prevention, clustering failures, and closing the loop into measurable improvement.
The Feedback Signal Hierarchy
Not all signals carry equal information density. Understanding this hierarchy shapes what you build and how you weight different signals when training or evaluating:
| Signal Type | Value | Volume | Friction | Best Use |
|---|---|---|---|---|
| Correction with full content | Highest | Very low (0.1-0.5%) | High | DPO/RLHF training data |
| Thumbs down + reason category | High | Low (2-4%) | Medium | Targeted prompt fixes |
| Thumbs down (bare) | Medium | Low-Medium (4-8%) | Low | Quality monitoring |
| Thumbs up | Low-Medium | Medium (3-7%) | Low | Quality monitoring |
| Reformulation detection | Medium | Medium (auto) | None | Systematic failure detection |
| Copy to clipboard | Low-Medium | High (auto) | None | Broad satisfaction signal |
| Edit detection | Medium | Low-Medium (auto) | None | Preference pair generation |
| Quick abandonment | Low | High (auto) | None | Session-level quality signal |
| Automated eval score | Low-Medium | All traffic | None | Continuous monitoring |
The fundamental trade-off: higher-value signals require higher friction, which means lower volume. A 5-star rating modal with a required text explanation collects extremely rich data from the 0.2% of users who complete it and systematically misses the 99.8% who don't - and those 99.8% include users who are silently dissatisfied. The winning strategy is always a layered approach: capture high-friction explicit signals for model improvement, and low-friction implicit signals for broad quality monitoring.
Explicit Feedback: UI Patterns and Backend
Pattern 1: Inline Thumbs with Reason Picker
The highest-value-to-friction ratio for explicit signals. Every thumbs-down should immediately present a reason picker - a horizontal set of one-click tags:
# backend/feedback/explicit.py
import anthropic
import uuid
from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional
import asyncpg
app = FastAPI()
anthropic_client = anthropic.Anthropic()
class ThumbsFeedback(BaseModel):
run_id: str = Field(..., description="Trace ID from observability backend")
thumbs_up: bool
reason_category: Optional[str] = Field(
None,
description="For thumbs-down: 'wrong_info', 'unhelpful', 'bad_tone', 'too_long', 'too_short', 'other'"
)
comment: Optional[str] = Field(None, max_length=2000)
session_id: str
user_id: str
feature: str = Field(..., description="Which product feature - 'writing_assist', 'search', 'summarize'")
model_version: Optional[str] = None
prompt_version: Optional[str] = None
class CorrectionFeedback(BaseModel):
run_id: str
original_response: str
corrected_response: str
correction_type: str = Field(
...,
description="'factual', 'tone', 'format', 'completeness', 'other'"
)
user_id: str
session_id: str
feature: str
class SessionFeedback(BaseModel):
session_id: str
task_completed: bool
helpfulness_score: Optional[int] = Field(None, ge=1, le=5)
difficulty_score: Optional[int] = Field(None, ge=1, le=5)
user_id: str
feature: str
@app.post("/api/feedback/thumbs")
async def submit_thumbs_feedback(
feedback: ThumbsFeedback,
db: asyncpg.Connection = Depends(get_db)
):
"""
Log thumbs feedback. For thumbs-down, the reason_category dramatically
increases the actionability of the signal.
"""
feedback_id = str(uuid.uuid4())
score = 1.0 if feedback.thumbs_up else 0.0
# Determine feedback type for schema
if feedback.thumbs_up:
fb_type = "thumbs_up"
elif feedback.reason_category:
fb_type = "thumbs_down_with_reason"
else:
fb_type = "thumbs_down"
await db.execute("""
INSERT INTO feedback_records (
id, run_id, session_id, user_id, feedback_type,
score, is_positive, reason_category, comment,
model_version, prompt_version, feature, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
""",
feedback_id, feedback.run_id, feedback.session_id,
feedback.user_id, fb_type, score, feedback.thumbs_up,
feedback.reason_category, feedback.comment,
feedback.model_version, feedback.prompt_version,
feedback.feature, datetime.now(timezone.utc)
)
# If thumbs-down with reason, check if we should sample this trace for
# human review queue
if not feedback.thumbs_up and feedback.reason_category == "wrong_info":
await enqueue_for_human_review(feedback.run_id, priority="high", db=db)
return {"status": "recorded", "feedback_id": feedback_id}
@app.post("/api/feedback/correction")
async def submit_correction(
correction: CorrectionFeedback,
db: asyncpg.Connection = Depends(get_db)
):
"""
Log an explicit correction - the highest-value feedback signal.
Stores (original, corrected) as a preference pair for future DPO training.
"""
feedback_id = str(uuid.uuid4())
await db.execute("""
INSERT INTO feedback_records (
id, run_id, session_id, user_id, feedback_type,
score, is_positive, original_response, corrected_response,
correction_type, feature, created_at
) VALUES ($1, $2, $3, $4, 'correction', 0.0, false, $5, $6, $7, $8, $9)
""",
feedback_id, correction.run_id, correction.session_id,
correction.user_id, correction.original_response,
correction.corrected_response, correction.correction_type,
correction.feature, datetime.now(timezone.utc)
)
# Save as a DPO training pair
await db.execute("""
INSERT INTO preference_pairs (
id, run_id, rejected, chosen, correction_type,
user_id, signal_type, feature, created_at
) VALUES ($1, $2, $3, $4, $5, $6, 'correction', $7, $8)
""",
str(uuid.uuid4()), correction.run_id,
correction.original_response, correction.corrected_response,
correction.correction_type, correction.user_id,
correction.feature, datetime.now(timezone.utc)
)
return {"status": "recorded", "preference_pair_saved": True}
@app.post("/api/feedback/session-complete")
async def session_complete_feedback(
feedback: SessionFeedback,
db: asyncpg.Connection = Depends(get_db)
):
"""
Collect session-level feedback after a user marks a task as done.
Delayed rating - less disruptive than inline, captures overall impression.
"""
await db.execute("""
INSERT INTO session_feedback (
id, session_id, user_id, feature, task_completed,
helpfulness_score, difficulty_score, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
""",
str(uuid.uuid4()), feedback.session_id, feedback.user_id,
feedback.feature, feedback.task_completed,
feedback.helpfulness_score, feedback.difficulty_score,
datetime.now(timezone.utc)
)
return {"status": "recorded"}
:::tip Design Rule: The One-Click Threshold Every step beyond one click reduces feedback completion by approximately 50%. A thumbs-down with an optional reason picker (pre-populated tags, no typing required) gets ~60% completion on the reason. A thumbs-down with a required free-text comment gets 8-12% completion. Design for the highest-friction signal that users will realistically complete. :::
Pattern 2: Correction Flow (Highest-Value Signal)
The correction flow - where a user can edit the AI's response to the correct version - requires careful UX design. The response must be pre-populated in an editable field, the diff must be clearly visible, and submission must be one click:
# backend/feedback/correction_validator.py
import anthropic
from dataclasses import dataclass
@dataclass
class CorrectionValidationResult:
is_genuine_correction: bool
quality_score: float # 0-1: is the corrected version actually better?
reason: str
should_use_for_training: bool
async def validate_correction_quality(
original_prompt: str,
original_response: str,
corrected_response: str,
correction_type: str
) -> CorrectionValidationResult:
"""
Use Claude to validate that a user's correction is actually an improvement.
Prevents spam corrections and low-quality edits from polluting training data.
"""
client = anthropic.Anthropic()
validation_prompt = f"""You are validating a user correction to an AI response.
Original question/prompt: {original_prompt}
Original AI response:
{original_response}
User's corrected version:
{corrected_response}
Correction type claimed: {correction_type}
Assess:
1. Is the corrected version actually better than the original? (yes/no + brief reason)
2. Is this a genuine improvement or is it spam/noise? (genuine/spam)
3. Quality score for the corrected version (0.0-1.0)
4. Should this be used as training data? (yes/no)
Respond in exactly this format:
BETTER: yes/no
REASON: one sentence
TYPE: genuine/spam
QUALITY: 0.X
TRAIN: yes/no"""
message = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=200,
messages=[{"role": "user", "content": validation_prompt}]
)
response_text = message.content[0].text
lines = {
line.split(": ")[0]: line.split(": ", 1)[1]
for line in response_text.strip().split("\n")
if ": " in line
}
is_better = lines.get("BETTER", "no").lower() == "yes"
is_genuine = lines.get("TYPE", "spam").lower() == "genuine"
quality = float(lines.get("QUALITY", "0.5"))
should_train = lines.get("TRAIN", "no").lower() == "yes"
return CorrectionValidationResult(
is_genuine_correction=is_genuine,
quality_score=quality,
reason=lines.get("REASON", ""),
should_use_for_training=should_train and is_better and is_genuine and quality >= 0.7
)
Implicit Signal Collection
Implicit signals require no user action - they are inferred from behavioral events tracked by the frontend. They are noisier than explicit signals but collected from 100% of users, making them essential for broad quality monitoring.
# backend/feedback/implicit.py
import anthropic
import difflib
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass
from typing import Optional
import asyncpg
@dataclass
class ImplicitSignalEvent:
run_id: str
user_id: str
session_id: str
event_type: str
score: float
metadata: dict
created_at: datetime
class ImplicitSignalTracker:
"""
Tracks behavioral signals that reveal response quality without
requiring any explicit user action. Integrates with the feedback
records table using the same schema as explicit signals.
"""
def __init__(self, db: asyncpg.Connection):
self.db = db
async def track_copy(
self,
run_id: str,
user_id: str,
session_id: str,
content_length: int,
feature: str
) -> None:
"""
User copied the AI's response to clipboard.
Strong positive signal - they found the content valuable enough to use.
Score: 0.8 (not 1.0 because copying could be for any reason)
"""
await self._persist_event(ImplicitSignalEvent(
run_id=run_id,
user_id=user_id,
session_id=session_id,
event_type="copy",
score=0.8,
metadata={"content_length": content_length, "feature": feature},
created_at=datetime.now(timezone.utc)
))
async def track_edit(
self,
run_id: str,
user_id: str,
session_id: str,
original: str,
edited: str,
feature: str
) -> Optional[dict]:
"""
User edited the AI's output before using it.
The extent of the edit determines the signal value.
Light editing (typos, formatting) = neutral.
Heavy editing (structural rewrite) = strong negative.
"""
ratio = difflib.SequenceMatcher(None, original, edited).ratio()
edit_extent = 1.0 - ratio # 0 = unchanged, 1 = completely rewritten
# Minor stylistic edits are not informative
if edit_extent < 0.05:
return None
# Score: 0 = heavily edited (bad original), 0.5 = lightly edited (ok original)
score = max(0.0, 0.5 - edit_extent)
await self._persist_event(ImplicitSignalEvent(
run_id=run_id,
user_id=user_id,
session_id=session_id,
event_type="edit",
score=score,
metadata={
"edit_extent": round(edit_extent, 3),
"original_length": len(original),
"edited_length": len(edited),
"feature": feature
},
created_at=datetime.now(timezone.utc)
))
# Substantial edits generate preference pair data
if edit_extent > 0.3:
await self._save_preference_pair(
run_id=run_id,
rejected=original,
chosen=edited,
user_id=user_id,
signal_type="implicit_edit",
edit_extent=edit_extent,
feature=feature
)
return {"preference_pair_saved": True, "edit_extent": edit_extent}
return {"preference_pair_saved": False, "edit_extent": edit_extent}
async def track_reformulation(
self,
session_id: str,
original_query: str,
reformulated_query: str,
original_run_id: str,
time_between_seconds: int,
feature: str
) -> None:
"""
User asked the same question differently within the same session.
If the queries are semantically similar and close in time, this is a
strong signal that the original answer did not satisfy the user.
Uses embedding cosine similarity to detect reformulations vs. topic changes.
"""
# Compute semantic similarity to distinguish reformulations from new questions
similarity = await self._compute_semantic_similarity(
original_query, reformulated_query
)
# Only flag as reformulation if:
# - Semantically similar (same underlying question)
# - Within 3 minutes (likely retry, not a new session direction)
if similarity > 0.72 and time_between_seconds < 180:
await self._persist_event(ImplicitSignalEvent(
run_id=original_run_id,
user_id="", # session-level, may not have user_id
session_id=session_id,
event_type="reformulation",
score=0.1, # strong negative: original answer clearly failed
metadata={
"similarity": round(similarity, 3),
"time_between_seconds": time_between_seconds,
"reformulated_query": reformulated_query[:500],
"feature": feature
},
created_at=datetime.now(timezone.utc)
))
async def track_abandonment(
self,
session_id: str,
last_run_id: str,
time_since_response_seconds: int,
feature: str
) -> None:
"""
User abandoned the session shortly after receiving a response.
Ambiguous signal: could mean they got what they needed, or they gave up.
Context matters: on a complex task, quick abandonment = negative.
On a quick lookup task, quick abandonment = positive.
Use with caution and low weight in aggregation.
"""
if time_since_response_seconds < 8:
score = 0.3 # very quick: ambiguous, slight negative lean
elif time_since_response_seconds < 20:
score = 0.4 # moderate: somewhat ambiguous
else:
return # not a meaningful abandonment signal
await self._persist_event(ImplicitSignalEvent(
run_id=last_run_id,
user_id="",
session_id=session_id,
event_type="abandonment",
score=score,
metadata={
"time_since_response_seconds": time_since_response_seconds,
"feature": feature
},
created_at=datetime.now(timezone.utc)
))
async def track_share_export(
self,
run_id: str,
user_id: str,
session_id: str,
export_type: str, # "share_link", "copy_all", "export_pdf", "email"
feature: str
) -> None:
"""
User shared or exported the AI's response.
Strong positive signal - they found it valuable enough to share externally.
"""
await self._persist_event(ImplicitSignalEvent(
run_id=run_id,
user_id=user_id,
session_id=session_id,
event_type="share_export",
score=0.9, # highest positive implicit signal
metadata={"export_type": export_type, "feature": feature},
created_at=datetime.now(timezone.utc)
))
async def _compute_semantic_similarity(self, text1: str, text2: str) -> float:
"""
Use a lightweight embedding model to compute cosine similarity.
For production: use a cached embedding service, not per-request computation.
"""
# In practice, use sentence-transformers with a cached model
# or call an embedding API with caching
# Example with sentence-transformers:
try:
from sentence_transformers import SentenceTransformer, util
model = SentenceTransformer("all-MiniLM-L6-v2")
emb1 = model.encode(text1, convert_to_tensor=True)
emb2 = model.encode(text2, convert_to_tensor=True)
return float(util.cos_sim(emb1, emb2))
except ImportError:
# Fallback: word overlap Jaccard similarity
words1 = set(text1.lower().split())
words2 = set(text2.lower().split())
intersection = words1 & words2
union = words1 | words2
return len(intersection) / len(union) if union else 0.0
async def _persist_event(self, event: ImplicitSignalEvent) -> None:
await self.db.execute("""
INSERT INTO feedback_records (
id, run_id, session_id, user_id, feedback_type,
score, is_positive, extra_metadata, created_at
) VALUES (
gen_random_uuid(), $1, $2, $3, $4,
$5, $6, $7::jsonb, $8
)
""",
event.run_id, event.session_id, event.user_id,
event.event_type, event.score, event.score >= 0.5,
str(event.metadata), event.created_at
)
async def _save_preference_pair(
self, run_id: str, rejected: str, chosen: str,
user_id: str, signal_type: str, edit_extent: float, feature: str
) -> None:
await self.db.execute("""
INSERT INTO preference_pairs (
id, run_id, rejected, chosen, user_id,
signal_type, edit_extent, feature, created_at
) VALUES (gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8)
""",
run_id, rejected, chosen, user_id,
signal_type, edit_extent, feature, datetime.now(timezone.utc)
)
Feedback Data Schema
A well-designed feedback schema must accommodate all signal types in a single table for unified analysis, while preserving the type-specific fields that make each signal actionable:
# db/schema.py
from sqlalchemy import (
Column, String, Float, DateTime, JSON, Boolean,
Enum, Text, Integer, ForeignKey, Index
)
from sqlalchemy.orm import declarative_base
import enum
Base = declarative_base()
class FeedbackType(str, enum.Enum):
THUMBS_UP = "thumbs_up"
THUMBS_DOWN = "thumbs_down"
THUMBS_DOWN_WITH_REASON = "thumbs_down_with_reason"
CORRECTION = "correction"
EDIT = "edit"
COPY = "copy"
SHARE_EXPORT = "share_export"
REFORMULATION = "reformulation"
ABANDONMENT = "abandonment"
SESSION_RATING = "session_rating"
AUTOMATED_EVAL = "automated_eval"
class FeedbackRecord(Base):
"""
Unified feedback table - all signal types stored here.
Enables cross-signal analysis (e.g., sessions where thumbs-down
AND reformulation occurred = highest-priority failure traces).
"""
__tablename__ = "feedback_records"
id = Column(String, primary_key=True)
# Linkage to observability backend
run_id = Column(String, index=True, nullable=False) # trace ID
session_id = Column(String, index=True)
user_id = Column(String, index=True)
# Core signal fields
feedback_type = Column(Enum(FeedbackType), nullable=False)
score = Column(Float) # 0.0 (very bad) to 1.0 (very good)
is_positive = Column(Boolean) # simplified bucketing
# Explicit signal fields
reason_category = Column(String) # wrong_info, unhelpful, bad_tone, etc.
comment = Column(Text)
# Correction / edit fields
original_response = Column(Text)
corrected_response = Column(Text)
correction_type = Column(String) # factual, tone, format, completeness
edit_extent = Column(Float) # 0-1: how much was changed
# Automated eval fields
eval_metric = Column(String) # faithfulness, relevance, toxicity, etc.
eval_model = Column(String) # claude-haiku-4-5-20251001, etc.
eval_score = Column(Float)
eval_reasoning = Column(Text)
# Attribution / versioning
model_version = Column(String)
prompt_version = Column(String)
feature = Column(String, index=True) # writing_assist, search, etc.
user_tier = Column(String) # free, pro, enterprise
ab_experiment_id = Column(String) # if part of an A/B test
# Behavioral context
response_latency_ms = Column(Integer)
session_position = Column(Integer) # which response in the session (1st, 2nd, ...)
created_at = Column(DateTime, nullable=False, index=True)
extra_metadata = Column(JSON)
class PreferencePair(Base):
"""
Stores (rejected, chosen) pairs for DPO/RLHF training.
Populated by corrections, edits, and validated comparison data.
"""
__tablename__ = "preference_pairs"
id = Column(String, primary_key=True)
run_id = Column(String, ForeignKey("feedback_records.run_id"))
rejected = Column(Text, nullable=False)
chosen = Column(Text, nullable=False)
prompt = Column(Text)
# Provenance
signal_type = Column(String) # correction, implicit_edit, comparison
user_id = Column(String)
edit_extent = Column(Float)
feature = Column(String)
# Quality validation
validated = Column(Boolean, default=False)
validation_score = Column(Float) # judge model score of chosen vs rejected
validation_model = Column(String)
created_at = Column(DateTime, nullable=False)
# Indexes for common query patterns
Index("idx_feedback_run_id_type", FeedbackRecord.run_id, FeedbackRecord.feedback_type)
Index("idx_feedback_feature_date", FeedbackRecord.feature, FeedbackRecord.created_at)
Index("idx_feedback_type_score", FeedbackRecord.feedback_type, FeedbackRecord.score)
Avoiding Annotation Bias
Poorly designed feedback systems collect systematically biased data. Biased data creates models that are good at satisfying biased raters, not at being genuinely helpful. There are five primary bias sources:
Preventing Anchoring Bias
# Never show previous ratings when asking for new ones
# BAD:
async def get_rating_ui_bad(run_id: str, user_id: str, db) -> dict:
prev_rating = await db.get_user_rating(run_id, user_id)
return {
"show_rating": True,
"previous_rating": prev_rating, # this anchors the new rating to the old one
}
# GOOD:
async def get_rating_ui_good(run_id: str, user_id: str) -> dict:
return {
"show_rating": True,
# No previous rating shown - each rating is independent
}
Randomizing Position in A/B Comparisons
import random
from dataclasses import dataclass
@dataclass
class ComparisonPair:
left_response: str
left_id: str # "a" or "b" in the experiment
right_response: str
right_id: str
position_swapped: bool # metadata for analysis
def get_comparison_ui(
response_a: str,
response_b: str,
pair_id: str
) -> ComparisonPair:
"""
Present two responses for comparison without position bias.
Randomly determine which response is shown on the left vs. right.
Store whether positions were swapped so you can correct for any
residual position bias in analysis.
"""
swap = random.random() > 0.5
if swap:
return ComparisonPair(
left_response=response_b, left_id="b",
right_response=response_a, right_id="a",
position_swapped=True
)
else:
return ComparisonPair(
left_response=response_a, left_id="a",
right_response=response_b, right_id="b",
position_swapped=False
)
Annotator Calibration System
For human annotation campaigns (dedicated labelers, not end users), you need a systematic approach to measuring and maintaining annotation quality:
# feedback/calibration.py
import random
from dataclasses import dataclass, field
from typing import Optional
import statistics
@dataclass
class CalibrationExample:
"""A known-quality example with an established ground-truth rating."""
example_id: str
prompt: str
response: str
ground_truth_score: float # established by expert consensus
ground_truth_category: str # what makes it good/bad
difficulty: str # easy, medium, hard
@dataclass
class AnnotatorStats:
annotator_id: str
agreement_rate: float # fraction that match ground truth within tolerance
total_calibration_items: int
avg_deviation: float # average distance from ground truth
is_calibrated: bool # above threshold agreement rate
flag_for_review: bool # below minimum threshold
class AnnotatorCalibrationSystem:
"""
Maintains annotation quality by embedding calibration examples in
annotation batches and measuring annotator agreement with known-quality items.
"""
AGREEMENT_THRESHOLD = 0.75 # 75% agreement = calibrated annotator
REVIEW_THRESHOLD = 0.60 # below 60% = needs retraining
INJECTION_RATE = 0.10 # 10% of each batch are calibration items
def __init__(self, calibration_examples: list[CalibrationExample]):
self.calibration_examples = {
ex.example_id: ex for ex in calibration_examples
}
def inject_calibration_items(
self,
batch: list[dict],
annotator_id: str
) -> list[dict]:
"""
Inject calibration items into an annotation batch.
Annotators do not know which items are calibration vs. real.
"""
result = []
calibration_pool = list(self.calibration_examples.values())
random.shuffle(calibration_pool)
cal_iter = iter(calibration_pool)
for item in batch:
result.append(item)
# Inject calibration item after this item with INJECTION_RATE probability
if random.random() < self.INJECTION_RATE:
try:
cal_example = next(cal_iter)
calibration_item = {
"prompt": cal_example.prompt,
"response": cal_example.response,
"_calibration_item": True, # hidden from annotator
"_calibration_id": cal_example.example_id,
"_annotator_id": annotator_id,
}
result.append(calibration_item)
except StopIteration:
pass
return result
def compute_annotator_stats(
self,
annotator_id: str,
calibration_responses: list[dict]
) -> AnnotatorStats:
"""
Compute an annotator's quality metrics from their calibration responses.
calibration_responses: list of {
"calibration_id": str,
"annotator_score": float
}
"""
if not calibration_responses:
return AnnotatorStats(
annotator_id=annotator_id,
agreement_rate=0.0,
total_calibration_items=0,
avg_deviation=1.0,
is_calibrated=False,
flag_for_review=True
)
agreements = []
deviations = []
for response in calibration_responses:
cal_id = response["calibration_id"]
annotator_score = response["annotator_score"]
if cal_id not in self.calibration_examples:
continue
ground_truth = self.calibration_examples[cal_id].ground_truth_score
# Agreement = within 1 point on a 5-point scale (or 0.2 on 0-1 scale)
deviation = abs(ground_truth - annotator_score)
deviations.append(deviation)
agreements.append(deviation <= 0.2)
agreement_rate = sum(agreements) / len(agreements) if agreements else 0.0
avg_deviation = statistics.mean(deviations) if deviations else 1.0
return AnnotatorStats(
annotator_id=annotator_id,
agreement_rate=agreement_rate,
total_calibration_items=len(agreements),
avg_deviation=round(avg_deviation, 3),
is_calibrated=agreement_rate >= self.AGREEMENT_THRESHOLD,
flag_for_review=agreement_rate < self.REVIEW_THRESHOLD
)
def weight_annotation(self, annotator_id: str, raw_score: float, db) -> float:
"""
Weight an annotator's scores by their calibration agreement rate.
Poor calibrators have their scores down-weighted toward the mean.
"""
stats = db.get_annotator_stats(annotator_id)
if not stats:
return raw_score # unknown annotator: accept as-is but track
# Linear interpolation: calibrated annotators get full weight,
# poor calibrators get pulled toward 0.5 (neutral)
weight = max(0.0, min(1.0, stats.agreement_rate))
return raw_score * weight + 0.5 * (1 - weight)
:::warning Selection Bias Is Unavoidable - Compensate for It Only 2-8% of users provide explicit feedback. These users are systematically different: more engaged, more opinionated, more likely to be power users or frequent complainers. Their feedback does not represent the silent majority. Compensate by: (1) supplementing explicit with implicit signals from all users, (2) running periodic random sampling surveys via in-app banners with random user selection, (3) building separate quality metrics for the explicit-feedback segment vs. all users to measure the gap. :::
Feedback Loop: From Signal to Model Improvement
The entire value of a feedback system is realized only if the signals actually change the product. Most teams collect feedback that sits in a database forever. The closed loop requires four stages:
Failure Mode Clustering
The key insight that most teams miss: individual bad responses are noise. Clustered bad responses reveal the type of failure - and types of failures have systematic fixes.
# feedback/failure_clustering.py
import anthropic
import numpy as np
from sklearn.cluster import KMeans, DBSCAN
from sklearn.preprocessing import normalize
from dataclasses import dataclass
from typing import Optional
import asyncpg
@dataclass
class FailureCluster:
cluster_id: int
size: int
sample_queries: list[str]
sample_comments: list[str]
centroid: np.ndarray
failure_summary: str # LLM-generated summary of the failure pattern
likely_root_cause: str # "prompt", "knowledge_gap", "model_capability"
suggested_action: str # specific next step
async def cluster_failure_modes(
db: asyncpg.Connection,
feature: str,
days: int = 7,
min_cluster_size: int = 5
) -> list[FailureCluster]:
"""
Find systematic failure patterns by clustering negative feedback
using semantic embeddings. Clusters represent distinct failure categories
that can be addressed systematically.
"""
# Fetch negative feedback with associated queries
rows = await db.fetch("""
SELECT fr.run_id, fr.comment, fr.reason_category,
t.user_query, t.ai_response
FROM feedback_records fr
JOIN traces t ON t.run_id = fr.run_id
WHERE fr.feedback_type IN ('thumbs_down', 'thumbs_down_with_reason', 'reformulation')
AND fr.feature = $1
AND fr.created_at > NOW() - INTERVAL '%s days' % $2
AND fr.score < 0.4
ORDER BY fr.created_at DESC
LIMIT 2000
""", feature, days)
if len(rows) < 20:
return [] # Not enough data for meaningful clustering
# Build text representation for embedding
texts = []
for row in rows:
query = row["user_query"] or ""
comment = row["comment"] or ""
reason = row["reason_category"] or ""
text = f"Query: {query[:300]}\nReason: {reason}\nComment: {comment[:200]}"
texts.append(text)
# Compute embeddings
embeddings = await _batch_embed(texts)
embeddings_normalized = normalize(embeddings)
# Use DBSCAN for density-based clustering (handles noise better than KMeans
# when failure modes have different frequencies)
clustering = DBSCAN(eps=0.3, min_samples=min_cluster_size, metric="cosine")
labels = clustering.fit_predict(embeddings_normalized)
unique_labels = set(labels) - {-1} # -1 = noise/outliers
clusters = []
for label in unique_labels:
mask = labels == label
cluster_indices = np.where(mask)[0]
cluster_rows = [rows[i] for i in cluster_indices]
# Sample queries and comments for the cluster summary
sample_queries = [r["user_query"] for r in cluster_rows[:8] if r["user_query"]]
sample_comments = [r["comment"] for r in cluster_rows[:8] if r["comment"]]
# Use Claude to generate a concise failure summary and root cause
failure_summary, root_cause, suggested_action = await _summarize_cluster(
sample_queries, sample_comments
)
clusters.append(FailureCluster(
cluster_id=int(label),
size=len(cluster_indices),
sample_queries=sample_queries[:5],
sample_comments=sample_comments[:5],
centroid=embeddings_normalized[cluster_indices].mean(axis=0),
failure_summary=failure_summary,
likely_root_cause=root_cause,
suggested_action=suggested_action
))
# Sort largest cluster first (biggest impact)
clusters.sort(key=lambda c: -c.size)
return clusters
async def _summarize_cluster(
sample_queries: list[str],
sample_comments: list[str]
) -> tuple[str, str, str]:
"""Use Claude to generate a human-readable cluster summary."""
client = anthropic.Anthropic()
queries_text = "\n".join(f"- {q}" for q in sample_queries[:5])
comments_text = "\n".join(f"- {c}" for c in sample_comments[:5])
prompt = f"""Analyze these negative feedback examples from an AI assistant.
User queries that got negative feedback:
{queries_text}
User comments/reasons:
{comments_text}
Provide:
1. SUMMARY: One sentence describing what type of failure this cluster represents.
2. ROOT_CAUSE: One of: "prompt_issue" | "knowledge_gap" | "model_capability" | "retrieval_failure" | "format_issue"
3. ACTION: One specific concrete next step to fix this.
Format exactly:
SUMMARY: <one sentence>
ROOT_CAUSE: <one of the options above>
ACTION: <one sentence>"""
message = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=250,
messages=[{"role": "user", "content": prompt}]
)
response = message.content[0].text
lines = {}
for line in response.strip().split("\n"):
if ": " in line:
key, val = line.split(": ", 1)
lines[key.strip()] = val.strip()
return (
lines.get("SUMMARY", "Unknown failure pattern"),
lines.get("ROOT_CAUSE", "unknown"),
lines.get("ACTION", "Investigate manually")
)
async def _batch_embed(texts: list[str]) -> np.ndarray:
"""Batch embedding with sentence-transformers or API."""
try:
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
return model.encode(texts, batch_size=64, show_progress_bar=False)
except ImportError:
# Fallback: use Anthropic embedding via voyage (or OpenAI)
# For simplicity, return random embeddings here as placeholder
return np.random.randn(len(texts), 384)
Building a Fine-Tuning Dataset from Feedback
# feedback/training_dataset.py
import anthropic
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta
import json
import asyncpg
@dataclass
class TrainingPair:
prompt: str
chosen: str
rejected: str
source: str # "correction", "implicit_edit", "comparison"
quality_score: float
feature: str
async def build_dpo_dataset(
db: asyncpg.Connection,
feature: str,
min_quality_score: float = 0.7,
days: int = 30
) -> list[TrainingPair]:
"""
Build a DPO (Direct Preference Optimization) training dataset
from accumulated feedback signals.
Sources:
1. Explicit corrections: user provided the correct version
2. Validated implicit edits: user heavily edited (edit_extent > 0.3)
and the edit was validated by a judge model
3. Comparison data from A/B experiments
"""
client = anthropic.Anthropic()
# Fetch preference pairs
rows = await db.fetch("""
SELECT pp.id, pp.run_id, pp.rejected, pp.chosen, pp.signal_type,
pp.feature, t.user_query as prompt,
pp.validation_score
FROM preference_pairs pp
JOIN traces t ON t.run_id = pp.run_id
WHERE pp.feature = $1
AND pp.created_at > NOW() - INTERVAL '%s days' % $2
AND (pp.validation_score IS NULL OR pp.validation_score >= $3)
AND pp.chosen IS NOT NULL
AND pp.rejected IS NOT NULL
AND pp.chosen != pp.rejected
ORDER BY pp.created_at DESC
""", feature, days, min_quality_score)
training_pairs = []
for row in rows:
if not row["prompt"] or not row["chosen"] or not row["rejected"]:
continue
# Validate unvalidated pairs with judge model
quality_score = row["validation_score"]
if quality_score is None:
quality_score = await _validate_pair_quality(
client=client,
prompt=row["prompt"],
chosen=row["chosen"],
rejected=row["rejected"]
)
# Update validation score in DB
await db.execute(
"UPDATE preference_pairs SET validation_score = $1, "
"validation_model = 'claude-haiku-4-5-20251001', validated = true "
"WHERE id = $2",
quality_score, row["id"]
)
if quality_score < min_quality_score:
continue
training_pairs.append(TrainingPair(
prompt=row["prompt"],
chosen=row["chosen"],
rejected=row["rejected"],
source=row["signal_type"],
quality_score=quality_score,
feature=row["feature"]
))
# Deduplicate: same prompt with multiple corrections -> keep highest quality
seen_prompts: dict[str, TrainingPair] = {}
for pair in training_pairs:
key = pair.prompt[:200] # hash on truncated prompt
if key not in seen_prompts or pair.quality_score > seen_prompts[key].quality_score:
seen_prompts[key] = pair
deduplicated = list(seen_prompts.values())
print(f"DPO dataset: {len(training_pairs)} raw pairs -> {len(deduplicated)} after dedup")
print(f"Source breakdown: {_count_by_source(deduplicated)}")
return deduplicated
async def _validate_pair_quality(
client: anthropic.Anthropic,
prompt: str,
chosen: str,
rejected: str
) -> float:
"""
Use a judge model to confirm that 'chosen' is actually better than 'rejected'.
Returns a quality score 0-1.
"""
judge_prompt = f"""You are evaluating whether a human's preferred AI response is genuinely better than the original.
User prompt: {prompt[:500]}
Original AI response (rejected by user):
{rejected[:800]}
Human's preferred version (chosen):
{chosen[:800]}
Is the chosen version genuinely better? Consider accuracy, helpfulness, and completeness.
Respond with just a number from 0.0 to 1.0 where:
0.0 = chosen is worse than rejected
0.5 = roughly equal
1.0 = chosen is clearly better
Score:"""
message = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=10,
messages=[{"role": "user", "content": judge_prompt}]
)
try:
return float(message.content[0].text.strip())
except ValueError:
return 0.5
def _count_by_source(pairs: list[TrainingPair]) -> dict:
counts: dict[str, int] = {}
for pair in pairs:
counts[pair.source] = counts.get(pair.source, 0) + 1
return counts
def export_dpo_dataset_jsonl(pairs: list[TrainingPair], output_path: str) -> None:
"""Export training pairs in standard DPO JSONL format."""
with open(output_path, "w") as f:
for pair in pairs:
record = {
"prompt": pair.prompt,
"chosen": pair.chosen,
"rejected": pair.rejected,
"source": pair.source,
"quality_score": round(pair.quality_score, 3),
"feature": pair.feature
}
f.write(json.dumps(record) + "\n")
print(f"Exported {len(pairs)} training pairs to {output_path}")
Aggregating Signals: The Feedback Score
Multiple signals about the same response need to be aggregated into a single quality estimate. The key is signal weighting based on reliability and information content:
# feedback/aggregation.py
from dataclasses import dataclass, field
from typing import Optional
# Signal weights: higher = more reliable / more information
SIGNAL_WEIGHTS = {
"correction": 10.0, # user provided the correct version
"thumbs_down_with_reason": 5.0,
"thumbs_down": 3.0,
"reformulation": 2.5,
"automated_eval": 2.0,
"thumbs_up": 2.0,
"edit": 1.5, # noisier: could be style preference
"share_export": 1.5,
"copy": 1.0,
"abandonment": 0.5, # most ambiguous
}
@dataclass
class AggregatedFeedbackScore:
run_id: str
weighted_score: float # 0-1, weighted by signal reliability
confidence: float # 0-1, based on total signal weight
signal_count: int
dominant_signal: Optional[str]
breakdown: dict = field(default_factory=dict)
def aggregate_feedback_signals(
run_id: str,
signals: list[dict]
) -> AggregatedFeedbackScore:
"""
Aggregate multiple feedback signals for a single run into a
single weighted quality score.
signals: list of {
"feedback_type": str,
"score": float,
}
"""
if not signals:
return AggregatedFeedbackScore(
run_id=run_id,
weighted_score=0.5, # unknown = neutral
confidence=0.0,
signal_count=0,
dominant_signal=None
)
total_weight = 0.0
weighted_sum = 0.0
breakdown = {}
for signal in signals:
fb_type = signal["feedback_type"]
score = signal["score"]
weight = SIGNAL_WEIGHTS.get(fb_type, 1.0)
total_weight += weight
weighted_sum += weight * score
if fb_type not in breakdown:
breakdown[fb_type] = {"count": 0, "avg_score": 0.0, "total_weight": 0.0}
breakdown[fb_type]["count"] += 1
breakdown[fb_type]["avg_score"] = (
breakdown[fb_type]["avg_score"] * (breakdown[fb_type]["count"] - 1) + score
) / breakdown[fb_type]["count"]
breakdown[fb_type]["total_weight"] += weight
weighted_score = weighted_sum / total_weight if total_weight > 0 else 0.5
# Confidence: caps at 1.0 when total weight reaches ~20 (5+ explicit signals)
confidence = min(1.0, total_weight / 20.0)
# Find the signal type that contributed the most weight
dominant_signal = max(breakdown.keys(), key=lambda k: breakdown[k]["total_weight"])
return AggregatedFeedbackScore(
run_id=run_id,
weighted_score=round(weighted_score, 3),
confidence=round(confidence, 3),
signal_count=len(signals),
dominant_signal=dominant_signal,
breakdown=breakdown
)
Common Mistakes
:::danger Never Use Raw Feedback Data Directly for Training Raw user feedback is polluted with spam, trolling, misunderstandings, and off-topic content. Before using feedback as training signal: (1) deduplicate - same user + same response + multiple ratings → keep only one, (2) filter by annotator reliability using calibration scores, (3) remove corrections where the "corrected" version is actually worse (use a judge model to validate), (4) balance across failure categories - don't over-represent any single failure mode. :::
:::warning Don't Treat Thumbs-Down Without Context as Actionable A bare thumbs-down tells you the user was dissatisfied, but not why. Was it wrong information? Poor tone? Too long? The response was fine but the user expected something different? Without the reason, you can't fix it. Always present a one-click reason picker after thumbs-down. Even a simplified 4-option picker (Wrong info / Unhelpful / Bad tone / Other) transforms the signal from "user unhappy" to "user unhappy because X". :::
:::danger Conflating Implicit and Explicit Signals Ruins Your Metrics A user copying an AI response means they found it valuable. But it might also mean they're copying it to show a colleague why it's wrong. A user abandoning a session might mean the response was terrible, or they got exactly what they needed in one turn. Implicit signals are probabilistic, not deterministic. Always weight them much lower than explicit signals, and always analyze them in aggregate (not individually). Never fire an alert based on a single implicit signal event. :::
:::warning Feedback Collection Is a Privacy Risk User feedback contains the exact queries users asked and the AI's exact responses - including potentially sensitive information. Design your schema with PII in mind: (1) hash or tokenize user IDs before storing, (2) scan response content for PII patterns before storing in feedback tables, (3) implement data retention policies (feedback data for model improvement doesn't need to live forever - 90-180 days is usually sufficient), (4) audit who has access to raw feedback data - it's more sensitive than aggregate metrics. :::
Interview Q&A
Q1: What feedback signals are most valuable for improving LLM systems, and how do you architect a system to collect them?
Answer: The signal hierarchy from most to least valuable: (1) Explicit correction with content - user rewrites the AI's response with the correct version. This gives you a (rejected, chosen) pair directly usable for DPO training without needing a labeler. Collection requires an "improve this" button below every response that opens an editable field pre-populated with the AI's output. (2) Thumbs-down with reason category - a one-click reason picker transforms a binary signal into an actionable category. (3) Bare thumbs-down - directional but not immediately actionable. (4) Behavioral reformulation - user asks the same question differently within the same session. Automated detection via embedding similarity (threshold ~0.75) with a time window (< 3 minutes) is highly reliable as a failure signal. (5) Copy to clipboard - positive implicit signal, high volume, easy to track via clipboard API events.
The architecture: a feedback ingestion API (FastAPI) receives all signals and writes to a unified feedback_records table. Explicit corrections also write to a preference_pairs table. Implicit signals are generated by frontend JavaScript event tracking (clipboard API, edit detection on contenteditable elements, session duration tracking) that POSTs to /api/feedback/implicit. A background worker runs nightly to validate new preference pairs with a judge model and export validated pairs for fine-tuning.
Key design principle: collect everything, but weight appropriately. A correction from one user is worth more than 20 bare thumbs-downs in training data, but 1,000 reformulation events are worth more than 50 corrections for broad quality monitoring.
Q2: How do you prevent bias in feedback data, and why does it matter?
Answer: Bias in feedback data creates a model that's good at satisfying biased raters, not at being genuinely helpful. The five major bias sources and their mitigations:
Selection bias: The 2-8% of users who provide explicit feedback are systematically different from the average user - more engaged, more opinionated, more likely to be power users or frustrated users. Mitigation: supplement explicit feedback with implicit signals from all users. Also run periodic random-sample surveys (shown to a random 1% of users, not just the most active) to get a less biased population.
Anchoring bias: If you show users their previous rating when asking them to re-rate, the new rating anchors to the old one. Simple fix: never display previous ratings in the feedback UI.
Position bias in A/B comparisons: When showing response A vs. B side-by-side, users prefer the response on the left by ~5-10% on average, regardless of quality. Mitigation: randomize position, and run each pair in both orders when doing systematic evaluation (each response appears left half the time).
Recency bias: Users only rate recent responses; older responses go unrated even if they were poor. Mitigation: use annotation queues that surface historical responses for rating, weighted by how likely they are to be informative (e.g., responses that triggered reformulations).
Adversarial / spam: Users deliberately rating incorrectly. Mitigation: annotator calibration tests - embed known-quality examples and measure agreement with ground truth. Down-weight or flag annotators below 70% agreement rate.
Why it matters: a model fine-tuned on biased data will optimize for the biased feedback rather than genuine quality. For example, if power users systematically prefer verbose responses, fine-tuning on their corrections will make the model more verbose - which may not be what average users want.
Q3: Walk me through building a closed feedback loop - from signal to model improvement.
Answer: The feedback loop has four stages. Most teams do stages 1-2 well and fail at 3-4:
Stage 1 - Capture: Multi-source collection. Explicit: thumbs, corrections, session ratings. Implicit: copy events, edit events, reformulation detection, abandonment. Automated: LLM-as-judge evaluations on sampled production traffic. Everything goes to the feedback_records table with consistent schema.
Stage 2 - Triage and aggregate: Daily job computes rolling quality metrics by feature. Identifies runs with multiple negative signals (e.g., thumbs-down AND reformulation on the same response - highest confidence failure). Sends high-priority failures to human review queue.
Stage 3 - Cluster and attribute (where most teams fail): Individual bad responses are noise. Cluster failures using embedding similarity (DBSCAN works well for varying cluster densities) to find types of failures. Use a judge model (Claude Haiku) to summarize each cluster in one sentence and attribute the likely root cause: prompt issue, knowledge gap, model capability, or retrieval failure. This step turns "we have 340 thumbs-downs this week" into "we have 3 systematic failure categories: (1) pricing calculations wrong when multiple discounts stack, (2) refund policy misquoted when asked in past tense, (3) escalation path for enterprise customers described incorrectly."
Stage 4 - Intervene and measure: Prompt issues: update the prompt, run your eval suite (which should include examples from the failure cluster), deploy if quality improves on the cluster. Knowledge gaps in RAG: update the knowledge base, re-embed, measure retrieval quality before/after. Model behavior issues: batch corrections into DPO training data (minimum ~500 validated pairs per category), fine-tune, evaluate on the held-out cluster before deploying.
Measure loop effectiveness: after each intervention, compute quality metrics specifically on the affected failure cluster. If the intervention worked, that cluster should shrink in the next cycle. Track cluster size week-over-week as the primary metric for improvement velocity.
Q4: What is the difference between explicit and implicit feedback, and when would you rely more on one versus the other?
Answer: Explicit feedback is intentionally provided: thumbs-up/down, corrections, star ratings, written comments. It's high signal-to-noise (each piece is directly interpretable) but low volume (2-8% of interactions), and subject to selection bias (power users, frustrated users, opinionated users are over-represented).
Implicit feedback is inferred from behavior: copying the response (positive), editing it (neutral to negative), asking the same question again (negative), sharing or exporting (strong positive), abandoning shortly after (weak negative). It's available for 100% of users but is ambiguous - a single copy event or abandonment event could mean many things. Implicit signals are only reliable in aggregate.
When to rely more on explicit: Early-stage products where you need high-quality ground truth for your evaluation dataset. When you want to build DPO training data (implicit edits can generate preference pairs, but explicit corrections are more reliable). When you need to understand why something failed (explicit feedback provides reason categories; implicit doesn't).
When to rely more on implicit: Large-scale products where volume makes aggregate statistics reliable. When you want to monitor quality for all users, not just the opinionated 5%. When explicit feedback rates have dropped (users become habituated to feedback prompts over time and ignore them). When you're monitoring very specific behaviors like "does the user use the AI's output or rewrite it entirely?" - only implicit edit tracking captures this.
In practice: use both. Use explicit as your primary signal for model improvement decisions (corrections → DPO data, thumbs-down with reason → root cause analysis). Use implicit as your primary signal for broad quality monitoring (daily reformulation rate, copy rate, abandonment rate after AI responses). Explicit catches quality issues earlier with less noise; implicit catches them with less bias.
Q5: How would you design the feedback schema for a production LLM system, and what fields are most important?
Answer: The schema must balance several competing requirements: store all signal types in one place for unified analysis, preserve type-specific fields that make each signal actionable, support efficient querying by feature/date/signal-type, and enable linkage back to the observability traces.
The critical fields:
Linkage fields - run_id (trace ID in your observability backend, enables linking feedback to the full request/response trace), session_id (enables session-level analysis, finding all signals in the same conversation), user_id (enables per-user quality analysis and bias filtering).
Signal fields - feedback_type (enum of all signal types: thumbs_up, thumbs_down, correction, edit, copy, reformulation, etc.), score (normalized 0-1 float so all signal types are comparable in aggregate queries), is_positive (boolean shortcut for fast filtering).
Actionability fields - reason_category (for thumbs-down: wrong_info, unhelpful, bad_tone - the most important field for root cause analysis), original_response and corrected_response (for corrections - enables DPO dataset creation), edit_extent (for edits - how much was changed, distinguishes light style edits from structural rewrites).
Attribution fields - model_version, prompt_version, feature, ab_experiment_id. These enable you to filter "what's the thumbs-down rate for model v2 vs v3?" and "does the new prompt reduce wrong_info ratings?". Without these, you can't do controlled quality analysis.
Privacy fields - The schema should NOT store raw user identity in plaintext. Hash the user_id before storage, or use a separate identity table with access controls. Implement a retention policy (delete raw feedback after 90-180 days; keep aggregated statistics permanently).
Indexing strategy: index on (run_id, feedback_type) for trace lookups, (feature, created_at) for time-series analysis, and (feedback_type, score) for quality metric queries. Avoid composite indexes with more than 3 columns - they become expensive to maintain at feedback-volume scale.