Exhaustive search

An exhaustive search scores every length-H sequence of a declared, finite action set and returns the best one. It also returns a CompletenessCertificate saying every sequence was scored. EFESelector samples a continuous range and reports CORROBORATED. A search decides over its declared set and reports PROVED. The names here are imported from cpomdp.enumeration.

FiniteActionSet dataclass

FiniteActionSet(actions: ArrayLike, *, version: str)

A finite, declared, versioned set of actions to enumerate over.

A distinct type from the continuous grid's (lo, hi, n_candidates) config, so the enumerated (PROVED) and sampled (CORROBORATED) families cannot be mixed up at a call site. Construction-time only (not a pytree).

Parameters:

Name	Type	Description	Default
actions	ArrayLike	the declared actions, shape (A, p) — A actions of dimension p. At least two (a set of one is no choice to search), every entry finite. A repeated action counts toward A, so it is scored twice.	required
version	str	a non-empty tag, so an action added later shows up in the diff.	required

Raises:

Type	Description
ValueError	If actions is not 2-D, holds fewer than two actions or a non-finite entry, or if version is empty.

Source code in src/cpomdp/enumeration.py

def __init__(self, actions: ArrayLike, *, version: str) -> None:
    object.__setattr__(self, "actions", jnp.asarray(actions, dtype=float))
    object.__setattr__(self, "version", version)
    self._validate()

size property

size: int

The action count A (the base of |A|^H).

n_controls property

n_controls: int

The action dimension p, named as the model names it.

Two searches, one interface

The two searches take the same model, set and horizon, and answer the same evaluate. Swapping one for the other changes the memory a plan needs and no line that reads the result. EnumeratedEFESearch holds every policy and every score, and is the one to reach for first. ChunkedEFESearch scores in blocks at a footprint its block sets. Use it past the front-loaded search's budget, which is 2^21 policies unless set.

In plain terms

Give the search a menu of moves and a number of steps. It writes down every way of chaining that many moves from the menu, scores each chain by expected free energy from the current belief, and hands back the lowest. Nothing is sampled. Every chain on the menu is tried, so "this is the best plan on the menu" is a decided fact rather than a likely one, and the certificate records the count that makes it so.

The number of chains is the menu's size raised to the number of steps, so it grows fast. The front-loaded search keeps every chain and every score in memory at once, which is quick and lets you look at all the scores afterwards. Past a budget it refuses at construction instead of running the machine out of memory, and its error says how big the set was and what to use instead. The chunked search works through the chains a block at a time and keeps only the best so far. It returns the same plan, bit for bit, and never holds the whole set.

One convention to know. A plan's cost is reported per plan, as cost_per_plan. What an agent step costs depends on how often the agent plans, so only the selectors below report cost_per_cycle.

ExhaustiveEFESearch

Bases: ABC

An EFE search that scores every length-H sequence of a declared action set.

The kind the drivers take. EnumeratedEFESearch holds every policy and every score at once, and ChunkedEFESearch scores the set a chunk at a time. Both earn the same PROVED certificate. Code that reads only what this class and SearchResult declare runs on either.

It is public as a type, for annotations and isinstance. Subclassing it outside cpomdp is not supported: a subclass records its set and horizon through the private _enumerate, and the PROVED it inherits holds only for a search that does.

n_policies property

n_policies: int

The enumerated policy count |A|^H, attributable work (RFC-001).

model property

model: LinearGaussianModel

The model every policy is scored under, on a graph the backend's own.

On a coupled graph it is the interface model, which leaves out the couplings the score reads off backend.

backend property

backend: GraphBackend | None

The graph backend the policies are scored on, or None on a flat model.

horizon property

horizon: int

The sequence length H.

cost_per_plan abstractmethod property

cost_per_plan: int

Step evaluations one evaluate runs (RFC-001).

warrant property

warrant: Warrant

PROVED: a finite set enumerated in full decides its universal.

evaluate abstractmethod

evaluate(
    belief: Belief, preference: Preference
) -> SearchResult

Score every policy from belief, and return the argmin with its evidence.

Parameters:

Name	Type	Description	Default
belief	Belief	the belief to plan from.	required
preference	Preference	the target and precision the EFE scores against.	required

Source code in src/cpomdp/enumeration.py

@abstractmethod
def evaluate(self, belief: Belief, preference: "Preference") -> SearchResult:
    """Score every policy from ``belief``, and return the argmin with its evidence.

    Args:
        belief: the belief to plan from.
        preference: the target and precision the EFE scores against.
    """

policies_at

policies_at(
    indices: ArrayLike,
) -> Float64[Array, "... H p"]

The policies indices name, decoded from the index rather than looked up.

The way to turn a result's best_index back into a plan, on either search. An index outside 0..|A|^H - 1 names no policy. Under jit or vmap it decodes to a NaN policy, the one a result with no winner carries.

Raises:

Type	Description
ValueError	If a concrete index lies outside the enumeration, including the -1 a result carries when no policy scored finitely. Decoded, it would wrap to a real policy.

Source code in src/cpomdp/enumeration.py

def policies_at(self, indices: ArrayLike) -> Float64[Array, "... H p"]:
    """The policies ``indices`` name, decoded from the index rather than looked up.

    The way to turn a result's ``best_index`` back into a plan, on either search.
    An index outside ``0..|A|^H - 1`` names no policy. Under ``jit`` or ``vmap`` it
    decodes to a NaN policy, the one a result with no winner carries.

    Raises:
        ValueError: If a concrete index lies outside the enumeration, including
            the -1 a result carries when no policy scored finitely. Decoded, it
            would wrap to a real policy.
    """
    indices = jnp.asarray(indices)
    outside = (indices < 0) | (indices >= self._n_policies)
    flagged = concrete(outside)  # None under a trace, where nothing can raise
    if flagged is not None and flagged.any():
        named = np.asarray(indices)[flagged.astype(bool)].tolist()
        raise ValueError(
            f"indices {named} lie outside 0..{self._n_policies - 1}, the "
            "enumeration's range."
        )
    policies = _decode_policies(indices, self._action_set, self._horizon)
    return jnp.where(outside[..., None, None], jnp.nan, policies)

SearchResult dataclass

SearchResult(
    best_policy: Float64[Array, "H p"],
    best_g: Float64[Array, ""],
    best_index: Int[Array, ""],
    certificate: CompletenessCertificate,
)

The outcome of one exhaustive search: the best plan and the evidence for it.

ChunkedEFESearch returns one. The front-loaded search returns EnumeratedSearchResult, which adds the score vector. Fields are read by name. A result does not unpack or index, so no position can name one field on one search and another on the other.

A JAX pytree whose certificate is static data, so a jitted or vmapped evaluate returns the whole result.

Attributes:

Name	Type	Description
best_policy	Float64[Array, 'H p']	the argmin sequence, shape (H, p).
best_g	Float64[Array, '']	the argmin's summed EFE, G.
best_index	Int[Array, '']	the argmin's position in the enumeration.
certificate	CompletenessCertificate	the completeness certificate (ADR-030).

When no policy scores finitely there is no argmin. best_policy and best_g are then NaN and best_index is -1. The certificate still holds, since it certifies that every policy was scored.

EnumeratedEFESearch

EnumeratedEFESearch(
    model: LinearGaussianModel,
    action_set: FiniteActionSet,
    *,
    horizon: int,
    max_policies: int | None = _DEFAULT_MAX_POLICIES,
)

Exhaustive EFE search over A^H for a declared finite action set (ADR-031).

Front-loads (ADR-002) the full |A|^H enumeration of varying sequences at construction, and checks its completeness there. evaluate scores them all and returns the argmin, its score and index, the certificate and the full G vector. Because the set is finite and fully enumerated the search decides its universal, warrant PROVED, distinct from EFESelector's CORROBORATED grid.

ChunkedEFESearch runs the same enumeration in blocks. The two take the same model, set and horizon, answer the same evaluate and drive the same selectors. Swapping one for the other changes the memory a plan needs and no line that reads the result.

Scoring is a strategy (ADR-034): the default constructor scores on a flat LinearGaussianModel (policy_efe); over_backend scores on an FFG backend (policy_efe_ffg, aimed at a node's block) so the same exhaustive search runs the coupled-tree crossover model a flat model cannot express. The enumeration, the certificate, and the cost are shared — only the per-policy scoring differs.

Unlike EFESelector this supports p >= 1 and varying sequences: it can express a detour-then-exploit policy the constant-action family cannot.

Memory grows with the set. While evaluate runs, every policy, its score and a belief per policy are resident at once, at least 8 · |A|^H · (H·p + 1 + n + n²) bytes before XLA's intermediates. A set past max_policies is refused at construction with that floor in the message. ChunkedEFESearch holds one block at a time.

Parameters:

Name	Type	Description	Default
model	LinearGaussianModel	the flat model to score on. It needs a control matrix.	required
action_set	FiniteActionSet	the declared actions, enumerated as every length-H sequence.	required
horizon	int	H, the sequence length.	required
max_policies	int | None	the most policies the search will hold, 2^21 = 2,097,152 unless set. None lifts the budget.	_DEFAULT_MAX_POLICIES

Raises:

Type	Description
TypeError	If model is a coupled graph backend's model, which leaves out the couplings.
PolicyBudgetError	If |A|^H exceeds max_policies.
ValueError	If the model has no control matrix, if the action set's dimension or the horizon does not fit, or if max_policies is below 1.

Source code in src/cpomdp/enumeration.py

def __init__(
    self,
    model: LinearGaussianModel,
    action_set: FiniteActionSet,
    *,
    horizon: int,
    max_policies: int | None = _DEFAULT_MAX_POLICIES,
) -> None:
    self._setup(self._flat_scorer(model), action_set, horizon, max_policies)

policies property

policies: Float64[Array, 'N H p']

The enumerated A^H policies, shape (|A|^H, H, p) (front-loaded).

A result's best_index goes through policies_at, which refuses the -1 of a result with no winner. Used to index this array, -1 picks the last policy.

cost_per_plan property

cost_per_plan: int

Step evaluations one evaluate runs, |A|^H * H (RFC-001).

n_policies property

n_policies: int

The enumerated policy count |A|^H, attributable work (RFC-001).

model property

model: LinearGaussianModel

The model every policy is scored under, on a graph the backend's own.

On a coupled graph it is the interface model, which leaves out the couplings the score reads off backend.

backend property

backend: GraphBackend | None

The graph backend the policies are scored on, or None on a flat model.

horizon property

horizon: int

The sequence length H.

warrant property

warrant: Warrant

PROVED: a finite set enumerated in full decides its universal.

over_backend classmethod

over_backend(
    backend: GraphBackend,
    action_set: FiniteActionSet,
    *,
    info_node: SupportsIndex
    | Sequence[SupportsIndex]
    | None = None,
    horizon: int,
    max_policies: int | None = _DEFAULT_MAX_POLICIES,
) -> EnumeratedEFESearch

Exhaustive search that scores on an FFG backend via policy_efe_ffg.

The FFG counterpart of the flat constructor: it enumerates the same A^H set with the same completeness certificate, but scores each policy on a coupling graph with the epistemic aimed at info_node's state, or the whole joint state when it is None. This runs the coupled-tree model the flat constructor cannot express. The policy budget is the flat constructor's.

Raises:

Type	Description
TypeError	If info_node is not a node index, or a list, tuple, range or 1-D array of them.
InadmissiblePartitionError	If the backend's carry partition severs an EFE-load-bearing edge.
PolicyBudgetError	If |A|^H exceeds max_policies.
ValueError	If the backend has no control matrix, if info_node is empty, repeats a node or names one outside the graph, if the action set's dimension or the horizon does not fit, or if max_policies is below 1.

Source code in src/cpomdp/enumeration.py

@classmethod
def over_backend(
    cls,
    backend: "GraphBackend",
    action_set: FiniteActionSet,
    *,
    info_node: SupportsIndex | Sequence[SupportsIndex] | None = None,
    horizon: int,
    max_policies: int | None = _DEFAULT_MAX_POLICIES,
) -> "EnumeratedEFESearch":
    """Exhaustive search that scores on an FFG ``backend`` via ``policy_efe_ffg``.

    The FFG counterpart of the flat constructor: it enumerates the same ``A^H`` set
    with the same completeness certificate, but scores each policy on a coupling
    graph with the epistemic aimed at ``info_node``'s state, or the whole joint
    state when it is ``None``. This runs the coupled-tree model the flat
    constructor cannot express. The policy budget is the flat constructor's.

    Raises:
        TypeError: If ``info_node`` is not a node index, or a list, tuple,
            range or 1-D array of them.
        InadmissiblePartitionError: If the backend's carry partition severs an
            EFE-load-bearing edge.
        PolicyBudgetError: If ``|A|^H`` exceeds ``max_policies``.
        ValueError: If the backend has no control matrix, if ``info_node`` is
            empty, repeats a node or names one outside the graph, if the action
            set's dimension or the horizon does not fit, or if ``max_policies`` is
            below 1.
    """
    search = cls.__new__(cls)
    scorer = cls._graph_scorer(backend, info_node)
    search._setup(scorer, action_set, horizon, max_policies)
    return search

evaluate

evaluate(
    belief: Belief, preference: Preference
) -> EnumeratedSearchResult

Score every enumerated policy, and return the argmin with its evidence.

One vmap of the scorer over the front-loaded |A|^H policies, then a NaN-safe argmin (a NaN-scoring policy must not silently win). The scorer holds the flat model or the FFG backend as a closure constant, so the vmap maps only over the policies. It is compiled once per search, so a plan after the first runs the compiled program.

Parameters:

Name	Type	Description	Default
belief	Belief	the belief to plan from.	required
preference	Preference	the target and precision the EFE scores against.	required

Returns:

Type	Description
EnumeratedSearchResult	The argmin, its score, its index, the certificate and the score vector.

Source code in src/cpomdp/enumeration.py

def evaluate(
    self, belief: Belief, preference: "Preference"
) -> EnumeratedSearchResult:
    """Score every enumerated policy, and return the argmin with its evidence.

    One ``vmap`` of the scorer over the front-loaded ``|A|^H`` policies, then a
    NaN-safe ``argmin`` (a NaN-scoring policy must not silently win). The scorer
    holds the flat model or the FFG backend as a closure constant, so the ``vmap``
    maps only over the policies. It is compiled once per search, so a plan after
    the first runs the compiled program.

    Args:
        belief: the belief to plan from.
        preference: the target and precision the EFE scores against.

    Returns:
        The argmin, its score, its index, the certificate and the score vector.
    """
    scored = self._score_every_policy(self._policies, belief, preference)
    return EnumeratedSearchResult(
        best_policy=scored.best_policy,
        g=scored.g,
        best_g=scored.best_g,
        best_index=scored.best_index,
        certificate=self._certificate,
    )

policies_at

policies_at(
    indices: ArrayLike,
) -> Float64[Array, "... H p"]

The policies indices name, decoded from the index rather than looked up.

The way to turn a result's best_index back into a plan, on either search. An index outside 0..|A|^H - 1 names no policy. Under jit or vmap it decodes to a NaN policy, the one a result with no winner carries.

Raises:

Type	Description
ValueError	If a concrete index lies outside the enumeration, including the -1 a result carries when no policy scored finitely. Decoded, it would wrap to a real policy.

Source code in src/cpomdp/enumeration.py

def policies_at(self, indices: ArrayLike) -> Float64[Array, "... H p"]:
    """The policies ``indices`` name, decoded from the index rather than looked up.

    The way to turn a result's ``best_index`` back into a plan, on either search.
    An index outside ``0..|A|^H - 1`` names no policy. Under ``jit`` or ``vmap`` it
    decodes to a NaN policy, the one a result with no winner carries.

    Raises:
        ValueError: If a concrete index lies outside the enumeration, including
            the -1 a result carries when no policy scored finitely. Decoded, it
            would wrap to a real policy.
    """
    indices = jnp.asarray(indices)
    outside = (indices < 0) | (indices >= self._n_policies)
    flagged = concrete(outside)  # None under a trace, where nothing can raise
    if flagged is not None and flagged.any():
        named = np.asarray(indices)[flagged.astype(bool)].tolist()
        raise ValueError(
            f"indices {named} lie outside 0..{self._n_policies - 1}, the "
            "enumeration's range."
        )
    policies = _decode_policies(indices, self._action_set, self._horizon)
    return jnp.where(outside[..., None, None], jnp.nan, policies)

EnumeratedSearchResult dataclass

EnumeratedSearchResult(
    best_policy: Float64[Array, "H p"],
    best_g: Float64[Array, ""],
    best_index: Int[Array, ""],
    certificate: CompletenessCertificate,
    g: Float64[Array, N],
)

Bases: SearchResult

The outcome of one front-loaded exhaustive search, every policy's score included.

A SearchResult plus the score vector, which only the front-loaded path holds. Code that reads the shared fields runs on either search.

Attributes:

Name	Type	Description
g	Float64[Array, N]	the summed EFE of every enumerated policy, in enumeration order, shape (|A|^H,).

ChunkedEFESearch

ChunkedEFESearch(
    model: LinearGaussianModel,
    action_set: FiniteActionSet,
    *,
    horizon: int,
    chunk: int = _DEFAULT_CHUNK,
)

Exhaustive EFE search over A^H in blocks, at O(chunk) residency.

The same enumeration as EnumeratedEFESearch and the same warrant, run so that neither the policy set nor the score vector is ever materialised. That is what puts sets past the front-loaded path's memory ceiling in reach, and the reason to have it is a set declared too large to hold, never a result that came out inconvenient.

Three things make the two paths interchangeable rather than merely similar. Blocks are visited in increasing index order and the argmin updates on a strict <, so ties resolve to the globally lowest index exactly as jnp.argmin does over the whole vector. Both paths decode an index to its policy with one function. And the trailing block is padded, with its padded lanes masked out of the argmin and out of visited, so completeness is a count of real work rather than of array slots.

It takes no policy budget. Its footprint is set by chunk, and its time grows with cost_per_plan.

evaluate composes with jit, vmap and lax.scan. Inside one, the loop's count exists only once the program runs. The certificate then states the expected count, and a callback checks the loop's count in the run, so a miscount fails the run instead of certifying it.

Parameters:

Name	Type	Description	Default
model	LinearGaussianModel	the flat model to score on. It needs a control matrix.	required
action_set	FiniteActionSet	the declared actions, enumerated as every length-H sequence.	required
horizon	int	H, the sequence length.	required
chunk	int	the most policies one block scores, 4096 unless set. Clamped to |A|^H. The answer does not depend on it.	_DEFAULT_CHUNK

Raises:

Type	Description
TypeError	If model is a coupled graph backend's model, which leaves out the couplings.
ValueError	If the model has no control matrix, if the action set's dimension or the horizon does not fit, if chunk is below 1, or if |A|^H exceeds the index dtype's ceiling.

Source code in src/cpomdp/enumeration.py

def __init__(
    self,
    model: LinearGaussianModel,
    action_set: FiniteActionSet,
    *,
    horizon: int,
    chunk: int = _DEFAULT_CHUNK,
) -> None:
    self._setup(self._flat_scorer(model), action_set, horizon, chunk)

chunk property

chunk: int

The block size actually used, clamped to |A|^H.

The answer does not depend on it. The residency does, and so does the wasted arithmetic in the padded tail.

n_blocks property

n_blocks: int

How many blocks the enumeration takes, the last one padded.

cost_per_plan property

cost_per_plan: int

Step evaluations one evaluate runs, |A|^H * H (RFC-001).

n_policies property

n_policies: int

The enumerated policy count |A|^H, attributable work (RFC-001).

model property

model: LinearGaussianModel

The model every policy is scored under, on a graph the backend's own.

On a coupled graph it is the interface model, which leaves out the couplings the score reads off backend.

backend property

backend: GraphBackend | None

The graph backend the policies are scored on, or None on a flat model.

horizon property

horizon: int

The sequence length H.

warrant property

warrant: Warrant

PROVED: a finite set enumerated in full decides its universal.

over_backend classmethod

over_backend(
    backend: GraphBackend,
    action_set: FiniteActionSet,
    *,
    info_node: SupportsIndex
    | Sequence[SupportsIndex]
    | None = None,
    horizon: int,
    chunk: int = _DEFAULT_CHUNK,
) -> ChunkedEFESearch

Chunked search that scores on an FFG backend via policy_efe_ffg.

Raises:

Type	Description
TypeError	If info_node is not a node index, or a list, tuple, range or 1-D array of them.
InadmissiblePartitionError	If the backend's carry partition severs an EFE-load-bearing edge.
ValueError	If the backend has no control matrix, if info_node is empty, repeats a node or names one outside the graph, if the action set's dimension or the horizon does not fit, or if chunk is below 1.

Source code in src/cpomdp/enumeration.py

@classmethod
def over_backend(
    cls,
    backend: "GraphBackend",
    action_set: FiniteActionSet,
    *,
    info_node: SupportsIndex | Sequence[SupportsIndex] | None = None,
    horizon: int,
    chunk: int = _DEFAULT_CHUNK,
) -> "ChunkedEFESearch":
    """Chunked search that scores on an FFG ``backend`` via ``policy_efe_ffg``.

    Raises:
        TypeError: If ``info_node`` is not a node index, or a list, tuple,
            range or 1-D array of them.
        InadmissiblePartitionError: If the backend's carry partition severs an
            EFE-load-bearing edge.
        ValueError: If the backend has no control matrix, if ``info_node`` is
            empty, repeats a node or names one outside the graph, if the action
            set's dimension or the horizon does not fit, or if ``chunk`` is below 1.
    """
    search = cls.__new__(cls)
    search._setup(cls._graph_scorer(backend, info_node), action_set, horizon, chunk)
    return search

evaluate

evaluate(
    belief: Belief, preference: Preference
) -> SearchResult

Score every policy in blocks, and return the argmin with its evidence.

One lax.scan over ceil(|A|^H / chunk) blocks. Each block decodes its own indices, scores them under one vmap, and folds the result into running scalars, so nothing of size |A|^H is ever alive. The loop is compiled once per search, so a plan after the first runs the compiled program.

Parameters:

Name	Type	Description	Default
belief	Belief	the belief to plan from.	required
preference	Preference	the target and precision the EFE scores against.	required

Returns:

Type	Description
SearchResult	The argmin, its score, its index, and the certificate the loop's own count
SearchResult	earns.

Source code in src/cpomdp/enumeration.py

def evaluate(self, belief: Belief, preference: "Preference") -> SearchResult:
    """Score every policy in blocks, and return the argmin with its evidence.

    One ``lax.scan`` over ``ceil(|A|^H / chunk)`` blocks. Each block decodes its own
    indices, scores them under one ``vmap``, and folds the result into running
    scalars, so nothing of size ``|A|^H`` is ever alive. The loop is compiled once
    per search, so a plan after the first runs the compiled program.

    Args:
        belief: the belief to plan from.
        preference: the target and precision the EFE scores against.

    Returns:
        The argmin, its score, its index, and the certificate the loop's own count
        earns.
    """
    argmin, best_policy, _ = self._scan_compiled(belief, preference)
    return self._result(argmin, best_policy)

policies_at

policies_at(
    indices: ArrayLike,
) -> Float64[Array, "... H p"]

The policies indices name, decoded from the index rather than looked up.

The way to turn a result's best_index back into a plan, on either search. An index outside 0..|A|^H - 1 names no policy. Under jit or vmap it decodes to a NaN policy, the one a result with no winner carries.

Raises:

Type	Description
ValueError	If a concrete index lies outside the enumeration, including the -1 a result carries when no policy scored finitely. Decoded, it would wrap to a real policy.

Source code in src/cpomdp/enumeration.py

def policies_at(self, indices: ArrayLike) -> Float64[Array, "... H p"]:
    """The policies ``indices`` name, decoded from the index rather than looked up.

    The way to turn a result's ``best_index`` back into a plan, on either search.
    An index outside ``0..|A|^H - 1`` names no policy. Under ``jit`` or ``vmap`` it
    decodes to a NaN policy, the one a result with no winner carries.

    Raises:
        ValueError: If a concrete index lies outside the enumeration, including
            the -1 a result carries when no policy scored finitely. Decoded, it
            would wrap to a real policy.
    """
    indices = jnp.asarray(indices)
    outside = (indices < 0) | (indices >= self._n_policies)
    flagged = concrete(outside)  # None under a trace, where nothing can raise
    if flagged is not None and flagged.any():
        named = np.asarray(indices)[flagged.astype(bool)].tolist()
        raise ValueError(
            f"indices {named} lie outside 0..{self._n_policies - 1}, the "
            "enumeration's range."
        )
    policies = _decode_policies(indices, self._action_set, self._horizon)
    return jnp.where(outside[..., None, None], jnp.nan, policies)

PolicyBudgetError

Bases: ValueError

A front-loaded search asked to hold more policies than its budget allows.

EnumeratedEFESearch holds every policy and every score at once, so its footprint grows with |A|^H. The count is exact before anything is allocated, and an oversized set is refused there rather than exhausting the host. The message gives the count, the step evaluations one plan costs and a floor on the memory. It names ChunkedEFESearch, which runs the same enumeration at a footprint its block sets and drives the same selectors.

IncompleteEnumerationError

Bases: Exception

Fewer than |A|^H policies enumerated: a sample, not a decision.

Driving an agent

A search returns a whole plan. The two selectors below turn one into an action per step for Agent(selector=...), and each takes any ExhaustiveEFESearch. RecedingHorizonSelector plans every step and applies the first action. OpenLoopSelector plans once and applies the whole sequence. Their cost_per_cycle differs by a factor of H for that reason.

RecedingHorizonSelector

RecedingHorizonSelector(search: ExhaustiveEFESearch)

Re-plans every step and applies the first action of the best sequence.

An ActionSelector over an exhaustive search. Every select re-runs the full A^H search on the current belief, so the belief drives every choice. It costs a whole plan per step, cost_per_cycle = |A|^H * H. A measurement has to declare which mode it ran, so this and OpenLoopSelector stay separate (ADR-031).

Parameters:

Name	Type	Description	Default
search	ExhaustiveEFESearch	the search to plan with. Either shipped search gives the same actions.	required

Raises:

Type	Description
TypeError	If search is not an ExhaustiveEFESearch.

Source code in src/cpomdp/enumeration.py

def __init__(self, search: ExhaustiveEFESearch) -> None:
    if not isinstance(search, ExhaustiveEFESearch):
        raise TypeError(
            f"{type(self).__name__} drives an ExhaustiveEFESearch, such as "
            f"EnumeratedEFESearch or ChunkedEFESearch, got {type(search).__name__}."
        )
    self._search = search
    self._certificate: CompletenessCertificate | None = None

replan_interval property

replan_interval: int

Agent steps between two plans. 1: this driver plans every step.

horizon property

horizon: int

The planning horizon H.

model property

model: LinearGaussianModel

The model the search plans under.

backend property

backend: GraphBackend | None

The graph backend the search scores on, or None on a flat model.

warrant property

warrant: Warrant

PROVED, inherited from the exhaustive search.

certificate property

certificate: CompletenessCertificate | None

The last plan's completeness certificate (ADR-030), None before one.

The chunked search earns its certificate by running, so a driver has one only once it has planned.

cost_per_plan property

cost_per_plan: int

Step evaluations one search costs, |A|^H * H (RFC-001).

cost_per_cycle property

cost_per_cycle: int

Step evaluations per agent step, one plan spread over its interval.

select

select(
    belief: Belief, preference: Preference
) -> Float64[Array, p]

Re-plan on belief and return the first action of the best sequence.

Source code in src/cpomdp/enumeration.py

def select(self, belief: Belief, preference: "Preference") -> Float64[Array, "p"]:
    """Re-plan on ``belief`` and return the first action of the best sequence."""
    return self._plan(belief, preference)[0]

OpenLoopSelector

OpenLoopSelector(search: ExhaustiveEFESearch)

Plans once, then applies the whole best sequence without looking at beliefs.

An ActionSelector over an exhaustive search. It plans on the belief of the first select and returns the plan's actions in order, ignoring the interim beliefs. It re-plans only when the committed sequence is spent, or after reset. Ignoring the interim beliefs is what makes it open-loop. One plan covers H steps, so cost_per_cycle = |A|^H. A measurement has to declare which mode it ran, so this and RecedingHorizonSelector stay separate (ADR-031).

Stateful: it holds the committed plan and a step counter. The Agent drives it eagerly, one select per step, so the mutation is safe.

Parameters:

Name	Type	Description	Default
search	ExhaustiveEFESearch	the search to plan with. Either shipped search gives the same actions.	required

Raises:

Type	Description
TypeError	If search is not an ExhaustiveEFESearch.

Source code in src/cpomdp/enumeration.py

def __init__(self, search: ExhaustiveEFESearch) -> None:
    super().__init__(search)
    self._committed: Float64[Array, "H p"] | None = None
    self._step = 0

replan_interval property

replan_interval: int

Agent steps between two plans, H. It commits to the whole sequence.

horizon property

horizon: int

The planning horizon H.

model property

model: LinearGaussianModel

The model the search plans under.

backend property

backend: GraphBackend | None

The graph backend the search scores on, or None on a flat model.

warrant property

warrant: Warrant

PROVED, inherited from the exhaustive search.

certificate property

certificate: CompletenessCertificate | None

The last plan's completeness certificate (ADR-030), None before one.

The chunked search earns its certificate by running, so a driver has one only once it has planned.

cost_per_plan property

cost_per_plan: int

Step evaluations one search costs, |A|^H * H (RFC-001).

cost_per_cycle property

cost_per_cycle: int

Step evaluations per agent step, one plan spread over its interval.

select

select(
    belief: Belief, preference: Preference
) -> Float64[Array, p]

Return the next committed action, re-planning on belief when spent.

Source code in src/cpomdp/enumeration.py

def select(self, belief: Belief, preference: "Preference") -> Float64[Array, "p"]:
    """Return the next committed action, re-planning on ``belief`` when spent."""
    if self._committed is None or self._step >= self._search.horizon:
        self._committed = self._plan(belief, preference)
        self._step = 0
    action = self._committed[self._step]
    self._step += 1
    return action

reset

reset() -> None

Drop the committed plan and its certificate, so the next select plans.

Source code in src/cpomdp/enumeration.py

def reset(self) -> None:
    """Drop the committed plan and its certificate, so the next ``select`` plans."""
    self._committed = None
    self._step = 0
    self._certificate = None