langgoap.ActionSpec

langgoap.ActionSpec#

class ActionSpec(name, preconditions=<factory>, effects=<factory>, effect_keys=None, cost=1.0, execute=None, aexecute=None, effect_validator=None, max_retries=0, qos=None, can_rerun=True, read_only=False, utility=None, require_human_approval=False, human_input_key=None, resources=None, duration=None, metadata=None)[source]#

Specification of a GOAP action for planning.

Parameters:

  • name (str)

  • preconditions (Mapping[str, Any])

  • effects (Mapping[str, Any] | Callable[[Mapping[str, Any]], Mapping[str, Any]])

  • effect_keys (frozenset[str] | None)

  • cost (float | CostFunction)

  • execute (Callable[[...], Any] | None)

  • aexecute (Callable[[...], Any] | None)

  • effect_validator (Callable[[dict[str, Any], dict[str, Any]], bool] | None)

  • max_retries (int)

  • qos (ActionQos | None)

  • can_rerun (bool)

  • read_only (bool)

  • utility (float | CostFunction | None)

  • require_human_approval (bool | type)

  • human_input_key (str | None)

  • resources (Mapping[str, float] | None)

  • duration (timedelta | None)

  • metadata (Mapping[str, Any] | None)

name#

Unique identifier for the action.

Type:

str

preconditions#

World state conditions that must hold before execution. Stored as an immutable MappingProxyType to match the frozen contract of this dataclass.

Type:

collections.abc.Mapping[str, Any]

effects#

World state changes produced by execution. Either a static mapping (stored as an immutable MappingProxyType) or a callable (world_state) -> Mapping[str, Any] for state-dependent transitions. Callable effects must declare effect_keys so the planner’s static analysis passes can reason about which state keys the action may touch.

Type:

collections.abc.Mapping[str, Any] | Callable[[collections.abc.Mapping[str, Any]], collections.abc.Mapping[str, Any]]

effect_keys#

Required when effects is callable; lists the state keys the callable may produce. Must be None for static effects.

Type:

frozenset[str] | None

cost#

Static cost or callable that computes cost from world state.

Type:

float | langgoap.types.CostFunction

execute#

The function to call when the action is executed (sync).

Type:

Callable[[…], Any] | None

aexecute#

Optional async execute callable. When set, the async executor will await this instead of calling execute.

Type:

Callable[[…], Any] | None

effect_validator#

Optional runtime postcondition checker. Signature: (pre_state: dict, post_state: dict) -> bool. Called by the executor after execution to verify actual effects. Returning False signals that the action did not produce its declared effects (treated as a soft failure prompting replanning).

Type:

Callable[[dict[str, Any], dict[str, Any]], bool] | None

max_retries#

Number of allowed retries before blacklisting. 0 (default) means the action is blacklisted on its first failure. N allows N retries before blacklisting (i.e., the action is blacklisted after N+1 total failures).

Type:

int

require_human_approval#

When True, the executor calls interrupt() before running this action, pausing the graph until a human resumes via Command(resume=...). Requires a checkpointer. The resume value is interpreted as approval: None, True, or {"approved": True} continue execution; False or {"approved": False} deny the action and trigger replanning.

Type:

bool | type

resources#

Estimated resource consumption for this action. Used by the CSP optimizer. Example: {"tokens": 500, "cost_usd": 0.02, "api_calls": 1}.

Type:

collections.abc.Mapping[str, float] | None

duration#

Estimated wall-clock duration. Used by the CSP optimizer for temporal scheduling.

Type:

datetime.timedelta | None

metadata#

Arbitrary key-value metadata (description, tags, etc.).

Type:

collections.abc.Mapping[str, Any] | None

__init__(name, preconditions=<factory>, effects=<factory>, effect_keys=None, cost=1.0, execute=None, aexecute=None, effect_validator=None, max_retries=0, qos=None, can_rerun=True, read_only=False, utility=None, require_human_approval=False, human_input_key=None, resources=None, duration=None, metadata=None)#
Parameters:

  • name (str)

  • preconditions (Mapping[str, Any])

  • effects (Mapping[str, Any] | Callable[[Mapping[str, Any]], Mapping[str, Any]])

  • effect_keys (frozenset[str] | None)

  • cost (float | CostFunction)

  • execute (Callable[[...], Any] | None)

  • aexecute (Callable[[...], Any] | None)

  • effect_validator (Callable[[dict[str, Any], dict[str, Any]], bool] | None)

  • max_retries (int)

  • qos (ActionQos | None)

  • can_rerun (bool)

  • read_only (bool)

  • utility (float | CostFunction | None)

  • require_human_approval (bool | type)

  • human_input_key (str | None)

  • resources (Mapping[str, float] | None)

  • duration (timedelta | None)

  • metadata (Mapping[str, Any] | None)

Return type:

None

Methods

__init__(name[, preconditions, effects, ...])

effect_key_set()

Return the set of state keys this action may touch.

get_cost([world_state])

Resolve the cost, calling the cost function if dynamic.

get_effects(world_state)

Resolve effects against world_state.

has_effects()

Return True if this action declares at least one effect.

is_async_execute()

Return True if this action has an async execute path.

validate_effects(pre_state, post_state)

Verify that the action's effects were actually achieved.

Attributes

name

preconditions

effects

effect_keys

cost

execute

aexecute

effect_validator

max_retries

qos

can_rerun

read_only

utility

require_human_approval

human_input_key

resources

duration

metadata

has_dynamic_effects

Return True if effects is resolved from live world state.