Building Your First Stateful Agent Graph
A production-shaped LangGraph agent starts with a typed state schema, small nodes that return partial updates, explicit edges, and a compiled graph you can invoke or stream.
This recipe walks that path with a research-style loop: plan, act (mock tool), and decide whether to stop.
Summary
Define state with reducers where lists grow, implement pure-ish nodes, wire START / conditional edges / END, compile, then invoke with a clear input dict and optional thread_id when you add a checkpointer.
Recipe
- Choose the state fields the whole run must share (messages, plan, findings, step count).
- Attach reducers (
add_messages,operator.add) for append-only channels. - Implement one concern per node (plan, tool, synthesize, guard).
- Draw fixed edges for always-next steps; use conditional edges for stop vs continue.
compile()once; inject a checkpointer when you need resume or HITL.- Call
invokeorstreamwith initial state; assert final fields in tests. - Cap loops with a step counter or recursion limit before connecting real tools.
Working Example
from typing import Annotated, Literal, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class ResearchState(TypedDict):
messages: Annotated[list, add_messages]
query: str
notes: Annotated[list[str], operator.add]
steps: int
def plan(state: ResearchState) -> dict:
q = state["query"]
return {
"messages": [{"role": "assistant", "content": f"Plan: search for '{q}'"}],
"steps": state.get("steps", 0) + 1,
}
def search(state: ResearchState) -> dict:
# Replace with a real tool / API (verify integrations at build)
hit = f"result for: {state['query']}"
return {
"notes": [hit],
"messages": [{"role": "tool", "content": hit, "name": "search"}],
"steps": state.get("steps", 0) + 1,
}
def synthesize(state: ResearchState) -> dict:
summary = "; ".join(state.get("notes", [])) or "no notes"
return {
"messages": [{"role": "assistant", "content": f"Summary: {summary}"}],
"steps": state.get("steps", 0) + 1,
}
def should_continue(state: ResearchState) -> Literal["search", "synthesize"]:
if state.get("steps", 0) >= 3 or state.get("notes"):
return "synthesize"
return "search"
builder = StateGraph(ResearchState)
builder.add_node("plan", plan)
builder.add_node("search", search)
builder.add_node("synthesize", synthesize)
builder.add_edge(START, "plan")
builder.add_conditional_edges(
"plan",
should_continue,
{"search": "search", "synthesize": "synthesize"},
)
builder.add_edge("search", "synthesize")
builder.add_edge("synthesize", END)
graph = builder.compile()
final = graph.invoke({
"messages": [{"role": "user", "content": "LangGraph checkpoints"}],
"query": "LangGraph checkpoints",
"notes": [],
"steps": 0,
})
print(final["messages"][-1])Swap search for a LangChain tool node or HTTP client when you leave the sandbox.
Keep the same graph shape so tests can still drive routing with fixtures.
Deep Dive
Designing state
Put in state only what more than one node needs or what you must checkpoint.
Ephemeral debug flags can live in config metadata instead of state if they must not pollute snapshots.
Prefer explicit types over untyped dicts so editors and FAQs stay honest.
Node contracts
A node should:
- Read the subset of state it needs
- Perform one unit of work
- Return a partial update
- Avoid hidden global mutation
Side effects (email, payments) belong in dedicated nodes with clear names so HITL can gate them later.
Edges as policy
Edges are product policy encoded as code.
"After plan, always search" is a fixed edge.
"After agent, tools or end" is a conditional edge.
Document why each branch exists; future you will treat mystery branches as bugs.
Compile options
compile(checkpointer=...) enables thread resume.
interrupt_before / interrupt_after help debug static breakpoints.
Per-node timeout, retry_policy, and error_handler (LangGraph 1.2+, verify at build) harden production nodes without changing the topology.
Testing strategy
Unit-test routers and nodes with plain dict inputs.
Integration-test the compiled graph with fixed fake tools.
Trace one golden path in LangSmith before attaching irreversible tools.
Gotchas
- Overwriting lists - forgetting
Annotated[..., add_messages]drops prior turns. - Returning full state - accidental large returns are fine functionally but confuse "what changed" diffs; prefer partial updates.
- Router key mismatch - conditional map keys must match what the router returns and node names you registered.
- Missing END path - cycles without a stop condition burn tokens forever.
- Checkpointer without thread_id - resume and multi-turn fail in confusing ways.
- Mutating state in place - always return new update dicts; in-place mutation may not checkpoint cleanly.
- Fat nodes - stuffing plan+tool+format into one node defeats tracing and HITL placement.
Alternatives
| Approach | Strength | Weakness |
|---|---|---|
Raw StateGraph (this recipe) | Full control | More boilerplate |
| LangChain prebuilt agent helpers | Fast ReAct loop | Harder custom topology |
| CrewAI / multi-agent role frameworks | Role UX | Different mental model |
| Hand-rolled while-loop | Zero deps | No checkpoints / streaming integration |
| Workflow DAG only | Deterministic | Weak adaptive tool use |
FAQs
How big should my first graph be?
Three to five nodes is enough: entry, one tool or work node, exit, plus a router. Grow only when a branch needs a named home.
Should every field use a reducer?
No. Use reducers for append/merge channels. Scalar fields like query or status usually overwrite.
Can nodes be async?
Yes. Prefer async when you add timeouts (timeouts are async-oriented in modern LangGraph) or concurrent I/O. Verify version-specific rules at build.
Where do system prompts live?
Often as a constant or factory used inside the model node, or as a state field if the prompt is dynamic per tenant. Keep secrets out of state you log.
How do I pass user identity?
Use config["configurable"] (and sometimes a store keyed by user id) rather than trusting the model to echo ids from the prompt alone.
What is the difference between invoke and stream?
invoke waits for final state. stream yields intermediate chunks for UI and debugging. Same graph; different consumer.
Do I need MessagesState?
MessagesState is a convenience base with a messages channel. Custom TypedDict is better when you have domain fields beyond chat.
How do subgraphs fit later?
Compile a child graph and add_node("child", child_graph). Start with one graph until boundaries are clear.
Related
Related: LangChain & LangGraph Basics
Related: The LangGraph Mental Model
Related: Branching and Conditional Edges
Related: Persistence and Checkpoints
Related: LangChain & LangGraph Best Practices
Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana
@solana/web3.js+@solana/spl-token.