Early-termination policies

Early-termination policies#

Halt a planning loop based on cost, wall-clock time, or any world-state predicate. LangGOAP ships five built-in policies plus two composites (FirstOfPolicy, AllOfPolicy). Backed by tests/integration/test_early_termination.py.

1. A five-step plan with per-action cost#

Each action adds $0.30 to world_state['total_cost_usd'] and flips one boolean. The goal needs all five flags.

from typing import Any
from langgoap.actions import ActionSpec
from langgoap.goals import GoalSpec
from langgoap.graph.builder import GoapGraph

def make_step(i: int) -> ActionSpec:
    def execute(ws: dict[str, Any]) -> dict[str, Any]:
        return {f"step_{i}": True,
                "total_cost_usd": float(ws.get("total_cost_usd", 0.0)) + 0.30}
    pre = {f"step_{i-1}": True} if i > 1 else {}
    return ActionSpec(
        name=f"step_{i}",
        preconditions=pre,
        effects={f"step_{i}": True, "total_cost_usd": 0.0},
        execute=execute,
    )

actions = [make_step(i) for i in range(1, 6)]
goal = GoalSpec(conditions={f"step_{i}": True for i in range(1, 6)})
[a.name for a in actions]

/Users/brian.sam-bodden/Code/langgoap/.venv/lib/python3.12/site-packages/langgraph/checkpoint/serde/encrypted.py:5: LangChainPendingDeprecationWarning: The default value of `allowed_objects` will change in a future version. Pass an explicit value (e.g., allowed_objects='messages' or allowed_objects='core') to suppress this warning.
  from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

['step_1', 'step_2', 'step_3', 'step_4', 'step_5']

2. MaxCostPolicy — halt when total_cost_usd exceeds budget#

With a $0.50 budget the loop completes step 1 (cost rises to 0.30), executes step 2 (cost rises to 0.60), and then the observer fires the policy. No further steps run.

from langgoap.termination import MaxCostPolicy

graph = GoapGraph(actions, termination_policies=[MaxCostPolicy(usd=0.50)])
result = graph.invoke(goal=goal, world_state={"total_cost_usd": 0.0})
(result["status"], len(result["execution_history"]),
 result["world_state"]["total_cost_usd"])

('terminated', 2, 0.6)

3. MaxActionsPolicy — halt after N executed actions#

Hard cap on the number of actions regardless of world state. Useful as a safety net against runaway replanning.

from langgoap.termination import MaxActionsPolicy

graph = GoapGraph(actions, termination_policies=[MaxActionsPolicy(2)])
result = graph.invoke(goal=goal, world_state={"total_cost_usd": 0.0})
(result["status"], len(result["execution_history"]))

('terminated', 2)

4. FirstOfPolicy — terminate on the first policy to fire#

Compose multiple budgets. Whichever fires first stops the run. Embabel’s equivalent is EarlyTerminationPolicy.firstOf(...).

from langgoap.termination import FirstOfPolicy

graph = GoapGraph(
    actions,
    termination_policies=[
        FirstOfPolicy(
            MaxCostPolicy(usd=10.00),   # would not fire (budget high)
            MaxActionsPolicy(3),         # fires after 3 actions
        )
    ],
)
result = graph.invoke(goal=goal, world_state={"total_cost_usd": 0.0})
(result["status"], len(result["execution_history"]))

('terminated', 3)

5. The termination decision is observable#

The terminal state carries a human-readable reason under result['replan_reason'] so logs and traces explain exactly why a run stopped.

result["replan_reason"]

'MaxActionsPolicy: Max actions of 3 reached'

Next steps#

  • Hook the goal-driven research example end-to-end with real LLM cost accounting: see examples/tutorials/cost_bounded_research_agent.ipynb.

  • Combine policies with AllOfPolicy for both budgets must fire semantics.

  • Use OnStuckPolicy to terminate when planning fails instead of escalating to a stuck handler.