langgoap.goap_action

Contents

langgoap.goap_action#

goap_action(*, preconditions=None, effects=None, cost=1.0, name=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]#

Decorator that converts a function into an ActionSpec.

Works with both sync and async functions. Async functions are stored in aexecute and must be invoked via ainvoke() — the sync executor will raise a clear error if you accidentally call invoke() with an async-only action.

Sync usage:

@goap_action(
    preconditions={"has_data": True},
    effects={"report_ready": True},
    cost=2.0,
    resources={"tokens": 500, "cost_usd": 0.02},
    duration=timedelta(seconds=2),
)
def generate_report(state: dict) -> dict:
    return {"report_ready": True}

Async usage — requires GoapGraph.ainvoke():

@goap_action(
    preconditions={"has_question": True},
    effects={"answer_ready": True},
    cost=3.0,
)
async def ask_llm(state: dict) -> dict:
    response = await llm.ainvoke(state["question"])
    return {"answer_ready": True, "answer": response.content}
Parameters:

  • preconditions (dict[str, Any] | None)

  • effects (dict[str, Any] | None)

  • cost (float | CostFunction)

  • name (str | 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 (dict[str, float] | None)

  • duration (timedelta | None)

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

Return type:

Callable[[Callable[[…], Any]], ActionSpec]

Contents