langgraph reference

Pinned version: langgraph==1.2.11 (released 2026-08-11; it pulls in langgraph-checkpoint 4.2.0, langgraph-prebuilt 1.1.0 and langgraph-sdk 0.4.4; the SqliteSaver example also needs langgraph-checkpoint-sqlite==3.1.1). Everything below was derived by inspecting this exact installed package, and every example was executed against it on Python 3.13. Each heading links to reference.langchain.com, which is unversioned and always serves the latest 1.x. Links open in a new tab.

Part of a five-page set — see how the LangChain stack fits together for how langchain-core, langgraph and langchain fit together.

What langgraph is

• It is a runtime that executes a graph of Python functions over a shared, typed state. You declare the state as a TypedDict, register functions as nodes, and wire them with edges; the runtime decides execution order, runs independent nodes in parallel, merges each node's returned dict into the state (through a reducer per key, or by overwriting), and can save the state after every step so a run can pause, resume, or be replayed.
• "Low level" means the level at which agent architecture is expressed — where you decide the loop, the branches and the pause points — not "hard mode".
• It is model-agnostic. The runtime does not care what a node does. Every example below is a plain Python function that imports nothing but langgraph and stdlib typing.
• In LangChain 1.x, langchain depends on langgraph — not the other way around. langchain.agents.create_agent returns a CompiledStateGraph, so invoke, stream, checkpointing and interrupt on an agent behave exactly as documented below.

Layer order

1langchain  ──depends on──▶  langgraph  ──depends on──▶  langchain-core

Nothing below depends on anything above it. Verified against the installed packages:

• langchain_core does not import langgraph. After importing every module of langchain-core 1.6.2, 'langgraph' in sys.modules is False.
• langgraph does not import langchain. langchain is absent from its Requires-Dist, and importing langgraph.graph, langgraph.types, langgraph.checkpoint.memory and langgraph.prebuilt leaves 'langchain' in sys.modules False. Two optional code paths try it lazily inside try:/except ImportError: (init_chat_model for "provider:model" strings in the deprecated create_react_agent, and init_embeddings in the store) and fail soft without it.
• langgraph does depend on langchain-core (<2,>=1.4.7) — that is how a node can return message objects the rest of the ecosystem understands, and how MessagesState converts dicts into HumanMessage/AIMessage.

Every example in this file was executed in a venv where langchain is installed but never imported (checked via sys.modules).

Old tutorials warning: langgraph.prebuilt.create_react_agent still imports, but calling it emits LangGraphDeprecatedSinceV10: "create_react_agent has been moved to langchain.agents. Please update your import to from langchain.agents import create_agent. Deprecated in LangGraph V1.0 to be removed in V2.0." The other prebuilt pieces (ToolNode, tools_condition, InjectedState, ToolRuntime) remain, and create_agent uses them internally.

Index

StateGraph

The builder. StateGraph(State) takes the state schema — a TypedDict whose keys are the state and whose Annotated[..., reducer] metadata says how a key is merged (no reducer = overwrite). Optional input_schema=/output_schema= narrow what callers pass in and get back, and context_schema= types the read-only runtime.context nodes can receive. Then:

• add_node(name, fn) registers a function fn(state) -> dict (or -> Command). The function receives the whole state and returns only the keys it changes. Options: retry_policy, cache_policy, defer=True (run after all other branches finish), destinations (declares where a Command-returning node may goto, for drawing).
• add_edge(a, b) always runs b after a. add_edge([a, b], c) waits for both.
• add_conditional_edges(source, router, path_map) runs router(state) after source; its return value is looked up in path_map (a dict) or used as a node name directly (a list).
• add_sequence([f, g, h]) adds nodes and the edges between them in one call.
• compile(checkpointer=, store=, interrupt_before=, interrupt_after=, name=, cache=) validates the graph (every node reachable, every edge target exists) and returns a CompiledStateGraph.

 1from typing import TypedDict
 2
 3from langgraph.graph import END, START, StateGraph
 4
 5
 6class State(TypedDict):
 7    n: int
 8    label: str
 9
10
11def double(state: State) -> dict:
12    return {"n": state["n"] * 2}
13
14
15def classify(state: State) -> str:
16    return "even" if state["n"] % 2 == 0 else "odd"
17
18
19def mark_even(state: State) -> dict:
20    return {"label": "even"}
21
22
23def mark_odd(state: State) -> dict:
24    return {"label": "odd"}
25
26
27builder = StateGraph(State)
28builder.add_node("double", double)
29builder.add_node("mark_even", mark_even)
30builder.add_node("mark_odd", mark_odd)
31builder.add_edge(START, "double")
32builder.add_conditional_edges("double", classify, {"even": "mark_even", "odd": "mark_odd"})
33builder.add_edge("mark_even", END)
34builder.add_edge("mark_odd", END)
35
36graph = builder.compile()
37print(graph.invoke({"n": 5, "label": ""}))   # {'n': 10, 'label': 'even'}
38print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

START and END

Two virtual nodes that mark where execution enters and leaves the graph. They are not classes — they are the strings '__start__' and '__end__', which is exactly why they can be passed anywhere a node name is expected. add_edge(START, "x") makes x the entry point (the older set_entry_point("x") does the same); an edge or Command(goto=END) into END finishes the run. A graph may have several edges into END, and a node with no outgoing edge at all is a build error.

1from langgraph.graph import END, START
2
3print(repr(START), repr(END))   # '__start__' '__end__'
4print(isinstance(START, str))   # True

↩ back to index

CompiledStateGraph

What .compile() hands back, and the only thing you actually run. It is a Runnable, so invoke, batch, stream and their async twins work as on any other. On top of that it adds:

• Streaming modes. stream(input, stream_mode=…): "updates" (default) yields {node_name: returned_dict} as each node finishes; "values" yields the full state after each step; "messages" yields model tokens as they arrive; "custom" yields what nodes write via get_stream_writer(); "debug" yields everything.
• The state API, all taking config={"configurable": {"thread_id": …}} and requiring a checkpointer: get_state(config) returns a StateSnapshot (.values, .next — the nodes about to run — .tasks, .config); get_state_history(config) iterates snapshots newest-first; update_state(config, values, as_node=…) writes a new checkpoint as if a node had returned values.
• Introspection. get_graph() returns a drawable graph object (draw_mermaid(), draw_mermaid_png(), draw_ascii()); get_subgraphs() lists nested compiled graphs.

 1from typing import TypedDict
 2
 3from langgraph.graph import END, START, StateGraph
 4
 5
 6class State(TypedDict):
 7    n: int
 8
 9
10def double(state: State) -> dict:
11    return {"n": state["n"] * 2}
12
13
14def bump(state: State) -> dict:
15    return {"n": state["n"] + 1}
16
17
18builder = StateGraph(State)
19builder.add_node("double", double)
20builder.add_node("bump", bump)
21builder.add_edge(START, "double")
22builder.add_edge("double", "bump")
23builder.add_edge("bump", END)
24graph = builder.compile()
25
26print(type(graph).__name__)           # CompiledStateGraph
27print(graph.invoke({"n": 3}))         # {'n': 7}
28for chunk in graph.stream({"n": 3}):                        # default stream_mode="updates"
29    print(chunk)                      # {'double': {'n': 6}} then {'bump': {'n': 7}}
30for snapshot in graph.stream({"n": 3}, stream_mode="values"):
31    print(snapshot)                   # {'n': 3} then {'n': 6} then {'n': 7}
32print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

MessagesState

A ready-made state schema for the common case of "the state is a conversation". It is a TypedDict with a single key, messages: Annotated[list[AnyMessage], add_messages], so nodes return the new messages only and the runtime appends them. Because the reducer is add_messages , message-like dicts ({"role": "user", "content": ...}) are converted to HumanMessage/AIMessage objects on the way in — no model or langchain import needed. Subclass it to add keys: class State(MessagesState): summary: str. langchain.agents.AgentState is such a subclass.

 1from langgraph.graph import END, MessagesState, START, StateGraph
 2
 3
 4def greet(state: MessagesState) -> dict:
 5    return {"messages": [{"role": "assistant", "content": "Hello!"}]}
 6
 7
 8builder = StateGraph(MessagesState)
 9builder.add_node("greet", greet)
10builder.add_edge(START, "greet")
11builder.add_edge("greet", END)
12graph = builder.compile()
13
14result = graph.invoke({"messages": [{"role": "user", "content": "Hi"}]})
15for m in result["messages"]:
16    print(type(m).__name__, "|", m.content)
17# HumanMessage | Hi
18# AIMessage | Hello!
19print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

add_messages

The reducer behind MessagesState. A reducer is a function (old, new) -> merged attached to a state key via Annotated; without one, a node's return value overwrites the key. What this one does with the incoming list, item by item:

• coerces a dict or (role, text) tuple into the matching message class, and assigns a random id to any message that has none;
• appends a message whose id is new;
• replaces in place a message whose id already exists in the old list — that is how history is edited rather than duplicated;
• deletes the message named by a RemoveMessage(id=…), and clears the whole list on RemoveMessage(id=REMOVE_ALL_MESSAGES).

 1from langchain_core.messages import RemoveMessage
 2from langgraph.graph import add_messages
 3from langgraph.graph.message import REMOVE_ALL_MESSAGES
 4
 5history = add_messages(
 6    [{"role": "user", "content": "Hi", "id": "1"}],
 7    [{"role": "assistant", "content": "Hello!", "id": "2"}],
 8)
 9print([(type(m).__name__, m.content) for m in history])
10# [('HumanMessage', 'Hi'), ('AIMessage', 'Hello!')]
11
12edited = add_messages(history, [{"role": "user", "content": "Hey", "id": "1"}])
13print([(type(m).__name__, m.content) for m in edited])
14# [('HumanMessage', 'Hey'), ('AIMessage', 'Hello!')]
15
16shorter = add_messages(edited, [RemoveMessage(id="2")])
17print([m.content for m in shorter])                                  # ['Hey']
18
19print(add_messages(edited, [RemoveMessage(id=REMOVE_ALL_MESSAGES)]))  # []

↩ back to index

Command

A dataclass a node can return instead of a plain dict. Its fields, all optional:

• update — the state changes (same shape as a returned dict).
• goto — the next node name, a list of names (run in parallel), a Send, or END. This replaces a separate routing function when the node already knows where to go.
• graph — Command.PARENT makes goto target a node of the parent graph when the node lives inside a subgraph.
• resume — used from outside, as the input to invoke, to continue a run paused by interrupt ; its value is what the interrupt() call returns.

Annotating the return type as Command[Literal["big", "small"]] (or passing destinations= to add_node) is what lets get_graph() draw the possible jumps — without it the diagram shows no edges out of the node, though execution is unaffected.

 1from typing import Literal, TypedDict
 2
 3from langgraph.graph import END, START, StateGraph
 4from langgraph.types import Command
 5
 6
 7class State(TypedDict):
 8    n: int
 9    path: str
10
11
12def route(state: State) -> Command[Literal["big", "small"]]:
13    if state["n"] > 10:
14        return Command(goto="big", update={"path": "big"})
15    return Command(goto="small", update={"path": "small"})
16
17
18def big(state: State) -> dict:
19    return {"n": state["n"] + 100}
20
21
22def small(state: State) -> dict:
23    return {"n": state["n"] + 1}
24
25
26builder = StateGraph(State)
27builder.add_node("route", route)
28builder.add_node("big", big)
29builder.add_node("small", small)
30builder.add_edge(START, "route")
31builder.add_edge("big", END)
32builder.add_edge("small", END)
33graph = builder.compile()
34
35print(graph.invoke({"n": 42, "path": ""}))   # {'n': 142, 'path': 'big'}
36print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

interrupt

A function called inside a node: interrupt(value). What happens, in order:

1. The runtime writes a checkpoint (so a checkpointer is required — without one it raises) and stops the run. invoke returns the state so far plus an __interrupt__ key holding a list of Interrupt objects, each with .value (what you passed) and .id.
2. The caller does whatever the pause is for — shows the value to a person, waits a week.
3. The caller invokes the same thread with Command(resume=payload). The interrupted node runs again from its first line; this time interrupt() does not pause but returns payload, and the node continues to its return.

The re-run in step 3 is the detail that bites: code before the interrupt() call executes twice, so keep side effects after it or make them idempotent. The example counts the node's runs to show this.

 1from typing import TypedDict
 2
 3from langgraph.checkpoint.memory import InMemorySaver
 4from langgraph.graph import END, START, StateGraph
 5from langgraph.types import Command, interrupt
 6
 7runs = {"review": 0}
 8
 9
10class State(TypedDict):
11    draft: str
12    approved: str
13
14
15def write(state: State) -> dict:
16    return {"draft": "ship it"}
17
18
19def review(state: State) -> dict:
20    runs["review"] += 1                                     # executes on both passes
21    decision = interrupt({"question": "Approve this draft?", "draft": state["draft"]})
22    return {"approved": decision}
23
24
25builder = StateGraph(State)
26builder.add_node("write", write)
27builder.add_node("review", review)
28builder.add_edge(START, "write")
29builder.add_edge("write", "review")
30builder.add_edge("review", END)
31graph = builder.compile(checkpointer=InMemorySaver())
32
33config = {"configurable": {"thread_id": "1"}}
34paused = graph.invoke({"draft": "", "approved": ""}, config)
35print(paused["__interrupt__"][0].value)
36# {'question': 'Approve this draft?', 'draft': 'ship it'}
37print(graph.get_state(config).next)                        # ('review',)  <- where it will resume
38
39resumed = graph.invoke(Command(resume="yes"), config)
40print(resumed)                                             # {'draft': 'ship it', 'approved': 'yes'}
41print("review ran", runs["review"], "times")               # review ran 2 times
42print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

InMemorySaver

A checkpointer: pass one to compile(checkpointer=…) and the graph writes a snapshot of the state after every step, filed under the thread_id in config["configurable"]. That single change is what turns a one-shot graph into a resumable, inspectable conversation:

• Continuation. Invoking the same thread_id again starts from the saved state and merges the new input into it through the reducers, so a MessagesState graph remembers earlier turns (a key without a reducer, like count below, is simply overwritten by the new input).
• Pausing. interrupt and interrupt_before/interrupt_after need somewhere to park the run.
• History and time travel. get_state_history(config) returns every checkpoint; invoking with one of those checkpoints' config replays from that point.

It implements BaseCheckpointSaver, the interface every backend satisfies: put (store a checkpoint), put_writes (store a node's pending writes), get_tuple (load the latest, or a specific one), list (iterate a thread's checkpoints), delete_thread, and async twins. InMemorySaver keeps all of it in a dict, so it is for development and tests; langgraph-checkpoint-sqlite and langgraph-checkpoint-postgres are the production swaps and change nothing else in your code.

 1from typing import TypedDict
 2
 3from langgraph.checkpoint.base import BaseCheckpointSaver
 4from langgraph.checkpoint.memory import InMemorySaver
 5from langgraph.graph import END, START, StateGraph
 6
 7
 8class State(TypedDict):
 9    count: int
10
11
12def bump(state: State) -> dict:
13    return {"count": state["count"] + 1}
14
15
16builder = StateGraph(State)
17builder.add_node("bump", bump)
18builder.add_edge(START, "bump")
19builder.add_edge("bump", END)
20
21checkpointer = InMemorySaver()
22print(isinstance(checkpointer, BaseCheckpointSaver))   # True
23graph = builder.compile(checkpointer=checkpointer)
24
25config = {"configurable": {"thread_id": "session-1"}}
26print(graph.invoke({"count": 0}, config))     # {'count': 1}
27print(graph.get_state(config).values)         # {'count': 1}
28print(graph.get_state(config).next)           # () — the run finished
29
30for snap in graph.get_state_history(config):   # newest first: 3 checkpoints for one run
31    print(snap.metadata["step"], snap.values, snap.next)
32# 1 {'count': 1} ()            <- after bump
33# 0 {'count': 0} ('bump',)     <- input written, bump about to run
34# -1 {} ('__start__',)         <- the empty thread
35
36print(graph.invoke({"count": 10}, config))    # {'count': 11}  <- same thread, run again
37print(len(list(checkpointer.list(config))))   # 6 — the saver holds both runs
38graph.update_state(config, {"count": 100}, as_node="bump")
39print(graph.get_state(config).values)         # {'count': 100} <- written as if bump returned it
40print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index

SqliteSaver

The first checkpointer that outlives the process. It comes from the separate package langgraph-checkpoint-sqlite (import langgraph.checkpoint.sqlite) and implements exactly the BaseCheckpointSaver interface InMemorySaver does — put, put_writes, get_tuple, list, delete_thread, and async twins in AsyncSqliteSaver — on top of a sqlite3 connection. What that changes:

• The thread is a row on disk. A run paused by interrupt can be resumed by a different process: open the same file, compile the same graph, invoke the same thread_id with Command(resume=…). get_state(config).next tells you where it parked.
• Construction. SqliteSaver(conn) takes an open sqlite3.Connection; SqliteSaver.from_conn_string(path) is a context manager that opens the connection (with check_same_thread=False) and closes it on exit — ":memory:" gives a throwaway database. Tables are created on first use (setup() does it explicitly).
• What it is for. Local tools, labs, single-machine services. It serialises with the same msgpack serde as InMemorySaver, so the same TypeError for unserializable state applies. Multi-process concurrency is where langgraph-checkpoint-postgres takes over; the code you wrote does not change.

 1import os
 2import tempfile
 3from typing import TypedDict
 4
 5from langgraph.checkpoint.base import BaseCheckpointSaver
 6from langgraph.checkpoint.sqlite import SqliteSaver
 7from langgraph.graph import END, START, StateGraph
 8from langgraph.types import Command, interrupt
 9
10
11class State(TypedDict):
12    count: int
13    ok: str
14
15
16def bump(state: State) -> dict:
17    return {"count": state["count"] + 1}
18
19
20def confirm(state: State) -> dict:
21    return {"ok": interrupt("continue?")}
22
23
24builder = StateGraph(State)
25builder.add_node("bump", bump)
26builder.add_node("confirm", confirm)
27builder.add_edge(START, "bump")
28builder.add_edge("bump", "confirm")
29builder.add_edge("confirm", END)
30
31path = os.path.join(tempfile.mkdtemp(), "checkpoints.sqlite")
32config = {"configurable": {"thread_id": "t1"}}
33
34with SqliteSaver.from_conn_string(path) as saver:            # "process 1"
35    print(isinstance(saver, BaseCheckpointSaver))            # True
36    graph = builder.compile(checkpointer=saver)
37    paused = graph.invoke({"count": 0, "ok": ""}, config)
38    print(paused["__interrupt__"][0].value)                  # continue?
39
40print(os.path.getsize(path) > 0)                             # True — the run is on disk
41
42with SqliteSaver.from_conn_string(path) as saver:            # "process 2": a fresh connection
43    graph = builder.compile(checkpointer=saver)
44    print(graph.get_state(config).next)                      # ('confirm',)  <- parked, on disk
45    print(graph.invoke(Command(resume="yes"), config))       # {'count': 1, 'ok': 'yes'}
46    print(len(list(saver.list(config))))                     # 4 — the whole thread, both halves

↩ back to index

Send

Send(node, arg) is an instruction to run node once with arg as its input instead of the shared state. Return a list of them from a conditional-edge router (or as Command(goto=[...])) and the runtime schedules one task per Send, all in the same step and in parallel; each task's returned dict is merged back into the shared state through the key's reducer. That is the map half of map-reduce, and the way to branch N ways when N is only known at runtime. The receiving node's parameter is whatever you passed as arg, so type it accordingly, not as the graph's State.

 1from typing import Annotated, TypedDict
 2
 3from langgraph.graph import END, START, StateGraph
 4from langgraph.types import Send
 5
 6
 7class State(TypedDict):
 8    items: list
 9    results: Annotated[list, lambda a, b: a + b]
10
11
12def fan_out(state: State):
13    return [Send("square", {"value": i}) for i in state["items"]]
14
15
16def square(state: dict) -> dict:           # receives {"value": i}, not the State
17    return {"results": [state["value"] ** 2]}
18
19
20builder = StateGraph(State)
21builder.add_node("square", square)
22builder.add_conditional_edges(START, fan_out, ["square"])
23builder.add_edge("square", END)
24graph = builder.compile()
25
26print(graph.invoke({"items": [1, 2, 3], "results": []}))
27# {'items': [1, 2, 3], 'results': [1, 4, 9]}
28print(graph.get_graph().draw_mermaid(with_styles=False))

Diagram — generated by the code above:

↩ back to index