Policy API Reference

This page contains the auto-generated API documentation for the policy module.

Base Classes

PolicyChecker

PolicyChecker() dataclass

Bases: ABC

Abstract base class for policy checkers that validate execution constraints.

PolicyCheckers enforce constraints before instruction execution. Subclasses must implement check_before method to define custom validation logic.

check_before(history)

Validate constraints before instruction execution.

Parameters:

Name	Type	Description	Default
history	History	The execution history up to this point.	required

Returns:

Type	Description
bool	True if validation passes.

Raises:

Type	Description
RuntimeError	If validation fails.

Source code in arbiteros_alpha/policy.py

def check_before(self, history: History) -> bool:
    """Validate constraints before instruction execution.

    Args:
        history: The execution history up to this point.

    Returns:
        True if validation passes.

    Raises:
        RuntimeError: If validation fails.
    """
    pass

PolicyRouter

PolicyRouter() dataclass

Bases: ABC

Abstract base class for policy routers that dynamically route execution flow.

PolicyRouters analyze execution history and decide whether to redirect the execution flow to a different node based on policy conditions.

route_after(history)

Determine the next node to execute based on policy conditions.

Parameters:

Name	Type	Description	Default
history	History	The execution history including the just-executed instruction.	required

Returns:

Type	Description
str	The name of the target node to route to, or None to continue normal flow.

Source code in arbiteros_alpha/policy.py

def route_after(self, history: History) -> str:
    """Determine the next node to execute based on policy conditions.

    Args:
        history: The execution history including the just-executed instruction.

    Returns:
        The name of the target node to route to, or None to continue normal flow.
    """
    pass

Implementations

HistoryPolicyChecker

HistoryPolicyChecker(name, bad_sequence) dataclass

Bases: PolicyChecker

Policy checker that validates against blacklisted instruction sequences.

This checker prevents specific sequences of instructions from being executed by maintaining a blacklist of forbidden instruction chains.

Example

from arbiteros_alpha.instructions import CognitiveCore, ExecutionCore checker = HistoryPolicyChecker( ... "no_direct_toolcall", ... [CognitiveCore.GENERATE, ExecutionCore.TOOL_CALL] ... )

This will raise RuntimeError if GENERATE is followed by TOOL_CALL

check_before(history)

Check if the current history contains any blacklisted sequences.

Parameters:

Name	Type	Description	Default
history	History	The execution history to validate.	required

Returns:

Type	Description
bool	True if no blacklisted sequences are detected.

Raises:

Type	Description
RuntimeError	If a blacklisted sequence is detected in the history.

Source code in arbiteros_alpha/policy.py

def check_before(self, history: History) -> bool:
    """Check if the current history contains any blacklisted sequences.

    Args:
        history: The execution history to validate.

    Returns:
        True if no blacklisted sequences are detected.

    Raises:
        RuntimeError: If a blacklisted sequence is detected in the history.
    """

    n_work = len(history.entries)

    n_pat = len(self.bad_sequence)

    logger.debug(
        f"HistoryPolicyChecker '{self.name}': checking {n_work} completed supersteps against pattern of length {n_pat}"
    )
    logger.debug(f"Bad sequence pattern: {[i.name for i in self.bad_sequence]}")

    if n_pat > n_work:
        return True

    for i in range(n_work - n_pat + 1):
        match = True
        for j in range(n_pat):
            # i for workflow offset，j for pattern index
            current_stage_workers = history.entries[i + j]
            current_stage_workers = [
                item.instruction for item in current_stage_workers
            ]
            target_worker = self.bad_sequence[j]

            logger.debug(
                f"  Window[{i}][{j}]: checking if {target_worker.name} in {[w.name for w in current_stage_workers]}"
            )

            if target_worker not in current_stage_workers:
                match = False
                break

        if match:
            logger.debug(
                f"Blacklisted sequence detected: {self.name}:[{self.bad_sequence}]"
            )
            return False

    return True

MetricThresholdPolicyRouter

MetricThresholdPolicyRouter(name, key, threshold, target) dataclass

Bases: PolicyRouter

Policy router that redirects execution flow based on threshold conditions.

This router monitors a specific metric in the instruction output and redirects to a target node when the metric value falls below a specified threshold. Commonly used for quality control patterns like regeneration on low confidence scores.

Attributes:

Name	Type	Description
name	str	Human-readable name for this policy router.
key	str	The key in the output state dictionary to monitor.
threshold	float	The minimum acceptable value for the monitored metric.
target	str	The node name to route to when value is below threshold.

Example

Create a router that triggers regeneration when confidence is low:

router = MetricThresholdPolicyRouter( ... name="regenerate_on_low_confidence", ... key="confidence", ... threshold=0.6, ... target="generate" ... )

If output["confidence"] < 0.6, routes back to "generate" node

route_after(history)

Route to target node if the monitored metric is below threshold.

Extracts the specified metric from the most recent instruction's output state and compares it against the configured threshold. If the metric value is below the threshold, returns the target node for routing.

Parameters:

Name	Type	Description	Default
history	History	The execution history including the just-executed instruction. The last entry's output_state is checked for the metric.	required

Returns:

Type	Description
str | None	The target node name if metric value < threshold, None otherwise.
str | None	When None is returned, execution continues with normal flow.

Source code in arbiteros_alpha/policy.py

def route_after(self, history: History) -> str | None:
    """Route to target node if the monitored metric is below threshold.

    Extracts the specified metric from the most recent instruction's output
    state and compares it against the configured threshold. If the metric
    value is below the threshold, returns the target node for routing.

    Args:
        history: The execution history including the just-executed instruction.
            The last entry's output_state is checked for the metric.

    Returns:
        The target node name if metric value < threshold, None otherwise.
        When None is returned, execution continues with normal flow.
    """
    if not history.entries or not history.entries[-1]:
        return None
    last_entry = history.entries[-1][-1]
    output = last_entry.output_state
    confidence = output.get(self.key, 1.0)
    if confidence < self.threshold:
        return self.target
    return None