Learn LangChain - Advanced LangGraph

#Introduction

In episode 13 you assembled create_agent and composed agent logic as a StateGraph with nodes, edges, conditional edges, even interrupts for human-in-the-loop; in episode 7 you also touched on simple checkpointing via MemorySaver and thread_id. Those two foundations are the springboard for the next step: graphs that persist in production. This episode dives deeper into LangGraph — starting with typed states and reducers so the state contract is explicit, checkpointing that saves a snapshot at every step, time travel to review, replay, and rewrite history, subgraphs to compose large graphs from small ones, then PostgreSQL persistence with namespaces and TTL to control the storage lifecycle.

#Typed States and Reducers

The StateGraph in episode 13 used a plain dict. In production, state must have a clear contract — that's what typed states are for: declaring the state schema with TypedDict and marking each channel with an update rule via reducers.

Typed state with a reducer

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages
class BlogState(TypedDict):
    pesan: Annotated[list, add_messages]
    draf: str
    jumlah_update: Annotated[int, add]

Without a reducer, the default behavior of every channel is last-write-wins: the newest value written by a node overwrites the old value. A reducer changes that behavior into merge. add_messages from langgraph.graph.message merges message lists without duplicates, operator.add merges values (suitable for lists and integers), and a reducer can be any custom function that takes the old value and new value and returns the merged result:

Custom reducer

def ambil_terpanjang(lama: str, baru: str) -> str:
    return baru if len(baru) >= len(lama) else lama
 
class DraffState(TypedDict):
    draf: Annotated[str, ambil_terpanjang]

The most important reducer in the agent world is add_messages — it's what keeps conversation history growing instead of being overwritten every turn. Choose reducers to match the channel semantics: accumulating lists use operator.add, messages use add_messages, and special scenarios just mean writing your own function.

#Checkpointing: Snapshot at Every Step

A checkpoint is a thread-state snapshot at every super-step — the boundary between graph execution steps. The checkpointer stores these snapshots and fuels memory, human-in-the-loop, time travel, and fault tolerance. Without a checkpointer, state is lost as soon as invoke finishes.

Compiling a graph with a checkpointer

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
 
def tulis_draf(state: BlogState):
    return {"pesan": [("user", "mulai menulis")], "draf": "Draf pertama.", "jumlah_update": 1}
builder = StateGraph(BlogState)
builder.add_node("tulis", tulis_draf)
builder.add_edge(START, "tulis")
builder.add_edge("tulis", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "thread-1"}}
graph.invoke({"pesan": [], "draf": "", "jumlah_update": 0}, config=config)

Every call includes thread_id in the configurable section — this is the primary storage key. The next invoke with the same thread_id continues from the last checkpoint. For production, MemorySaver isn't enough because it stores in RAM and is lost when the process restarts. Replace it with a persistent checkpointer: SqliteSaver (local file, pip install langgraph-checkpoint-sqlite) for development, and PostgresSaver for production — we'll cover that at the end.

#Time Travel: Reviewing, Replaying, and Rewriting History

Because every step is stored as a checkpoint, you can "travel through time". get_state fetches the latest snapshot, get_state_history returns the entire history (newest first), and each snapshot carries a config containing checkpoint_id.

Time travel with checkpoint_id

snapshot = graph.get_state(config)
print(snapshot.values)
print(snapshot.next)
history = list(graph.get_state_history(config))
titik_input = history[-1]
balik = {"configurable": {"thread_id": "thread-1", "checkpoint_id": titik_input.config["configurable"]["checkpoint_id"]}}
graph.invoke(None, config=balik)
graph.update_state(config, {"draf": "Draf revisi dari masa lalu"})

Adding checkpoint_id to the config makes invoke replay the execution from that point: nodes before the checkpoint are skipped (their results are already stored), nodes after it are re-executed. update_state writes a new value to the state and creates a new checkpoint without changing the old one — that's how you rewrite history or fork toward an alternate path. The update value still passes through the reducer, so channels like pesan accumulate rather than overwrite.

#Subgraphs: Graphs Within Graphs

A subgraph is a compiled graph installed as a node in another graph. This is the composition pattern: break a large workflow into small graphs that can be tested separately, then assemble them into one.

Subgraph as a node

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class ChildState(TypedDict):
    tugas: str
    hasil: str
 
def kerjakan(state: ChildState):
    return {"hasil": "diproses: " + state["tugas"]}
child_graph = StateGraph(ChildState)
child_graph.add_node("kerjakan", kerjakan)
child_graph.add_edge(START, "kerjakan")
child_graph.add_edge("kerjakan", END)
child = child_graph.compile()
 
class ParentState(TypedDict):
    tugas: str
    hasil: str
 
def jalankan_child(state: ParentState):
    return child.invoke({"tugas": state["tugas"]})
parent = StateGraph(ParentState)
parent.add_node("child", jalankan_child)
parent.add_edge(START, "child")
parent.add_edge("child", END)
graph = parent.compile(checkpointer=checkpointer)

State moves between parent and child by matching key names: keys that exist in the child schema are sent along, and keys the child writes that also exist in the parent schema are mapped back. Each subgraph also manages its own checkpoint namespace (checkpoint_ns holds the node name), so state changes inside a subgraph aren't necessarily visible to the parent right away — for data that must cross graph boundaries, consider a Store. Also important: only compile the parent graph with a checkpointer; a subgraph compiled with its own checkpointer breaks the namespace and bloats storage.

#Production Persistence with langgraph-checkpoint-postgres

For production, memory and SQLite aren't adequate: you need a durable checkpointer that supports async and is shared across instances. langgraph-checkpoint-postgres provides PostgresSaver and AsyncPostgresSaver, which store checkpoints in PostgreSQL. Install with pip install langgraph-checkpoint-postgres.

Installing the PostgreSQL checkpointerPostgresSaver with from_conn_string

pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver
 
DB_URI = "postgresql://langgraph:password@localhost:5432/langgraph"
 
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

setup() creates the checkpoints table and its indexes — run it once when first using a new database. Every invoke with a thread_id now writes a checkpoint to Postgres, survives restarts, and is accessible from any instance. For async applications, use AsyncPostgresSaver from langgraph.checkpoint.postgres.aio. Under high load, avoid a single connection; use a connection pool:

Connection pool for production load

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
pool = ConnectionPool(DB_URI, max_size=10, kwargs={"autocommit": True, "row_factory": dict_row})
checkpointer = PostgresSaver(pool)
checkpointer.setup()

The two required settings above — autocommit and a dict_row row_factory — are needed so setup() can commit the table and so PostgresSaver can access columns using dict syntax. This pattern is used in production deployments because connections are shared and don't time out on long-running workflows.

#Namespaces and TTL: Data Isolation and Lifecycle

Every checkpoint has a namespace (checkpoint_ns). A value of "" means the parent graph; subgraphs use the node name plus a unique id, and nested subgraphs are joined with the | separator. This namespace is what separates the parent graph's checkpoints from subgraph checkpoints within the same thread — and this isolation matters for multi-tenancy: design thread_id with a clear scheme (user id, workflow type, timestamp) so threads don't overlap. PostgresSaver stores thread_id in a length-limited column, so keep it under 255 characters, for example with a UUID or hash. TTL (time-to-live) controls how long checkpoints are retained. On the LangGraph Platform, the retention policy is configured in langgraph.json:

{
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "checkpointer": {
    "ttl": {
      "strategy": "delete",
      "sweep_interval_minutes": 60,
      "default_ttl": 43200
    }
  }
}

default_ttl is expressed in minutes (43200 = 30 days) and counted from the last activity — each new run resets the countdown. The strategy value delete removes the entire thread along with its checkpoints and writes, while keep_latest retains the thread and its latest checkpoint while pruning old history; sweep_interval_minutes controls how often a background process checks for expired threads. TTL only applies to threads created after the configuration is applied, not retroactively. If you self-host and use PostgresSaver directly, TTL isn't automatic: checkpoints pile up without limit, so set up scheduled pruning — a cron job that deletes checkpoints rows older than the limit, or calls adelete_thread for stale threads. Establish a retention policy from the start, because data that's never cleaned up slows down every get_state_history and bloats storage costs.

#Conclusion

Episode 17 turned a simple StateGraph into an operable engine: typed states and reducers make the state contract explicit, checkpointing stores a snapshot at every step, time travel unlocks the ability to review, replay, and rewrite history, subgraphs break large graphs into testable components, and PostgreSQL persistence with namespaces and TTL ensures state lasts long, stays isolated, and doesn't grow out of control.

Key takeaways:

• Typed state with TypedDict plus reducers (add_messages, operator.add, or custom functions) determines how updates are merged; the default is last-write-wins.
• The checkpointer stores a checkpoint per super-step and always needs thread_id; MemorySaver is for development only, production needs PostgresSaver or AsyncPostgresSaver.
• Time travel uses get_state, get_state_history, and checkpoint_id for replay, plus update_state to rewrite or fork history.
• Subgraphs share state via key matching and have their own checkpoint namespace; compile only the parent with a checkpointer.
• TTL is configured via checkpointer.ttl in langgraph.json on the platform; when self-hosting, scheduled pruning is required because checkpoints pile up without limit.

Checkpointing and time travel add storage and latency overhead that isn't free. In episode 18 we work on Performance & Cost Optimization: streaming, batching, and parallelism to reduce latency, plus token usage tracking, prompt caching, model tiering, and caching layers to reduce costs. See you there!