Agent Learning from Feedback

Why This Matters

Feedback loops are how agents evolve from brittle rule-followers into adaptive, intelligent systems. Without learning from feedback, agents repeat mistakes, fail to improve, and cannot align with human preferences. RLHF (Reinforcement Learning from Human Feedback) is the technology that made ChatGPT conversational—turning a raw language model into a helpful assistant.

Real-World Analogy: Think of training a dog. You don't write code for every behavior—instead, you reward good actions and correct bad ones. Over time, the dog learns what you want. RLHF does the same for AI: humans provide feedback (rewards), and the agent learns to maximize those rewards while following constraints.

RLHF Architecture Overview

Reward Model Implementation

import torch
import torch.nn as nn
import torch.nn.functional as F
from dataclasses import dataclass
from typing import List, Tuple, Optional
import numpy as np
import logging

logger = logging.getLogger(__name__)


@dataclass
class PreferencePair:
    prompt: str
    chosen: str
    rejected: str


class RewardModel(nn.Module):
    def __init__(self, model_name: str = "gpt2", hidden_size: int = 768):
        super().__init__()
        from transformers import AutoModel
        self.backbone = AutoModel.from_pretrained(model_name)
        self.reward_head = nn.Linear(hidden_size, 1)
        self.dropout = nn.Dropout(0.1)

    def forward(self, input_ids, attention_mask=None):
        outputs = self.backbone(input_ids=input_ids, attention_mask=attention_mask)
        hidden = outputs.last_hidden_state[:, -1, :]
        hidden = self.dropout(hidden)
        reward = self.reward_head(hidden).squeeze(-1)
        return reward

    def compute_reward(self, input_ids, attention_mask=None):
        with torch.no_grad():
            return self.forward(input_ids, attention_mask)


class RewardModelTrainer:
    def __init__(self, model: RewardModel, learning_rate: float = 1e-5):
        self.model = model
        self.optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
        self.loss_history: List[float] = []

    def compute_loss(self, chosen_rewards, rejected_rewards):
        loss = -F.logsigmoid(chosen_rewards - rejected_rewards).mean()
        return loss

    def train_step(self, batch: dict) -> float:
        self.model.train()
        self.optimizer.zero_grad()

        chosen_rewards = self.model(
            batch["chosen_input_ids"],
            batch["chosen_attention_mask"],
        )
        rejected_rewards = self.model(
            batch["rejected_input_ids"],
            batch["rejected_attention_mask"],
        )

        loss = self.compute_loss(chosen_rewards, rejected_rewards)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
        self.optimizer.step()

        self.loss_history.append(loss.item())
        return loss.item()

    def compute_accuracy(self, batch: dict) -> float:
        self.model.eval()
        with torch.no_grad():
            chosen_rewards = self.model(
                batch["chosen_input_ids"],
                batch["chosen_attention_mask"],
            )
            rejected_rewards = self.model(
                batch["rejected_input_ids"],
                batch["rejected_attention_mask"],
            )
            correct = (chosen_rewards > rejected_rewards).float().mean()
            return correct.item()


class RewardModelPipeline:
    def __init__(self, model_name: str = "gpt2"):
        self.model = RewardModel(model_name)
        self.trainer = RewardModelTrainer(self.model)
        self.preference_data: List[PreferencePair] = []

    def add_preference(self, prompt: str, chosen: str, rejected: str):
        self.preference_data.append(PreferencePair(prompt, chosen, rejected))

    def train(self, epochs: int = 3, batch_size: int = 8):
        for epoch in range(epochs):
            total_loss = 0.0
            num_batches = 0
            
            for i in range(0, len(self.preference_data), batch_size):
                batch_data = self.preference_data[i:i + batch_size]
                if len(batch_data) < 2:
                    continue
                
                batch = self._collate_batch(batch_data)
                loss = self.trainer.train_step(batch)
                total_loss += loss
                num_batches += 1

            avg_loss = total_loss / max(num_batches, 1)
            logger.info(f"Epoch {epoch + 1}/{epochs}, Loss: {avg_loss:.4f}")

    def _collate_batch(self, batch: List[PreferencePair]) -> dict:
        from transformers import AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained("gpt2")
        tokenizer.pad_token = tokenizer.eos_token

        chosen_texts = [p.chosen for p in batch]
        rejected_texts = [p.rejected for p in batch]

        chosen_encodings = tokenizer(
            chosen_texts, truncation=True, padding=True, max_length=512, return_tensors="pt"
        )
        rejected_encodings = tokenizer(
            rejected_texts, truncation=True, padding=True, max_length=512, return_tensors="pt"
        )

        return {
            "chosen_input_ids": chosen_encodings["input_ids"],
            "chosen_attention_mask": chosen_encodings["attention_mask"],
            "rejected_input_ids": rejected_encodings["input_ids"],
            "rejected_attention_mask": rejected_encodings["attention_mask"],
        }

DPO (Direct Preference Optimization) Implementation

import torch
import torch.nn.functional as F
from transformers import AutoModelForCausalLM, AutoTokenizer
import logging

logger = logging.getLogger(__name__)


class DPOTrainer:
    def __init__(
        self,
        model_name: str = "gpt2",
        beta: float = 0.1,
        learning_rate: float = 1e-5,
    ):
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.ref_model = AutoModelForCausalLM.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.beta = beta
        self.optimizer = torch.optim.AdamW(self.model.parameters(), lr=learning_rate)

    def compute_logprobs(self, model, input_ids, attention_mask):
        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
        logits = outputs.logits[:, :-1, :]
        labels = input_ids[:, 1:]
        log_probs = F.log_softmax(logits, dim=-1)
        token_log_probs = torch.gather(log_probs, 2, labels.unsqueeze(-1)).squeeze(-1)
        mask = attention_mask[:, 1:].float()
        sequence_log_probs = (token_log_probs * mask).sum(dim=-1)
        return sequence_log_probs

    def dpo_loss(self, chosen_logps, rejected_logps, ref_chosen_logps, ref_rejected_logps):
        chosen_logratios = chosen_logps - ref_chosen_logps
        rejected_logratios = rejected_logps - ref_rejected_logps
        logits = self.beta * (chosen_logratios - rejected_logratios)
        loss = -F.logsigmoid(logits).mean()
        return loss

    def train_step(self, batch: dict) -> float:
        self.model.train()
        self.optimizer.zero_grad()

        chosen_ids = batch["chosen_input_ids"]
        chosen_mask = batch["chosen_attention_mask"]
        rejected_ids = batch["rejected_input_ids"]
        rejected_mask = batch["rejected_attention_mask"]

        chosen_logps = self.compute_logprobs(self.model, chosen_ids, chosen_mask)
        rejected_logps = self.compute_logprobs(self.model, rejected_ids, rejected_mask)

        with torch.no_grad():
            ref_chosen_logps = self.compute_logprobs(
                self.ref_model, chosen_ids, chosen_mask
            )
            ref_rejected_logps = self.compute_logprobs(
                self.ref_model, rejected_ids, rejected_mask
            )

        loss = self.dpo_loss(chosen_logps, rejected_logps, ref_chosen_logps, ref_rejected_logps)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
        self.optimizer.step()

        return loss.item()

Human Feedback Collector

import asyncio
import uuid
import time
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class FeedbackType(Enum):
    BINARY = "binary"
    RANKING = "ranking"
    RATING = "rating"
    COMPARISON = "comparison"


@dataclass
class FeedbackRequest:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    prompt: str = ""
    responses: list[str] = field(default_factory=list)
    feedback_type: FeedbackType = FeedbackType.BINARY
    metadata: dict = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)


@dataclass
class FeedbackResponse:
    request_id: str
    feedback_type: FeedbackType
    scores: dict[str, float] = field(default_factory=dict)
    ranking: list[str] = field(default_factory=list)
    binary_label: Optional[bool] = None
    annotator_id: str = ""
    timestamp: float = field(default_factory=time.time)
    notes: str = ""


class HumanFeedbackCollector:
    def __init__(self):
        self.pending_requests: dict[str, FeedbackRequest] = {}
        self.completed_feedback: list[FeedbackResponse] = []
        self.feedback_handlers: dict[FeedbackType, Callable] = {}

    def create_feedback_request(
        self,
        prompt: str,
        responses: list[str],
        feedback_type: FeedbackType = FeedbackType.BINARY,
        metadata: dict = None,
    ) -> FeedbackRequest:
        request = FeedbackRequest(
            prompt=prompt,
            responses=responses,
            feedback_type=feedback_type,
            metadata=metadata or {},
        )
        self.pending_requests[request.id] = request
        return request

    async def collect_feedback(
        self,
        request: FeedbackRequest,
        timeout: float = 300.0,
    ) -> Optional[FeedbackResponse]:
        handler = self.feedback_handlers.get(request.feedback_type)
        if not handler:
            raise ValueError(f"No handler for feedback type: {request.feedback_type}")

        try:
            response = await asyncio.wait_for(
                handler(request),
                timeout=timeout,
            )
            if response:
                self.completed_feedback.append(response)
                del self.pending_requests[request.id]
                return response
        except asyncio.TimeoutError:
            logger.warning(f"Feedback timeout for request {request.id}")
        return None

    def register_handler(
        self,
        feedback_type: FeedbackType,
        handler: Callable[[FeedbackRequest], Coroutine],
    ):
        self.feedback_handlers[feedback_type] = handler

    def get_feedback_stats(self) -> dict:
        return {
            "total_collected": len(self.completed_feedback),
            "pending": len(self.pending_requests),
            "by_type": {
                ft.value: sum(
                    1 for f in self.completed_feedback
                    if f.feedback_type == ft
                )
                for ft in FeedbackType
            },
        }

    def export_training_data(self) -> list[dict]:
        return [
            {
                "prompt": self.pending_requests.get(f.request_id, FeedbackRequest()).prompt,
                "feedback": {
                    "type": f.feedback_type.value,
                    "scores": f.scores,
                    "ranking": f.ranking,
                    "binary": f.binary_label,
                },
            }
            for f in self.completed_feedback
        ]

Performance Considerations

Method	Training Cost	Data Needs	Quality	Scalability
RLHF (PPO)	Very High	Medium	High	Low
DPO	Medium	Medium	High	Medium
RLAIF	Low	Low	Medium	Very High
Explicit Ranking	N/A	High	Very High	Low
Implicit Signals	N/A	High	Medium	Very High

Security Considerations

Feedback Integrity: Validate and sanitize all human feedback to prevent adversarial attacks
Model Safety: Implement guardrails to prevent reward hacking and unintended behaviors
Data Privacy: Protect human annotator data and ensure compliance with privacy regulations
Audit Trails: Log all feedback and training decisions for accountability and debugging

Mathematical Foundation

Bradley-Terry Model (Reward Model):

DPO Loss Function:

KL Penalty (prevents deviation from reference policy):

Interview Questions

1. What is the difference between RLHF and DPO?

Answer: RLHF trains a separate reward model then optimizes the policy using PPO, requiring two stages and significant compute. DPO directly optimizes the policy on preference data using a contrastive loss. DPO is simpler to implement, requires less compute, and often achieves comparable results. RLHF may be preferred when you need an explicit reward model for other purposes or when DPO's implicit reward assumption doesn't hold.

2. How do you collect high-quality human feedback at scale?

Answer: Strategies: 1) Use clear annotation guidelines with examples, 2) Implement inter-annotator agreement metrics (Cohen's kappa), 3) Pay annotators fairly and provide training, 4) Use consensus from multiple annotators per example, 5) Detect and remove spammers with attention checks, 6) Balance expert and crowd workers, 7) Provide context and let annotators flag uncertainty, 8) Use active learning to prioritize informative examples.

3. What is reward hacking and how do you prevent it?

Answer: Reward hacking occurs when the policy learns to exploit flaws in the reward model to achieve high scores without actually improving quality. Examples: generating overly verbose responses that score high on length-based rewards. Prevention: 1) KL penalty to prevent large policy deviations, 2) Ensemble of reward models, 3) Regular reward model retraining, 4) Human evaluation of high-reward samples, 5) Constrained optimization with safety bounds.

4. How does the Bradley-Terry model relate to reward modeling?

Answer: The Bradley-Terry model assumes pairwise preferences follow a logistic function of reward differences: P(A > B) = σ(r(A) - r(B)). This provides a principled probabilistic framework for learning reward functions from comparisons. The model is trained by maximizing log-likelihood of observed preferences. Key properties: transitivity, consistency, and convergence with sufficient data.

5. What are the challenges of RLHF in production?

Answer: Key challenges: 1) Reward model drift — Distribution shift between training and deployment, 2) Feedback quality — Noisy or inconsistent human labels, 3) Computational cost — PPO training is expensive, 4) Catastrophic forgetting — Policy loses general capabilities, 5) Reward hacking — Policy exploits reward model flaws, 6) Scalability — Human feedback is expensive to collect.

6. How would you implement online learning from user feedback?

Answer: Use a feedback loop: 1) Collect implicit signals (thumbs up/down, edit requests), 2) Buffer feedback with deduplication, 3) Periodically retrain reward model, 4) Fine-tune policy in batches (not continuously), 5) Use replay buffers to prevent catastrophic forgetting, 6) A/B test against baseline, 7) Implement rollback if metrics degrade.

7. What is the relationship between RLHF and constitutional AI?

Answer: Constitutional AI (CAI) uses AI feedback (RLAIF) instead of human feedback. An AI system evaluates its own outputs against a set of principles (constitution), generating preference data for training. Benefits: scalable, consistent, and less expensive than human feedback. In practice, combine both: use CAI for initial alignment, then RLHF for fine-tuning to human preferences.

8. How do you evaluate if RLHF is actually improving the model?

Answer: Multi-faceted evaluation: 1) Automated metrics — Reward model scores, win rate against baseline, 2) Human evaluation — Side-by-side comparisons, 3) Task performance — Maintain benchmark scores, 4) Safety metrics — Toxicity, bias, refusal rates, 5) A/B testing — Online user preference signals, 6) Reward-hacking detection — Monitor for abnormal reward distributions.

Common Pitfalls

Pitfall	Solution
Reward hacking/exploitation	KL penalty, ensemble reward models, human audits
Feedback quality degradation	Annotator training, inter-rater reliability checks
Catastrophic forgetting	Replay buffers, conservative fine-tuning, EWC
High computational cost	DPO instead of PPO, LoRA, gradient accumulation
Feedback collection bottleneck	AI feedback (RLAIF), implicit signals, active learning
Distribution shift	Regular model updates, domain adaptation
Policy collapse	Diversity regularization, entropy bonuses
Reward model overfitting	Regularization, cross-validation, held-out evaluation

Summary with Key Takeaways

RLHF trains a reward model on human preferences then optimizes the policy with PPO
DPO directly optimizes on preference data without a separate reward model; simpler and often sufficient
Bradley-Terry model provides the probabilistic foundation for learning from pairwise comparisons
KL penalty prevents the policy from deviating too far from the reference model
Reward hacking is a critical failure mode requiring ensemble methods and human oversight
Online feedback loops enable continuous improvement but require careful architecture
Constitutional AI uses AI feedback for scalable alignment, complemented by human feedback
Evaluation requires multiple metrics: automated scores, human judgment, and safety monitoring

KnowledgeCheck

What does RLHF stand for?
- a) Reinforcement Learning from Human Feedback
- b) Recurrent Learning from Human Features
- c) Random Learning from High Frequencies
- d) Reinforced Learning from Hybrid Functions
What is the Bradley-Terry model used for in RLHF?
- a) Generating text responses
- b) Training the reward function from preferences
- c) Tokenizing input text
- d) Optimizing the policy with PPO
What is reward hacking?
- a) Improving reward model accuracy
- b) Policy exploiting flaws in the reward model
- c) Manually adjusting reward scores
- d) Using multiple reward models
How does DPO differ from RLHF?
- a) DPO uses a separate reward model
- b) DPO directly optimizes the policy on preference data
- c) DPO requires more compute than RLHF
- d) DPO doesn't need preference data
What prevents the policy from deviating too far from the reference model?
- a) Learning rate scheduling
- b) KL divergence penalty
- c) Gradient clipping
- d) Batch normalization
What is RLAIF?
- a) Reinforcement Learning with AI Feedback
- b) Random Learning from AI Features
- c) Reinforcement Learning for Image Fusion
- d) Recurrent Learning from AI Functions

Answers: 1-a, 2-b, 3-b, 4-b, 5-b, 6-a

Agent Learning from Feedback: RLHF, Reward Modeling & Human Feedback