| name | aworld-distributed-agent-training |
| title | AWorld Distributed Training Recipe for Agentic AI |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2508.20404 |
| keywords | ["distributed-training","agent-learning","reinforcement-learning","scalability","experience-collection"] |
| description | Accelerate agentic AI training by distributing task execution across clusters, achieving 14.6x speedup in experience collection and enabling practical large-scale agent development |
AWorld: Orchestrating Training Recipe for Agentic AI
Core Concept
AWorld addresses the bottleneck in reinforcement learning for agents: the slow collection of environmental interactions. By distributing task execution across multiple cluster nodes, rather than sequential single-machine execution, the system dramatically accelerates experience generation. This enables practical training of capable agents on complex benchmarks like GAIA, where computational efficiency is critical.
Architecture Overview
- Distributed Executor: Parallel task execution across cluster nodes
- Centralized Experience Aggregation: Collects trajectories from all workers
- Efficient Communication: Minimal overhead for distributed coordination
- Scalable RL Pipeline: From experience collection through model training
- Production-Ready Implementation: Open-source reference implementation
Implementation Steps
Stage 1: Design Distributed Task Execution Framework
Create infrastructure for parallel task distribution and result collection.
import asyncio
from typing import Dict, List, Any
import pickle
import queue
from concurrent.futures import ProcessPoolExecutor
class DistributedExecutor:
"""Execute tasks across multiple nodes in parallel"""
def __init__(self, num_workers: int, max_queue_size: int = 1000):
self.num_workers = num_workers
self.executor = ProcessPoolExecutor(max_workers=num_workers)
self.task_queue = queue.Queue(maxsize=max_queue_size)
self.result_queue = queue.Queue()
self.running_tasks = {}
def submit_task(self, task_id: str, task_fn, *args, **kwargs):
"""Submit a task for distributed execution"""
future = self.executor.submit(task_fn, *args, **kwargs)
self.running_tasks[task_id] = future
return task_id
def collect_results(self, timeout: float = None) -> Dict:
"""Collect completed task results"""
completed = {}
for task_id, future in list(self.running_tasks.items()):
future.done():
:
result = future.result(timeout=timeout)
completed[task_id] = {: , : result}
Exception e:
completed[task_id] = {: , : (e)}
.running_tasks[task_id]
completed
():
.executor.shutdown(wait=)
:
():
.nodes = nodes
.node_load = {node: node nodes}
() -> :
node = (.nodes, key= n: .node_load[n])
.node_load[node] +=
task_id = .submit_remote_task(node, task_fn, *args, **kwargs)
task_id
() -> :
rpc_client
client = rpc_client.connect(node)
serialized_fn = pickle.dumps(task_fn)
serialized_args = pickle.dumps((args, kwargs))
task_id = client.submit_task(
function=serialized_fn,
arguments=serialized_args
)
task_id
Stage 2: Implement Task and Trajectory Definition
Define what constitutes a task and how to collect trajectories from execution.
from dataclasses import dataclass
from typing import Optional
@dataclass
class Task:
"""Single task for agent to solve"""
task_id: str
instruction: str
tools: List[str]
ground_truth: Any
metadata: Dict = None
def to_prompt(self) -> str:
"""Convert task to agent prompt"""
return f"Instruction: {self.instruction}\nAvailable tools: {self.tools}"
@dataclass
class Trajectory:
"""Single execution trace from task"""
task_id: str
agent_id: str
steps: List[Dict]
final_result: Any
success: bool
token_count: int
execution_time: float
metadata: Dict = None
def to_dict(self):
return {
"task_id": self.task_id,
: .steps,
: .success,
: .token_count,
: .execution_time
}
Stage 3: Create Parallel Task Executor with Agent Integration
Execute tasks on distributed workers with agent inference.
import time
from typing import Callable
class ParallelTaskExecutor:
"""Execute tasks in parallel across workers"""
def __init__(
self,
agent_model: Any,
distributed_executor: DistributedExecutor,
task_loader: Callable
):
self.agent = agent_model
self.executor = distributed_executor
self.task_loader = task_loader
self.trajectories = []
def execute_single_task(
self,
task: Task,
max_steps: int = 10,
timeout: float = 60
) -> Trajectory:
"""
Execute a single task with agent.
Can run in parallel across workers.
"""
steps = []
current_state = {"instruction": task.instruction}
total_tokens = 0
start_time = time.time()
success = False
try:
for step_idx in range(max_steps):
action, action_tokens = self.agent.generate_action(
current_state,
max_tokens=256
)
total_tokens += action_tokens
observation = self.execute_action(action, task)
steps.append({
: step_idx,
: action,
: observation,
: action_tokens
})
current_state[] = action
current_state[] = observation
observation.get():
success = observation.get(, )
Exception e:
steps.append({
: ,
: (e)
})
execution_time = time.time() - start_time
trajectory = Trajectory(
task_id=task.task_id,
agent_id=.agent.model_id,
steps=steps,
final_result=current_state.get(),
success=success,
token_count=total_tokens,
execution_time=execution_time
)
trajectory
() -> :
tool_name = .parse_tool_name(action)
tool_args = .parse_tool_args(action)
:
tool_name task.tools:
result = .call_tool(tool_name, tool_args)
{: result, : }
:
{: , : }
Exception e:
{: , : }
() -> [Trajectory]:
submitted_tasks = {}
task tasks:
task_id = .executor.submit_task(
task.task_id,
.execute_single_task,
task
)
submitted_tasks[task_id] = task
trajectories = []
submitted_tasks:
results = .executor.collect_results(timeout=)
task_id, result results.items():
result[]:
trajectories.append(result[])
:
()
submitted_tasks[task_id]
trajectories
Stage 4: Implement Experience Aggregation and RL Training
Collect experiences from all workers and train the model.
import numpy as np
class ExperienceBuffer:
"""Buffer for collecting trajectories"""
def __init__(self, max_size: int = 100000):
self.trajectories = []
self.max_size = max_size
def add_trajectory(self, traj: Trajectory):
"""Add trajectory to buffer"""
self.trajectories.append(traj)
if len(self.trajectories) > self.max_size:
self.trajectories.pop(0)
def add_batch(self, trajs: List[Trajectory]):
"""Add multiple trajectories"""
for traj in trajs:
self.add_trajectory(traj)
def sample_batch(self, batch_size: int) -> List[Trajectory]:
"""Sample random batch"""
indices = np.random.choice(len(self.trajectories), batch_size)
return [self.trajectories[i] for i in indices]
def compute_returns():
traj .trajectories:
returns = []
cumulative =
step (traj.steps):
reward = step.get(, traj.success )
cumulative = reward + discount * cumulative
returns.insert(, cumulative)
traj.returns = returns
:
():
.model = agent_model
.optimizer = agent_model.optimizer_class(
agent_model.parameters(), lr=learning_rate
)
() -> :
total_loss =
traj batch:
step_idx, step (traj.steps):
action_tokens = step.get(, )
log_prob = .model.get_log_prob(
step[],
traj.steps[:step_idx]
)
advantage = traj.returns[step_idx]
loss = -(log_prob * advantage)
total_loss += loss
.optimizer.zero_grad()
total_loss.backward()
.optimizer.step()
(total_loss / (batch)).item()
:
():
.agent = agent_model
.executor = DistributedExecutor(num_workers)
.task_executor = ParallelTaskExecutor(
agent_model,
.executor,
: task_dataset
)
.experience_buffer = ExperienceBuffer()
.trainer = RLTrainer(agent_model)
.task_dataset = task_dataset
():
selected_tasks = np.random.choice(
.task_dataset,
size=(tasks_per_epoch, (.task_dataset)),
replace=
).tolist()
()
trajectories = .task_executor.distribute_tasks(selected_tasks)
()
.experience_buffer.add_batch(trajectories)
.experience_buffer.compute_returns()
()
batch_size =
_ ((trajectories) // batch_size):
batch = .experience_buffer.sample_batch(batch_size)
loss = .trainer.train_step(batch)
{
: (trajectories),
: (t.success t trajectories) / (trajectories),
: np.mean([t.token_count t trajectories])
}
Practical Guidance
Scaling Configuration
- Worker Count: Start with 8-16 workers; scale up to 64+ for large benchmarks
- Task Batch Size: 100-1000 tasks per distribution cycle
- Experience Buffer: Keep last 10,000-100,000 trajectories for training stability
- Training Frequency: After each 1000 task executions, perform 1-5 training epochs
Performance Optimization
- Network Efficiency: Compress trajectories before transfer (usually 10-100KB each)
- Load Balancing: Use least-loaded routing to avoid hotspots
- Fault Tolerance: Implement retry logic for failed tasks
- Memory Management: Stream results to disk if buffer exceeds RAM
Baseline Metrics
- Single-Node Baseline: ~10 tasks/minute on 1 GPU
- 16-Worker Cluster: ~140-150 tasks/minute (14.6x speedup)
- GAIA Performance: 32.23% pass@1 with Qwen3-32B agent (vs 27.91% GPT-4o)
When to Use
- Training agents on complex benchmarks (GAIA, ScienceBoard)
- Settings with access to multi-node clusters
- Problems requiring millions of task executions
- Scenarios where wall-clock training time matters
When NOT to Use
- Single-machine environments without cluster access
- Tasks with very fast execution (<1 second)
- Strongly sequential dependencies between tasks
- Scenarios with limited network bandwidth
Reference
AWorld: Orchestrating Training Recipe for Agentic AI. arXiv:2508.20404
