subgoal-driven-long-horizon-agents

name	subgoal-driven-long-horizon-agents
title	A Subgoal-driven Framework for Improving Long-Horizon LLM Agents
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2603.19685
keywords	["Long-Horizon Planning","Reinforcement Learning","Hierarchical Decomposition","Agent Training"]
description	Improve long-horizon task success via subgoal decomposition and dense milestone-based rewards, dramatically outperforming sparse-reward RL and standard baselines.

Subgoal-Driven Framework for Long-Horizon LLM Agents

Long-horizon tasks present two fundamental challenges for LLM agents. First, the agent can lose sight of the ultimate goal as intermediate steps accumulate, making suboptimal local decisions that push toward failure. Second, when rewards arrive only at the end (sparse rewards), the agent cannot learn which actions contribute to success—a credit assignment crisis.

This framework solves both through hierarchical planning with dense rewards. At the start of execution, decompose the task into explicit subgoals using an external planner. Then, during RL training, provide dense rewards whenever the agent reaches a subgoal (milestone). This transforms the sparse-reward problem into a sequence of easier intermediate problems, enabling dramatically faster learning.

Core Concept

The framework combines two reinforcement learning innovations:

Subgoal-Driven Online Planning: Use an external planner to decompose complex objectives into concrete intermediate milestones. This gives the agent explicit waypoints—reducing goal drift and providing structure for exploration.

Milestoning Reinforcement Learning Enhanced Agent (MiRA): Replace sparse endpoint rewards with dense per-milestone rewards. When the agent reaches a milestone, it immediately receives positive reward feedback, enabling credit assignment without delayed signals.

Together, these enable learning on long-horizon tasks that would fail with standard RL training (0% success) to achieve 43%+ success rates.

Architecture Overview

• Task Planner: External model that decomposes high-level objectives into subgoals
• Milestone Tracker: Monitors agent progress against planned subgoals
• Dense Reward Generator: Provides per-milestone rewards based on progress
• Milestone-Conditioned Policy: RL policy that learns to reach specific subgoals
• Subgoal Validator: Checks whether claimed milestone achievements are legitimate
• Fallback Mechanism: If agent gets stuck, replan or skip unreachable milestones

Implementation Steps

Step 1: Task Decomposition into Subgoals

Convert high-level objectives into concrete, verifiable subgoals.

from typing import List, Dict, Optional
from dataclasses import dataclass
import json

@dataclass
class Subgoal:
    """Represents an intermediate milestone in task execution."""
    index: int  # Order in sequence
    description: str  # What must be achieved
    preconditions: List[str]  # What must be true first
    success_criteria: str  # How to verify completion
    estimated_steps: int  # Rough # steps expected
    required_for_task: bool = True  # Skip if impossible?

    def to_prompt_segment(self) -> str:
        return f"""
Subgoal {self.index}: {self.description}
- Verify by: {self.success_criteria}
- Expected steps: ~{self.estimated_steps}
"""

class TaskPlanner:
    """
    Decomposes complex tasks into subgoals.
    Uses external LLM for planning.
    """

    def __init__(self, planning_model):
        self.planning_model = planning_model

    def decompose_task(self, task_description: str, num_subgoals:  = ) - [Subgoal]:
        
        planning_prompt = 

        response = .planning_model.generate(planning_prompt)
        subgoals_data = ._parse_subgoal_json(response)

        subgoals = []
         idx, sg_data  (subgoals_data):
            subgoal = Subgoal(
                index=idx,
                description=sg_data[],
                preconditions=sg_data.get(, []),
                success_criteria=sg_data[],
                estimated_steps=sg_data.get(, ),
                required_for_task=
            )
            subgoals.append(subgoal)

         subgoals

     () -> []:
        
         re
        json_match = re.search(, response, re.DOTALL)
         json_match:
             json.loads(json_match.group())
         []

     () -> [[Subgoal]]:
        
        remaining_task = 

         .decompose_task(remaining_task, num_subgoals=)

Step 2: Milestone Tracking and Verification

Monitor progress toward subgoals and verify legitimate achievement.

class MilestoneTracker:
    """
    Tracks agent progress against planned subgoals.
    Verifies that claimed achievements are real.
    """

    def __init__(self, environment, verifier_model):
        self.environment = environment
        self.verifier_model = verifier_model
        self.completed_subgoals: List[Subgoal] = []
        self.current_subgoal_idx = 0

    def get_current_milestone(self, subgoals: List[Subgoal]) -> Optional[Subgoal]:
        """What is the agent currently working toward?"""
        if self.current_subgoal_idx < len(subgoals):
            return subgoals[self.current_subgoal_idx]
        return None

    def verify_milestone_achievement(self, subgoal: Subgoal,
                                     environment_state: Dict) -> bool:
        """
        Check if a claimed subgoal achievement is legitimate.
        Uses both environment observation + LLM verification.
        """

        # Method 1: Structured environment check
        env_verification = self._check_environment_state(subgoal, environment_state)

        # Method 2: LLM verification (for complex criteria)
        llm_verification = self._verify_with_llm(subgoal, environment_state)

        # Require both forms of agreement
        return env_verification and llm_verification

     () -> :
        
        criteria_parts = subgoal.success_criteria.split()

         criterion  criteria_parts:
            criterion = criterion.strip().lower()

            
               criterion    state.get(, {}):
                 
               criterion  state.get() == subgoal.description.split()[-]:
                 
               criterion  state.get():
                 

         

     () -> :
        
        verification_prompt = 

        response = .verifier_model.generate(verification_prompt).strip().upper()
           response

     () -> :
        
         .verify_milestone_achievement(subgoal, environment_state):
            .completed_subgoals.append(subgoal)
            .current_subgoal_idx += 
             

         

     () -> :
        
         {
            : (.completed_subgoals),
            : (subgoals),
            :  * (.completed_subgoals) / ((subgoals), ),
            : subgoals[.current_subgoal_idx].description  .current_subgoal_idx < (subgoals)  
        }

Step 3: Dense Reward Generation (MiRA)

Compute per-milestone rewards for effective RL training.

import numpy as np

class MilestoneRewardGenerator:
    """
    Generates dense per-milestone rewards.
    Replaces sparse endpoint rewards with intermediate signals.
    """

    def __init__(self, milestone_tracker: MilestoneTracker):
        self.milestone_tracker = milestone_tracker

    def compute_reward(self, previous_progress: int, current_progress: int,
                       episode_step: int, max_steps: int,
                       subgoals: List[Subgoal]) -> float:
        """
        Compute reward for this step.

        Rewards are primarily given for reaching milestones,
        with small bonuses for efficiency.
        """

        reward = 0.0

        # Major reward: reaching a new milestone
        if current_progress > previous_progress:
            # Milestone reward proportional to sequence position (later milestones worth more)
            milestone_bonus = 10.0 * (current_progress / max(len(subgoals), 1))
            reward += milestone_bonus

            # Efficiency bonus: reward reaching milestone quickly
            steps_available = subgoals[current_progress - 1].estimated_steps * 2
            if episode_step < steps_available:
                efficiency_bonus = 2.0 * (1.0 - episode_step / steps_available)
                reward += efficiency_bonus

        # Small per-step cost: encourage efficiency
        step_cost = -0.01

        # Penalty if running out of steps
         episode_step > max_steps * :
            timeout_penalty = -
            reward += timeout_penalty

         reward + step_cost

     () -> :
        
        total_return = 
        previous_progress = 

         step_idx, step_data  (trajectory):
            current_progress = step_data[]
            reward = .compute_reward(
                previous_progress, current_progress,
                step_idx, (trajectory), subgoals
            )
            total_return += ( ** step_idx) * reward  
            previous_progress = current_progress

        
         previous_progress == (subgoals):
            total_return += 

         total_return

Step 4: Milestone-Conditioned Policy Training

Train the RL policy to reach specific subgoals.

import torch
import torch.nn as nn
from torch.optim import Adam

class MilestoneConditionedPolicy:
    """
    RL policy that learns to reach specific milestones.
    Conditioned on: current state + target milestone.
    """

    def __init__(self, state_dim: int, action_dim: int, hidden_dim: int = 256):
        self.policy_net = nn.Sequential(
            nn.Linear(state_dim + 64, hidden_dim),  # +64 for milestone embedding
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )

        self.value_net = nn.Sequential(
            nn.Linear(state_dim + 64, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

        self.milestone_encoder = nn.Embedding(10, 64)  # Encode milestone index
        self.optimizer = Adam(
            list(self.policy_net.parameters()) +
            list(self.value_net.parameters()),
            lr=1e-4
        )

    def forward(self, state: torch.Tensor, milestone_idx: int) -> torch.Tensor:
        """Compute action distribution conditioned on target milestone."""
        milestone_embedding = self.milestone_encoder(torch.tensor(milestone_idx))
        state_milestone = torch.cat([state, milestone_embedding], dim=-1)

        logits = .policy_net(state_milestone)
        action_probs = torch.softmax(logits, dim=-)

         action_probs

     () -> torch.Tensor:
        
        milestone_embedding = .milestone_encoder(torch.tensor(milestone_idx))
        state_milestone = torch.cat([state, milestone_embedding], dim=-)

        value = .value_net(state_milestone)
         value

     ():
        

        
        values = []
         step_data  trajectory:
            state = torch.tensor(step_data[], dtype=torch.float32)
            milestone_idx = step_data[]
            value = .estimate_value(state, milestone_idx)
            values.append(value.item())

        
        advantages = []
        gae = 
         t  (((trajectory))):
            td_error = rewards[t] +  * values[t + ] - values[t]  t < (trajectory) -   rewards[t] - values[t]
            gae = td_error +  *  * gae
            advantages.insert(, gae)

        advantages = torch.tensor(advantages, dtype=torch.float32)
        advantages = (advantages - advantages.mean()) / (advantages.std() + )

        
        policy_loss = 
         t  ((trajectory)):
            state = torch.tensor(trajectory[t][], dtype=torch.float32)
            action = trajectory[t][]
            milestone_idx = trajectory[t][]

            action_probs = .forward(state.unsqueeze(), milestone_idx)
            log_prob = torch.log(action_probs[, action] + )
            policy_loss = policy_loss - log_prob * advantages[t]

        
        value_loss = 
         t  ((trajectory)):
            state = torch.tensor(trajectory[t][], dtype=torch.float32)
            milestone_idx = trajectory[t][]
            value = .estimate_value(state.unsqueeze(), milestone_idx)
            target_value = torch.tensor(values[t], dtype=torch.float32)
            value_loss = value_loss + (value - target_value).()

        
        total_loss = policy_loss +  * value_loss

        .optimizer.zero_grad()
        total_loss.backward()
        torch.nn.utils.clip_grad_norm_(.policy_net.parameters(), )
        torch.nn.utils.clip_grad_norm_(.value_net.parameters(), )
        .optimizer.step()

         total_loss.item()

Step 5: Main Training Loop

Integrate planning, tracking, and RL training.

class SubgoalDrivenLongHorizonAgent:
    """Complete agent with subgoal planning and milestone-based RL."""

    def __init__(self, environment, planning_model, verifier_model,
                 rl_policy, state_dim: int):
        self.environment = environment
        self.task_planner = TaskPlanner(planning_model)
        self.milestone_tracker = MilestoneTracker(environment, verifier_model)
        self.reward_generator = MilestoneRewardGenerator(self.milestone_tracker)
        self.rl_policy = rl_policy

    def train(self, task_description: str, num_episodes: int = 100):
        """Train agent on task with subgoal decomposition and dense rewards."""

        # Initial task decomposition
        subgoals = self.task_planner.decompose_task(task_description)
        print(f"Decomposed task into {len(subgoals)} subgoals")

        episode_returns = []

        for episode in range(num_episodes):
            # Reset
            state = self.environment.reset()
            self.milestone_tracker = MilestoneTracker(self.environment, self.verifier_model)
            trajectory = []
            rewards_list = []
            previous_progress = 0

            for step in range(500):  
                
                current_milestone = .milestone_tracker.get_current_milestone(subgoals)
                  current_milestone:
                      

                
                state_tensor = torch.tensor(state, dtype=torch.float32)
                milestone_idx = current_milestone.index
                action_probs = .rl_policy.forward(state_tensor, milestone_idx)
                action = torch.multinomial(action_probs, ).item()

                
                next_state, _ = .environment.step(action)

                
                progress_advanced = .milestone_tracker.update_progress(
                    current_milestone, .environment.get_state()
                )
                current_progress = .milestone_tracker.current_subgoal_idx

                
                reward = .reward_generator.compute_reward(
                    previous_progress, current_progress, step, , subgoals
                )

                
                trajectory.append({
                    : state,
                    : action,
                    : next_state,
                    : milestone_idx
                })
                rewards_list.append(reward)

                previous_progress = current_progress
                state = next_state

            
             trajectory:
                loss = .rl_policy.train_on_trajectory(trajectory, subgoals, rewards_list)

            
            episode_return = .reward_generator.compute_episode_return(trajectory, subgoals)
            episode_returns.append(episode_return)

             episode %  == :
                progress = .milestone_tracker.get_progress_summary(subgoals)
                ()

            
             previous_progress ==   episode > :
                ()
                subgoals = .task_planner.replan_from_milestone(
                    task_description, [], .environment.get_state()
                )

         episode_returns

Practical Guidance

Hyperparameters:

• Number of subgoals: 4-7 (task dependent; too many fragments focus)
• Milestone bonus: 10x baseline per-step reward (makes milestones the dominant signal)
• Efficiency bonus multiplier: 2.0-5.0 (encourages quick milestone reaching)
• Policy learning rate: 1e-4 to 1e-3 (stable for milestone-conditioned learning)
• GAE parameter: λ=0.95 (balance bias-variance in advantage estimation)

When to Use:

• Long-horizon tasks (>50 steps) where credit assignment is hard
• Environments where intermediate milestones are identifiable
• Scenarios with high penalty for task failure (structure reduces random exploration)
• Training agent on single long task (not multi-task)

When NOT To Use:

• Short-horizon tasks (<10 steps) where subgoal overhead dominates
• Environments where milestones are ambiguous or hard to verify
• Tasks requiring rapid exploration (dense rewards can narrow behavior too much)
• Online learning where replanning is prohibitively expensive

Pitfalls:

• Subgoal specification critical: bad milestones mislead the agent; validate with model predictions
• Verification bugs: if milestone detection is wrong, rewards become noise
• Reward shaping can harm: if endpoint rewards still matter, milestone rewards may conflict
• Replanning cost: replanning too frequently disrupts learning; do it sparingly

Reference

Paper: arxiv.org/abs/2603.19685