LangGraph basics
What is LangGraph?
LangGraph is an open-source library from the LangChain team for building stateful, multi-step LLM applications as graphs. Nodes do work, edges decide what runs next, and a shared state object flows through. It adds loops, persistence, human-in-the-loop pauses and streaming, which plain chains lack.
Why use a graph instead of a simple chain?
A chain runs fixed steps in one direction. Agents need loops (call a tool, look at the result, maybe call another), branches, retries, pauses for approval and the ability to resume after a crash. A graph with explicit state models all of that and keeps the control flow visible and testable.
What is a StateGraph?
StateGraph is the main graph builder. You give it a state schema, add nodes and edges, then compile it. Every node receives the current state and returns an update, and the graph merges that update into the state before the next node runs.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
question: str
answer: str
def respond(state: State):
return {"answer": f"You asked: {state['question']}"}
builder = StateGraph(State)
builder.add_node("respond", respond)
builder.add_edge(START, "respond")
builder.add_edge("respond", END)
graph = builder.compile()
print(graph.invoke({"question": "What is RAG?"}))
What is a node in LangGraph?
A node is a Python function, or any Runnable, that takes the state and returns a dict with only the keys it wants to change. Nodes can call a model, run a tool, query a database or apply plain logic, and can be sync or async.
What are START and END?
START is the virtual entry point: an edge from START names the first node to run. END is the virtual exit: routing to END finishes the run and returns the final state. A graph can have several paths to END.
Why do you have to call compile()?
compile() checks the graph structure, such as missing nodes or dead ends, and turns the builder into a runnable graph with invoke, stream and async methods. It is also where you attach a checkpointer, a store and static breakpoints.
How is LangGraph related to LangChain?
They are separate libraries that work together. LangGraph does not require LangChain, but nodes often use LangChain chat models, tools and retrievers. Since the 1.0 releases, LangChain's own agent builder runs on LangGraph underneath.
State and reducers
How do you define state in LangGraph?
Usually with a TypedDict, or with a Pydantic model or dataclass if you want validation and defaults. Each key is a channel that nodes can read and update. Keep state small and explicit: inputs, intermediate results, and whatever the next nodes need to decide what to do.
What is a reducer?
A reducer decides how an update to a key is combined with the existing value. Without one, the new value replaces the old. With one, declared through Annotated, values can be combined instead, for example appending to a list, which matters when parallel nodes write the same key.
import operator
from typing import Annotated, TypedDict
class State(TypedDict):
question: str # overwritten
notes: Annotated[list[str], operator.add] # appended
What does add_messages do?
add_messages is the built-in reducer for chat history. It appends new messages, updates an existing message when a new one has the same ID, and converts dicts to message objects. MessagesState is a ready-made state with a messages key that uses it.
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class ChatState(TypedDict):
messages: Annotated[list, add_messages]
What does a node return, and what happens to keys it leaves out?
A node returns a partial update: only the keys it changed. Keys it does not return keep their current values. Returning the whole state is allowed but makes it easy to overwrite another node's work by accident, especially with parallel branches.
What are input, output and private state schemas?
A graph can accept a smaller input schema and return a smaller output schema than its internal state, so callers only see what they should. Nodes can also read or write private keys that never appear in the output, such as scratch data passed between two internal steps.
What is a Command in LangGraph?
A node can return Command(update=..., goto=...) to change the state and choose the next node in one step, instead of a separate conditional edge. It is useful for agent handoffs, where one agent decides which agent runs next and passes it data.
from langgraph.types import Command
def triage(state):
if "refund" in state["question"].lower():
return Command(update={"team": "billing"}, goto="billing_agent")
return Command(update={"team": "support"}, goto="support_agent")
How do you handle state that grows too large, such as long chat history?
Trim or summarise old messages in a dedicated node before calling the model, remove messages with RemoveMessage through the add_messages reducer, and keep large data such as documents in external storage with only IDs in the state. Checkpoints store the state, so a huge state also costs storage.
Control flow
How do conditional edges work?
add_conditional_edges takes a source node and a routing function. After the source node runs, the function reads the state and returns the name of the next node, or END. An optional mapping translates return values to node names, which also helps the graph diagram.
def route(state):
return "tools" if state["messages"][-1].tool_calls else END
builder.add_conditional_edges("agent", route, ["tools", END])
How do you build the classic tool-calling agent loop?
Add an agent node that calls a model bound to tools and a ToolNode that runs requested tools. A conditional edge goes from the agent to the tools when the last message has tool calls, otherwise to END, and an edge from the tools goes back to the agent. tools_condition is the prebuilt router.
from langgraph.prebuilt import ToolNode, tools_condition
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
How do you stop an agent from looping forever?
Every run has a recursion limit on the number of steps, set in the config; when it is hit, LangGraph raises GraphRecursionError. Also add your own guard, such as a step counter in state that routes to a fallback answer after a few tool calls.
graph.invoke(inputs, config={"recursion_limit": 15})
How do nodes run in parallel?
If several edges leave one node, all target nodes run in the same step, called a superstep, and the graph waits for all of them before moving on. Keys that more than one parallel node writes need a reducer, otherwise LangGraph raises an error about conflicting updates.
What is the Send API used for?
Send handles map-reduce when the number of parallel tasks is known only at run time. A conditional edge returns a list of Send objects, each naming a node and its own input, such as one per document. The results flow back through a reducer, often into a list.
from langgraph.types import Send
def fan_out(state):
return [Send("summarise", {"doc": doc}) for doc in state["docs"]]
builder.add_conditional_edges("load", fan_out, ["summarise"])
What is a subgraph?
A subgraph is a compiled graph used as a node inside another graph. It lets teams build and test parts separately, such as a research agent inside a larger assistant. If parent and child share state keys, you add the compiled subgraph directly; otherwise a wrapper node maps state in and out.
How do you retry a node that calls a flaky API?
Attach a retry policy when adding the node, with the maximum attempts, backoff and which exceptions to retry. Only that node reruns, not the whole graph, and with a checkpointer the steps that already succeeded are not repeated after a crash.
Persistence and memory
What is a checkpointer?
A checkpointer saves the graph state after every step. With it, a conversation can continue later, a failed run can resume from the last good step, a person can inspect or edit state, and interrupts work. In-memory savers suit tests; SQLite and Postgres savers suit real apps.
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [("user", "Hi, I am Riya")]}, config)
graph.invoke({"messages": [("user", "What is my name?")]}, config)
What is a thread_id?
thread_id in the config names one conversation or run history. The checkpointer stores checkpoints per thread, so the same graph can serve many users at once while each thread keeps its own state. A new thread_id starts fresh.
What is the difference between short-term and long-term memory in LangGraph?
Short-term memory is the state of one thread, saved by the checkpointer, such as this chat's messages. Long-term memory lives in a store shared across threads, organised by namespaces such as a user ID, and holds facts like preferences that should survive into new conversations.
How do you inspect or rewind a past run?
get_state(config) returns the latest checkpoint for a thread, including the values and which nodes run next. get_state_history(config) lists earlier checkpoints. You can invoke from an older checkpoint ID to replay or branch from that point, which helps debugging.
How do you change a thread's state from outside the graph?
Call update_state(config, values) with the update you want, which is applied through the same reducers as a node update and saved as a new checkpoint. It is used to fix a wrong tool argument, add a human correction or skip ahead before resuming the run.
Which checkpointer would you use in production?
A durable one backed by a database your team already runs, typically Postgres, so threads survive restarts and work across several server instances. Keep checkpoint tables indexed and pruned, because every step of every thread writes a row.
What happens if the server crashes in the middle of a run?
With a durable checkpointer, the state after the last completed step is already saved. Invoking the same thread again, with None as input, continues from there, so finished steps such as expensive tool calls are not repeated. Without a checkpointer the run is lost and must start over.
Human-in-the-loop
How does interrupt() work?
Calling interrupt(value) inside a node pauses the run, saves the state through the checkpointer and returns value to the caller, for example a draft or a question. The run waits until you invoke the graph again with Command(resume=...), and the resume value becomes the return value of interrupt().
from langgraph.types import interrupt, Command
def approve_refund(state):
decision = interrupt({"refund": state["amount"], "ask": "Approve?"})
return {"approved": decision == "yes"}
# later, after a person answers
graph.invoke(Command(resume="yes"), config)
What is a common mistake with interrupt()?
When the run resumes, the node starts again from the top, not from the interrupt line. Any code before interrupt runs twice, so side effects such as sending an email or charging a card must come after the interrupt or live in a separate node.
How would you add human approval before an agent calls a risky tool?
Put an approval node, or an interrupt inside the tool node, before the tool runs. Show the reviewer the tool name and arguments, then resume with approve, edit or reject. On edit, update the tool call's arguments; on reject, send a ToolMessage explaining why so the model can respond.
What are static breakpoints?
interrupt_before and interrupt_after, given at compile time or per run, pause the graph before or after named nodes without changing node code. They are handy for debugging and step-by-step runs; for real approval flows, interrupt() inside a node is more flexible because it can pass data out.
How do you let a person edit a draft before it is sent?
A drafting node writes the draft into state, then a review node calls interrupt() with the draft. The reviewer resumes with either approval or edited text, the review node stores the final text, and only a later send node performs the side effect.
Streaming
Which stream modes does LangGraph support?
values streams the full state after each step, updates streams only what each node changed, messages streams model tokens with metadata about which node produced them, and custom streams data you emit yourself. You can request several modes at once.
for mode, chunk in graph.stream(inputs, config, stream_mode=["updates", "messages"]):
print(mode, chunk)
How do you stream tokens from an LLM inside a node?
Use stream_mode="messages". LangGraph captures tokens from chat model calls made inside nodes, even if the node calls invoke, and yields each token with metadata such as the node name, so the UI can show only the final answer node's tokens.
How do you send progress updates from inside a long node?
Get a stream writer inside the node and write custom data, such as "fetched 3 of 10 pages", then stream with stream_mode="custom". The client sees progress while the node is still running, which matters for slow research or tool steps.
from langgraph.config import get_stream_writer
def research(state):
writer = get_stream_writer()
for i, url in enumerate(state["urls"], start=1):
writer({"progress": f"{i}/{len(state['urls'])}"})
fetch(url)
return {"done": True}
How do you stream a graph run to a web frontend?
Call astream in an async endpoint with the stream modes the UI needs, convert each chunk to JSON and send it as Server-Sent Events. Include the thread_id so the browser can reconnect or resume after an interrupt, and send a final event when the run ends.
Agents, multi-agent design and production
What prebuilt agent does LangGraph offer?
A prebuilt tool-calling (ReAct-style) agent builds the model-tools loop for you from a model, a list of tools and a prompt. Older code uses create_react_agent from langgraph.prebuilt; since LangChain 1.0, create_agent in langchain.agents is the recommended builder and runs on LangGraph.
What is the supervisor pattern?
One supervisor agent receives the request and decides which specialist agent, such as research, coding or billing, should act next. Specialists report back, and the supervisor either delegates again or finishes. It is easy to reason about but adds an extra model call per step.
How do handoffs between agents work?
An agent hands off by returning a Command that names the next agent and passes any data it needs, often exposed to the model as a "transfer to X" tool. When agents are subgraphs, the Command can target a node in the parent graph so control moves across graphs.
When is a multi-agent design worse than a single agent?
When one agent with a few well-described tools can do the job. Each extra agent adds model calls, latency, cost and more places for context to get lost. Split into several agents only when tool sets or instructions are large and distinct enough that one prompt handles them badly.
How do you test a LangGraph application?
Test nodes as plain functions with sample states, test routing functions with states that should take each branch, and run the compiled graph with a fake model for the control flow. Then run an evaluation dataset against the real model and check final answers and tool-call sequences.
How would you deploy a LangGraph agent?
Expose the compiled graph behind an API such as FastAPI, use a Postgres checkpointer so state survives restarts and is shared across instances, stream results to the client, and run long jobs in background workers. LangChain also offers a managed deployment service for graphs.
How do you observe and debug agents in production?
Trace every run with a tool such as LangSmith or an OpenTelemetry-based tracer to see each node, prompt, tool call and token count. Track step counts, tool error rates, interrupts, latency and cost per thread, and replay failed threads from their checkpoints.
How does LangGraph compare with other agent frameworks?
LangGraph is low level: you define state and control flow explicitly, which gives precise control, persistence and human-in-the-loop support. Higher-level frameworks such as CrewAI or AutoGen give role-based agents and conversations with less code but less control. Choose by how much you need to control the flow.