POMDPPlanners.planners package

Policy factory module for creating POMDP policies.

class POMDPPlanners.planners.BetaZero(environment, discount_factor, depth, name, action_sampler, k_a=1.0, alpha_a=0.5, k_o=1.0, alpha_o=0.5, exploration_constant=1.0, time_out_in_seconds=None, n_simulations=None, min_visit_count_per_action=1, network=None, belief_representation=None, state_dim=None, z_q=1.0, z_n=1.0, temperature=1.0, n_buffer=1, training_batch_size=256, training_epochs=10, learning_rate=0.001, weight_decay=0.0001, hidden_sizes=(128, 128), track_gradients=False, normalize_inputs=True, normalize_values=True, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaDoubleProgressiveWideningMCTSPolicy, TrainablePolicy

BetaZero: Neural MCTS for POMDPs.

Extends DoubleProgressiveWideningMCTSPolicy with three key innovations from the BetaZero paper:

  1. PUCT selection: Replaces UCB1 using learned policy priors.

  2. Neural value estimation: Replaces random rollouts at leaf nodes.

  3. Policy iteration via ``fit()``: Collects episodes, computes Q-weighted policy targets, and trains the network.

The planner has two modes: - Online planning via action(belief): builds an MCTS tree with

PUCT and network value estimates.

  • Offline training via fit(): alternates data collection and network training.

Parameters:
network

Dual-head neural network for policy and value prediction.

belief_representation

Belief → feature-vector mapping φ(b).

z_q

Exponent for Q-value term in policy target.

z_n

Exponent for visit-count term in policy target.

temperature

Temperature τ for sharpening/smoothing policy target.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> from POMDPPlanners.utils.action_samplers import DiscreteActionSampler
>>> from POMDPPlanners.planners.mcts_planners.beta_zero.beta_zero import BetaZero
>>>
>>> env = TigerPOMDP(discount_factor=0.95)
>>> sampler = DiscreteActionSampler(env.get_actions())
>>> planner = BetaZero(
...     environment=env,
...     discount_factor=0.95,
...     depth=3,
...     name="BetaZero_Tiger",
...     action_sampler=sampler,
...     n_simulations=20,
...     state_dim=1,
... )
>>> belief = get_initial_belief(env, n_particles=10)
>>> actions, run_data = planner.action(belief)
>>> actions[0] in env.get_actions()
True
action(belief)[source]

Select an action via MCTS with PUCT and network value estimates.

If data collection is active (during fit()), also stores a pending training example from the tree root.

Return type:

Tuple[List[Any], PolicyRunData]

Parameters:

belief (Belief)

begin_collecting()[source]

Signal the start of a data-collection phase.

Return type:

None

buffer_size()[source]

Return the number of examples currently in the replay buffer.

Return type:

int

collect_episodes_batched(initial_belief_fn, n_episodes, episode_length)[source]

Collect training data using fast batched (network-only) rollouts.

Parameters:

  • initial_belief_fn (Callable[[], Belief]) – Callable returning a fresh initial belief.

  • n_episodes (int) – Number of episodes to collect.

  • episode_length (int) – Maximum steps per episode.

Return type:

None

end_collecting()[source]

Signal the end of a data-collection phase.

Return type:

None

finalize_episode(history)[source]

Process a completed episode into the replay buffer.

Parameters:

history – The History returned by the episode runner.

Return type:

None

get_metric_keys()[source]

Return the loss-metric key names produced by train_step().

Return type:

List[str]

get_network()[source]

Return the underlying trainable network, or None if not applicable.

Override in concrete policies to enable weight-histogram logging in TensorBoardCallback.

Return type:

AbstractBetaZeroNetwork

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.

load_normalization_stats(filepath)[source]

Restore normalization statistics from a saved directory.

Should be called after network.load_weights() when loading a checkpoint that was saved with normalisation enabled.

Parameters:

filepath – Directory previously returned by save().

Return type:

None

prepare_episode()[source]

Reset per-episode scratch state before an episode begins.

Return type:

None

save(filepath=None)[source]

Save policy config and network weights to a directory.

Parameters:

filepath – Directory path. If None, uses default.

Return type:

Path

Returns:

Directory where the policy was saved.

train_step()[source]

Train the network on the current replay buffer.

Return type:

Dict[str, List[float]]

Returns:

Per-key lists of loss values produced during training.

class POMDPPlanners.planners.ConstrainedZero(environment, discount_factor, depth, name, action_sampler, failure_fn, delta_0=0.01, eta=1e-05, delta_compounding=1.0, k_a=1.0, alpha_a=0.5, k_o=1.0, alpha_o=0.5, exploration_constant=1.0, time_out_in_seconds=None, n_simulations=None, min_visit_count_per_action=1, network=None, belief_representation=None, state_dim=None, z_q=1.0, z_n=1.0, temperature=1.0, n_buffer=1, training_batch_size=256, training_epochs=10, learning_rate=0.001, weight_decay=0.0001, hidden_sizes=(128, 128), use_dropout=True, p_dropout=0.2, track_gradients=False, normalize_inputs=True, normalize_values=True, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: BetaZero

ConstrainedZero: Neural MCTS for Chance-Constrained POMDPs.

Extends BetaZero with:

  1. 3-head network: Adds a failure probability head alongside policy and value.

  2. SPUCT selection: Safety-constrained PUCT that masks unsafe actions.

  3. Adaptive Delta (conformal inference): Calibrates the failure threshold during tree search using online conformal inference.

  4. Failure propagation: Tracks failure probability per action node using p = p_immediate + delta_compounding * (1 - p_immediate) * p_next.

  5. Constrained policy targets: Applies safety mask during target computation.

Parameters:
failure_fn

User-provided function state -> bool defining failure.

delta_0

Nominal failure probability threshold.

eta

Learning rate for adaptive Delta calibration.

delta_compounding

Discount factor for failure propagation.

Example

>>> import numpy as np
>>> np.random.seed(42)
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> from POMDPPlanners.utils.action_samplers import DiscreteActionSampler
>>> from POMDPPlanners.planners.mcts_planners.constrained_zero.constrained_zero import ConstrainedZero
>>>
>>> env = TigerPOMDP(discount_factor=0.95)
>>> sampler = DiscreteActionSampler(env.get_actions())
>>> planner = ConstrainedZero(
...     environment=env,
...     discount_factor=0.95,
...     depth=3,
...     name="CZ_Tiger",
...     action_sampler=sampler,
...     n_simulations=20,
...     state_dim=1,
...     failure_fn=lambda s: False,
... )
>>> belief = get_initial_belief(env, n_particles=10)
>>> actions, run_data = planner.action(belief)
>>> actions[0] in env.get_actions()
True
get_metric_keys()[source]

Return the loss-metric key names produced by train_step().

Return type:

List[str]

network: ConstrainedZeroNetwork
class POMDPPlanners.planners.DiscreteActionSequencesPlanner(environment, discount_factor, name, depth, n_return_samples, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: Policy

Open-loop planner for discrete action spaces using exhaustive sequence search.

This planner uses an open-loop strategy to find optimal action sequences by enumerating all possible action sequences up to a specified depth and selecting the sequence with the highest expected return. It’s particularly useful for problems with small action spaces and short planning horizons.

The algorithm works by: 1. Generating all possible action sequences of the specified depth 2. For each sequence, estimating the expected return through Monte Carlo sampling 3. Selecting the sequence with the maximum expected return 4. Returning the first action in the optimal sequence

Open-Loop vs Closed-Loop Planning: - Open-loop: Plans a complete action sequence without considering future observations - Closed-loop: Re-plans at each step based on new observations (like MCTS algorithms)

This approach is computationally intensive (O(|A|^depth)) but provides optimal solutions for the open-loop setting when the action space is manageable.

Parameters:

  • environment (DiscreteActionsEnvironment) – The discrete actions POMDP environment

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

  • name (str) – Identifier for the planner instance

  • depth (int) – Planning horizon (number of actions in sequence)

  • n_return_samples (int) – Number of Monte Carlo samples for return estimation

  • log_path (Optional[Path]) – Optional path for logging planner execution details

  • debug (bool) – Enable debug mode for detailed execution traces

  • use_queue_logger (bool)

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Create environment and planner
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> planner = DiscreteActionSequencesPlanner(
...     environment=tiger,
...     discount_factor=0.95,
...     name="ExamplePlanner",
...     depth=2,
...     n_return_samples=10
... )
>>>
>>> # Basic planner interface usage
>>> planner.name
'ExamplePlanner'
>>>
>>> # Action selection from belief
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> # Planner space information
>>> space_info = DiscreteActionSequencesPlanner.get_space_info()
>>> space_info.action_space.name
'DISCRETE'
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.

estimate_return(action_sequence, belief)[source]
Return type:

float

Parameters:
classmethod get_info_variable_names()[source]

Get names of policy info variables.

Discrete action sequences planner does not produce any info variables.

Return type:

List[str]

Returns:

Empty list as this planner produces no info variables

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.

search(belief)[source]
Return type:

Any

Parameters:

belief (Belief)

class POMDPPlanners.planners.ICVaRSparseSampling(environment, branching_factor, depth, alpha, name='ICVaRSparseSampling')[source]

Bases: SparseSamplingDiscreteActionsPlanner

Risk-sensitive sparse sampling planner using CVaR for value backups.

This planner extends the standard sparse sampling algorithm by replacing the expected value (mean) in Q-value computation with the Conditional Value at Risk (CVaR). CVaR focuses on the worst-alpha fraction of outcomes, making the planner risk-sensitive.

The standard Q-value update uses:

Q = immediate_cost + gamma * mean(child_v_values)

The ICVaR variant replaces this with:

Q = immediate_cost + gamma * CVaR_alpha(child_v_values)

Parameters:
alpha

CVaR confidence level (0 < alpha <= 1). Lower alpha means more risk-sensitive (focuses on worse outcomes). alpha=1.0 recovers the standard expected value.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Create environment and risk-sensitive planner
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> planner = ICVaRSparseSampling(
...     environment=tiger,
...     branching_factor=2,
...     depth=2,
...     alpha=0.3,
...     name="ICVaRPlanner"
... )
>>>
>>> # Basic planner interface usage
>>> planner.name
'ICVaRPlanner'
>>> planner.alpha
0.3
>>>
>>> # Action selection from belief
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> # Planner space information
>>> space_info = ICVaRSparseSampling.get_space_info()
>>> space_info.action_space.name
'DISCRETE'
classmethod get_info_variable_names()[source]

Get names of policy info variables.

Return type:

List[str]

Returns:

Empty list as this planner produces no info variables.

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.ICVaR_PFT_DPW(environment, name, depth, action_sampler, discount_factor=0.95, time_out_in_seconds=None, n_simulations=None, alpha=0.1, delta=0.1, belief_child_num=5, min_immediate_cost=0.0, max_immediate_cost=1.0, min_visit_count_per_action=1, exploration_constant=1.0, k_a=1.0, alpha_a=0.5, k_o=1.0, alpha_o=0.5, visit_count_penalty=0.0, reserve_capacity=0)[source]

Bases: ArenaPathSimulationPolicyCostSetting

ICVaR PFT-DPW operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Parameters:
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.

is_terminal_belief(belief)[source]

Return True if all particles in belief are terminal states.

Return type:

bool

Parameters:

belief (Belief)

update_nodes(tree, belief_id, action_id)[source]
Return type:

None

Parameters:
class POMDPPlanners.planners.ICVaR_POMCPOW(environment, discount_factor, depth, exploration_constant, k_o, k_a, alpha_o, alpha_a, min_immediate_cost, max_immediate_cost, min_visit_count_per_action, delta, name, action_sampler, time_out_in_seconds=None, n_simulations=None, alpha=0.05, min_samples_per_node=10, reserve_capacity=0, log_path=None, debug=False, visit_count_penalty=0.0)[source]

Bases: ArenaPathSimulationPolicyCostSetting

ICVaR POMCPOW operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Parameters:
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.PFT_DPW(environment, discount_factor, depth, name, action_sampler, k_a=1.0, alpha_a=0.5, k_o=1.0, alpha_o=0.5, exploration_constant=1.0, time_out_in_seconds=None, n_simulations=None, min_visit_count_per_action=1, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaDoubleProgressiveWideningMCTSPolicy

PFT-DPW operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> from POMDPPlanners.utils.action_samplers import DiscreteActionSampler
>>> np.random.seed(42)
>>>
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> action_sampler = DiscreteActionSampler(tiger.get_actions())
>>> planner = PFT_DPW(
...     environment=tiger,
...     discount_factor=0.95,
...     depth=5,
...     name="ExamplePlanner",
...     action_sampler=action_sampler,
...     k_a=2.0,
...     alpha_a=0.5,
...     n_simulations=10
... )
>>> planner.name
'ExamplePlanner'
>>>
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> space_info = PFT_DPW.get_space_info()
>>> space_info.action_space.name
'MIXED'
Parameters:
sample_existing_belief_node(tree, belief_id, action_id)[source]
Return type:

Tuple[int, float]

Parameters:
class POMDPPlanners.planners.POMCP(environment, discount_factor, depth, exploration_constant, name, time_out_in_seconds=None, n_simulations=None, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaPathSimulationPolicy

POMCP operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> np.random.seed(42)
>>>
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> planner = POMCP(
...     environment=tiger,
...     discount_factor=0.95,
...     depth=5,
...     exploration_constant=1.0,
...     name="ExamplePlanner",
...     n_simulations=10
... )
>>> planner.name
'ExamplePlanner'
>>>
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> space_info = POMCP.get_space_info()
>>> space_info.action_space.name
'DISCRETE'
Parameters:
get_explored_action_node(tree, belief_id)[source]

Pick an action child via UCB1; if any child has zero visits, pick uniformly from those.

Return type:

int

Parameters:
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.

random_rollout(state, depth)[source]
Return type:

float

Parameters:
update_nodes(tree, belief_id, action_id, return_sample, state)[source]
Return type:

None

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

Bases: ArenaDoubleProgressiveWideningMCTSPolicy

POMCPOW operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> from POMDPPlanners.utils.action_samplers import DiscreteActionSampler
>>> np.random.seed(42)
>>>
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> action_sampler = DiscreteActionSampler(tiger.get_actions())
>>> planner = POMCPOW(
...     environment=tiger,
...     discount_factor=0.95,
...     depth=5,
...     exploration_constant=1.0,
...     k_o=3.0,
...     k_a=3.0,
...     alpha_o=0.5,
...     alpha_a=0.5,
...     action_sampler=action_sampler,
...     n_simulations=10,
...     name="ExamplePlanner"
... )
>>> planner.name
'ExamplePlanner'
>>>
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
Parameters:
class POMDPPlanners.planners.POMCP_DPW(environment, discount_factor, depth, exploration_constant, k_o, k_a, alpha_o, alpha_a, name, action_sampler, time_out_in_seconds=None, n_simulations=None, min_visit_count_per_action=1, reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaDoubleProgressiveWideningMCTSPolicy

POMCP_DPW operating on the arena Tree + integer node IDs.

See module docstring for algorithm details and reference.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> from POMDPPlanners.utils.action_samplers import DiscreteActionSampler
>>> np.random.seed(42)
>>>
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> action_sampler = DiscreteActionSampler(tiger.get_actions())
>>> planner = POMCP_DPW(
...     environment=tiger,
...     discount_factor=0.95,
...     depth=5,
...     exploration_constant=1.0,
...     k_o=3.0,
...     k_a=3.0,
...     alpha_o=0.5,
...     alpha_a=0.5,
...     action_sampler=action_sampler,
...     n_simulations=10,
...     name="ExamplePlanner"
... )
>>> planner.name
'ExamplePlanner'
>>>
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> space_info = POMCP_DPW.get_space_info()
>>> space_info.action_space.name
'MIXED'
Parameters:
class POMDPPlanners.planners.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.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.

class POMDPPlanners.planners.SparsePFT(environment, discount_factor, depth, c_ucb, beta_ucb, belief_child_num, time_out_in_seconds=None, n_simulations=None, name='SparsePFT', reserve_capacity=0, log_path=None, debug=False, use_queue_logger=False)[source]

Bases: ArenaPathSimulationPolicy

Sparse-PFT operating on the arena Tree + integer node IDs.

See module docstring for algorithm details.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> np.random.seed(42)
>>>
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> planner = SparsePFT(
...     environment=tiger,
...     discount_factor=0.95,
...     depth=5,
...     c_ucb=1.0,
...     beta_ucb=2.0,
...     belief_child_num=3,
...     n_simulations=10,
...     name="ExamplePlanner"
... )
>>> planner.name
'ExamplePlanner'
>>>
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> space_info = SparsePFT.get_space_info()
>>> space_info.action_space.name
'DISCRETE'
Parameters:
get_explored_action_node(tree, belief_id)[source]
Return type:

int

Parameters:
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.

random_rollout(state, depth)[source]
Return type:

float

Parameters:
update_nodes(tree, belief_id, action_id, return_sample)[source]
Return type:

None

Parameters:
class POMDPPlanners.planners.SparseSamplingDiscreteActionsPlanner(environment, branching_factor, depth, name='SparseSamplingDiscreteActionsPlanner')[source]

Bases: BaseSparseSamplingDiscreteActionsPlanner

Standard implementation of sparse sampling for POMDP planning.

This concrete implementation of sparse sampling uses standard value updates: - Q-values for actions are computed as immediate cost plus discounted future value - V-values for beliefs are computed as the minimum Q-value over actions (cost formulation) - Leaf nodes use only immediate cost estimates

The algorithm provides theoretical guarantees: with probability 1-δ, the computed policy is ε-optimal, where ε decreases with increasing depth and branching factor.

Example

>>> import numpy as np
>>> from POMDPPlanners.environments.tiger_pomdp import TigerPOMDP
>>> from POMDPPlanners.core.belief import get_initial_belief
>>> np.random.seed(42)  # For reproducible results
>>>
>>> # Create environment and planner
>>> tiger = TigerPOMDP(discount_factor=0.95)
>>> planner = SparseSamplingDiscreteActionsPlanner(
...     environment=tiger,
...     branching_factor=2,
...     depth=2,
...     name="ExamplePlanner"
... )
>>>
>>> # Basic planner interface usage
>>> planner.name
'ExamplePlanner'
>>>
>>> # Action selection from belief
>>> initial_belief = get_initial_belief(tiger, n_particles=10)
>>> actions, run_data = planner.action(initial_belief)
>>>
>>> # Planner space information
>>> space_info = SparseSamplingDiscreteActionsPlanner.get_space_info()
>>> space_info.action_space.name
'DISCRETE'
Parameters:

Subpackages