Chapter 10
Atropos RL Framework: From Trajectories to Capabilities
Chapter 10: The Atropos RL Framework โ From Trajectories to Capabilities
The history of reinforcement learning is filled with triumphs on simple games, but applying it to open-ended Agent tasks has remained an elusive challenge. Atropos represents the first systematic engineering answer to this challenge in the open-source community.
10.1 Design Goals and Philosophy
10.1.1 The Name's Metaphor
In Greek mythology, Atropos is one of the three Fates, responsible for cutting the thread of life โ deciding where fate ends and turns. NousResearch chose this name to reflect the framework's core function: judging which Agent behavior trajectories are worth keeping, and which should be discarded.
This is not merely poetic naming โ it mirrors the system's design: Atropos evaluates the quality of Agent trajectories and uses those judgments to improve the model itself.
10.1.2 Three Core Design Goals
Goal 1: Scalable Trajectory Collection
Traditional RL systems rely on closed simulated environments (Atari games, board games). Atropos must collect trajectories in real tool-calling environments โ where each step might invoke real code executors, web search, or file systems with unpredictable results.
Goal 2: Reliable Trajectory Evaluation
For tasks an Agent completes ("help me analyze this report"), there is no single correct answer โ evaluation is inherently subjective. Atropos needs a scalable, consistent judging mechanism that provides reasonable scores without relying heavily on human labor.
Goal 3: Closed-Loop Improvement
The ultimate goal is for the model to continuously improve through its own experience โ much like humans developing a new skill through repeated practice and feedback.
10.1.3 Design Constraints
| Constraint | Requirement | Solution |
|---|---|---|
| Compute cost | Large-scale trajectory collection is expensive | Async parallel collection + priority queues |
| Judgment consistency | Different judges may score the same trajectory differently | Multi-judge voting + judge calibration |
| Distribution shift | Old trajectories lose value as model improves | Online data mixing + experience replay |
| Safety boundaries | Agent actions can cause real harm | Sandbox isolation + action whitelisting |
10.2 Trajectory Collection Mechanism
10.2.1 Trajectory Definition and Structure
In Atropos, a trajectory is a complete task execution sequence:
@dataclass
class ToolCall:
tool_name: str
parameters: Dict[str, Any]
result: str
error: Optional[str] = None
latency_ms: int = 0
@dataclass
class TrajectoryStep:
step_id: int
thought: Optional[str] # <think> content
tool_call: Optional[ToolCall]
observation: str # environment feedback
timestamp: float
@dataclass
class Trajectory:
task_id: str
task_description: str
model_id: str
steps: List[TrajectoryStep] = field(default_factory=list)
final_response: Optional[str] = None
success: Optional[bool] = None
judge_score: Optional[float] = None # 0.0 - 1.0
token_count: int = 0
wall_time_seconds: float = 0.0
10.2.2 Collection Pipeline Architecture
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Atropos Trajectory Collection Pipeline โ
โ โ
โ โโโโโโโโโโโโโโโ โ
โ โ Task Pool โ โ Human-designed + auto-generated โ
โ โโโโโโโโฌโโโโโโโ โ
โ โ task dispatch โ
โ โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ Agent Worker Pool (parallel) โ โ
โ โ โโโโโโโโโโ โโโโโโโโโโ โโโโโโโโโ โ
โ โ โWorker 1โ โWorker 2โ โWorker Nโโ โ
โ โ โโโโโฌโโโโโ โโโโโฌโโโโโ โโโโโฌโโโโ โ
โ โโโโโโโโผโโโโโโโโโโโผโโโโโโโโโโโผโโโโ โ
โ โ raw trajectories โ โ
โ โ โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ Trajectory Collector โ โ
โ โโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโ โ
โ โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ Judge Cluster โ โ
โ โ โโโโโโโโโโ โโโโโโโโโโ โ โ
โ โ โJudge A โ โJudge B โ ... โ โ
โ โ โ(GPT-4) โ โ(Claude)โ โ โ
โ โ โโโโโโโโโโ โโโโโโโโโโ โ โ
โ โโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโ โ
โ โ scored trajectories โ
โ โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ Training Dataset โ โ
โ โ Positives (high score) + Negativesโ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
10.2.3 Agent Execution Phase
class AtroposWorker:
async def execute_trajectory(self, task: str) -> Trajectory:
trajectory = Trajectory(task_id=generate_id(), task_description=task)
context = [{"role": "user", "content": task}]
for step_num in range(self.max_steps):
response = await self.model.generate(context, tools=self.tools.get_schemas())
thought, tool_call = parse_model_response(response)
if tool_call is None:
trajectory.final_response = response
break
try:
result = await self.sandbox.execute(
tool_call.tool_name, tool_call.parameters, timeout=30
)
observation, error = str(result), None
except Exception as e:
observation, error = f"Error: {str(e)}", str(e)
trajectory.steps.append(TrajectoryStep(
step_id=step_num, thought=thought,
tool_call=ToolCall(tool_call.tool_name, tool_call.parameters, observation, error),
observation=observation, timestamp=time.time()
))
context.extend([
{"role": "assistant", "content": response},
{"role": "tool", "content": observation}
])
return trajectory
10.2.4 Multi-Judge Scoring
class TrajectoryJudge:
async def score_trajectory(self, trajectory: Trajectory) -> float:
scores = []
for judge in self.judges:
prompt = self._build_judge_prompt(trajectory)
response = await judge.generate(prompt)
scores.append(self._parse_score(response))
# Trim outliers and average for robustness
if len(scores) > 2:
scores = sorted(scores)[1:-1]
return sum(scores) / len(scores)
def _build_judge_prompt(self, trajectory: Trajectory) -> str:
return f"""You are an expert AI Agent task evaluator. Assess the following agent's performance.
Task: {trajectory.task_description}
Agent Trajectory:
{self._format_trajectory(trajectory)}
Score each dimension (0-10):
1. Task completion: Does the final response fully address the user's need?
2. Reasoning quality: Is the reasoning process logical and efficient?
3. Tool usage: Are tool selections and parameters appropriate?
4. Error handling: Are errors handled correctly when encountered?
5. Efficiency: Are unnecessary steps avoided?
Final composite score (0.0-1.0):"""
10.3 Fundamental Differences from RLHF/DPO
10.3.1 Comparison of Three Alignment Methods
| Dimension | RLHF | DPO | Atropos RL |
|---|---|---|---|
| Data Source | Human preference annotations | Human preference pairs | Agent self-executed trajectories |
| Evaluator | Human annotators | Implicit (preference pairs) | LLM judges |
| Optimization Target | Maximize human preference | Maximize preference log-ratio | Maximize task completion rate |
| Temporal Modeling | Single-step / short sequences | Single-step / short sequences | Complete multi-step trajectories |
| Scalability | Bottlenecked by human labor | Medium (requires annotated pairs) | High (automatic collection) |
| Counterfactual Learning | Limited | Explicit positive/negative examples | Success/failure trajectory contrast |
| Cost | Very high | Medium | Medium (primarily compute) |
10.3.2 The Fundamental Limitation of RLHF
RLHF assumes: humans know what a good response is. For complex Agent tasks, this assumption often fails:
Task: Optimize the performance of this Python program
RLHF Dilemma:
- Response A: Thorough theoretical analysis, but actual speedup unknown
- Response B: Large code changes, but actually 3x faster in practice
RLHF cannot distinguish โ it has no ability to execute tools.
Atropos can: run the code directly, measure actual speedup.
10.3.3 Why Atropos Uses GRPO
Atropos uses GRPO (Group Relative Policy Optimization) to handle temporal credit assignment:
def grpo_update(model, trajectories_batch, gamma=0.99):
"""
GRPO: Compare trajectories within the same task group
rather than scoring them on an absolute scale
"""
task_groups = group_by_task(trajectories_batch)
losses = []
for task_id, group in task_groups.items():
scores = [t.judge_score for t in group]
mean_score = sum(scores) / len(scores)
std_score = statistics.stdev(scores) + 1e-8
for trajectory, score in zip(group, scores):
# Advantage relative to group average
advantage = (score - mean_score) / std_score
# Discounted returns for each step
discounted_returns = compute_discounted_returns(
trajectory.steps, final_reward=advantage, gamma=gamma
)
loss = policy_gradient_loss(model, trajectory.steps, discounted_returns)
losses.append(loss)
return torch.stack(losses).mean()
10.4 Continuous Improvement Loop Engineering
10.4.1 Online + Offline Hybrid Training
class AtroposTrainer:
def training_step(self):
new_trajectories = self.collect_trajectories(n=256)
scored = [self.judge(t) for t in new_trajectories]
# Add high-quality trajectories to replay buffer
for t in scored:
if t.judge_score > 0.7:
self.replay_buffer.add(t, priority=t.judge_score)
# Mix online (70%) and offline replay (30%)
online_batch = scored[:int(256 * 0.7)]
offline_batch = self.replay_buffer.sample(int(256 * 0.3))
loss = grpo_update(self.model, online_batch + offline_batch)
self.optimizer.step(loss)
10.4.2 Task Curriculum Design
class TaskCurriculum:
difficulty_levels = {
1: ["single tool call", "simple file ops", "basic calculation"],
2: ["multi-step retrieval", "code debugging", "data conversion"],
3: ["complex analysis", "multi-tool coordination", "error recovery"],
4: ["cross-system integration", "long-horizon planning", "task clarification"],
5: ["open-ended research", "creative problem solving", "autonomous project planning"]
}
def should_advance(self, recent_scores: List[float]) -> bool:
return sum(recent_scores[-100:]) / 100 > 0.75
10.4.3 Collapse Prevention
class SafetyMonitor:
def check_regression(self, model):
current_scores = run_benchmarks(model, self.baselines)
regressions = {
bench: {"baseline": base, "current": current_scores[bench]}
for bench, base in self.baselines.items()
if current_scores[bench] < base * 0.95
}
if regressions:
self.trainer.adjust_data_ratio(capability_data_ratio=0.5)
return regressions
10.5 Community Interfaces
10.5.1 Open-Source Strategy
| Component | Open-Source Status | Notes |
|---|---|---|
| Trajectory collection API | Open | Community can contribute trajectories |
| Judge interface | Open | Supports custom evaluation criteria |
| Core training loop | Open (simplified) | Full version internal use |
| Pre-collected trajectory dataset | Partially open | ~2M trajectories public |
| Hermes 4 weights | Open | Apache 2.0 |
10.5.2 Custom Task Contribution
from atropos import TrajectoryContributor, TaskSchema
class SecurityAnalysisTask(TaskSchema):
task_type = "security_analysis"
difficulty = 3
required_tools = ["code_exec", "web_search", "file_read"]
def generate_task(self) -> str:
return f"Analyze the following code for security vulnerabilities:\n{self.get_vulnerable_code()}"
def evaluate(self, trajectory: Trajectory) -> float:
response = trajectory.final_response or ""
score = sum(
1.0 / len(self.known_vulnerabilities)
for vuln in self.known_vulnerabilities
if vuln.cve_id in response or vuln.name in response
)
return score
contributor = TrajectoryContributor(api_key="your_api_key")
contributor.register_task(SecurityAnalysisTask)
contributor.submit_trajectories(task=SecurityAnalysisTask(), n_trajectories=100)
10.5.3 Local Atropos Training
pip install atropos-rl
cat > atropos_config.yaml << 'EOF'
model:
base_model: "NousResearch/Hermes-4-405B"
quantization: "q4_k_m"
training:
max_steps: 1000
batch_size: 8
learning_rate: 1e-5
trajectory_collection:
n_workers: 4
max_steps_per_trajectory: 30
tools: ["python_exec", "web_search", "file_ops"]
judge:
model: "gpt-4o"
n_judges: 3
tasks:
task_file: "./my_tasks.json"
EOF
atropos train --config atropos_config.yaml --output ./my_hermes_ft
10.6 Limitations and Future Directions
10.6.1 Current Limitations
Judge Bias: Using GPT-4 as evaluator introduces GPT-4's preferences โ known issues include preference for longer/better-formatted responses.
Sparse Rewards: Binary success/failure signals are sparse. Atropos partially addresses this with process rewards:
def compute_process_rewards(trajectory: Trajectory) -> List[float]:
rewards = []
for step in trajectory.steps:
r = 0.0
if step.tool_call and not step.tool_call.error:
r += 0.05 # small bonus for successful tool call
if step.thought and len(step.thought) > 50:
r += 0.02 # small bonus for substantive reasoning
rewards.append(r)
rewards[-1] += trajectory.judge_score # terminal reward
return rewards
10.6.2 Future Directions
- Self-judging: Hermes model evaluates its own trajectories for fully autonomous improvement
- Multi-Agent trajectories: Training on trajectories where multiple Agents collaborate
- World model assistance: Reduce real-environment interactions via predictive models
- Meta-learning: Teaching the model to "learn new tasks faster"
Chapter Summary
- Atropos's core innovation extends RL alignment from "single-turn dialogue" to "complete Agent trajectories"
- The collection pipeline consists of an Agent worker pool, judge cluster, and training optimizer
- Compared to RLHF/DPO, Atropos handles temporal credit assignment and real tool-calling environments
- GRPO uses within-group relative ranking, avoiding instability of absolute scoring
- Communities can contribute trajectory data via standard APIs or run a local simplified version of Atropos
Discussion Questions
- Atropos uses LLMs as judges, but LLMs have biases. How would you design a more objective evaluation mechanism?
- GRPO updates policy based on relative rankings within a group rather than absolute scores. What problem does this design solve, and what risks does it introduce?
- If designing Atropos tasks for a "medical Q&A" scenario, how would you define a "successful trajectory"? How would you design evaluation criteria?
- How should the mixing ratio between online learning (new trajectories) and experience replay (old trajectories) be dynamically adjusted? Design an adaptive strategy.