State and reducers
Nodes return partial updates, never the whole state. A key without a reducer is overwritten; a key annotated with a reducer is merged. add_messages appends new messages and replaces one whose id matches, which is how you edit a message.
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import add_messages
class State(TypedDict):
question: str # no reducer: a new value overwrites the old one
log: Annotated[list[str], operator.add] # reducer: lists are concatenated
messages: Annotated[list, add_messages] # append, or replace by message id
print(operator.add(["start"], ["agent"]))
old = [HumanMessage("Hi", id="1"), AIMessage("Draft", id="2")]
print([m.content for m in add_messages(old, [AIMessage("Final", id="2")])])
print([m.content for m in add_messages(old, [AIMessage("More")])])
['start', 'agent']
['Hi', 'Final']
['Hi', 'Draft', 'More']
MessagesState is the ready-made state with just messages and add_messages; subclass it to add keys.
Nodes, edges and conditional edges
- A node is a function (sync or async) that takes the state and returns a dict of updates.
add_edge(a, b) always goes from a to b. START and END mark the entry and exit.
add_conditional_edges(a, router) calls router(state) and goes where it returns. Pass a list or dict of targets so the graph can be drawn.
- A node can also return
Command(goto="b", update={...}) to update state and route in one place.
- An edge back to an earlier node makes a cycle. That loop is the difference between a graph and a chain.
Cycles and the recursion limit
Every super-step counts towards recursion_limit. When a run hits it, LangGraph raises GraphRecursionError instead of looping forever. Set it per call in the config.
from typing import TypedDict
from langgraph.errors import GraphRecursionError
from langgraph.graph import START, StateGraph
class State(TypedDict):
n: int
builder = StateGraph(State)
builder.add_node("loop", lambda s: {"n": s["n"] + 1})
builder.add_edge(START, "loop")
builder.add_edge("loop", "loop") # a cycle with no exit
graph = builder.compile()
try:
graph.invoke({"n": 0}, {"recursion_limit": 5})
except GraphRecursionError as e:
print(type(e).__name__, str(e).splitlines()[0])
GraphRecursionError Recursion limit of 5 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
Older releases defaulted to 25 steps. In langgraph 1.2 the default is 10,007 (LANGGRAPH_DEFAULT_RECURSION_LIMIT), and create_agent sets 9,999, so set your own limit for agents that could loop.
Checkpointers, threads and time travel
Compile with a checkpointer and every super-step is saved as a checkpoint under the thread_id in the config. That gives you conversation memory, resumable runs after a crash, interrupts, and time travel: go back to an old checkpoint, change it, and run forward on a new branch.
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
class State(TypedDict):
n: int
builder = StateGraph(State)
builder.add_node("double", lambda s: {"n": s["n"] * 2})
builder.add_node("inc", lambda s: {"n": s["n"] + 1})
builder.add_edge(START, "double")
builder.add_edge("double", "inc")
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t"}}
print(graph.invoke({"n": 5}, config))
history = list(graph.get_state_history(config)) # newest first
before_inc = next(c for c in history if c.next == ("inc",))
print(before_inc.values)
fork = graph.update_state(before_inc.config, {"n": 100})
print(graph.invoke(None, fork))
{'n': 11}
{'n': 10}
{'n': 101}
Use InMemorySaver in tests and a database saver in production, such as PostgresSaver from langgraph-checkpoint-postgres.
Interrupts
interrupt(value) inside a node pauses the run and returns value to the caller under __interrupt__. It needs a checkpointer.
- Resume with
graph.invoke(Command(resume=answer), config) on the same thread. The node restarts from its first line, and interrupt() returns answer.
- Because of the restart, keep side effects after the
interrupt() call, or make them idempotent.
interrupt_before and interrupt_after at compile time pause around whole nodes; they are mostly for debugging.
Send: map-reduce and parallel branches
When the number of branches is only known at run time, return a list of Send(node, input) from a conditional edge. Each one runs the node with its own input, in parallel in the same super-step, and a reducer collects the results.
import operator
from typing import Annotated, TypedDict
from langgraph.graph import START, StateGraph
from langgraph.types import Send
class State(TypedDict):
topics: list[str]
summaries: Annotated[list[str], operator.add]
def fan_out(state: State):
return [Send("summarise", {"topic": t}) for t in state["topics"]]
def summarise(item: dict):
return {"summaries": [f"summary of {item['topic']}"]}
builder = StateGraph(State)
builder.add_node("summarise", summarise)
builder.add_conditional_edges(START, fan_out, ["summarise"])
graph = builder.compile()
print(graph.invoke({"topics": ["cats", "dogs", "owls"]}))
{'topics': ['cats', 'dogs', 'owls'], 'summaries': ['summary of cats', 'summary of dogs', 'summary of owls']}
For a fixed set of branches you don’t need Send: add several edges out of one node and the targets run in parallel.
Subgraphs
A compiled graph is a Runnable, so it can be a node in another graph. If both share state keys, pass it straight to add_node. If the schemas differ, call it inside a node function and map the state in and out. Subgraphs are how you build multi-agent systems from smaller, testable agents.
from typing import TypedDict
from langgraph.graph import START, StateGraph
class State(TypedDict):
text: str
def clean(state: State):
return {"text": state["text"].strip()}
inner = StateGraph(State)
inner.add_node("clean", clean)
inner.add_edge(START, "clean")
cleaner = inner.compile()
outer = StateGraph(State)
outer.add_node("cleaner", cleaner) # a compiled graph is a node
outer.add_node("shout", lambda s: {"text": s["text"].upper()})
outer.add_edge(START, "cleaner")
outer.add_edge("cleaner", "shout")
print(outer.compile().invoke({"text": " hi "}))
{'text': 'HI'}
Long-term memory: the Store
A checkpointer remembers one thread. A Store keeps JSON documents under namespaces, shared across threads: user preferences, learned facts. Nodes reach it, and the per-run context, through the Runtime argument.
from dataclasses import dataclass
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
def remember(state: MessagesState, runtime: Runtime[Context]):
ns = ("users", runtime.context.user_id)
runtime.store.put(ns, "prefs", {"language": "Romanian"})
return {}
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node("remember", remember)
builder.add_edge(START, "remember")
store = InMemoryStore()
graph = builder.compile(store=store)
graph.invoke({"messages": []}, context=Context(user_id="ana"))
print(store.get(("users", "ana"), "prefs").value)
{'language': 'Romanian'}
With an embedding index configured, store.search(namespace, query=...) finds memories by meaning.