adu2021 Online Experiential Learning for Language Models
Deployed language models generate valuable experiential data through real-world user interactions, yet this rich signal remains unexploited in typical offline training paradigms. Online Experiential Learning enables continuous model improvement by: (1) gathering interaction trajectories during actual deployment, (2) extracting transferable knowledge from these experiences, and (3) consolidating knowledge into model parameters via on-policy context distillation. The approach maintains on-policy consistency without requiring access to the user-side environment, enabling true deployment-driven improvement loops.
The key insight: experiential knowledge (patterns extracted from trajectories) is significantly more effective than raw trajectories, enabling efficient parameter updates from deployment data.
Core Concept
Online Experiential Learning operates through an iterative cycle:
- Experience Collection — Gather interaction trajectories from deployed model
- Knowledge Extraction — Identify generalizable patterns and insights from trajectories
- On-Policy Distillation — Consolidate extracted knowledge into improved model
- Iterative Deployment — Improved model goes into production, generates new experiences
This creates a positive feedback loop where each generation of deployed model generates better experiences for the next iteration.
Architecture Overview
- Deployment Trajectory Logger — Record user interactions and model responses
- Experience Analyzer — Identify successful patterns and failure modes
- Knowledge Extractor — Synthesize generalizable insights from experiences
- On-Policy Distiller — Distill knowledge into student model with consistency preservation
- Experiential Knowledge Encoder — Represent extracted knowledge for model consumption
- Deployment Manager — Orchestrate model updates and rollout cycles
- Generalization Validator — Test out-of-distribution performance
Implementation Steps
Start by setting up trajectory collection and analysis infrastructure.
from dataclasses import dataclass
from typing import List, Dict, Tuple
import json
@dataclass
class InteractionTrajectory:
"""Single user interaction with deployed model."""
user_id: str
query: str
model_response: str
user_feedback: str # Reward or quality signal
success: bool
timestamp: float
model_version: str
class TrajectoryLogger:
"""Collect and store deployment trajectories."""
def __init__(self, log_file='deployment_trajectories.jsonl'):
self.log_file = log_file
self.buffer = []
self.stats = {
'total_interactions': 0,
'successful': 0,
'failed': 0,
'avg_response_length': 0
}
def log_interaction(self, trajectory: InteractionTrajectory):
"""Record single interaction."""
self.buffer.append(trajectory)
self.stats['total_interactions'] += 1
if trajectory.success:
self.stats['successful'] += 1
else:
self.stats['failed'] += 1
# Periodically flush to disk
if len(self.buffer) >= 100:
self.flush()
def flush(self):
"""Write buffered trajectories to disk."""
with open(self.log_file, 'a') as f:
for traj in self.buffer:
json.dump({
'query': traj.query,
'response': traj.model_response,
'feedback': traj.user_feedback,
'success': traj.success,
'timestamp': traj.timestamp
}, f)
f.write('\n')
self.buffer = []
def get_statistics(self) -> Dict:
"""Return statistics on collected data."""
if self.stats['total_interactions'] > 0:
self.stats['success_rate'] = (
self.stats['successful'] / self.stats['total_interactions']
)
return self.stats
class ExperienceAnalyzer:
"""Extract patterns from trajectories."""
def __init__(self):
self.trajectories = []
def load_trajectories(self, log_file: str, num_recent=1000):
"""Load trajectories from log file."""
trajectories = []
with open(log_file, 'r') as f:
for line in f:
traj_dict = json.loads(line)
trajectories.append(traj_dict)
self.trajectories = trajectories[-num_recent:]
def analyze_success_patterns(self) -> List[str]:
"""Identify common patterns in successful interactions."""
successful = [t for t in self.trajectories if t['success']]
if not successful:
return []
# Extract key phrases from successful responses
key_phrases = {}
for traj in successful:
response = traj['response'].lower()
words = response.split()
for phrase in self._extract_ngrams(words, n=3):
key_phrases[phrase] = key_phrases.get(phrase, 0) + 1
# Sort by frequency
sorted_phrases = sorted(key_phrases.items(),
key=lambda x: x[1], reverse=True)
return [phrase for phrase, count in sorted_phrases[:10]]
def analyze_failure_modes(self) -> List[Tuple[str, str]]:
"""Identify common failure patterns."""
failed = [t for t in self.trajectories if not t['success']]
failure_modes = []
for traj in failed:
query = traj['query']
feedback = traj['feedback']
failure_modes.append((query, feedback))
return failure_modes[:10]
def _extract_ngrams(self, words: List[str], n: int) -> List[str]:
"""Extract n-grams from word sequence."""
ngrams = []
for i in range(len(words) - n + 1):
ngram = ' '.join(words[i:i+n])
ngrams.append(ngram)
return ngrams
def synthesize_insights(self) -> str:
"""Generate text summarizing extracted knowledge."""
success_patterns = self.analyze_success_patterns()
failure_modes = self.analyze_failure_modes()
insights = "Extracted deployment insights:\n\n"
insights += "Success patterns:\n"
for phrase in success_patterns[:5]:
insights += f" - {phrase}\n"
insights += "\nCommon failure modes:\n"
for query, feedback in failure_modes[:5]:
insights += f" - Query: {query[:50]}\n"
insights += f" Feedback: {feedback[:50]}\n"
return insights
Now implement the on-policy distillation that consolidates extracted knowledge.
import torch
import torch.nn as nn
from torch.optim import AdamW
class ExperientialKnowledgeDistiller:
"""Distill deployed experiences into improved model."""
def __init__(self, student_model, teacher_model=None):
self.student = student_model
self.teacher = teacher_model
self.optimizer = AdamW(student_model.parameters(), lr=5e-6)
def distill_from_experiences(self, trajectories: List[Dict],
extracted_insights: str,
num_epochs=3):
"""Update student model using extracted knowledge."""
for epoch in range(num_epochs):
total_loss = 0
for i, trajectory in enumerate(trajectories):
query = trajectory['query']
successful_response = trajectory['response']
success_flag = trajectory['success']
# Compute loss for replicating successful behavior
if success_flag:
# On-policy: only update on successful trajectories
loss = self._compute_response_loss(query, successful_response)
else:
# Off-policy correction: learn to avoid failures
loss = self._compute_correction_loss(query,
trajectory['feedback'])
self.optimizer.zero_grad()
loss.backward()
# Gradient clipping for stability
torch.nn.utils.clip_grad_norm_(self.student.parameters(), 1.0)
self.optimizer.step()
total_loss += loss.item()
if (i + 1) % 100 == 0:
avg_loss = total_loss / (i + 1)
print(f"Batch {i+1}: Distillation Loss = {avg_loss:.4f}")
# Incorporate extracted insights as auxiliary objective
insight_loss = self._compute_insight_alignment(extracted_insights)
self.optimizer.zero_grad()
insight_loss.backward()
self.optimizer.step()
def _compute_response_loss(self, query: str,
successful_response: str) -> torch.Tensor:
"""Loss for replicating successful responses."""
# Generate response from student
student_logits = self.student.forward(query, return_logits=True)
# Tokenize target response
target_tokens = self.student.tokenizer.encode(successful_response)
# Cross-entropy loss
loss = nn.functional.cross_entropy(
student_logits.view(-1, student_logits.size(-1)),
torch.tensor(target_tokens).view(-1)
)
return loss
def _compute_correction_loss(self, query: str, failure_feedback: str):
"""Loss for learning from failures."""
# Generate response from student
student_response = self.student.generate(query, max_length=100)
# Penalty for responses similar to failure case
similarity = self._compute_similarity(student_response, failure_feedback)
# Minimize similarity to failed patterns
loss = torch.tensor(similarity, dtype=torch.float32)
return loss
def _compute_insight_alignment(self, insights: str) -> torch.Tensor:
"""Loss for aligning model behavior with extracted insights."""
# Encode insights
insight_embedding = self.student.encode_text(insights)
# This is a placeholder; in practice you would verify that
# model outputs align with encoded insights
alignment_loss = torch.tensor(0.0)
return alignment_loss
def _compute_similarity(self, response1: str, response2: str) -> float:
"""Simple text similarity (BLEU-like)."""
tokens1 = set(response1.lower().split())
tokens2 = set(response2.lower().split())
if not tokens1 or not tokens2:
return 0.0
intersection = len(tokens1 & tokens2)
union = len(tokens1 | tokens2)
return intersection / union
class OnlineExperientialLearner:
"""Full pipeline for online learning from deployment."""
def __init__(self, model, log_file='trajectories.jsonl'):
self.model = model
self.logger = TrajectoryLogger(log_file)
self.analyzer = ExperienceAnalyzer()
self.distiller = ExperientialKnowledgeDistiller(model)
def deployment_step(self, query: str, user_feedback: str) -> str:
"""Single interaction during deployment."""
# Generate response
response = self.model.generate(query, max_length=100)
# Evaluate success (simplified; real systems use more sophisticated metrics)
success = len(user_feedback) > 0 and 'good' not in user_feedback.lower()
# Log trajectory
trajectory = InteractionTrajectory(
user_id='anon',
query=query,
model_response=response,
user_feedback=user_feedback,
success=success,
timestamp=time.time(),
model_version='v0'
)
self.logger.log_interaction(trajectory)
return response
def update_cycle(self, num_interactions=1000):
"""Periodic update from collected experiences."""
# Load trajectories
self.analyzer.load_trajectories(self.logger.log_file,
num_recent=num_interactions)
# Extract insights
insights = self.analyzer.synthesize_insights()
print(insights)
# Get trajectories for distillation
trajectories = [
{
'query': t['query'],
'response': t['response'],
'feedback': t['feedback'],
'success': t['success']
}
for t in self.analyzer.trajectories
]
# Distill knowledge
print("Distilling knowledge from experiences...")
self.distiller.distill_from_experiences(trajectories, insights)
# Test on OOD validation set
print("Validating generalization...")
self._validate_generalization()
def _validate_generalization(self):
"""Test that improvements generalize."""
# Load validation set (not seen during deployment)
validation_queries = self._load_validation_set()
successes = 0
for query in validation_queries[:100]:
response = self.model.generate(query)
success = self._evaluate_response(query, response)
if success:
successes += 1
success_rate = successes / min(100, len(validation_queries))
print(f"Validation success rate: {success_rate:.1%}")
def _load_validation_set(self):
"""Load held-out validation queries."""
# Placeholder
return []
def _evaluate_response(self, query: str, response: str) -> bool:
"""Simple evaluation (real systems use learning-to-rank)."""
return len(response) > 20 # Placeholder
Practical Guidance
Hyperparameters and When to Use:
- Update cycle frequency: every 1000-10000 interactions (every 1-7 days depending on volume)
- On-policy ratio: weight successful trajectories 3-5x higher than corrected failures
- Insight extraction threshold: use only patterns appearing in >5% of successful interactions
- Apply when you have continuous deployment generating high-volume interaction data
- Particularly effective for conversational models and interactive systems
When NOT to use:
- For offline-only systems without user interaction data
- When deployment scenarios are drastically different from training (distribution shift)
- For safety-critical applications without explicit human oversight of updates
- When data collection violates privacy constraints
Common Pitfalls:
- Distribution shift: deployment data differs from training; use domain adaptation techniques
- Feedback bias: user signals may not reflect true quality; validate with held-out eval sets
- Catastrophic forgetting: focus on recent experiences too much; maintain replay buffer
- Positive feedback loops: successful model generates easier data, suppressing diversity; periodically reset to baseline
- Privacy concerns: ensure user data is properly anonymized before extraction and distillation