adu2021 Train Agentic Systems to Solve Complex Tasks with Tools
Outcome: Build agents that iteratively reason, call external tools, observe results, and adapt across diverse problem domains using a single unified framework.
Problem Context
Existing approaches to agentic AI fragment tool use into domain-specific systems. A knowledge QA system handles search differently than a code execution system; SQL agents use different APIs than visual reasoning systems. When researchers want agents to solve multi-domain tasks—or when practitioners need to add new tool capabilities—the cost of integration is high: custom pipelines, reimplemented coordination logic, and repeated infrastructure investment.
Single-turn language models also cannot naturally handle the sequential decision-making required for tool use: acting, observing results from that action, and choosing the next action based on new information. Reinforcement learning (RL) can optimize these multi-turn trajectories, but extending RL frameworks to support diverse tools requires careful trajectory representation, observation tokenization, and reward alignment across modalities.
VerlTool solves this by introducing a unified, modular framework that extends Reinforcement Learning with Verifiable Rewards (RLVR) to multi-turn agentic settings with tool use.
Core Concept
VerlTool treats tool use in RL as a trajectory of alternating actions and observations: the agent selects an action (tool call with arguments), the tool executes and returns an observation, and this cycle repeats until the task is solved. The key insight is that observations in tool-use trajectories are environment-generated facts outside the agent's control—they should not influence the policy gradient during training, only provide information for subsequent decisions.
The framework is "holistic" because it unifies:
- Multiple tool types (code, search, SQL, vision) under one API
- Multiple modalities (text, images, video) in observation tokens
- Multi-turn RL training with asynchronous execution for efficiency
- Upstream compatibility with VeRL for seamless maintenance
Instead of building separate agents for code execution, search, and SQL, teams develop a single agent that learns to compose any combination of these tools effectively.
Architecture Overview
VerlTool splits into two primary subsystems that communicate via standardized APIs:
VeRL Workflow (Training Side): Orchestrates policy training, reward computation, and model updates. Inherits VeRL as a submodule to stay aligned with upstream improvements and avoid duplicating core RL logic.
Tool Server (Execution Side): Manages execution of tool calls. Processes rollouts asynchronously on a trajectory-by-trajectory basis instead of enforcing synchronous batch alignment, eliminating idle waiting and achieving approximately 2× speedup during rollout phases.
Modular Tool Registration: Each tool implements a common BaseTool interface with methods for parsing actions, managing environment state, and executing operations. New tools are added by creating lightweight Python definition files without modifying training code. This design separates concerns and enables domain experts to contribute tools without deep RL knowledge.
Tokenization Strategy: A critical detail for stability: action and observation strings are tokenized separately, then concatenated. This prevents boundary-related token mismatches that can cause training instability during multi-turn rollouts. The framework ensures consistent token alignment across all turns.
Observation Masking in Policy Optimization: Since observations are environment-generated and off-policy with respect to the model being trained, the framework masks observation tokens during gradient computation. Only action tokens and the final reward contribute to policy updates. This prevents the model from learning spurious correlations with stale observation data and maintains RL stability.
Implementation
1. Set Up VerlTool Modular Architecture
Begin by inheriting VeRL and structuring the tool server as a separate component. This ensures training logic remains decoupled from tool execution infrastructure.
# verltool/config.py
# Configuration for dual-component architecture
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from verl.config import BaseConfig
@dataclass
class ToolServerConfig(BaseConfig):
"""Configuration for asynchronous tool execution server."""
host: str = "localhost"
port: int = 8888
max_workers: int = 32 # Async worker threads for tool calls
timeout_seconds: int = 300
enable_async_rollout: bool = True # Critical for 2× speedup
@dataclass
class VerlToolConfig(BaseConfig):
"""Top-level VerlTool configuration."""
verl_config: Dict = field(default_factory=dict) # Inherited VeRL settings
tool_server: ToolServerConfig = field(default_factory=ToolServerConfig)
modalities: List[str] = field(default_factory=lambda: ["text"]) # text, image, video
observation_masking_enabled: bool = True # Mask off-policy observations
tokenization_mode: str = "separate" # Separate action/observation tokens
2. Implement Unified Tool Interface
Define the BaseTool abstraction that all tools inherit from. This enables dynamic registration and consistent handling of diverse tools.
# verltool/tools/base.py
# Unified tool interface for consistent behavior across domains
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, Union
from dataclasses import dataclass
@dataclass
class ToolAction:
"""Structured representation of a tool invocation."""
tool_name: str
arguments: Dict[str, Any]
timestamp: Optional[float] = None
@dataclass
class ToolObservation:
"""Result returned by tool execution."""
content: Union[str, bytes]
modality: str = "text" # text, image, video
token_count: Optional[int] = None
error: Optional[str] = None
class BaseTool(ABC):
"""Base class for all tools in VerlTool ecosystem."""
def __init__(self, name: str, modalities: List[str] = None):
self.name = name
self.modalities = modalities or ["text"]
self.state = {} # Maintains context across turns
@abstractmethod
def parse_action(self, action_string: str) -> ToolAction:
"""Convert raw action string to structured ToolAction."""
pass
@abstractmethod
def execute(self, action: ToolAction) -> ToolObservation:
"""Execute the tool and return observation."""
pass
def update_state(self, observation: ToolObservation):
"""Update internal state based on execution result."""
self.state['last_result'] = observation
def reset(self):
"""Clear state for new trajectory."""
self.state = {}
3. Register Tools Dynamically
Create a registry that discovers and manages tools without hardcoding them into training logic.
# verltool/tools/registry.py
# Dynamic tool registration system for extensibility
from typing import Dict, Type, Optional
from verltool.tools.base import BaseTool
import importlib
import os
class ToolRegistry:
"""Registry for tool discovery and management."""
def __init__(self, tool_dir: str = "verltool/tools"):
self.tools: Dict[str, BaseTool] = {}
self.tool_dir = tool_dir
def register(self, tool_class: Type[BaseTool]):
"""Explicitly register a tool class."""
instance = tool_class()
self.tools[instance.name] = instance
return self
def discover_from_directory(self):
"""Auto-discover tools from tool_dir by importing modules."""
for filename in os.listdir(self.tool_dir):
if filename.endswith('_tool.py') and not filename.startswith('_'):
module_name = filename[:-3]
try:
module = importlib.import_module(f'verltool.tools.{module_name}')
# Assume each module exports TOOL_CLASS
if hasattr(module, 'TOOL_CLASS'):
self.register(module.TOOL_CLASS)
except Exception as e:
print(f"Warning: Could not load tool {module_name}: {e}")
return self
def get_tool(self, name: str) -> Optional[BaseTool]:
"""Retrieve a tool by name."""
return self.tools.get(name)
def list_tools(self) -> Dict[str, BaseTool]:
"""Return all registered tools."""
return self.tools.copy()
4. Implement Code Execution Tool
Demonstrate a concrete tool implementation for executing Python code, a common requirement in reasoning and software engineering tasks.
# verltool/tools/code_execution_tool.py
# Code execution tool with sandboxed environment support
from verltool.tools.base import BaseTool, ToolAction, ToolObservation
from typing import Dict, Any
import subprocess
import tempfile
import os
class CodeExecutionTool(BaseTool):
"""Execute Python code and capture output."""
def __init__(self, timeout_seconds: int = 30):
super().__init__(
name="code_execution",
modalities=["text"]
)
self.timeout = timeout_seconds
self.execution_history = []
def parse_action(self, action_string: str) -> ToolAction:
"""Extract Python code from action string."""
# Expected format: <code>python_code_here</code>
start = action_string.find('<code>')
end = action_string.find('</code>')
if start == -1 or end == -1:
return ToolAction(
tool_name=self.name,
arguments={"code": action_string}
)
code = action_string[start + 6:end].strip()
return ToolAction(
tool_name=self.name,
arguments={"code": code}
)
def execute(self, action: ToolAction) -> ToolObservation:
"""Run code in subprocess and capture output."""
code = action.arguments.get("code", "")
if not code:
return ToolObservation(
content="Error: No code provided",
modality="text",
error="empty_code"
)
try:
with tempfile.NamedTemporaryFile(
mode='w',
suffix='.py',
delete=False
) as f:
f.write(code)
temp_file = f.name
result = subprocess.run(
['python', temp_file],
capture_output=True,
text=True,
timeout=self.timeout
)
output = result.stdout
if result.stderr:
output += "\nStderr:\n" + result.stderr
self.execution_history.append({
"code": code,
"output": output,
"returncode": result.returncode
})
return ToolObservation(
content=output or "(No output)",
modality="text",
error=None if result.returncode == 0 else "execution_error"
)
except subprocess.TimeoutExpired:
return ToolObservation(
content=f"Error: Code execution timeout (>{self.timeout}s)",
modality="text",
error="timeout"
)
except Exception as e:
return ToolObservation(
content=f"Error: {str(e)}",
modality="text",
error="execution_failed"
)
finally:
if os.path.exists(temp_file):
os.remove(temp_file)
TOOL_CLASS = CodeExecutionTool
5. Build Multi-Turn Trajectory Handling with Observation Masking
Implement trajectory construction that alternates actions and observations, with proper masking for off-policy observations.
# verltool/training/trajectory.py
# Multi-turn trajectory representation with observation masking
from typing import List, Optional, Dict, Any
from dataclasses import dataclass, field
import torch
@dataclass
class MultiTurnTrajectory:
"""Represents a complete multi-turn agent trajectory."""
actions: List[str] = field(default_factory=list) # Agent actions/tool calls
observations: List[str] = field(default_factory=list) # Environment observations
action_tokens: List[torch.Tensor] = field(default_factory=list)
observation_tokens: List[torch.Tensor] = field(default_factory=list)
observation_masks: List[bool] = field(default_factory=list) # True = mask out
reward: float = 0.0
episode_done: bool = False
def add_turn(
self,
action: str,
observation: str,
action_token_ids: torch.Tensor,
observation_token_ids: torch.Tensor,
mask_observation: bool = True
):
"""Add a single action-observation pair to trajectory."""
self.actions.append(action)
self.observations.append(observation)
self.action_tokens.append(action_token_ids)
self.observation_tokens.append(observation_token_ids)
# Observation tokens are off-policy and should be masked
self.observation_masks.append(mask_observation)
def get_concatenated_tokens(self) -> torch.Tensor:
"""Interleave action and observation tokens."""
sequence = []
for i in range(len(self.actions)):
sequence.append(self.action_tokens[i])
if i < len(self.observation_tokens):
sequence.append(self.observation_tokens[i])
return torch.cat(sequence, dim=0)
def get_loss_mask(self) -> torch.Tensor:
"""Create mask: True where loss should be computed, False otherwise."""
mask = []
for i in range(len(self.actions)):
# Action tokens contribute to loss
mask.append(torch.ones_like(self.action_tokens[i], dtype=torch.bool))
# Observation tokens are masked (do not contribute to loss)
if i < len(self.observation_masks):
is_masked = self.observation_masks[i]
obs_mask = torch.zeros_like(
self.observation_tokens[i],
dtype=torch.bool
) if is_masked else torch.ones_like(
self.observation_tokens[i],
dtype=torch.bool
)
mask.append(obs_mask)
return torch.cat(mask, dim=0)
class TrajectoryBuilder:
"""Construct trajectories from agent rollouts."""
def __init__(self, tokenizer):
self.tokenizer = tokenizer
def build_from_rollout(
self,
actions: List[str],
observations: List[str],
reward: float,
mask_all_observations: bool = True
) -> MultiTurnTrajectory:
"""Convert rollout data into a trajectory with proper tokenization."""
trajectory = MultiTurnTrajectory(reward=reward)
for i, action in enumerate(actions):
# Tokenize separately to avoid boundary issues
action_tokens = self.tokenizer.encode(action)
action_tensor = torch.tensor(action_tokens)
if i < len(observations):
obs = observations[i]
obs_tokens = self.tokenizer.encode(obs)
obs_tensor = torch.tensor(obs_tokens)
trajectory.add_turn(
action=action,
observation=obs,
action_token_ids=action_tensor,
observation_token_ids=obs_tensor,
mask_observation=mask_all_observations
)
return trajectory
6. Implement Asynchronous Tool Server for 2× Speedup
Deploy tool execution asynchronously so rollouts don't block waiting for tool completion.
# verltool/server/async_tool_server.py
# Asynchronous tool execution with trajectory-level batching
import asyncio
from typing import List, Dict, Any, Callable
from concurrent.futures import ThreadPoolExecutor
import queue
from verltool.tools.registry import ToolRegistry
from verltool.tools.base import ToolAction, ToolObservation
class AsyncToolServer:
"""Non-blocking tool execution server for efficient rollouts."""
def __init__(self, tool_registry: ToolRegistry, max_workers: int = 32):
self.registry = tool_registry
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.pending_tasks: Dict[str, asyncio.Future] = {}
self.task_counter = 0
def execute_tool_async(
self,
tool_name: str,
action_string: str
) -> str:
"""Submit tool execution without blocking, return task ID."""
tool = self.registry.get_tool(tool_name)
if not tool:
return None
task_id = f"task_{self.task_counter}"
self.task_counter += 1
def _run():
action = tool.parse_action(action_string)
observation = tool.execute(action)
tool.update_state(observation)
return observation
future = asyncio.get_event_loop().run_in_executor(
self.executor,
_run
)
self.pending_tasks[task_id] = future
return task_id
async def wait_for_result(self, task_id: str, timeout: int = 300) -> ToolObservation:
"""Await tool completion and retrieve result."""
if task_id not in self.pending_tasks:
raise ValueError(f"Unknown task: {task_id}")
try:
result = await asyncio.wait_for(
self.pending_tasks[task_id],
timeout=timeout
)
del self.pending_tasks[task_id]
return result
except asyncio.TimeoutError:
return ToolObservation(
content=f"Timeout waiting for {task_id}",
modality="text",
error="timeout"
)
async def process_trajectory_batch(
self,
trajectories: List[Dict[str, Any]]
) -> List[List[ToolObservation]]:
"""Execute all tool calls in a batch of trajectories concurrently."""
all_observations = []
for traj in trajectories:
tool_calls = traj.get("tool_calls", [])
tasks = [
self.execute_tool_async(call["tool"], call["action"])
for call in tool_calls
]
observations = []
for task_id in tasks:
result = await self.wait_for_result(task_id)
observations.append(result)
all_observations.append(observations)
return all_observations
7. Configure Reward and Loss Computation
Define how rewards are computed for tool-use trajectories and how the loss respects observation masking.
# verltool/training/reward.py
# Reward computation for multi-turn agentic trajectories
from typing import Optional, Dict, Any
import torch
import torch.nn as nn
class VerlToolRewardComputer:
"""Compute rewards from verifiable task outcomes."""
def __init__(self, reward_fn: callable, use_per_step_rewards: bool = False):
"""
reward_fn: Function taking (final_output, expected_output) -> float
use_per_step_rewards: If True, grant intermediate rewards for progress
"""
self.reward_fn = reward_fn
self.use_per_step_rewards = use_per_step_rewards
def compute(
self,
final_output: str,
expected_output: Optional[str] = None,
intermediate_outputs: Optional[Dict[int, str]] = None
) -> float:
"""
Compute scalar reward for trajectory.
Follows RLVR principle: reward only depends on verifiable task outcome.
"""
if expected_output is None:
# Fallback: treat any valid output as success
return 1.0 if final_output else 0.0
base_reward = self.reward_fn(final_output, expected_output)
if self.use_per_step_rewards and intermediate_outputs:
# Optional: add small bonuses for intermediate progress
bonus = 0.0
for step, output in intermediate_outputs.items():
if output and step < len(intermediate_outputs) - 1:
bonus += 0.05 # Small step reward
return min(base_reward + bonus, 1.0)
return base_reward
class PolicyLoss(nn.Module):
"""Compute policy gradient loss with observation masking."""
def __init__(self, model: nn.Module):
super().__init__()
self.model = model
def forward(
self,
trajectory, # MultiTurnTrajectory instance
logits: torch.Tensor
) -> torch.Tensor:
"""
Compute policy loss, masking observation tokens.
logits: Model output logits for entire sequence
"""
loss_mask = trajectory.get_loss_mask() # Bool tensor
# Shift logits for language modeling objective
shift_logits = logits[:-1]
target_tokens = trajectory.get_concatenated_tokens()[1:]
# Cross-entropy loss only on unmasked (action) tokens
ce_loss = torch.nn.functional.cross_entropy(
shift_logits.view(-1, logits.shape[-1]),
target_tokens.view(-1),
reduction='none'
)
# Apply mask to loss
ce_loss = ce_loss * loss_mask[1:].float()
# Weight by trajectory reward
trajectory_reward = max(trajectory.reward, 0.0)
weighted_loss = ce_loss.mean() * trajectory_reward
return weighted_loss
Practical Guidance
Hyperparameter Configuration:
| Parameter | Domain | Recommended | Rationale |
|---|---|---|---|
max_turns |
Math/SQL | 8–12 | Allows sufficient tool calls without excessive sequences |
max_turns |
Code execution | 5–8 | Shorter: limited debugging iterations needed |
max_turns |
Web search | 4–6 | Search queries return quick results |
observation_masking |
All | True | Prevents off-policy observation leakage into gradients |
async_workers |
All | 32–64 | Balances concurrency; 2× speedup with ≥32 |
tokenization_mode |
All | "separate" | Avoids token boundary mismatches across turns |
timeout_seconds |
Code/search | 30–60 | Prevents runaway tool calls; adjust by domain |
learning_rate |
All | 1e-5 to 5e-5 | Tool-use RL requires careful tuning |
When to Use VerlTool:
- Multi-domain agents where code, search, SQL, and vision tools must coexist
- Organizations with shared RL infrastructure that multiple teams extend
- Tasks requiring multi-turn reasoning where outcomes are verifiable (rewards are computable)
- Scenarios demanding fast rollout throughput (async execution provides 2× improvement)
- Environments where observation tokens are abundant and masking prevents noise in gradients
When NOT to Use VerlTool:
- Single-task systems where domain-specific optimization is critical (specialized agents may outperform)
- Scenarios where tool observations directly determine rewards and policy (observation masking would discard signal)
- Real-time systems with strict latency budgets (asynchronous design adds queueing overhead)
- Environments with non-verifiable rewards (RLVR extension assumes ground-truth task outcomes)
- Teams without RL expertise managing training infrastructure (requires careful reward engineering and convergence tuning)
Pitfalls to Avoid:
Neglecting Tool Timeout Configuration: Set realistic timeouts per tool. Code execution might need 60s; search should timeout faster. Timeouts that are too long create training bottlenecks; too short causes spurious failures.
Improper Reward Alignment: Verify that reward functions reflect the actual task goal. Weak reward signals lead to high variance in RL and poor convergence. Use verification functions that match your evaluation metric.
Forgetting to Reset Tool State: Each trajectory should start fresh. Stale state from previous episodes corrupts observations. Always call
tool.reset()between trajectories.Mixing Observation Masking Strategies: Once you enable observation masking, apply it consistently across training. Inconsistent masking destabilizes gradients and causes sudden performance drops.
Inadequate Batch Size for Async Server: The speedup assumes enough concurrent trajectories to saturate workers. Small batch sizes (< 8) negate async benefits. Use batch sizes ≥ 16 when enabling async rollout.
Skipping Tool Registry Discovery: Hardcoding tools into training loops defeats modularity. Use the registry's
discover_from_directory()method to enable true plug-and-play tool integration.
Reference
Paper: VerlTool: Towards Holistic Agentic Reinforcement Learning with Tool Use. Jiang, D., Lu, Y., et al. (2025). arXiv:2509.01055 [cs.AI].
Full Citation: https://arxiv.org/abs/2509.01055
Key Contributions: The framework demonstrates that unified, modular tool-use RL matches or exceeds domain-specific systems across six diverse benchmarks (mathematical reasoning at 62.2%, knowledge QA at 45.9%, SQL generation matching SkyRL-SQL, visual reasoning at 82.7%, web search at 34.0% GAIA accuracy, and software engineering at 19.5% SWE-Verified). Its 2× rollout speedup via async execution and observation masking strategy provide both efficiency and training stability for multi-turn agentic RL.