Skip to content

Core API Reference

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

The main components are implemented in arbiteros_alpha.core and re-exported from the package root for convenience.

ArbiterOSAlpha

ArbiterOSAlpha(backend='langgraph')

Main ArbiterOS coordinator for policy-driven LangGraph execution.

ArbiterOSAlpha provides a lightweight governance layer on top of LangGraph, enabling policy-based validation and dynamic routing without modifying the underlying graph structure.

Attributes:

Name Type Description
backend

The execution backend in use.
history History

List of execution history entries with timestamps and I/O.
policy_checkers list[PolicyChecker]

List of PolicyChecker instances for validation.
policy_routers list[PolicyRouter]

List of PolicyRouter instances for dynamic routing.
Example

os = ArbiterOSAlpha(backend="langgraph") os.add_policy_checker(HistoryPolicyChecker("require_verification",["generate", "execute"])) os.add_policy_router(ConfidencePolicyRouter("confidence", 0.5, "retry")) @os.instruction("generate") ... def generate(state): return {"result": "output"}

Initialize the ArbiterOSAlpha instance.

Parameters:

Name Type Description Default
backend Literal['langgraph', 'vanilla']

The execution backend to use. - "langgraph": Use an agent based on the LangGraph framework. - "vanilla": Use the framework-less ('from scratch') agent implementation.

 'langgraph'
Source code in arbiteros_alpha/core.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(self, backend: Literal["langgraph", "vanilla"] = "langgraph"):
    """Initialize the ArbiterOSAlpha instance.

    Args:
        backend: The execution backend to use.
            - "langgraph": Use an agent based on the LangGraph framework.
            - "vanilla": Use the framework-less ('from scratch') agent implementation.
    """
    self.backend = backend
    self.history: History = History()
    self.policy_checkers: list[PolicyChecker] = []
    self.policy_routers: list[PolicyRouter] = []

    if self.backend == "langgraph":
        self._patch_pregel_loop()

add_policy_checker(checker)

Register a policy checker for validation.

Parameters:

Name Type Description Default
checker PolicyChecker

A PolicyChecker instance to validate execution constraints.

 required
Source code in arbiteros_alpha/core.py 
67
68
69
70
71
72
73
74
def add_policy_checker(self, checker: PolicyChecker) -> None:
    """Register a policy checker for validation.

    Args:
        checker: A PolicyChecker instance to validate execution constraints.
    """
    logger.debug(f"Adding policy checker: {checker}")
    self.policy_checkers.append(checker)

add_policy_router(router)

Register a policy router for dynamic flow control.

Policy routers are only supported when using the "langgraph" backend.

Parameters:

Name Type Description Default
router PolicyRouter

A PolicyRouter instance to dynamically route execution.

 required

Raises:

Type Description
RuntimeError

If the backend is not "langgraph".
Source code in arbiteros_alpha/core.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def add_policy_router(self, router: PolicyRouter) -> None:
    """Register a policy router for dynamic flow control.

    Policy routers are only supported when using the "langgraph" backend.

    Args:
        router: A PolicyRouter instance to dynamically route execution.

    Raises:
        RuntimeError: If the backend is not "langgraph".
    """
    if self.backend != "langgraph":
        raise RuntimeError(
            "Policy routers are only supported with the 'langgraph' backend."
        )
    logger.debug(f"Adding policy router: {router}")
    self.policy_routers.append(router)

_check_before()

Execute all policy checkers before instruction execution.

Returns:

Type Description
dict[str, bool]

A dictionary mapping checker names to their validation results.
bool

A final boolean indicating if all checkers passed.
Source code in arbiteros_alpha/core.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def _check_before(self) -> tuple[dict[str, bool], bool]:
    """Execute all policy checkers before instruction execution.

    Returns:
        A dictionary mapping checker names to their validation results.
        A final boolean indicating if all checkers passed.
    """
    results = {}
    logger.debug(f"Running {len(self.policy_checkers)} policy checkers (before)")
    for checker in self.policy_checkers:
        result = checker.check_before(self.history)

        if result is False:
            results[checker.name] = result
            logger.error(f"Policy checker {checker} failed validation.")

    return results, all(results.values())

_route_after()

Determine if execution should be routed to a different node.

Consults all registered policy routers in order. Returns the first non-None routing decision.

Returns:

Type Description
dict[str, str | None]

A dictionary mapping checker names to their route destination.
str | None

A final str indicating the final route destination.
Source code in arbiteros_alpha/core.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def _route_after(self) -> tuple[dict[str, str | None], str | None]:
    """Determine if execution should be routed to a different node.

    Consults all registered policy routers in order. Returns the first
    non-None routing decision.

    Returns:
        A dictionary mapping checker names to their route destination.
        A final str indicating the final route destination.
    """
    results = {}
    destination = None
    used_router = None
    logger.debug(f"Checking {len(self.policy_routers)} policy routers")
    for router in self.policy_routers:
        decision = router.route_after(self.history)

        if decision:
            results[router.name] = decision
            used_router = router
            destination = decision

    decision_count = sum(1 for v in results.values() if v is not None)
    if decision_count > 1:
        logger.error(
            "Multiple routers decided to route. Fallback to first decision."
        )

    if destination is not None:
        logger.warning(f"Router {used_router} decision made to: {destination}")
    return results, destination

instruction(instruction_type)

Decorator to wrap LangGraph node functions with policy governance.

This decorator adds policy validation, execution history tracking, and dynamic routing to LangGraph node functions. It's the core integration point between ArbiterOS and LangGraph.

Parameters:

Name Type Description Default
instruction_type InstructionType

An instruction type from one of the Core enums (CognitiveCore, MemoryCore, ExecutionCore, NormativeCore, MetacognitiveCore, AdaptiveCore, SocialCore, or AffectiveCore).

 required

Returns:

Type Description
Callable[[Callable], Callable]

A decorator function that wraps the target node function.
Example

from arbiteros_alpha.instructions import CognitiveCore @os.instruction(CognitiveCore.GENERATE) ... def generate(state: State) -> State: ... return {"field": "value"}

Function now includes policy checks and history tracking

Source code in arbiteros_alpha/core.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def instruction(
    self, instruction_type: InstructionType
) -> Callable[[Callable], Callable]:
    """Decorator to wrap LangGraph node functions with policy governance.

    This decorator adds policy validation, execution history tracking,
    and dynamic routing to LangGraph node functions. It's the core
    integration point between ArbiterOS and LangGraph.

    Args:
        instruction_type: An instruction type from one of the Core enums
            (CognitiveCore, MemoryCore, ExecutionCore, NormativeCore,
            MetacognitiveCore, AdaptiveCore, SocialCore, or AffectiveCore).

    Returns:
        A decorator function that wraps the target node function.

    Example:
        >>> from arbiteros_alpha.instructions import CognitiveCore
        >>> @os.instruction(CognitiveCore.GENERATE)
        ... def generate(state: State) -> State:
        ...     return {"field": "value"}
        >>> # Function now includes policy checks and history tracking
    """
    # Validate that instruction_type is a valid InstructionType enum
    if not isinstance(instruction_type, InstructionType.__args__):
        raise TypeError(
            f"instruction_type must be an instance of one of the Core enums, got {type(instruction_type)}"
        )

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            logger.debug(
                f"Executing instruction: {instruction_type.__class__.__name__}.{instruction_type.name}"
            )

            history_item = HistoryItem(
                timestamp=datetime.datetime.now(),
                instruction=instruction_type,
                input_state=args[0],
            )

            if self.backend == "vanilla":
                self.history.enter_next_superstep([instruction_type.name])

            self.history.add_entry(history_item)

            history_item.check_policy_results, all_passed = self._check_before()

            result = func(*args, **kwargs)
            logger.debug(f"Instruction {instruction_type.name} returned: {result}")
            history_item.output_state = result

            history_item.route_policy_results, destination = self._route_after()

            if destination:
                return Command(update=result, goto=destination)

            return result

        return wrapper

    return decorator

History

History()

Initialize an empty execution history.

Source code in arbiteros_alpha/history.py 
36
37
38
39
def __init__(self) -> None:
    """Initialize an empty execution history."""
    self.entries: list[SuperStep] = []
    self.next_superstep: list[str] = []

enter_next_superstep(nodes)

Source code in arbiteros_alpha/history.py 
41
42
43
44
45
46
def enter_next_superstep(self, nodes: list[str]) -> None:
    if "__start__" in nodes or "__end__" in nodes:
        return
    logger.debug(f"Entering next superstep with nodes: {nodes}")
    self.next_superstep = nodes
    self.entries.append([])

add_entry(entry)

Source code in arbiteros_alpha/history.py 
48
49
50
51
52
53
54
55
56
def add_entry(self, entry: HistoryItem) -> None:
    if not self.entries or len(self.entries[-1]) >= len(self.next_superstep):
        raise RuntimeError(
            "All nodes for the current superstep have already recorded entries.\n"
            "Hint: Did you forget to call \n"
            "    - register_compiled_graph() for langgraph backend or \n"
            "    - enter_next_superstep() for vanilla backend?"
        )
    self.entries[-1].append(entry)

pprint()

Source code in arbiteros_alpha/history.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def pprint(self) -> None:
    import yaml
    from rich.console import Console

    console = Console()
    console.print("\n[bold cyan]📋 Arbiter OS Execution History[/bold cyan]")
    console.print("=" * 80)

    for superstep_idx, superstep in enumerate(self.entries, 1):
        console.print(
            f"\n[bold magenta]╔═══ SuperStep {superstep_idx} ═══╗[/bold magenta]"
        )

        for entry_idx, entry in enumerate(superstep, 1):
            # Format policy results
            check_results = entry.check_policy_results
            route_results = entry.route_policy_results

            # Header with instruction name
            console.print(
                f"\n[bold cyan]  [{superstep_idx}.{entry_idx}] {entry.instruction.name}[/bold cyan]"
            )
            console.print(f"[dim]    Timestamp: {entry.timestamp}[/dim]")

            # Format input state as YAML
            console.print("    [yellow]Input:[/yellow]")
            input_yaml = yaml.dump(
                entry.input_state, default_flow_style=False, sort_keys=False
            )
            for line in input_yaml.strip().split("\n"):
                console.print(f"      [dim]{line}[/dim]")

            # Format output state as YAML
            console.print("    [yellow]Output:[/yellow]")
            output_yaml = yaml.dump(
                entry.output_state, default_flow_style=False, sort_keys=False
            )
            for line in output_yaml.strip().split("\n"):
                console.print(f"      [dim]{line}[/dim]")

            # Show detailed policy check results
            console.print("    [yellow]Policy Checks:[/yellow]")
            if check_results:
                for policy_name, result in check_results.items():
                    status = "[green]✓[/green]" if result else "[red]✗[/red]"
                    console.print(f"      {status} {policy_name}")
            else:
                console.print("      [dim](none)[/dim]")

            # Show detailed policy route results
            console.print("    [yellow]Policy Routes:[/yellow]")
            if route_results:
                for policy_name, destination in route_results.items():
                    if destination:
                        console.print(
                            f"      [magenta]→[/magenta] {policy_name} [bold magenta]⇒ {destination}[/bold magenta]"
                        )
                    else:
                        console.print(f"      [dim]— {policy_name}[/dim]")
            else:
                console.print("      [dim](none)[/dim]")

        console.print(
            f"[bold magenta]╚{'═' * (len(f'SuperStep {superstep_idx}') + 9)}╝[/bold magenta]"
        )

    console.print("\n" + "=" * 80 + "\n")

HistoryItem

HistoryItem(timestamp, instruction, input_state, output_state=dict(), check_policy_results=dict(), route_policy_results=dict()) dataclass

The minimal OS metadata for tracking instruction execution.

Attributes:

Name Type Description
timestamp datetime

When the instruction was executed.
instruction InstructionType

The instruction type that was executed.
input_state dict[str, Any]

The state passed to the instruction.
output_state dict[str, Any]

The state returned by the instruction.
check_policy_results dict[str, bool]

Results of policy checkers (name -> passed/failed).
route_policy_results dict[str, str | None]

Results of policy routers (name -> target or None).