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.
//
Search across all documentation pages
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.
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.
add_messages, operator.add) for append-only channels.compile() once; inject a checkpointer when you need resume or HITL.invoke or stream with initial state; assert final fields in tests.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.
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.
A node should:
Side effects (email, payments) belong in dedicated nodes with clear names so HITL can gate them later.
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(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.
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.
Annotated[..., add_messages] drops prior turns.| 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 |
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.
No. Use reducers for append/merge channels. Scalar fields like query or status usually overwrite.
Yes. Prefer async when you add timeouts (timeouts are async-oriented in modern LangGraph) or concurrent I/O. Verify version-specific rules at build.
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.
Use config["configurable"] (and sometimes a store keyed by user id) rather than trusting the model to echo ids from the prompt alone.
invoke waits for final state. stream yields intermediate chunks for UI and debugging. Same graph; different consumer.
MessagesState is a convenience base with a messages channel. Custom TypedDict is better when you have domain fields beyond chat.
Compile a child graph and add_node("child", child_graph). Start with one graph until boundaries are clear.
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.
Reviewed by Chris St. John·Last updated Jul 16, 2026