Scheduled delivery confirmer

Scheduled delivery confirmer#

Collect a structured payload from a human before an action runs, using LangGraph’s interrupt() plus a Pydantic schema.

A boolean approve/deny gate isn’t enough for a same-day delivery workflow: the dispatcher needs a typed payload (window, gate code, signature waiver, instructions) so downstream systems can route the truck without anyone transcribing free-form text.

Typed-form HITL in LangGOAP:

  1. The action declares require_human_approval=DeliveryConfirmation, a Pydantic BaseModel.

  2. The executor calls interrupt() with the model’s JSON schema, so any UI or chat client can render a real form.

  3. On Command(resume=...), the executor validates the payload via Model.model_validate. On success, the parsed form is merged into world state under human_input_key and the action proceeds. On failure, the action is denied with a structured human_form_invalid error.

Backed by tests/integration/test_form_hitl.py.

The form schema#

Everything the dispatcher needs in one Pydantic model. When the executor pauses, the interrupt payload includes Model.model_json_schema() so any client can render a typed form.

from datetime import date
from typing import Any

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from pydantic import BaseModel, Field

from langgoap import ActionSpec, GoalSpec, GoapGraph


class DeliveryConfirmation(BaseModel):
    """Typed payload the customer must complete before dispatch."""

    delivery_date: date = Field(description='Confirmed delivery date.')
    window_start: str = Field(description='30-minute window start, HH:MM.')
    notes: str = Field(description='Special instructions for the driver.')
    gate_code: str | None = Field(default=None, description='Optional gate code.')
    signature_waiver: bool = Field(
        default=False, description='Allow contactless drop-off.'
    )


DeliveryConfirmation.model_json_schema()

{'description': 'Typed payload the customer must complete before dispatch.',
 'properties': {'delivery_date': {'description': 'Confirmed delivery date.',
   'format': 'date',
   'title': 'Delivery Date',
   'type': 'string'},
  'window_start': {'description': '30-minute window start, HH:MM.',
   'title': 'Window Start',
   'type': 'string'},
  'notes': {'description': 'Special instructions for the driver.',
   'title': 'Notes',
   'type': 'string'},
  'gate_code': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
   'default': None,
   'description': 'Optional gate code.',
   'title': 'Gate Code'},
  'signature_waiver': {'default': False,
   'description': 'Allow contactless drop-off.',
   'title': 'Signature Waiver',
   'type': 'boolean'}},
 'required': ['delivery_date', 'window_start', 'notes'],
 'title': 'DeliveryConfirmation',
 'type': 'object'}

Two-action plan: prepare → confirm_delivery (gated)#

def prepare(world_state: dict[str, Any]) -> dict[str, Any]:
    return {'package_packed': True}


def dispatch(world_state: dict[str, Any]) -> dict[str, Any]:
    confirmation = world_state.get('confirmation', {})
    print('  [dispatch] dispatching with confirmed payload:')
    print(f'    window: {confirmation.get("window_start")} on '
          f'{confirmation.get("delivery_date")}')
    print(f'    notes:  {confirmation.get("notes")!r}')
    print(f'    waive signature: {confirmation.get("signature_waiver")}')
    return {'dispatched': True}


actions = [
    ActionSpec(
        name='prepare',
        preconditions={},
        effects={'package_packed': True},
        execute=prepare,
    ),
    ActionSpec(
        name='confirm_delivery',
        preconditions={'package_packed': True},
        effects={'dispatched': True},
        execute=dispatch,
        require_human_approval=DeliveryConfirmation,
        human_input_key='confirmation',  # where the parsed form lands
    ),
]

graph = GoapGraph(actions).compile(checkpointer=MemorySaver())
config = {'configurable': {'thread_id': 'demo-delivery-1'}}

Phase 1 — agent runs, then pauses on the typed-form interrupt#

The first invoke runs prepare, then hits the form gate before confirm_delivery. We inspect the interrupt payload and confirm the JSON schema is right there for any UI / chat client to render.

first = graph.invoke(
    {
        'goal': GoalSpec(conditions={'dispatched': True}),
        'world_state': {},
    },
    config=config,
)

interrupt_obj = first['__interrupt__'][0]
payload = interrupt_obj.value
print(f"interrupt type:  {payload['type']}")
print(f"action:          {payload['action']}")
print(f"model name:      {payload['model_name']}")
print(f"required fields: {payload['form_schema']['required']}")

interrupt type:  goap_action_form
action:          confirm_delivery
model name:      DeliveryConfirmation
required fields: ['delivery_date', 'window_start', 'notes']

Phase 2 — customer responds with a valid form#

The customer (or chat-client UI) responds with the typed payload. Command(resume=...) carries it back; the executor validates against the Pydantic schema and merges the parsed form into world state under confirmation.

customer_response = {
    'delivery_date': '2026-05-15',
    'window_start': '14:30',
    'notes': 'Leave at side door near garage',
    'gate_code': '4521#',
    'signature_waiver': True,
}

second = graph.invoke(Command(resume=customer_response), config=config)

print(f"\nstatus:        {second['status']}")
print(f"dispatched:    {second['world_state']['dispatched']}")
print(f"confirmation:  {second['world_state']['confirmation']}")

  [dispatch] dispatching with confirmed payload:
    window: 14:30 on 2026-05-15
    notes:  'Leave at side door near garage'
    waive signature: True

status:        goal_achieved
dispatched:    True
confirmation:  {'delivery_date': '2026-05-15', 'window_start': '14:30', 'notes': 'Leave at side door near garage', 'gate_code': '4521#', 'signature_waiver': True}

Phase 3 — invalid form is rejected with a structured error#

If the customer submits a payload that fails Pydantic validation (missing required field, malformed date, etc.), the executor refuses to run the action and surfaces a human_form_invalid failure. This matches Embabel’s IllegalStateException on invalid form submission.

graph2 = GoapGraph(actions).compile(checkpointer=MemorySaver())
bad_config = {'configurable': {'thread_id': 'demo-delivery-bad'}}

graph2.invoke(
    {
        'goal': GoalSpec(conditions={'dispatched': True}),
        'world_state': {},
    },
    config=bad_config,
)

# Customer submits something invalid: bad date format, missing fields.
bad_response = {'delivery_date': 'tomorrow ish', 'window_start': '14:30'}

rejection = graph2.invoke(Command(resume=bad_response), config=bad_config)

print(f"status:        {rejection['status']}")
print(f"dispatched?    {'dispatched' in rejection['world_state']}")
confirm_failure = next(
    h for h in rejection['execution_history']
    if h.action_name == 'confirm_delivery'
)
print(f'\nfailure error: {confirm_failure.error}')

Action 'confirm_delivery' failed: human_form_invalid: form payload failed validation: [{'type': 'date_from_datetime_parsing', 'loc': ('delivery_date',), 'msg': 'Input should be a valid date or datetime, invalid character in year', 'input': 'tomorrow ish', 'ctx': {'error': 'invalid character in year'}, 'url': 'https://errors.pydantic.dev/2.13/v/date_from_datetime_parsing'}, {'type': 'missing', 'loc': ('notes',), 'msg': 'Field required', 'input': {'delivery_date': 'tomorrow ish', 'window_start': '14:30'}, 'url': 'https://errors.pydantic.dev/2.13/v/missing'}]

status:        executing
dispatched?    False

failure error: human_form_invalid: form payload failed validation: [{'type': 'date_from_datetime_parsing', 'loc': ('delivery_date',), 'msg': 'Input should be a valid date or datetime, invalid character in year', 'input': 'tomorrow ish', 'ctx': {'error': 'invalid character in year'}, 'url': 'https://errors.pydantic.dev/2.13/v/date_from_datetime_parsing'}, {'type': 'missing', 'loc': ('notes',), 'msg': 'Field required', 'input': {'delivery_date': 'tomorrow ish', 'window_start': '14:30'}, 'url': 'https://errors.pydantic.dev/2.13/v/missing'}]

What this gives a production logistics system#

  • Deterministic UI contract. The interrupt payload includes model_json_schema() — any client (web form, Slack modal, mobile app) can render a real typed form without LangGOAP knowing or caring.

  • No string parsing. The validated payload arrives downstream as structured fields ready for the routing system / GPS / billing.

  • Validation at the gate. An invalid submission fails the action immediately and gets blacklisted (matches the existing approve/deny denial path), so a misbehaving client can’t poison the dispatch.

  • Backwards compatible. Setting require_human_approval=True still gives you the boolean approve/deny gate — typed forms are the new opt-in mode.

Next steps#

  • See basics/typed_form_hitl.ipynb for the minimal interrupt + resume pattern in isolation.

  • For a plain approve/deny gate, set require_human_approval=True (a bool) instead of a Pydantic model class.

  • Persist the typed payload across process restarts by swapping MemorySaver for AsyncPostgresSaver or RedisSaver.