菜单
Improve deployed language models by learning from real-world user interactions. Extract transferable knowledge from interaction trajectories and consolidate via on-policy context distillation without needing environment access.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/ADu2021/skillXiv --skill online-experiential-learning-lms命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | online-experiential-learning-lms |
| title | Online Experiential Learning for Language Models |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2603.16856 |
| keywords | ["Online Learning","Deployment Experience","Knowledge Distillation","Continuous Improvement","On-Policy Learning"] |
| description | Improve deployed language models by learning from real-world user interactions. Extract transferable knowledge from interaction trajectories and consolidate via on-policy context distillation without needing environment access. |
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.
Online Experiential Learning operates through an iterative cycle:
This creates a positive feedback loop where each generation of deployed model generates better experiences for the next iteration.
Start by setting up trajectory collection and analysis infrastructure.
from dataclasses import dataclass
from typing import List, Dict, Tuple
json
:
user_id:
query:
model_response:
user_feedback:
success:
timestamp:
model_version:
:
():
.log_file = log_file
.buffer = []
.stats = {
: ,
: ,
: ,
:
}
():
.buffer.append(trajectory)
.stats[] +=
trajectory.success:
.stats[] +=
:
.stats[] +=
(.buffer) >= :
.flush()
():
(.log_file, ) f:
traj .buffer:
json.dump({
: traj.query,
: traj.model_response,
: traj.user_feedback,
: traj.success,
: traj.timestamp
}, f)
f.write()
.buffer = []
() -> :
.stats[] > :
.stats[] = (
.stats[] / .stats[]
)
.stats
:
():
.trajectories = []
():
trajectories = []
(log_file, ) f:
line f:
traj_dict = json.loads(line)
trajectories.append(traj_dict)
.trajectories = trajectories[-num_recent:]
() -> []:
successful = [t t .trajectories t[]]
successful:
[]
key_phrases = {}
traj successful:
response = traj[].lower()
words = response.split()
phrase ._extract_ngrams(words, n=):
key_phrases[phrase] = key_phrases.get(phrase, ) +
sorted_phrases = (key_phrases.items(),
key= x: x[], reverse=)
[phrase phrase, count sorted_phrases[:]]
() -> [[, ]]:
failed = [t t .trajectories t[]]
failure_modes = []
traj failed:
query = traj[]
feedback = traj[]
failure_modes.append((query, feedback))
failure_modes[:]
() -> []:
ngrams = []
i ((words) - n + ):
ngram = .join(words[i:i+n])
ngrams.append(ngram)
ngrams
() -> :
success_patterns = .analyze_success_patterns()
failure_modes = .analyze_failure_modes()
insights =
insights +=
phrase success_patterns[:]:
insights +=
insights +=
query, feedback failure_modes[:]:
insights +=
insights +=
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
Hyperparameters and When to Use:
When NOT to use:
Common Pitfalls: