aworld-distributed-agent-training

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.

# Distributed task execution framework
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.

# Task and trajectory definitions
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]  # Available tools
    ground_truth: Any  # Expected solution
    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]  # Each step: action, observation, reward
    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.

# Parallel task execution with agent
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):
                # Agent generates action
                action, action_tokens = self.agent.generate_action(
                    current_state,
                    max_tokens=256
                )
                total_tokens += action_tokens

                # Execute action
                observation = self.execute_action(action, task)

                # Record step
                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.

# Experience aggregation and RL training
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

• https://arxiv.org/abs/2508.20404