POMDPPlanners.planners.planners_utils package

Submodules

POMDPPlanners.planners.planners_utils.constrained_mcts_mixin module

Shared abstract layer for CPOMDP-DPW MCTS variants.

Mirrors Listing 1 from Jamgochian, Corso, & Kochenderfer (ICAPS 2023), Online Planning for Constrained POMDPs with Continuous Spaces through Dual Ascent (arXiv:2212.12154). The paper factors all three variants (CPOMCP-DPW, CPOMCPOW, CPFT-DPW) into:

  • A shared layer of PLAN, GREEDYPOLICY, ACTIONPROGWIDEN procedures with dual ascent on a Lagrange multiplier λ.

  • A variant-specific SIMULATE (Algorithm 1 for CPOMCPOW, Algorithm 2 for CPFT-DPW).

ConstrainedMCTSMixin implements the shared layer and declares _simulate_path_with_cost as abstract. Concrete subclasses (CPOMCPOW, CPFT_DPW, and any future variant) inherit the mixin first so its overrides of action and _construct_tree_using_* win over the underlying unconstrained planner’s implementations.

Classes:
ConstrainedMCTSMixin: ABC providing the shared dual-ascent + Lagrangian

action-selection scaffold, with one abstract method _simulate_path_with_cost.

class POMDPPlanners.planners.planners_utils.constrained_mcts_mixin.ConstrainedMCTSMixin[source]

Bases: ABC

Shared scaffold for constrained-POMDP MCTS planners.

Implements paper Listing 1’s PLAN, GREEDYPOLICY (Lagrangian UCB + root greedy pick), ACTIONPROGWIDEN, and the dual-ascent update on λ — together with constraint-parameter validation, cost-Q storage, and a boundary check on env-returned constraint cost vectors. Concrete planners implement only _simulate_path_with_cost() (the paper’s SIMULATE).

Inheritance pattern (mixin-first):

class CPOMCPOW(ConstrainedMCTSMixin, POMCPOW): ...
class CPFT_DPW(ConstrainedMCTSMixin, PFT_DPW): ...

Required host-class surface (provided by POMCPOW / PFT_DPW via their parent chain ArenaDoubleProgressiveWideningMCTSPolicy ArenaPathSimulationPolicy): environment, action_sampler, depth, k_a, alpha_a, exploration_constant, min_visit_count_per_action, discount_factor, n_simulations, time_out_in_seconds, _learn_tree, _is_terminal_belief, _sample_random_action.

action(belief)[source]

Plan one action by running n dual-ascent-coupled simulations.

Mirrors PLAN(b) from the paper: for each iteration, run SIMULATE (variant-specific), pick the greedy Lagrangian action a*, and update λ toward λ + α_i (QC(b, a*) ĉ) with non-negative projection. Returns the final greedy Lagrangian root action.

Return type:

Tuple[List[Any], PolicyRunData]

Parameters:

belief (Belief)

constrained_env: ConstrainedEnvironment
cost_budget: ndarray
lambda_init: ndarray
lambda_step: float
n_constraints: int
return_minimal_cost: bool

POMDPPlanners.planners.planners_utils.cvar_exploration module

POMDPPlanners.planners.planners_utils.cvar_exploration.get_explored_action_node(belief_node, min_immediate_cost, max_immediate_cost, depth, max_depth, gamma, exploration_constant, min_visit_count_per_action, alpha, delta, visit_count_penalty=0.0)[source]
Return type:

ActionNode

Parameters:
POMDPPlanners.planners.planners_utils.cvar_exploration.get_explored_action_node_arena(tree, belief_id, min_immediate_cost, max_immediate_cost, depth, max_depth, gamma, exploration_constant, min_visit_count_per_action, alpha, delta, visit_count_penalty=0.0)[source]

Arena variant of get_explored_action_node(). Returns action-node ID.

Return type:

int

Parameters:

POMDPPlanners.planners.planners_utils.cvar_progressive_widening module

POMDPPlanners.planners.planners_utils.cvar_progressive_widening.cvar_action_progressive_widening(belief_node, alpha_a, action_sampler, exploration_constant, k_a, min_immediate_cost, max_immediate_cost, depth, max_depth, gamma, min_visit_count_per_action, alpha, delta, discrete_actions=False, visit_count_penalty=0.0)[source]
Return type:

ActionNode

Parameters:
POMDPPlanners.planners.planners_utils.cvar_progressive_widening.cvar_action_progressive_widening_arena(tree, belief_id, alpha_a, action_sampler, exploration_constant, k_a, min_immediate_cost, max_immediate_cost, depth, max_depth, gamma, min_visit_count_per_action, alpha, delta, discrete_actions=False, visit_count_penalty=0.0, environment=None)[source]

Arena variant of cvar_action_progressive_widening().

Returns the action-node ID selected by CVaR-aware progressive widening from belief node belief_id in tree.

When environment is supplied, environment.hash_action is used to index action children in O(1). When omitted, falls back to the legacy linear-scan path for backwards compatibility.

Return type:

int

Parameters:

POMDPPlanners.planners.planners_utils.dpw module

class POMDPPlanners.planners.planners_utils.dpw.ActionSampler[source]

Bases: ABC

Abstract base class for action sampling strategies in PFT-DPW.

Action samplers provide domain-specific strategies for generating new actions during progressive widening. This allows PFT-DPW to work with continuous or large discrete action spaces by intelligently sampling promising actions.

The ActionSampler interface enables flexible action space exploration by allowing custom sampling strategies that can incorporate domain knowledge, belief state information, or specialized sampling distributions.

The class is serializable and can be safely pickled/unpickled for distributed computing, caching, or saving/loading configurations.

Examples

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>> from POMDPPlanners.planners.planners_utils.dpw import ActionSampler
>>>
>>> class ContinuousControlSampler(ActionSampler):
...     def __init__(self, action_bounds=(-1.0, 1.0), action_dim=2):
...         self.action_bounds = action_bounds
...         self.action_dim = action_dim
...
...     def sample(self, belief_node: Optional[BeliefNode] = None):
...         # Sample uniformly from action space
...         low, high = self.action_bounds
...         return np.random.uniform(low, high, size=self.action_dim)
>>>
>>> # Usage with PFT-DPW
>>> sampler = ContinuousControlSampler(action_bounds=(-2.0, 2.0), action_dim=4)
>>> action = sampler.sample()  # Returns 4D action vector
>>>
>>> # Serialization works automatically
>>> import pickle
>>> serialized = pickle.dumps(sampler)
>>> restored_sampler = pickle.loads(serialized)

Discrete action sampler with custom distribution:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ActionSampler

class WeightedDiscreteActionSampler(ActionSampler):
    def __init__(self, actions, probabilities=None):
        self.actions = actions
        # Use uniform probabilities if none provided
        if probabilities is None:
            self.probabilities = np.ones(len(actions)) / len(actions)
        else:
            self.probabilities = np.array(probabilities)
            self.probabilities /= np.sum(self.probabilities)  # Normalize

    def sample(self, belief_node: Optional[BeliefNode] = None):
        return np.random.choice(self.actions, p=self.probabilities)

# Prefer certain actions over others
actions = ["up", "down", "left", "right", "stay"]
probs = [0.2, 0.2, 0.2, 0.2, 0.2]  # Uniform
sampler = WeightedDiscreteActionSampler(actions, probs)

Belief-informed action sampler:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ActionSampler

class AdaptiveActionSampler(ActionSampler):
    def __init__(self, base_actions, exploration_noise=0.1):
        self.base_actions = base_actions
        self.exploration_noise = exploration_noise

    def sample(self, belief_node: Optional[BeliefNode] = None):
        if belief_node is not None and belief_node.visit_count > 10:
            # Use belief state to inform sampling
            best_action = self._get_best_action_from_belief(belief_node)
            # Add exploration noise
            noise = np.random.normal(0, self.exploration_noise, len(best_action))
            return best_action + noise
        else:
            # Random exploration for new nodes
            return np.random.choice(self.base_actions)

    def _get_best_action_from_belief(self, belief_node):
        # Simplified: return action from best child
        if belief_node.children:
            best_child = max(belief_node.children, key=lambda x: x.q_value)
            return best_child.action
        return np.random.choice(self.base_actions)

sampler = AdaptiveActionSampler([0, 1, 2, 3], exploration_noise=0.05)

Multi-modal action sampler for hybrid control:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ActionSampler

class MultiModalActionSampler(ActionSampler):
    def __init__(self, discrete_actions, continuous_bounds, mode_prob=0.5):
        self.discrete_actions = discrete_actions
        self.continuous_bounds = continuous_bounds
        self.mode_prob = mode_prob  # Probability of discrete vs continuous

    def sample(self, belief_node=None):
        if np.random.random() < self.mode_prob:
            # Sample discrete action
            return {"type": "discrete", "action": np.random.choice(self.discrete_actions)}
        else:
            # Sample continuous action
            low, high = self.continuous_bounds
            continuous_action = np.random.uniform(low, high, size=2)
            return {"type": "continuous", "action": continuous_action}

# For environments with both discrete and continuous actions
discrete_acts = ["stop", "emergency_brake", "lane_change"]
continuous_bounds = (-5.0, 5.0)  # Steering/acceleration range
sampler = MultiModalActionSampler(discrete_acts, continuous_bounds)

Goal-directed action sampler:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ActionSampler

class GoalDirectedActionSampler(ActionSampler):
    def __init__(self, goal_position, action_magnitude=1.0, goal_bias=0.7):
        self.goal_position = np.array(goal_position)
        self.action_magnitude = action_magnitude
        self.goal_bias = goal_bias

    def sample(self, belief_node=None):
        if np.random.random() < self.goal_bias and belief_node is not None:
            # Sample action towards goal based on current belief
            current_position = self._estimate_position(belief_node)
            direction = self.goal_position - current_position
            if np.linalg.norm(direction) > 0:
                direction = direction / np.linalg.norm(direction)
                return direction * self.action_magnitude

        # Random exploration
        angle = np.random.uniform(0, 2 * np.pi)
        return self.action_magnitude * np.array([np.cos(angle), np.sin(angle)])

    def _estimate_position(self, belief_node):
        # Simplified: use mean of particles in belief
        if hasattr(belief_node.belief, 'particles'):
            positions = [p[:2] for p in belief_node.belief.particles]  # First 2D as position
            return np.mean(positions, axis=0)
        return np.array([0.0, 0.0])

# Navigation towards specific goal
goal = [10.0, 5.0]
sampler = GoalDirectedActionSampler(goal, action_magnitude=2.0, goal_bias=0.8)
abstractmethod sample(belief_node=None)[source]

Sample a new action for progressive widening.

Parameters:

belief_node (Optional[BeliefNode]) – Optional belief node context for informed sampling

Return type:

Any

Returns:

A sampled action compatible with the environment’s action space

POMDPPlanners.planners.planners_utils.dpw.action_progressive_widening(belief_node, alpha_a, action_sampler, exploration_constant, k_a, min_visit_count_per_action=1)[source]

Select or add action using progressive widening strategy.

Progressive widening gradually expands the action space based on visit counts. New actions are added when ⌊n^α_a⌋ > ⌊(n-1)^α_a⌋, where n is the visit count. Otherwise, existing actions are selected using UCB1.

The progressive widening mechanism balances exploration and exploitation by: 1. Initially adding new actions frequently (exploration phase) 2. Gradually reducing the rate of new actions as visit count increases 3. Eventually relying primarily on UCB1 selection (exploitation phase)

Parameters:
  • belief_node (BeliefNode) – Current belief node to select action from

  • alpha_a (float) – Progressive widening exponent (0 < alpha_a ≤ 1). Lower values create fewer actions, higher values create more actions.

  • action_sampler (ActionSampler) – Action sampler for generating new actions

  • exploration_constant (float) – UCB1 exploration constant for existing actions

  • k_a (float)

  • min_visit_count_per_action (int)

Return type:

ActionNode

Returns:

Selected or newly created action node

Examples

Basic usage with continuous action sampler:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import (
    ActionSampler, action_progressive_widening
)
from POMDPPlanners.core.tree import BeliefNode
from POMDPPlanners.core.belief import WeightedParticleBelief

# Create action sampler
class SimpleActionSampler(ActionSampler):
    def sample(self, belief_node=None):
        return np.random.uniform(-1, 1, size=2)

# Create belief node
particles = [[0.0, 0.0], [1.0, 1.0]]
log_weights = np.log(np.array([0.5, 0.5]))
belief = WeightedParticleBelief(particles, log_weights)
belief_node = BeliefNode(belief=belief)

# Progressive widening
action_sampler = SimpleActionSampler()
action_node = action_progressive_widening(
    belief_node=belief_node,
    alpha_a=0.5,  # Moderate exploration
    action_sampler=action_sampler,
    exploration_constant=1.0
)

Comparing different alpha_a values:

# Conservative exploration (fewer new actions)
conservative_action = action_progressive_widening(
    belief_node=belief_node,
    alpha_a=0.25,  # Low alpha = fewer actions
    action_sampler=action_sampler,
    exploration_constant=1.0
)

# Aggressive exploration (more new actions)
aggressive_action = action_progressive_widening(
    belief_node=belief_node,
    alpha_a=0.75,  # High alpha = more actions
    action_sampler=action_sampler,
    exploration_constant=1.0
)

Progressive widening in a loop (simulating MCTS):

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ActionSampler, action_progressive_widening
from POMDPPlanners.core.tree import BeliefNode, ActionNode
from POMDPPlanners.core.belief import WeightedParticleBelief

# Setup
class DiscreteActionSampler(ActionSampler):
    def __init__(self, actions):
        self.actions = actions

    def sample(self, belief_node=None):
        return np.random.choice(self.actions)

particles = [[0], [1], [2]]
log_weights = np.log(np.array([1/3, 1/3, 1/3]))
belief = WeightedParticleBelief(particles, log_weights)
root_node = BeliefNode(belief=belief)

sampler = DiscreteActionSampler(['up', 'down', 'left', 'right'])

# Simulate multiple selections
for i in range(10):
    root_node.visit_count = i  # Simulate increasing visits
    action_node = action_progressive_widening(
        belief_node=root_node,
        alpha_a=0.5,
        action_sampler=sampler,
        exploration_constant=1.41  # sqrt(2)
    )
    print(f"Visit {i}: {len(root_node.children)} actions, selected {action_node.action}")

Tuning progressive widening parameters:

# Effect of alpha_a on action creation
visit_counts = range(1, 21)
alpha_values = [0.25, 0.5, 0.75, 1.0]

for alpha in alpha_values:
    action_counts = []
    for n in visit_counts:
        # Calculate when new actions would be created
        should_create = floor(n ** alpha) > floor((n-1) ** alpha) if n > 0 else True
        action_counts.append(1 if should_create else 0)

    total_new_actions = sum(action_counts)
    print(f"Alpha {alpha}: {total_new_actions} new actions in 20 visits")
POMDPPlanners.planners.planners_utils.dpw.action_progressive_widening_arena(tree, belief_id, alpha_a, action_sampler, exploration_constant, k_a, min_visit_count_per_action=1, environment=None)[source]

Arena variant of action_progressive_widening().

Returns the action-node ID (an int) selected by progressive widening + UCB1 from belief node belief_id in tree. Mirrors the semantics of the legacy helper but operates on the column-store tree.

When environment is supplied, environment.hash_action is used to index action children in O(1). When omitted, falls back to the legacy linear-scan path for backwards compatibility.

Return type:

int

Parameters:
POMDPPlanners.planners.planners_utils.dpw.ucb1_exploration(belief_node, exploration_constant)[source]

Select action from existing children using UCB1 criterion.

Uses Upper Confidence Bounds (UCB1) to balance exploration and exploitation: UCB1(a) = Q(a) + c * sqrt(log(N) / N(a)) where Q(a) is the average reward, N is parent visits, N(a) is action visits, and c is the exploration constant.

The UCB1 algorithm provides theoretical guarantees for multi-armed bandit problems and is widely used in Monte Carlo Tree Search algorithms. It automatically balances exploitation of good actions (high Q-values) with exploration of uncertain actions (low visit counts).

Parameters:
  • belief_node (BeliefNode) – Belief node with existing action children

  • exploration_constant (float) – Controls exploration vs exploitation trade-off. Higher values favor exploration, lower values favor exploitation. Common values: √2 ≈ 1.41 (theoretical optimum), 0.5-2.0 (practical range)

Return type:

ActionNode

Returns:

Action node with highest UCB1 value

Examples

Basic UCB1 action selection:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ucb1_exploration
from POMDPPlanners.core.tree import BeliefNode, ActionNode
from POMDPPlanners.core.belief import WeightedParticleBelief

# Create belief node with action children
particles = [[0.0], [1.0]]
log_weights = np.log(np.array([0.5, 0.5]))
belief = WeightedParticleBelief(particles, log_weights)
belief_node = BeliefNode(belief=belief)
belief_node.visit_count = 100

# Add action nodes with different Q-values and visit counts
actions_data = [
    {"action": "up", "q_value": 0.8, "visits": 30},
    {"action": "down", "q_value": 0.6, "visits": 20},
    {"action": "left", "q_value": 0.9, "visits": 40},
    {"action": "right", "q_value": 0.4, "visits": 10}
]

for data in actions_data:
    action_node = ActionNode(action=data["action"], parent=belief_node)
    action_node.q_value = data["q_value"]
    action_node.visit_count = data["visits"]

# Select action using UCB1
selected_action = ucb1_exploration(
    belief_node=belief_node,
    exploration_constant=1.41  # sqrt(2)
)
print(f"Selected action: {selected_action.action}")

Comparing exploration constants:

# Low exploration (favor exploitation)
conservative_action = ucb1_exploration(
    belief_node=belief_node,
    exploration_constant=0.1
)

# High exploration (favor exploration)
exploratory_action = ucb1_exploration(
    belief_node=belief_node,
    exploration_constant=3.0
)

# Balanced approach (theoretical optimum)
balanced_action = ucb1_exploration(
    belief_node=belief_node,
    exploration_constant=1.41  # sqrt(2)
)

Manual UCB1 calculation and verification:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ucb1_exploration

# Calculate UCB1 values manually
exploration_constant = 1.0
ucb1_values = []

for child in belief_node.children:
    exploration_term = exploration_constant * np.sqrt(
        np.log(belief_node.visit_count) / child.visit_count
    )
    ucb1 = child.q_value + exploration_term
    ucb1_values.append(ucb1)
    print(f"Action {child.action}: Q={child.q_value:.2f}, "
          f"exploration={exploration_term:.3f}, UCB1={ucb1:.3f}")

# Verify our function selects the highest UCB1
expected_best_idx = np.argmax(ucb1_values)
selected_action = ucb1_exploration(belief_node, exploration_constant)
actual_best_idx = belief_node.children.index(selected_action)

assert expected_best_idx == actual_best_idx, "UCB1 selection mismatch"

UCB1 in dynamic scenarios:

import numpy as np
from POMDPPlanners.planners.planners_utils.dpw import ucb1_exploration

# Simulate how UCB1 selection changes over time
belief_node.visit_count = 1

for round_num in range(1, 11):
    belief_node.visit_count = round_num * 10

    # Select action
    selected = ucb1_exploration(belief_node, exploration_constant=1.41)

    # Update the selected action (simulate learning)
    selected.visit_count += 1
    selected.q_value += (np.random.normal(0.5, 0.1) - selected.q_value) / selected.visit_count

    print(f"Round {round_num}: Selected {selected.action}, "
          f"Q={selected.q_value:.3f}, visits={selected.visit_count}")

Exploration vs exploitation analysis:

# Create scenario with clear best action vs uncertain actions
exploration_constants = [0.1, 0.5, 1.0, 1.41, 2.0, 5.0]
selection_counts = {c: {"up": 0, "down": 0, "left": 0, "right": 0} for c in exploration_constants}

for exploration_c in exploration_constants:
    # Run multiple selections
    for _ in range(100):
        selected = ucb1_exploration(belief_node, exploration_c)
        selection_counts[exploration_c][selected.action] += 1

    print(f"Exploration constant {exploration_c}:")
    for action, count in selection_counts[exploration_c].items():
        print(f"  {action}: {count}% selections")

UCB1 with confidence intervals:

import math
from POMDPPlanners.planners.planners_utils.dpw import ucb1_exploration

# Calculate confidence intervals for each action
exploration_constant = 1.41
confidence_level = 0.95

for child in belief_node.children:
    # UCB1 upper confidence bound
    confidence_radius = exploration_constant * math.sqrt(
        math.log(belief_node.visit_count) / child.visit_count
    )

    lower_bound = child.q_value - confidence_radius
    upper_bound = child.q_value + confidence_radius

    print(f"Action {child.action}: "
          f"Q={child.q_value:.3f} ± {confidence_radius:.3f} "
          f"[{lower_bound:.3f}, {upper_bound:.3f}]")

# The selected action has the highest upper bound
selected = ucb1_exploration(belief_node, exploration_constant)
print(f"Selected: {selected.action} (highest upper confidence bound)")
POMDPPlanners.planners.planners_utils.dpw.ucb1_exploration_arena(tree, belief_id, exploration_constant)[source]

Arena variant of ucb1_exploration(). Returns the action-node ID.

Return type:

int

Parameters:

POMDPPlanners.planners.planners_utils.path_simulations_policy module

class POMDPPlanners.planners.planners_utils.path_simulations_policy.DoubleProgressiveWideningMCTSPolicy(environment, discount_factor, depth, name, action_sampler, k_a, alpha_a, k_o, alpha_o, exploration_constant, min_visit_count_per_action=1, time_out_in_seconds=None, n_simulations=None, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: PathSimulationPolicy

Abstract base class for MCTS planners using double progressive widening.

This base class provides common initialization, parameter validation, and attributes for MCTS planners that use double progressive widening (both action and observation progressive widening). Subclasses implement their own simulation strategies while sharing common parameters and validation logic.

Progressive Widening Overview: Double progressive widening controls tree growth by limiting how many actions and observations are added to the tree based on visit counts: - Action widening: New actions added when ⌊k_a * n^α_a⌋ increases - Observation widening: Max observations limited by ⌊k_o * n^α_o⌋

Common Progressive Widening Parameters: - k_a, alpha_a: Control action progressive widening - k_o, alpha_o: Control observation progressive widening - exploration_constant: UCB1 exploration parameter

Parameters:
depth

Maximum search depth for tree expansion

exploration_constant

UCB1 exploration parameter (c in UCB1 formula)

action_sampler

Action sampling strategy for progressive widening

k_o

Observation progressive widening coefficient (k_o > 0)

k_a

Action progressive widening coefficient (k_a > 0)

alpha_o

Observation progressive widening exponent (0 < α_o ≤ 1)

alpha_a

Action progressive widening exponent (0 < α_a ≤ 1)

Subclasses:
  • POMCP_DPW: Uses unweighted particle beliefs with double progressive widening

  • POMCPOW: Uses weighted particle beliefs with double progressive widening

  • PFT_DPW: Uses progressive function transfer with custom simulation strategy

Note

This is an abstract base class and cannot be instantiated directly. Subclasses must implement the _simulate_path method.

classmethod get_space_info()[source]

Get information about action and observation spaces.

Default implementation returns MIXED space types, which is appropriate for most progressive widening MCTS planners that support both discrete and continuous action spaces through the action sampler interface.

Subclasses can override this method to specify different space requirements (e.g., PFT_DPW specifies CONTINUOUS action space).

Return type:

PolicySpaceInfo

Returns:

PolicySpaceInfo with MIXED space types for both actions and observations

class POMDPPlanners.planners.planners_utils.path_simulations_policy.PathSimulationPolicy(environment, discount_factor, name, n_simulations, time_out_in_seconds, action_sampler=None, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: Policy

Abstract base class for Monte Carlo Tree Search algorithms in POMDP planning.

This class provides a common framework for MCTS-based POMDP planners that build search trees through path simulations. It handles the core tree construction loop and provides hooks for algorithm-specific simulation strategies.

The class supports two termination criteria: 1. Simulation count: Run a fixed number of MCTS simulations 2. Time limit: Run simulations for a specified time duration

Key Components: - Tree construction with configurable termination criteria - Automatic tree metrics collection for analysis - Action selection from the constructed search tree - Abstract simulation interface for algorithm specialization

Subclass Responsibilities: Concrete implementations must provide the _simulate_path method that defines how individual MCTS simulations are performed, including: - Node expansion strategies - Action selection during tree traversal - Value estimation and backpropagation

Parameters:
environment

The POMDP environment for planning

discount_factor

Discount factor for future rewards (0 < γ ≤ 1)

n_simulations

Number of MCTS simulations to run (mutually exclusive with timeout)

time_out_in_seconds

Time limit for planning in seconds (mutually exclusive with n_simulations)

name

Identifier for the policy instance

Algorithm Integration: This base class is used by several MCTS algorithms in the framework: - POMCP: Uses UCB1 for action selection with particle filtering - PFT-DPW: Implements progressive widening for continuous action spaces - Sparse-PFT: Combines sparse sampling with progressive widening

The common interface allows easy comparison and benchmarking of different MCTS variants while sharing the core tree construction infrastructure.

action(belief)[source]

Select action(s) based on the current belief state.

This is the core method that implements the policy’s decision-making logic. It takes a belief state and returns the selected action(s) along with execution information and performance metrics.

Parameters:

belief (Belief) – Current belief state representing uncertainty over states

Returns:

  • List of selected actions (typically single action, but supports multiple)

  • PolicyRunData with execution metrics and performance information

Return type:

Tuple[List[Any], PolicyRunData]

Note

Subclasses must implement this method with their specific planning or decision-making algorithm.

classmethod get_info_variable_names()[source]

Get names of tree metric info variables produced by path simulation policies.

Return type:

List[str]

Returns:

List of metric names from tree statistics

class POMDPPlanners.planners.planners_utils.path_simulations_policy.PathSimulationPolicyCostSetting(environment, discount_factor, name, action_sampler=None, n_simulations=None, time_out_in_seconds=None, log_path=None, debug=False)[source]

Bases: PathSimulationPolicy

Parameters:
action(belief)[source]

Select action(s) based on the current belief state.

This is the core method that implements the policy’s decision-making logic. It takes a belief state and returns the selected action(s) along with execution information and performance metrics.

Parameters:

belief (Belief) – Current belief state representing uncertainty over states

Returns:

  • List of selected actions (typically single action, but supports multiple)

  • PolicyRunData with execution metrics and performance information

Return type:

Tuple[List[Any], PolicyRunData]

Note

Subclasses must implement this method with their specific planning or decision-making algorithm.

POMDPPlanners.planners.planners_utils.path_simulations_policy_arena module

Arena-tree variants of the MCTS path-simulation base classes.

Mirrors POMDPPlanners.planners.planners_utils.path_simulations_policy.PathSimulationPolicy and DoubleProgressiveWideningMCTSPolicy but operates on the column-store POMDPPlanners.core.tree.arena.Tree (integer node IDs) instead of the anytree-based BeliefNode/ActionNode objects.

Subclasses implement _simulate_path(tree, belief_id, depth) against the arena Tree API. The base class handles the simulation loop, root construction, action selection, and tree-metrics collection.

class POMDPPlanners.planners.planners_utils.path_simulations_policy_arena.ArenaDoubleProgressiveWideningMCTSPolicy(environment, discount_factor, depth, name, action_sampler, k_a, alpha_a, k_o, alpha_o, exploration_constant, min_visit_count_per_action=1, time_out_in_seconds=None, n_simulations=None, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaPathSimulationPolicy

Arena-tree variant of DoubleProgressiveWideningMCTSPolicy.

Adds the standard double-progressive-widening configuration (k_a, alpha_a, k_o, alpha_o, exploration_constant, min_visit_count_per_action) on top of ArenaPathSimulationPolicy. Same validation as the legacy class; subclasses (POMCPOW, PFT-DPW, etc.) implement _simulate_path() using the arena Tree API.

Parameters:
action_sampler: ActionSampler
classmethod get_space_info()[source]

Get space type requirements for this policy class.

This class method specifies what types of action and observation spaces this policy implementation can handle, enabling compatibility checking with environments.

Return type:

PolicySpaceInfo

Returns:

PolicySpaceInfo specifying required action and observation space types

Note

Subclasses must implement this method to declare their space compatibility. This is used for validation when pairing policies with environments.

class POMDPPlanners.planners.planners_utils.path_simulations_policy_arena.ArenaPathSimulationPolicy(environment, discount_factor, name, n_simulations, time_out_in_seconds, action_sampler=None, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: Policy

MCTS base class operating on the arena Tree + integer node IDs.

Mirrors PathSimulationPolicy’s shape and validation but uses (tree, belief_id) instead of node objects. Subclasses implement _simulate_path() against the arena Tree API.

Parameters:
action(belief)[source]

Select action(s) based on the current belief state.

This is the core method that implements the policy’s decision-making logic. It takes a belief state and returns the selected action(s) along with execution information and performance metrics.

Parameters:

belief (Belief) – Current belief state representing uncertainty over states

Returns:

  • List of selected actions (typically single action, but supports multiple)

  • PolicyRunData with execution metrics and performance information

Return type:

Tuple[List[Any], PolicyRunData]

Note

Subclasses must implement this method with their specific planning or decision-making algorithm.

classmethod get_info_variable_names()[source]

Get names of policy info variables that this policy produces.

This class method returns the names of metrics and performance data that the policy tracks during execution via PolicyInfoVariable objects. It enables users to discover what metrics are available for hyperparameter optimization before running simulations.

Return type:

List[str]

Returns:

List of info variable names that this policy produces during action selection

Note

Subclasses must implement this method to declare what metrics they track. Use an Enum to ensure consistency between the names returned here and the names used when creating PolicyInfoVariable objects in the action() method.

class POMDPPlanners.planners.planners_utils.path_simulations_policy_arena.ArenaPathSimulationPolicyCostSetting(environment, discount_factor, name, action_sampler=None, n_simulations=None, time_out_in_seconds=None, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaPathSimulationPolicy

Arena-tree variant of PathSimulationPolicyCostSetting.

Same shape as ArenaPathSimulationPolicy but selects the optimal action by minimum q_value (cost setting) instead of maximum (reward setting). Used by the ICVaR planners.

Parameters:
action(belief)[source]

Select action(s) based on the current belief state.

This is the core method that implements the policy’s decision-making logic. It takes a belief state and returns the selected action(s) along with execution information and performance metrics.

Parameters:

belief (Belief) – Current belief state representing uncertainty over states

Returns:

  • List of selected actions (typically single action, but supports multiple)

  • PolicyRunData with execution metrics and performance information

Return type:

Tuple[List[Any], PolicyRunData]

Note

Subclasses must implement this method with their specific planning or decision-making algorithm.

POMDPPlanners.planners.planners_utils.rollout module

POMDPPlanners.planners.planners_utils.rollout.cost_aware_random_rollout(state, depth, action_sampler, environment, discount_factor, max_depth, n_constraints)[source]

Random rollout that returns (discounted_reward, discounted_cost_vector).

Used by constrained planners (CPOMCPOW, CPFT_DPW) at leaf expansion so constraint cost is accumulated across the rollout trajectory rather than approximated as zero. The cost vector is read from ConstrainedEnvironment.constraint_cost() at every transition.

Iterative (no recursion) so deep rollouts don’t hit Python’s stack limit. Does NOT dispatch to native rollout kernels — those compute reward only and have no constraint-cost channel.

Parameters:
  • state (Any) – Starting state.

  • depth (int) – Current depth (rollout runs from depth to max_depth).

  • action_sampler (ActionSampler) – Sampler for rollout actions.

  • environment (ConstrainedEnvironment) – Constrained POMDP env. Must expose constraint_cost.

  • discount_factor (float) – Per-step discount.

  • max_depth (int) – Exclusive upper bound on rollout depth.

  • n_constraints (int) – Length of the constraint-cost vector. Used to size the zero return and validate per-step constraint_cost shape.

Return type:

Tuple[float, ndarray]

Returns:

(discounted_reward, discounted_cost_vector). The cost vector has shape (n_constraints,).

Raises:

ValueError – If constraint_cost returns a vector whose length does not match n_constraints.

POMDPPlanners.planners.planners_utils.rollout.python_random_rollout(state, depth, action_sampler, environment, discount_factor, max_depth=10)[source]

Recursive Python rollout that bypasses the env-level native dispatch.

Used as the fallback path inside env-specific simulate_random_rollout overrides (e.g. when the native kernel cannot handle a particular reward model or parameter configuration).

Return type:

float

Parameters:
POMDPPlanners.planners.planners_utils.rollout.random_rollout_action_sampler(state, depth, action_sampler, environment, discount_factor, max_depth=10)[source]

Perform random rollout to estimate value from leaf node.

Rollout policy samples random actions using the action_sampler until reaching maximum depth or terminal state. This provides value estimates for leaf nodes in the search tree during Monte Carlo Tree Search.

The rollout uses a random policy (via action_sampler) to quickly estimate the value of a state without expensive planning. This is a key component of MCTS algorithms where accurate value estimation is traded off against computational efficiency.

Parameters:
  • state (Any) – Current state to rollout from

  • depth (int) – Current depth in rollout (starts at 0)

  • action_sampler (ActionSampler) – Action sampler for selecting rollout actions

  • environment (Environment) – POMDP environment to simulate in

  • discount_factor (float) – Discount factor for future rewards (0 < γ ≤ 1)

  • max_depth (int) – Maximum rollout depth to prevent infinite loops

Return type:

float

Returns:

Total discounted return from rollout simulation

Examples

>>> import numpy as np
>>> np.random.seed(42)  # For reproducible results
>>> from POMDPPlanners.planners.planners_utils.rollout import random_rollout_action_sampler
>>> from POMDPPlanners.planners.planners_utils.dpw import ActionSampler
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>>
>>> # Simple action sampler for Tiger POMDP
>>> class TigerActionSampler(ActionSampler):
...     def sample(self, belief_node=None):
...         return np.random.choice(["listen", "open_left", "open_right"])
>>>
>>> # Create environment and sampler
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> action_sampler = TigerActionSampler()
>>>
>>> # Perform rollout from initial state
>>> initial_state = "tiger_left"
>>> rollout_value = random_rollout_action_sampler(
...     state=initial_state,
...     depth=0,
...     action_sampler=action_sampler,
...     environment=tiger,
...     discount_factor=0.95,
...     max_depth=10
... )