ADu2021/automated-tool-learning-rl

Feedback-Driven Tool-Use Improvements via Automated Build Environments

Core Concept

This skill improves LLM tool-use abilities by creating automated training environments that generate detailed, verifiable feedback without requiring external tool access. The system constructs realistic task environments through scenario decomposition, document generation, and function integration, then trains models using RL with rewards that evaluate both tool precision and task completion.

Architecture Overview

• Automated Environment Pipeline: Scenario decomposition, document generation, function integration
• Complexity Scaling: Gradually increase environment difficulty
• Localized Deployment: Self-contained environments for reproducible training
• Verifiable Rewards: Evaluate tool precision and task success simultaneously
• Trajectory-Based Learning: Learn from synthetic interaction sequences

Implementation Steps

Step 1: Design Scenario Decomposition

Break tasks into learnable subtasks:

# Pseudocode for scenario decomposition
class ScenarioDecomposer:
    def __init__(self):
        super().__init__()
        self.task_library = {}

    def decompose_task(self, complex_task):
        """
        Break complex task into subtasks.

        Args:
            complex_task: High-level task description

        Returns:
            subtasks: List of learnable subtasks
        """
        subtasks = []

        # Parse task intent
        intent = self._parse_intent(complex_task)

        # Identify required tools
        required_tools = self._identify_required_tools(intent)

        # Decompose into steps
        for tool in required_tools:
            subtask = {
                'tool': tool,
                'prerequisites': self._get_prerequisites(tool),
                'success_criteria': self._define_success(tool),
                'complexity': self._estimate_complexity(tool)
            }
            subtasks.append(subtask)

        return subtasks

    def _parse_intent(self, task):
        """
        Extract task intent from description.
        """
        # Use NLP or simple pattern matching
        keywords = {}
        for token in task.lower().split():
            keywords[token] = keywords.get(token, 0) + 1

        return keywords

    def _identify_required_tools(self, intent):
        """
        Determine which tools are needed.
        """
        tool_keywords = {
            'file': ['read', 'write', 'open', 'save', 'create'],
            'api': ['fetch', 'request', 'query', 'retrieve'],
            'database': ['query', 'insert', 'update', 'select'],
            'compute': ['calculate', 'process', 'analyze'],
        }

        required = []
        for tool, keywords in tool_keywords.items():
            if any(kw in intent for kw in keywords):
                required.append(tool)

        return required

    def _get_prerequisites(self, tool):
        """
        Get prerequisites for tool use.
        """
        prereq_map = {
            'file': ['file_path'],
            'api': ['endpoint', 'credentials'],
            'database': ['connection_string'],
            'compute': ['input_data']
        }
        return prereq_map.get(tool, [])

    def _define_success(self, tool):
        """
        Define success criteria for tool use.
        """
        criteria = {
            'file': ['file_exists', 'content_correct'],
            'api': ['status_code_200', 'response_valid'],
            'database': ['rows_affected > 0', 'query_valid'],
            'compute': ['result_accurate', 'performance_ok']
        }
        return criteria.get(tool, [])

    def _estimate_complexity(self, tool):
        """
        Estimate learning complexity.
        """
        complexity = {
            'file': 1,
            'api': 2,
            'database': 3,
            'compute': 2
        }
        return complexity.get(tool, 2)

Step 2: Implement Document Generation

Create tool documentation automatically:

# Pseudocode for automated documentation
class DocumentationGenerator:
    def __init__(self):
        super().__init__()
        self.doc_templates = {}

    def generate_tool_documentation(self, tool_name, tool_spec):
        """
        Generate documentation for tool.

        Args:
            tool_name: Name of tool
            tool_spec: Tool specification

        Returns:
            documentation: Generated markdown documentation
        """
        doc = f"# {tool_name.title()} Tool\n\n"

        # Overview
        doc += f"## Overview\n{tool_spec.get('description', 'Tool for ...')}\n\n"

        # Parameters
        doc += "## Parameters\n"
        for param, spec in tool_spec.get('parameters', {}).items():
            doc += f"- **{param}** ({spec.get('type', 'string')}): {spec.get('description', '')}\n"

        # Return value
        doc += f"\n## Returns\n{tool_spec.get('return_description', 'Result object')}\n"

        # Examples
        doc += f"\n## Examples\n```\n{self._generate_example(tool_name, tool_spec)}\n```\n"

        # Common errors
        doc += f"\n## Common Errors\n"
        for error in tool_spec.get('errors', []):
            doc += f"- {error}\n"

        return doc

    def _generate_example(self, tool_name, spec):
        """
        Generate example usage.
        """
        example = f"result = {tool_name}("
        params = []
        for param, pspec in spec.get('parameters', {}).items():
            example_val = pspec.get('example', 'value')
            params.append(f"{param}={example_val}")
        example += ', '.join(params) + ")"
        return example

    def generate_api_spec(self, tool_name, endpoints):
        """
        Generate API specification.

        Args:
            tool_name: API name
            endpoints: List of endpoints

        Returns:
            spec: API documentation
        """
        spec = f"# {tool_name} API Specification\n\n"

        for endpoint in endpoints:
            spec += f"## {endpoint['method']} {endpoint['path']}\n"
            spec += f"{endpoint.get('description', '')}\n"
            spec += f"**Parameters**: {endpoint.get('params', [])}\n"
            spec += f"**Response**: {endpoint.get('response', {})}\n\n"

        return spec

Step 3: Build Function Integration Layer

Create actual tool implementations:

# Pseudocode for function integration
class FunctionIntegration:
    def __init__(self):
        super().__init__()
        self.tool_functions = {}
        self.tool_results = {}

    def integrate_file_operations(self):
        """
        Integrate file operation tools.
        """
        def read_file(file_path: str) -> str:
            """Read file content."""
            try:
                with open(file_path, 'r') as f:
                    return f.read()
            except Exception as e:
                return f"Error: {str(e)}"

        def write_file(file_path: str, content: str) -> bool:
            """Write content to file."""
            try:
                with open(file_path, 'w') as f:
                    f.write(content)
                return True
            except Exception as e:
                return False

        self.tool_functions['read_file'] = read_file
        self.tool_functions['write_file'] = write_file

    def integrate_api_calls(self):
        """
        Integrate API calling tools.
        """
        def api_request(method: str, endpoint: str, params: dict = None):
            """Make API request."""
            import requests
            try:
                if method.upper() == 'GET':
                    response = requests.get(endpoint, params=params)
                elif method.upper() == 'POST':
                    response = requests.post(endpoint, json=params)
                else:
                    return {'error': f'Unknown method: {method}'}

                return {
                    'status_code': response.status_code,
                    'body': response.json(),
                    'success': response.status_code == 200
                }
            except Exception as e:
                return {'error': str(e), 'success': False}

        self.tool_functions['api_request'] = api_request

    def integrate_database_tools(self):
        """
        Integrate database tools.
        """
        def query_database(connection_str: str, query: str):
            """Execute database query."""
            # Simplified implementation
            return {
                'rows_affected': 1,
                'result': [],
                'success': True
            }

        self.tool_functions['query_database'] = query_database

    def execute_tool(self, tool_name: str, **kwargs):
        """
        Execute integrated tool.

        Args:
            tool_name: Name of tool
            **kwargs: Tool parameters

        Returns:
            result: Tool execution result
        """
        if tool_name not in self.tool_functions:
            return {'error': f'Tool not found: {tool_name}'}

        tool_fn = self.tool_functions[tool_name]

        try:
            result = tool_fn(**kwargs)
            self.tool_results[tool_name] = result
            return result
        except Exception as e:
            return {'error': str(e)}

Step 4: Implement Verifiable Reward Mechanism

Design reward that evaluates tool use:

# Pseudocode for reward computation
class VerifiableRewardMechanism:
    def __init__(self):
        super().__init__()
        self.task_success_evaluator = None

    def compute_tool_use_reward(self, action, tool_result, task_state, task_goal):
        """
        Compute reward for tool use action.

        Args:
            action: Tool call with parameters
            tool_result: Result from tool execution
            task_state: Current task state
            task_goal: Goal to achieve

        Returns:
            reward: Scalar reward value
        """
        # Component 1: Tool parameter precision
        parameter_reward = self._evaluate_parameter_precision(action)

        # Component 2: Tool execution success
        execution_reward = self._evaluate_execution_success(tool_result)

        # Component 3: Progress toward goal
        progress_reward = self._evaluate_progress(tool_result, task_state, task_goal)

        # Component 4: Efficiency (not using unnecessary tools)
        efficiency_reward = self._evaluate_efficiency(action, task_state)

        # Combined reward
        total_reward = (
            0.25 * parameter_reward +
            0.25 * execution_reward +
            0.35 * progress_reward +
            0.15 * efficiency_reward
        )

        return total_reward

    def _evaluate_parameter_precision(self, action):
        """
        Score parameter correctness.
        """
        # Check if parameters are well-formed
        params = action.get('parameters', {})

        valid_params = 0
        total_params = len(params)

        for param_name, param_value in params.items():
            if self._is_valid_parameter(param_name, param_value):
                valid_params += 1

        if total_params == 0:
            return 1.0

        return valid_params / total_params

    def _is_valid_parameter(self, param_name, param_value):
        """
        Check if parameter is valid.
        """
        # Simple validation
        if param_name and param_value is not None:
            return True
        return False

    def _evaluate_execution_success(self, tool_result):
        """
        Score execution outcome.
        """
        if not tool_result:
            return 0.0

        if tool_result.get('success'):
            return 1.0
        elif tool_result.get('status_code') == 200:
            return 1.0
        elif 'error' in tool_result:
            return 0.0
        else:
            return 0.5

    def _evaluate_progress(self, tool_result, task_state, task_goal):
        """
        Measure progress toward goal.
        """
        # Check if result brings us closer to goal
        task_components = task_goal.split()
        result_str = str(tool_result)

        matching_components = sum(
            1 for component in task_components
            if component.lower() in result_str.lower()
        )

        if len(task_components) == 0:
            return 0.5

        return matching_components / len(task_components)

    def _evaluate_efficiency(self, action, task_state):
        """
        Reward efficient tool use.
        """
        # Penalize redundant calls
        tool_name = action.get('tool')

        if task_state.get('last_tool') == tool_name:
            return 0.5  # Penalize repeated calls

        return 1.0

    def compute_trajectory_reward(self, trajectory, task_goal):
        """
        Compute reward for complete trajectory.

        Args:
            trajectory: List of (action, result) pairs
            task_goal: Goal statement

        Returns:
            reward: Total trajectory reward
        """
        step_rewards = []
        task_state = {}

        for action, result in trajectory:
            step_reward = self.compute_tool_use_reward(
                action,
                result,
                task_state,
                task_goal
            )

            step_rewards.append(step_reward)
            task_state['last_tool'] = action.get('tool')

        # Compute trajectory return
        trajectory_reward = sum(
            (0.99 ** i) * r for i, r in enumerate(step_rewards)
        )

        return trajectory_reward

Step 5: Implement RL Training Loop

Train tool use through RL:

# Pseudocode for RL training
class ToolUseRLTrainer:
    def __init__(self, model, tool_integration, reward_fn):
        super().__init__()
        self.model = model
        self.tools = tool_integration
        self.reward_fn = reward_fn

    def collect_trajectory(self, task_goal, max_steps=10):
        """
        Collect trajectory using model and tools.

        Returns:
            trajectory: List of (action, result, reward) tuples
        """
        trajectory = []
        task_state = {}
        cumulative_reward = 0

        for step in range(max_steps):
            # Model decides next action
            action = self.model.decide_action(task_goal, task_state)

            # Execute tool
            result = self.tools.execute_tool(
                action['tool'],
                **action.get('parameters', {})
            )

            # Compute reward
            step_reward = self.reward_fn.compute_tool_use_reward(
                action,
                result,
                task_state,
                task_goal
            )

            trajectory.append({
                'action': action,
                'result': result,
                'reward': step_reward,
                'step': step
            })

            cumulative_reward += step_reward
            task_state['last_result'] = result

            # Check if task complete
            if self._task_complete(result, task_goal):
                break

        return trajectory

    def train_on_trajectories(self, trajectories, num_epochs=3):
        """
        Train model on collected trajectories.

        Args:
            trajectories: List of collected trajectories
            num_epochs: Training epochs

        Returns:
            training_stats: Training statistics
        """
        optimizer = AdamW(self.model.parameters(), lr=1e-5)

        for epoch in range(num_epochs):
            total_loss = 0

            for trajectory in trajectories:
                # Compute returns
                returns = self._compute_returns(trajectory)

                for step_idx, step_data in enumerate(trajectory):
                    action = step_data['action']
                    return_val = returns[step_idx]

                    # Forward pass
                    action_logits = self.model.get_action_logits(
                        action['tool'],
                        action.get('parameters', {})
                    )

                    # Policy gradient loss
                    log_prob = F.log_softmax(action_logits, dim=-1).sum()
                    loss = -log_prob * return_val

                    optimizer.zero_grad()
                    loss.backward()
                    torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
                    optimizer.step()

                    total_loss += loss.item()

        return total_loss / len(trajectories)

    def _compute_returns(self, trajectory):
        """
        Compute discounted returns.
        """
        returns = []
        G = 0

        for step in reversed(trajectory):
            G = step['reward'] + 0.99 * G
            returns.insert(0, G)

        return returns

    def _task_complete(self, result, goal):
        """
        Check if task is complete.
        """
        return result.get('success', False)

Practical Guidance

Hyperparameters and Configuration:

• Maximum steps per trajectory: 10-20
• Number of trajectories: 100-1000
• RL learning rate: 1e-5 to 5e-5
• Reward weights: 0.25/0.25/0.35/0.15 (param/execution/progress/efficiency)
• Training epochs: 2-5

When to Use Automated Tool Learning:

• Training LLMs for tool use without external dependencies
• Scenarios with well-defined tool APIs and documentation
• Systems requiring reproducible, scalable training
• Applications needing verifiable tool use correctness

When NOT to Use:

• Complex, ill-defined tools (hard to automate)
• Real tools with external dependencies (database servers, APIs)
• Scenarios where only human feedback is reliable
• When tool complexity exceeds environment simulation

Implementation Notes:

• Automated environments should match real tool behavior closely
• Verifiable rewards are key to learning proper tool use
• Monitor parameter precision separately from task progress
• Consider curriculum: start simple tools, add complexity
• Validate learned tool use on actual external tools

Reference

Paper: Feedback-Driven Tool-Use Improvements via Automated Build Environments ArXiv: 2508.08791 Performance: RL training on synthetic environments improves tool-use capability while preserving general abilities