LangGraph Fundamentals

This content was generated with the assistance of AI and should be verified against the official documentation of each tool and framework before being relied on in production.

This section’s bibliography lists the reference material consulted while preparing these pages.

This page’s examples target Python 3.12 with LangGraph 1.2.12, langchain-core 1.6.5 and LangChain 1.4.3, as checked on 2026-09-29 (see Ecosystem & Packages for the package map). Ollama was not available while the page was written, so every output shown was produced with deterministic Python functions or LangChain’s GenericFakeChatModel, not with a real model; DocsAssistant’s real models remain llama3.1:8b for chat and nomic-embed-text for embeddings.

LangChain’s create_agent (see Agents with create_agent) is built on LangGraph: the agent is a compiled graph with a model node and a tools node. LangGraph is the layer to reach for when the control flow is not "loop until the model stops calling tools" — when you want fixed steps, explicit branches, parallel workers or a graph you can inspect, pause and resume. The framework-neutral reasoning behind those choices is in Agent Loop & Architectures; this page is the vocabulary and mechanics of the graph itself.

The model: state, nodes and edges

A LangGraph program is three things:

  • State — a shared, typed snapshot of the run (a TypedDict, dataclass or Pydantic model).

  • Nodes — functions that receive the current state and return a partial update to it.

  • Edges — the wiring that decides which node runs next.

Execution proceeds in discrete super-steps: every node scheduled for the step runs (in parallel when there are several), their updates are merged into the state, and the edges from those nodes select the next step. The run ends when control reaches END. Nothing runs until you compile() the builder, which validates the graph and returns a runnable with the usual invoke, stream and ainvoke methods.

flowchart LR S([START]) --> A[node: retrieve] A --> B[node: answer] B --> E([END]) ST[(State)] -. read .-> A A -. partial update .-> ST ST -. read .-> B B -. partial update .-> ST
Figure 1. Nodes read the shared state and return partial updates to it

StateGraph and the smallest graph

StateGraph is the builder for the Graph API. It takes the state schema, and you add nodes and edges to it before compiling. The smallest useful DocsAssistant graph retrieves and then answers, with a messages list and a hits counter as state:

A two-node StateGraph
from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]
    hits: int


def retrieve(state: State) -> dict:
    return {"messages": [AIMessage("3 sections found")], "hits": 3}


def answer(state: State) -> dict:
    return {"messages": [AIMessage(f"Answer based on {state['hits']} sections")]}


builder = StateGraph(State)
builder.add_node("retrieve", retrieve)
builder.add_node("answer", answer)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", END)
graph = builder.compile()

out = graph.invoke({"messages": [HumanMessage("What is HNSW?")], "hits": 0})
for m in out["messages"]:
    print(type(m).__name__, "|", m.content)
print(out["hits"])

Running it printed:

Output of the two-node graph
HumanMessage | What is HNSW?
AIMessage | 3 sections found
AIMessage | Answer based on 3 sections
3

START and END are the two virtual nodes that mark where the run enters and leaves. Node names are strings, and a node function may also take a config argument (for the thread_id and other run configuration) or a runtime object; see the guide below for the exact signatures. StateGraph also accepts separate input_schema and output_schema types so the graph’s public input and output can be narrower than its internal state, and a context_schema for read-only run-scoped data such as a user id.

See LangGraph — Graph API overview and LangGraph — Use the Graph API.

State and reducers

Each key in the state schema may carry a reducer: a function that says how an update to that key is merged into the existing value. Without a reducer, a node’s returned value simply overwrites the key (hits above). With one, concurrent or successive updates accumulate. A reducer is attached with Annotated[type, reducer]; any binary function works, and operator.add is the usual choice for lists.

The reducer you will meet most is add_messages. It appends new messages to the list, but if an incoming message has the same id as an existing one it replaces that message instead, which is what makes editing or removing a message from a node possible:

The add_messages reducer appends new messages and replaces same-id ones
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import add_messages

history = [HumanMessage("hi", id="1")]
print(add_messages(history, [AIMessage("hello", id="2"), HumanMessage("hi there", id="1")]))
Output: message 1 was replaced in place, message 2 appended
[HumanMessage(content='hi there', additional_kwargs={}, response_metadata={}, id='1'), AIMessage(content='hello', additional_kwargs={}, response_metadata={}, id='2', tool_calls=[], invalid_tool_calls=[])]

Because a state with a messages list is so common, LangGraph ships MessagesState, a TypedDict with that one key already wired to add_messages. Subclass it to add your own keys. The next example is a graph with a single model node, using a fake chat model in place of Ollama:

MessagesState with a model node (fake model)
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langgraph.graph import END, START, MessagesState, StateGraph

model = GenericFakeChatModel(messages=iter([AIMessage("HNSW is a graph-based ANN index.")]))


def call_model(state: MessagesState) -> dict:
    return {"messages": [model.invoke(state["messages"])]}


graph = (
    StateGraph(MessagesState)
    .add_node("model", call_model)
    .add_edge(START, "model")
    .add_edge("model", END)
    .compile()
)
print(graph.invoke({"messages": [("user", "What is HNSW?")]})["messages"][-1].content)

This printed HNSW is a graph-based ANN index.. With a real model, replace the fake with the chat model from Chat Models & Messages. Reducers matter most when nodes run in parallel: two workers that both return a list under the same key can only be merged safely if that key has a reducer (see Send below). The state can also be overwritten explicitly, bypassing a reducer, with the Overwrite wrapper documented in the Graph API guide.

See LangGraph — Graph API overview.

Conditional edges

A normal edge always goes to the same next node. A conditional edge calls a routing function with the state and returns the name of the next node (or a list of names to run in parallel). The optional mapping translates the function’s return values into node names:

A conditional edge that routes on the state
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph


class S(TypedDict):
    question: str
    hits: Annotated[list[str], operator.add]
    route: str


b2 = StateGraph(S)
b2.add_node("x", lambda s: {"hits": ["x"]})
b2.add_node("A", lambda s: {"route": "A"})
b2.add_node("B", lambda s: {"route": "B"})
b2.add_edge(START, "x")
b2.add_conditional_edges("x", lambda s: "A" if s["hits"] else "B", {"A": "A", "B": "B"})
b2.add_edge("A", END)
b2.add_edge("B", END)
g2 = b2.compile()
print(g2.invoke({"question": "q", "hits": [], "route": ""}))

The run printed {'question': 'q', 'hits': ['x'], 'route': 'A'}: node x recorded a hit, so the router chose A. In real graphs the routing function is where an LLM-produced decision (a grade, a classification, a tool_calls check) turns into control flow; annotating its return type with Literal[…​] documents the possible targets and makes them show up in the drawn graph. Loops are ordinary edges that point backwards — a run that never reaches END is stopped by the recursion_limit config value, which you can raise or lower per call.

See LangGraph — Graph API overview.

Command: update and route in one return

A conditional edge separates "what to change" (the node) from "where to go" (the router). When the same node knows both, return a Command instead: update is a partial state update and goto names the next node, so no edge needs to be declared for it. Annotate the return type with Command[Literal[…​]] so the graph knows the possible destinations. The example is DocsAssistant’s grading step: retrieve, and if nothing came back, rewrite the question and try again:

Command combines a state update and a routing decision
import operator
from typing import Annotated, Literal, TypedDict

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


class S(TypedDict):
    question: str
    hits: Annotated[list[str], operator.add]
    route: str


def grade(state: S) -> Command[Literal["answer", "rewrite"]]:
    if state["hits"]:
        return Command(update={"route": "answer"}, goto="answer")
    return Command(update={"route": "rewrite"}, goto="rewrite")


def rewrite(state: S) -> dict:
    return {"question": state["question"] + " pgvector", "hits": ["doc-1"]}


def answer(state: S) -> dict:
    return {"route": "done:" + ",".join(state["hits"])}


b = StateGraph(S)
b.add_node("grade", grade)
b.add_node("rewrite", rewrite)
b.add_node("answer", answer)
b.add_edge(START, "grade")
b.add_edge("rewrite", "grade")
b.add_edge("answer", END)
g = b.compile()
print(g.invoke({"question": "hnsw", "hits": [], "route": ""}))

Output: {'question': 'hnsw pgvector', 'hits': ['doc-1'], 'route': 'done:doc-1'}. The first pass found no hits and went to rewrite; the edge from rewrite back to grade closed the loop, and the second pass went to answer. Command is also the value a graph is resumed with after an interrupt (Command(resume=…​), see Human-in-the-Loop & Time Travel) and can carry graph=Command.PARENT to jump from inside a subgraph to a node in its parent. The corrective loop above is the core of the self-corrective RAG pattern developed in LangGraph Workflows; the retrieval side is in Agentic RAG.

See LangGraph — Graph API overview and LangGraph — Use the Graph API.

Send: map-reduce

Sometimes the number of parallel branches is only known at run time — one summary per documentation component, one fetch per search hit. Edges declared at build time cannot express that. A routing function can instead return a list of Send(node, payload) objects: each one launches node once with its own private payload as input. The workers' results flow back through a reducer, and a downstream node combines them (the "reduce" step).

flowchart LR S([START]) -->|Send x N| W1[summarise: vector-rag] S -->|Send x N| W2[summarise: mcp] S -->|Send x N| W3[summarise: agents] W1 --> C[combine] W2 --> C W3 --> C C --> E([END])
Figure 2. Send fans out one worker per item; a reducer collects the results
Send in a map-reduce graph
import operator
from typing import Annotated, TypedDict

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


class Overall(TypedDict):
    components: list[str]
    summaries: Annotated[list[str], operator.add]   # reducer: workers append here
    report: str


class Worker(TypedDict):
    component: str


def fan_out(state: Overall):
    return [Send("summarise", {"component": c}) for c in state["components"]]


def summarise(state: Worker) -> dict:
    return {"summaries": [f"summary of {state['component']}"]}


def combine(state: Overall) -> dict:
    return {"report": "\n".join(sorted(state["summaries"]))}


b = StateGraph(Overall)
b.add_node("summarise", summarise)
b.add_node("combine", combine)
b.add_conditional_edges(START, fan_out, ["summarise"])
b.add_edge("summarise", "combine")
b.add_edge("combine", END)
g = b.compile()
print(g.invoke({"components": ["vector-rag", "mcp", "agents"], "summaries": [], "report": ""})["report"])
Output: one summary per Send, merged by the reducer
summary of agents
summary of mcp
summary of vector-rag

Note that the worker’s input type (Worker) differs from the graph state: Send gives each invocation only what it needs. The sorting in combine is deliberate, because the order in which parallel workers finish is not part of the contract. The orchestrator-worker pattern in LangGraph Workflows is built on this same mechanism.

See LangGraph — Graph API overview (the Send section) and LangGraph — Use the Graph API.

Compiling, running and drawing a graph

compile() returns a runnable graph. It accepts a checkpointer for persistence across turns (see Short-Term Memory & Checkpointers) and a store for cross-thread memory (see Long-Term Memory Stores). A compiled graph can also draw itself; this is the quickest check that the wiring matches your intent. get_graph().draw_mermaid() on the routing graph above printed:

Mermaid text emitted by a compiled graph
graph TD;
	__start__([<p>__start__</p>]):::first
	x(x)
	A(A)
	B(B)
	__end__([<p>__end__</p>]):::last
	__start__ --> x;
	x -.-> A;
	x -.-> B;
	A --> __end__;
	B --> __end__;

Dotted arrows mark conditional edges and solid ones fixed edges. The output also carries a Mermaid config header and classDef styling lines, omitted here. Paste the text into any Mermaid renderer, as in the diagram below, which redraws the same graph:

flowchart TD start([__start__]) --> x[x] x -.-> A[A] x -.-> B[B] A --> fin([__end__]) B --> fin
Figure 3. The compiled routing graph: solid edges are fixed, dotted edges are conditional

draw_mermaid_png() renders the same graph to an image. Tracing a run in LangSmith shows the executed path (see Observability with LangSmith).

See LangGraph — Graph API overview.

Graph API vs. Functional API

The Graph API declares the control flow as data: nodes, edges and a state schema. The Functional API expresses the same runtime capabilities — persistence, interrupts, streaming, parallel execution — in ordinary Python control flow. An @entrypoint function is the workflow; each unit of work that should be cached, retried or run concurrently is an @task. Calling a task returns a future, so you fan out by calling it several times and then collecting .result():

The same idea in the Functional API
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task


@task
def search(q: str) -> list[str]:
    return [f"{q}-1", f"{q}-2"]


@task
def summarise(s: str) -> str:
    return s.upper()


@entrypoint(checkpointer=InMemorySaver())
def workflow(q: str) -> list[str]:
    hits = search(q).result()
    futures = [summarise(h) for h in hits]     # the two summaries run concurrently
    return [f.result() for f in futures]


print(workflow.invoke("hnsw", {"configurable": {"thread_id": "t1"}}))

This printed ['HNSW-1', 'HNSW-2']. The checkpointer records each task’s result under the thread_id, so a run that fails or is interrupted part-way through does not repeat completed tasks when resumed. Code that has side effects (calling an API, opening a GitHub issue) belongs inside a task for that reason.

Table 1. Choosing between the two APIs
Graph API Functional API

Control flow

Explicit nodes and edges; visible and drawable

Ordinary if, for and function calls

State

Declared schema with reducers

Function arguments and local variables; no schema

Parallelism

Parallel edges and Send

Calling several tasks before collecting results

Best for

Workflows you want to inspect, visualise or hand to others

Adding persistence and interrupts to existing procedural code with minimal change

The two APIs use the same runtime and can be mixed in one application. A reasonable default: start with the Graph API when the shape of the workflow is the point, and use the Functional API to retrofit durability onto code that is already a straight function.

See LangGraph — Functional API overview and LangGraph — Choosing between the Graph and Functional APIs.

Where to go next

The named workflow patterns (chaining, routing, parallelisation, orchestrator-worker, evaluator-optimizer, self-corrective RAG) are built from exactly these pieces in LangGraph Workflows; framework-neutral theory is in Workflow Patterns. Persistence is covered in Short-Term Memory & Checkpointers, and pausing for approval in Human-in-the-Loop & Time Travel.

References