Skip to main content ホーム クリエイター adu2021 skillxiv online-experiential-learning-lms
online-experiential-learning-lms 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.
インストールへ移動 Skills Marketplace コミュニティが作成したAIスキルを発見・探索
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/ADu2021/skillXiv --skill online-experiential-learning-lmsコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか?SkillsMP が現在取得できるファイルをダウンロードできます。
Zipをダウンロード ダウンロード中... 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.
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
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
import
@dataclass
class
InteractionTrajectory
"""Single user interaction with deployed model."""
str
str
str
str
bool
float
str
class
TrajectoryLogger
"""Collect and store deployment trajectories."""
def
__init__
self, log_file='deployment_trajectories.jsonl'
self
self
self
'total_interactions'
0
'successful'
0
'failed'
0
'avg_response_length'
0
def
log_interaction
self, trajectory: InteractionTrajectory
"""Record single interaction."""
self
self
'total_interactions'
1
if
self
'successful'
1
else
self
'failed'
1
if
len
self
100
self
def
flush
self
"""Write buffered trajectories to disk."""
with
open
self
'a'
as
for
in
self
'query'
'response'
'feedback'
'success'
'timestamp'
'\n'
self
def
get_statistics
self
Dict
"""Return statistics on collected data."""
if
self
'total_interactions'
0
self
'success_rate'
self
'successful'
self
'total_interactions'
return
self
class
ExperienceAnalyzer
"""Extract patterns from trajectories."""
def
__init__
self
self
def
load_trajectories
self, log_file: str , num_recent=1000
"""Load trajectories from log file."""
with
open
'r'
as
for
in
self
def
analyze_success_patterns
self
List
str
"""Identify common patterns in successful interactions."""
for
in
self
if
'success'
if
not
return
for
in
'response'
for
in
self
3
0
1
sorted
lambda
1
True
return
for
in
10
def
analyze_failure_modes
self
List
Tuple
str
str
"""Identify common failure patterns."""
for
in
self
if
not
'success'
for
in
'query'
'feedback'
return
10
def
_extract_ngrams
self, words: List [str ], n: int
List
str
"""Extract n-grams from word sequence."""
for
in
range
len
1
' '
return
def
synthesize_insights
self
str
"""Generate text summarizing extracted knowledge."""
self
self
"Extracted deployment insights:\n\n"
"Success patterns:\n"
for
in
5
f" - {phrase} \n"
"\nCommon failure modes:\n"
for
in
5
f" - Query: {query[:50 ]} \n"
f" Feedback: {feedback[:50 ]} \n"
return
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' ]
if success_flag:
loss = self ._compute_response_loss(query, successful_response)
else :
loss = self ._compute_correction_loss(query,
trajectory['feedback' ])
self .optimizer.zero_grad()
loss.backward()
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:.4 f} " )
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."""
student_logits = self .student.forward(query, return_logits=True )
target_tokens = self .student.tokenizer.encode(successful_response)
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."""
student_response = self .student.generate(query, max_length=100 )
similarity = self ._compute_similarity(student_response, failure_feedback)
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."""
insight_embedding = self .student.encode_text(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."""
response = self .model.generate(query, max_length=100 )
success = len (user_feedback) > 0 and 'good' not in user_feedback.lower()
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."""
self .analyzer.load_trajectories(self .logger.log_file,
num_recent=num_interactions)
insights = self .analyzer.synthesize_insights()
print (insights)
trajectories = [
{
'query' : t['query' ],
'response' : t['response' ],
'feedback' : t['feedback' ],
'success' : t['success' ]
}
for t in self .analyzer.trajectories
]
print ("Distilling knowledge from experiences..." )
self .distiller.distill_from_experiences(trajectories, insights)
print ("Validating generalization..." )
self ._validate_generalization()
def _validate_generalization (self ):
"""Test that improvements generalize."""
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."""
return []
def _evaluate_response (self, query: str , response: str ) -> bool :
"""Simple evaluation (real systems use learning-to-rank)."""
return len (response) > 20
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
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
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
Reference