Skip to main content Home Creators adu2021 skillxiv knowledge-agents-rl-synthesis
knowledge-agents-rl-synthesis Build knowledge agents that generalize across grounded reasoning tasks by combining agentic data synthesis via vector search with off-policy reinforcement learning. Use multi-task training on synthetic question-answer pairs to achieve robust cross-domain performance.
Jump to install Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Show prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/ADu2021/skillXiv --skill knowledge-agents-rl-synthesisThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... More from this repository
Related occupations SOC
Based on SOC occupation classification
name knowledge-agents-rl-synthesis title KARL: Knowledge Agents via Reinforcement Learning version 0.0.2 engine skillxiv-v0.0.2-claude-opus-4.6 license MIT url https://arxiv.org/abs/2603.05218 keywords ["Knowledge Agents","Synthetic Data","Reinforcement Learning","Multi-Task Training","Off-Policy Learning"] description Build knowledge agents that generalize across grounded reasoning tasks by combining agentic data synthesis via vector search with off-policy reinforcement learning. Use multi-task training on synthetic question-answer pairs to achieve robust cross-domain performance.
KARL: Knowledge Agents via Reinforcement Learning
Retrieval-augmented agents struggle with distribution shift: training on one task's synthetic data doesn't transfer well to different retrieval demands. KARL addresses this through two coordinated mechanisms: agentic data synthesis that dynamically explores corpora to generate challenging examples, and multi-task off-policy RL that improves generalization. The key insight is to treat data synthesis as an agent behavior itself, then learn policies that work across diverse reasoning patterns.
The core innovation combines active learning (agents explore corpora) with meta-learning (training across diverse synthetic distributions) to build robust knowledge agents.
Core Concept
KARL implements three coordinated components:
Agentic Data Synthesis : Question-Answer synthesizer explores document corpus via vector search, generates diverse QA pairs grounded in actual retrieval scenariosMulti-Task Training : Train on heterogeneous tasks (cross-document synthesis, constraint-driven search) simultaneously for generalizationOff-Policy RL : Large-batch off-policy learning avoids online instabilities and enables efficient multi-task optimization
Architecture Overview
Input : Document corpus + task specifications (e.g., TREC, BrowseComp)Synthesis Module : Agent explores corpus with vector search, generates challenging QA pairsMulti-Task Buffer : Accumulate synthetic data from multiple task typesRL Trainer : Optimize agent policy via advantage-based off-policy learning (OAPL)Output : Robust knowledge agent generalizing across tasks
Implementation Steps
Step 1: Design agentic data synthesis with vector search
Create a synthesis agent that explores the corpus actively to generate training data.
class SynthesisAgent :
"""
Dynamically synthesizes QA pairs by exploring corpus via retrieval.
"""
def __init__ (self, corpus, encoder_model ):
"""
corpus: list of documents
encoder_model: dense encoder for vector search (e.g., BERT, CLIP)
"""
self .corpus = corpus
self .encoder = encoder_model
self
for
in
self
self
def
_build_faiss_index
self
"""Build FAISS index for efficient similarity search."""
import
self
1
return
def
explore_and_synthesize
self, num_examples=100 , difficulty_schedule=None
"""
Actively explore corpus and generate synthetic QA pairs.
difficulty_schedule: optional function(iteration) -> difficulty_level
"""
for
in
range
if
else
0.5
0
len
self
1
self
self
self
self
5
self
for
in
0
self
'question'
'answer'
'retrieved_docs'
'difficulty'
return
def
_generate_question_from_doc
self, doc, difficulty
"""Generate question of specified difficulty from document."""
f"""
Document: {doc[:200 ]}
Generate a {'simple' if difficulty < 0.3 else 'moderate' if difficulty < 0.7 else 'complex' } question about this document.
Question:
"""
self
50
return
def
_synthesize_answer
self, question, retrieved_docs
"""Synthesize answer from retrieved documents."""
f"""
Question: {question}
Relevant documents:
{' ' .join(retrieved_docs[:3 ])}
Synthesize a concise answer based on the documents.
Answer:
"""
self
100
return
Step 2: Create multi-task training buffer
Accumulate synthetic data from different tasks and domains.
class MultiTaskBuffer :
"""
Replay buffer storing synthetic examples from multiple tasks.
Enables multi-task learning via random task sampling.
"""
def __init__ (self, max_size_per_task=10000 ):
self .tasks = {}
self .max_size_per_task = max_size_per_task
def add_task (self, task_name, task_description ):
"""Register a new task."""
self .tasks[task_name] = {
'description' : task_description,
'buffer' : []
}
def add_examples (self, task_name, examples ):
"""Add synthetic examples to a task."""
if task_name not in self .tasks:
self .add_task(task_name, "Unnamed task" )
buffer = self .tasks[task_name]['buffer' ]
buffer.extend(examples)
if len (buffer) > self .max_size_per_task:
self .tasks[task_name]['buffer' ] = buffer[-self .max_size_per_task:]
def sample_batch (self, batch_size, task_distribution=None ):
"""
Sample batch across tasks.
task_distribution: optional dict mapping task_name -> probability
"""
if task_distribution is None :
task_distribution = {
name: 1.0 / len (self .tasks)
for name in self .tasks
}
batch = []
for _ in range (batch_size):
task_name = np.random.choice(
list (task_distribution.keys()),
p=list (task_distribution.values())
)
task_buffer = self .tasks[task_name]['buffer' ]
example = random.choice(task_buffer)
batch.append((task_name, example))
return batch
def sample_diverse_batch (self, batch_size, diversity_metric='task' ):
"""
Sample diverse batch: ensure coverage across tasks.
diversity_metric: 'task' (equal task coverage) or 'difficulty'
"""
if diversity_metric == 'task' :
num_per_task = batch_size // len (self .tasks)
batch = []
for task_name in self .tasks:
task_buffer = self .tasks[task_name]['buffer' ]
task_samples = random.sample(
task_buffer,
min (num_per_task, len (task_buffer))
)
batch.extend([(task_name, ex) for ex in task_samples])
return batch
elif diversity_metric == 'difficulty' :
all_examples = []
for task_name in self .tasks:
for ex in self .tasks[task_name]['buffer' ]:
all_examples.append((task_name, ex))
all_examples.sort(key=lambda x: x[1 ].get('difficulty' , 0.5 ))
step = len (all_examples) // batch_size
return [all_examples[i * step] for i in range (batch_size)]
Step 3: Implement off-policy RL (OAPL)
Train knowledge agent using large-batch off-policy optimization.
def compute_advantages (rollouts, baseline_model, gamma=0.99 ):
"""
Compute advantage estimates for rollouts.
A_t = R_t - V_baseline(s_t)
"""
advantages = []
for rollout in rollouts:
trajectory = rollout['trajectory' ]
task_success = rollout['success' ]
returns = []
cumulative_return = 0.0
for t in reversed (range (len (trajectory))):
if t == len (trajectory) - 1 :
cumulative_return = float (task_success)
else :
cumulative_return = float (task_success) + gamma * cumulative_return
returns.insert(0 , cumulative_return)
values = []
for step in trajectory:
value = baseline_model(step['state' ]).item()
values.append(value)
step_advantages = [r - v for r, v in zip (returns, values)]
advantages.append(step_advantages)
return advantages
def off_policy_update (agent, baseline, buffer, batch_size=128 ,
learning_rate=1e-4 , num_epochs=3 ):
"""
Off-policy RL update using importance sampling.
Collects large batch, then performs multiple gradient steps.
"""
optimizer = torch.optim.AdamW(agent.parameters(), lr=learning_rate)
batch = buffer.sample_diverse_batch(batch_size, diversity_metric='task' )
rollouts = []
for task_name, example in batch:
trajectory, success = agent.rollout(example)
rollouts.append({
'trajectory' : trajectory,
'success' : success,
'task' : task_name
})
advantages = compute_advantages(rollouts, baseline)
for epoch in range (num_epochs):
total_loss = 0.0
for rollout, step_advantages in zip (rollouts, advantages):
for t, (step, advantage) in enumerate (
zip (rollout['trajectory' ], step_advantages)
):
action = step['action' ]
logprob = agent.compute_logprob(step['state' ], action)
pg_loss = -logprob * advantage
ref_logprob = agent.reference_model.compute_logprob(
step['state' ], action
)
kl_loss = logprob - ref_logprob
loss = pg_loss + 0.01 * kl_loss
total_loss += loss
optimizer.zero_grad()
(total_loss / len (rollouts)).backward()
torch.nn.utils.clip_grad_norm_(agent.parameters(), max_norm=1.0 )
optimizer.step()
return total_loss.item() / len (rollouts)
Step 4: Implement multi-task training loop
Alternate between data synthesis and RL updates.
def train_knowledge_agent (corpus, agent_model, baseline_model,
num_iterations=1000 ):
"""
Train knowledge agent via multi-task learning:
1. Synthesize data from different tasks
2. Update agent via off-policy RL
3. Iterate
"""
synthesizer = SynthesisAgent(corpus)
buffer = MultiTaskBuffer()
buffer.add_task('cross_document' , 'Synthesize answers from multiple documents' )
buffer.add_task('constraint_search' , 'Answer questions with specific constraints' )
for iteration in range (num_iterations):
if iteration % 10 == 0 :
print (f"Iteration {iteration} : Synthesizing data..." )
cross_doc_examples = synthesizer.explore_and_synthesize(
num_examples=50 ,
difficulty_schedule=lambda i: min (i / 50 , 1.0 )
)
buffer.add_examples('cross_document' , cross_doc_examples)
constraint_examples = synthesizer.explore_and_synthesize(
num_examples=30
)
buffer.add_examples('constraint_search' , constraint_examples)
loss = off_policy_update(
agent_model,
baseline_model,
buffer,
batch_size=32 ,
num_epochs=2
)
if (iteration + 1 ) % 100 == 0 :
print (f"Iteration {iteration + 1 } : Loss = {loss:.4 f} " )
return agent_model
Step 5: Evaluation on diverse benchmarks
Test generalization across different reasoning tasks.
def evaluate_generalization (agent, eval_tasks ):
"""
Evaluate: does agent generalize across diverse tasks?
"""
results = {}
for task_name, examples in eval_tasks.items():
correct = 0
total = len (examples)
for example in examples:
trajectory, success = agent.rollout(example)
predicted_answer = trajectory[-1 ]['output' ]
expected_answer = example['answer' ]
if predicted_answer.lower() in expected_answer.lower():
correct += 1
accuracy = correct / total
results[task_name] = accuracy
print (f"{task_name} : {accuracy * 100 :.1 f} %" )
accuracies = list (results.values())
harmonic_mean = len (accuracies) / sum (1.0 / (a + 1e-8 ) for a in accuracies)
print (f"\nOverall generalization score: {harmonic_mean * 100 :.1 f} %" )
return results
Practical Guidance Hyperparameter Selection:
Synthesis exploration budget : 50-200 examples per task. More = better coverage; diminishing returns beyond 100.Task distribution : Start uniform; can weight difficult tasks higher in later training.Off-policy batch size : 32-128. Larger batches = more stable gradients; memory-limited systems use smaller.KL regularization strength : 0.01-0.05. Prevents instability; too high can prevent improvement.Difficulty schedule : Linear 0→1 over synthesis examples; exponential schedules also work well.
Knowledge-grounded QA tasks requiring cross-document reasoning
Multi-domain scenarios where single-task training fails
Settings with access to large document corpus for synthesis
Tasks benefiting from synthetic data augmentation
Single-task, well-defined problems (simpler approaches sufficient)
Real-time systems with low latency budgets (synthesis is expensive)
Domains with limited or unreliable document corpus
Tasks where synthetic data distribution significantly differs from real distribution
Synthesis distribution mismatch : If synthesized QA pairs don't match evaluation distribution, generalization fails. Validate synthesis quality on held-out set.Task imbalance : If one task has much larger buffer, it dominates training. Use explicit task weighting or equal-size buffers.Baseline instability : Poor baseline estimates lead to high-variance advantages. Pre-train baseline on supervised data.Off-policy divergence : If policy drifts too far from data collection policy, importance weights explode. Monitor KL divergence; add explicit bounds.
Reference 