The Serverless Memory Table

You know the feeling of calling a support line, explaining your whole problem, getting cut off, and calling back — and the person who picks up has no idea who you are. A good one pulls up your ticket, reads what the last agent wrote, and carries on like nothing was lost. You never notice the handoff. The notes did the work.

That is, almost exactly, the problem this course solves. Except the "agent" here is an AWS Lambda function, and it does not even get the dignity of being disconnected — it just ends, on purpose, the moment it finishes a piece of work. Lambda keeps nothing in memory between runs, and it will not run past fifteen minutes, full stop. So the instant your LangGraph agent needs to pause — "check with a human before booking this slot, then pick this back up" — you are in the exact same spot as that support call: whoever answers next, possibly hours later, on a completely different machine, needs the full notes of where things stood.

This course builds those notes: a durable memory table in DynamoDB that lets a LangGraph agent pause for as long as it needs and then resume on a fresh Lambda run as if no time had passed at all. By the end you will know exactly why that is necessary, and you will have a complete, production-ready, multi-tenant version of it to deploy.

Start with the mental model, because everything else follows from it.

A plain chatbot is essentially stateless. You send the whole conversation, the model responds, and nothing needs to persist on the server — if the client keeps the message history, the server can forget you entirely between requests. This is why simple LLM apps scale so easily.

An agent is different in kind, not just degree. It runs a loop (you saw this in The ReAct Loop): think, act, observe, repeat. Along the way it accumulates real state — what it has tried, what tools returned, what it is waiting for. And crucially, a serious agent sometimes has to stop and wait for a human. Consider a booking agent that wants to override a customer's slot: responsible design does not let it act unilaterally. It proposes, then pauses for approval.

Process-aware state machine is the phrase worth holding onto. The instant your agent can pause and wait, it stops being a request and becomes a process. And a process needs a place to keep its state while it waits — somewhere durable, outside the memory of any single running function.

  LLM WRAPPER (stateless)
  request ──► model ──► response ──► forgotten

  AGENTIC PROCESS (stateful)
  request ──► think ──► act ──► observe ──► think ──► [PAUSE: wait for human]
                                                          │
                          ...minutes or hours later...    │
                                                          ▼
              [RESUME exactly here] ──► act ──► done

Two vocabulary items we will use throughout:

• State — the full data the workflow is carrying right now (the request, partial results, what it is waiting for).
• Checkpoint — a saved snapshot of that state at a point in time, durable enough to reload later.

Before we build the machinery, it is worth being precise about why it is needed — because "human in the loop" is often waved around as a virtue rather than a design decision. There are three distinct forces that put a person in front of an agent's action, and they show up in very different industries.

The first is law. Several jurisdictions now require a human in specific loops. The EU's GDPR, Article 22 gives people the right not to be subject to a decision based solely on automated processing where it produces legal or similarly significant effects — and an explicit right to obtain human intervention. The EU AI Act goes further for systems it classes as high-risk, requiring effective human oversight (Article 14). New York City's Local Law 144 requires bias audits and candidate notification for automated hiring tools. If your agent denies a loan, screens a CV, or closes an account, this is not a philosophical debate; it is a compliance requirement.

The second is irreversibility. Some actions cannot be taken back. Money that has settled, a deleted database, a sent email, a permanently banned account. Finance solved this long before AI existed with the four-eyes principle — a second authoriser for payments above a threshold — and the same logic transfers directly to an agent with payment tools.

The third is trust. Sometimes nothing forces the pause and you should build it anyway, because being technically right and socially wrong still loses you a customer. Overriding somebody's booking is the example we will build.

Now the harder question, and the one most teams get wrong: which actions deserve a gate? Gating everything feels safe and is actively dangerous — a human asked to approve forty trivial things a day stops reading by the fifth, and your real safety control decays into a reflex click. This is approval fatigue, and it is the same failure mode as alarm fatigue in hospitals.

Two properties predict regret better than anything else: can it be undone, and how far does it reach. Put them on two axes and the policy writes itself:

LangGraph models an agent as a graph: nodes are steps (functions), edges decide what runs next, and a shared state object flows through them. Instead of one opaque model call, you get an explicit state machine you can inspect, pause, and resume — which is exactly what a durable agent needs.

The component that gives a LangGraph app a memory is the checkpointer. You attach one when you compile the graph, and from then on LangGraph automatically saves a checkpoint after every step. Persistence is what unlocks the features we care about:

• Durability — the workflow survives a crash or a redeploy.
• Resumption — a paused graph continues from the exact step it stopped at.
• Human-in-the-loop — an interrupt pauses the graph; the checkpoint holds its place until a human replies.
• Time-travel & audit — every past state is on record.

LangGraph ships in-memory and SQLite/Postgres checkpointers. To run on Lambda with DynamoDB, we implement the same interface ourselves. That interface is a small class, BaseCheckpointSaver, with four methods that matter:

Every call is addressed by a config object carrying three identifiers, which become the backbone of our table:

• thread_id — the unique id of one workflow/session (one user's booking, say).
• checkpoint_ns — a namespace, usually empty, used by subgraphs.
• checkpoint_id — a specific snapshot; omit it to mean "the latest".

Now the constraint that makes this interesting. AWS Lambda runs your code in a container that is spun up on demand and then frozen or destroyed. It gives you two hard facts to design around:

1. It is stateless. Anything you hold in a Python variable is gone once the invocation ends. There is no reliable "next time" for in-memory state — the next request may hit a completely different container.
2. It is time-limited. A single invocation may run for at most fifteen minutes. You cannot simply "wait" inside the function for a human to click approve.

Put those together with a human-in-the-loop agent and the difficulty is stark. The graph runs, reaches the approval step, and must pause. But it cannot sit in memory waiting — the function has to return, and its memory will vanish. The approval might arrive an hour later, as a separate HTTP request that lands on a different Lambda container with no knowledge of the first.

The only way through is to move the state out of the function and into durable storage, so the second invocation can pick up exactly where the first left off. That store is our DynamoDB table — the "serverless memory table" of the title. DynamoDB is the natural partner for Lambda: fully managed, serverless itself, single-digit-millisecond reads, and priced per request so an idle workflow costs almost nothing while it waits.

  ── Invocation 1: start ─────────────────────────────
  API Gateway ─► Lambda ─► LangGraph ─┐
                                      ├─ save checkpoint ─► DynamoDB
                    reach approval ───┘   (status = PAUSED_FOR_HUMAN)
  Lambda returns "please approve"  ◄── function memory discarded

  ============ human decides, minutes/hours later ============

  ── Invocation 2: resume (a NEW container) ──────────
  Approval webhook ─► Lambda ─► load checkpoint ◄─ DynamoDB
                                LangGraph resumes at the approval step
                                finish work ─► save ─► DynamoDB (COMPLETED)

DynamoDB rewards a single-table design: one table holds several item types, distinguished by how the keys are shaped, so related data sits in the same partition and is fetched in one query. We will store three kinds of item — checkpoints, writes, and (implicitly) the operational metadata on the checkpoint rows.

The keys. Conceptually the workflow is identified by thread_id and a snapshot by checkpoint_id. Physically we shape the keys to do two jobs at once — co-locate a thread's data, and isolate tenants:

• Partition key pk = TENANT#thread_id. Baking the tenant into the partition key means a request literally cannot address another tenant's data — isolation by construction, not by a filter you might forget. (More on this in Chapter 7.)
• Sort key sk = a prefixed string that both distinguishes item types and orders snapshots:
  • CKPT#<ns>#<checkpoint_id> for a checkpoint
  • WRITE#<ns>#<checkpoint_id>#<task_id>#<idx> for an intermediate write

Because checkpoint_id values sort chronologically, "get the latest checkpoint" is a single query with ScanIndexForward=False, Limit=1, and "get the writes for that checkpoint" is a begins_with query in the same partition.

  pk = ACME#booking-9f3
  ┌──────────────────────────────────────────────────────────────┐
  │ sk = CKPT#  #01J...a   status=RUNNING           checkpoint,meta │
  │ sk = CKPT#  #01J...b   status=PAUSED_FOR_HUMAN  checkpoint,meta │  ◄─ latest
  │ sk = WRITE# #01J...b#task-7#0000                channel,value   │
  │ sk = WRITE# #01J...b#task-7#0001                channel,value   │
  └──────────────────────────────────────────────────────────────┘
     newest checkpoint = highest sk  →  one query, Limit 1

The attributes. Each checkpoint item carries the serialized state, the identifiers, a parent pointer (for time-travel), the operational metadata, and a TTL for automatic cleanup.

Write items are smaller: task_id, channel, type + value (compressed), plus the same keys and TTL.

Access patterns, the DynamoDB way (design the table around the questions you will ask):

Here is a complete BaseCheckpointSaver for the schema above. Read it once for shape, then we will walk the important parts.

import zlib
from datetime import datetime, timedelta, timezone
from typing import Any, Iterator, Optional, Tuple

import boto3
from boto3.dynamodb.conditions import Key
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
    CheckpointTuple,
    get_checkpoint_id,
)
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

TTL_DAYS = 30


def _ttl() -> int:
    return int((datetime.now(timezone.utc) + timedelta(days=TTL_DAYS)).timestamp())


class DynamoDBSaver(BaseCheckpointSaver):
    """A durable LangGraph checkpointer backed by a single DynamoDB table.

    Construct one per request, bound to the tenant_id from the verified JWT.
    """

    def __init__(self, table_name: str, tenant_id: str, region: Optional[str] = None):
        super().__init__(serde=JsonPlusSerializer())
        self.tenant_id = tenant_id
        self._table = boto3.resource("dynamodb", region_name=region).Table(table_name)

    # ---- key helpers ---------------------------------------------------
    def _pk(self, thread_id: str) -> str:
        return f"{self.tenant_id}#{thread_id}"          # tenant baked in

    def _ckpt_sk(self, ns: str, checkpoint_id: str) -> str:
        return f"CKPT#{ns}#{checkpoint_id}"

    def _write_sk(self, ns: str, checkpoint_id: str, task_id: str, idx: int) -> str:
        return f"WRITE#{ns}#{checkpoint_id}#{task_id}#{idx:04d}"

    # ---- serialize + compress -----------------------------------------
    def _dump(self, obj: Any) -> Tuple[str, bytes]:
        type_, blob = self.serde.dumps_typed(obj)      # (tag, bytes)
        return type_, zlib.compress(blob)

    def _load(self, type_: str, blob: bytes) -> Any:
        return self.serde.loads_typed((type_, zlib.decompress(bytes(blob))))

    # ---- WRITE: save a checkpoint --------------------------------------
    def put(self, config, checkpoint, metadata, new_versions) -> RunnableConfig:
        thread_id = config["configurable"]["thread_id"]
        ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = checkpoint["id"]
        parent_id = config["configurable"].get("checkpoint_id")

        c_type, c_blob = self._dump(checkpoint)
        m_type, m_blob = self._dump(metadata)

        self._table.put_item(Item={
            "pk": self._pk(thread_id),
            "sk": self._ckpt_sk(ns, checkpoint_id),
            "tenant_id": self.tenant_id,
            "thread_id": thread_id,
            "checkpoint_id": checkpoint_id,
            "parent_checkpoint_id": parent_id,
            "type": c_type,
            "checkpoint": c_blob,
            "metadata_type": m_type,
            "metadata": m_blob,
            "status": "RUNNING",
            "ttl": _ttl(),
        })
        return {"configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": ns,
            "checkpoint_id": checkpoint_id,
        }}

    # ---- WRITE: save intermediate writes ------------------------------
    def put_writes(self, config, writes, task_id, task_path="") -> None:
        thread_id = config["configurable"]["thread_id"]
        ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = config["configurable"]["checkpoint_id"]
        pk = self._pk(thread_id)
        with self._table.batch_writer() as batch:
            for idx, (channel, value) in enumerate(writes):
                v_type, v_blob = self._dump(value)
                batch.put_item(Item={
                    "pk": pk,
                    "sk": self._write_sk(ns, checkpoint_id, task_id, idx),
                    "tenant_id": self.tenant_id,
                    "task_id": task_id,
                    "channel": channel,
                    "type": v_type,
                    "value": v_blob,
                    "ttl": _ttl(),
                })

    # ---- READ: load a checkpoint (+ its pending writes) ---------------
    def get_tuple(self, config) -> Optional[CheckpointTuple]:
        thread_id = config["configurable"]["thread_id"]
        ns = config["configurable"].get("checkpoint_ns", "")
        pk = self._pk(thread_id)
        checkpoint_id = get_checkpoint_id(config)      # None means "latest"

        if checkpoint_id:
            item = self._table.get_item(
                Key={"pk": pk, "sk": self._ckpt_sk(ns, checkpoint_id)}
            ).get("Item")
        else:
            items = self._table.query(
                KeyConditionExpression=Key("pk").eq(pk)
                & Key("sk").begins_with(f"CKPT#{ns}#"),
                ScanIndexForward=False,                # newest first
                Limit=1,
            ).get("Items", [])
            item = items[0] if items else None

        if not item:
            return None

        ckpt_id = item["checkpoint_id"]
        checkpoint = self._load(item["type"], item["checkpoint"])
        metadata = self._load(item["metadata_type"], item["metadata"])

        writes = self._table.query(
            KeyConditionExpression=Key("pk").eq(pk)
            & Key("sk").begins_with(f"WRITE#{ns}#{ckpt_id}#"),
        ).get("Items", [])
        pending = [
            (w["task_id"], w["channel"], self._load(w["type"], w["value"]))
            for w in writes
        ]

        parent = item.get("parent_checkpoint_id")
        parent_config = (
            {"configurable": {"thread_id": thread_id, "checkpoint_ns": ns,
                              "checkpoint_id": parent}}
            if parent else None
        )
        return CheckpointTuple(
            config={"configurable": {"thread_id": thread_id, "checkpoint_ns": ns,
                                     "checkpoint_id": ckpt_id}},
            checkpoint=checkpoint,
            metadata=metadata,
            parent_config=parent_config,
            pending_writes=pending,
        )

    # ---- READ: list a thread's checkpoints ----------------------------
    def list(self, config, *, filter=None, before=None, limit=None) -> Iterator[CheckpointTuple]:
        thread_id = config["configurable"]["thread_id"]
        ns = config["configurable"].get("checkpoint_ns", "")
        pk = self._pk(thread_id)
        kwargs: dict = {
            "KeyConditionExpression": Key("pk").eq(pk)
            & Key("sk").begins_with(f"CKPT#{ns}#"),
            "ScanIndexForward": False,
        }
        if limit:
            kwargs["Limit"] = limit
        for item in self._table.query(**kwargs).get("Items", []):
            yield CheckpointTuple(
                config={"configurable": {"thread_id": thread_id, "checkpoint_ns": ns,
                                         "checkpoint_id": item["checkpoint_id"]}},
                checkpoint=self._load(item["type"], item["checkpoint"]),
                metadata=self._load(item["metadata_type"], item["metadata"]),
                parent_config=None,
                pending_writes=None,
            )

    # ---- operational metadata (denormalized, app-managed) -------------
    def mark_status(self, config, status: str, next_node: Optional[str] = None) -> None:
        thread_id = config["configurable"]["thread_id"]
        ns = config["configurable"].get("checkpoint_ns", "")
        checkpoint_id = config["configurable"]["checkpoint_id"]
        expr, names, vals = "SET #s = :s", {"#s": "status"}, {":s": status}
        if next_node is not None:
            expr += ", next_node = :n"
            vals[":n"] = next_node
        self._table.update_item(
            Key={"pk": self._pk(thread_id), "sk": self._ckpt_sk(ns, checkpoint_id)},
            UpdateExpression=expr,
            ExpressionAttributeNames=names,
            ExpressionAttributeValues=vals,
        )

A few parts deserve a closer look.

• Serialization is not optional glue — it is the whole trick. LangGraph state contains rich objects (messages, tool calls, custom types), not just strings. JsonPlusSerializer.dumps_typed turns any of it into a (tag, bytes) pair, and loads_typed turns it back. We wrap both in zlib compression, which matters for the next point.
• Compression buys you headroom. A DynamoDB item is capped at 400 KB. Agent state can grow (long histories, retrieved documents), and compression typically shaves it by half or more, keeping you under the limit and lowering write cost.
• get_tuple does two reads on purpose. First the checkpoint, then its pending writes — because LangGraph needs both to correctly resume a step that was mid-flight when the process stopped.
• The tenant is inside the key, always. Every read and write goes through _pk, which prepends self.tenant_id. There is no code path that touches a bare thread_id.

Now we spend the durability we just built. LangGraph's interrupt function stops the graph mid-node, hands a payload back to the caller, and — because a checkpointer is attached — saves the exact position so a later call can resume. On Lambda, "a later call" is a whole new invocation, and this is where the DynamoDB table earns its keep.

Here is a minimal booking graph with an approval gate:

from typing import Optional, TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt


class BookingState(TypedDict):
    request: str
    slot: Optional[str]
    approved: Optional[bool]


def propose_slot(state: BookingState) -> dict:
    # (call your model/tools to pick a slot)
    return {"slot": "2026-09-10T14:00"}


def human_approval(state: BookingState) -> dict:
    # Execution STOPS here and returns the payload to the caller.
    # It resumes on this exact line when a decision arrives.
    decision = interrupt({
        "question": "Override the customer's slot to this time?",
        "slot": state["slot"],
    })
    return {"approved": bool(decision.get("approve"))}


def finalize(state: BookingState) -> dict:
    if state["approved"]:
        ...  # actually perform the booking override
    return {}


def build_graph(checkpointer):
    g = StateGraph(BookingState)
    g.add_node("propose_slot", propose_slot)
    g.add_node("human_approval", human_approval)
    g.add_node("finalize", finalize)
    g.add_edge(START, "propose_slot")
    g.add_edge("propose_slot", "human_approval")
    g.add_edge("human_approval", "finalize")
    g.add_edge("finalize", END)
    return g.compile(checkpointer=checkpointer)   # ← durability attaches here

And the Lambda handler that runs it. Notice it handles both invocations — the initial start and the later resume — because on Lambda they are separate HTTP requests hitting the same code:

import json
import os

from langgraph.types import Command

from auth import tenant_and_thread     # verifies the Cognito JWT (Chapter 7)
from checkpointer import DynamoDBSaver
from graph import build_graph

TABLE = os.environ["AGENT_STATE_TABLE"]


def _resp(code: int, body: dict) -> dict:
    return {"statusCode": code, "body": json.dumps(body)}


def handler(event, context):
    # tenant_id comes ONLY from the verified token — never from the client body
    tenant_id, thread_id, body = tenant_and_thread(event)

    saver = DynamoDBSaver(TABLE, tenant_id=tenant_id)
    graph = build_graph(saver)
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}

    if body.get("resume") is not None:
        # INVOCATION 2 — a human answered; continue from the interrupt
        result = graph.invoke(Command(resume=body["resume"]), config)
    else:
        # INVOCATION 1 — start the workflow
        result = graph.invoke(body["input"], config)

    # After running, ask the graph where it stands
    snapshot = graph.get_state(config)
    latest = {"configurable": {**config["configurable"],
                              "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"]}}

    if snapshot.next:  # there are pending nodes → we paused (e.g. for approval)
        ask = result["__interrupt__"][0].value
        saver.mark_status(latest, "PAUSED_FOR_HUMAN", next_node=",".join(snapshot.next))
        return _resp(202, {"status": "PAUSED_FOR_HUMAN", "thread_id": thread_id, "ask": ask})

    saver.mark_status(latest, "COMPLETED")
    return _resp(200, {"status": "COMPLETED", "result": result})

Walk the lifecycle once and the whole course clicks into place:

  invoke(input) ─► propose ─► [interrupt] ─► put() ─► DynamoDB(PAUSED)
                                    │
                                    ▼
                             return 202 "approve?"     ← container gone

        · · · human approves later · · ·

  invoke(Command(resume)) ─► get_tuple() ◄─ DynamoDB
                             resume at interrupt line
                             finalize ─► put() ─► DynamoDB(COMPLETED)

That is the flow in the abstract. It is worth watching it concretely — stepping through the same run while seeing exactly which rows appear in DynamoDB, and what happens to the table at the moment the Lambda container is destroyed:

If more than one customer uses this system, a single leaked workflow is a serious breach. Multi-tenant isolation is not a feature to bolt on later; it is a property the schema and code must guarantee.

Rule one: the tenant identity comes only from a verified token. Put a Cognito authorizer in front of the API. It validates the JWT and hands your Lambda the verified claims. You read tenant_id from those claims — never from the request body, query string, or a header the client controls, all of which an attacker can forge.

def tenant_and_thread(event) -> tuple[str, str, dict]:
    # API Gateway (HTTP API) with a Cognito JWT authorizer places the
    # *verified* claims here. Trust these; trust nothing from the raw body.
    claims = event["requestContext"]["authorizer"]["jwt"]["claims"]
    tenant_id = claims["custom:tenant_id"]          # a custom Cognito attribute

    body = json.loads(event.get("body") or "{}")
    thread_id = body["thread_id"]                    # which workflow, not which tenant
    return tenant_id, thread_id, body

Rule two: isolate by construction, not by filter. Because our partition key is TENANT#thread_id, and tenant_id is fixed from the token at construction time, the saver cannot form a key that points at another tenant. There is no WHERE tenant = ? to forget — the wrong tenant's data is simply unaddressable. This is the single most valuable property of the schema.

Rule three: defence in depth. Isolation-by-key is strong, but layer more behind it:

• Least-privilege IAM. The Lambda role gets only GetItem, PutItem, Query, BatchWriteItem, UpdateItem on this one table — nothing else.
• Physical key scoping (advanced). For the strictest isolation, mint short-lived per-tenant credentials with STS and an IAM policy that uses the dynamodb:LeadingKeys condition, so even a compromised function cannot read outside its tenant's partitions.
• Encryption. Enable server-side encryption at rest (on by default; use a KMS key you control for sensitive tenants) and TLS in transit (default for the SDK).
• Don't log secrets. Log tenant_id and thread_id for tracing, never the deserialized state, which may hold personal data. This matters when we add observability next.

The code is done; now make it survive production. First, the table:

aws dynamodb create-table \
  --table-name Agent-State \
  --attribute-definitions \
      AttributeName=pk,AttributeType=S \
      AttributeName=sk,AttributeType=S \
  --key-schema \
      AttributeName=pk,KeyType=HASH \
      AttributeName=sk,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification Enabled=true

# Auto-expire old workflows to control cost and data retention
aws dynamodb update-time-to-live \
  --table-name Agent-State \
  --time-to-live-specification "Enabled=true, AttributeName=ttl"

# Recommended: turn on point-in-time recovery for a safety net
aws dynamodb update-continuous-backups --table-name Agent-State \
  --point-in-time-recovery-specification PointInTimeRecoveryEnabled=true

Then the operational realities, each with its defence:

You now have the full picture: why a durable memory is needed the instant an agent can pause, what a LangGraph checkpointer must do, and how to implement one on DynamoDB that is compressed, multi-tenant, and safe to resume across separate Lambda invocations. The natural next steps are to wire this into an approvals dashboard (the GSI already supports it), to add the async methods if you move to ainvoke, and to reuse the exact same pattern for any long-running, human-gated workflow — refunds, moderation, deployments — not just bookings.