This cheat sheet is a practical reference for designing LangGraph workflows that remain controlled, testable, and production-safe. LangGraph is most useful when the system is no longer a linear prompt pipeline and starts behaving like a stateful workflow engine with model reasoning inside it.
Installation
1
| pip install langgraph langchain-openai langgraph-checkpoint-sqlite
|
1. Core Concepts
| Concept | Type | Role |
|---|
StateGraph | Graph container | Holds nodes, edges, state schema, compilation |
TypedDict state | Shared memory | Typed contract between all nodes |
| Node | Python function state → state | One testable unit of work |
add_edge | Unconditional transition | Always goes from A to B |
add_conditional_edges | Branching | Routing function decides the next node |
MemorySaver | In-memory checkpointer | State persistence across turns (dev) |
SqliteSaver | SQLite checkpointer | State persistence (single-process prod) |
interrupt_before | Human-in-the-loop | Pause before a node executes |
Command | Node return type | Node that modifies graph flow explicitly |
2. Minimal Graph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| from langgraph.graph import StateGraph, END
from typing import TypedDict
from langchain_openai import ChatOpenAI
# 1. Define state schema
class State(TypedDict):
question: str
answer: str
# 2. Define nodes — each receives state and returns updated state
llm = ChatOpenAI(model="gpt-4o-mini")
def generate(state: State) -> State:
state["answer"] = llm.invoke(state["question"]).content
return state
# 3. Build and compile graph
graph = StateGraph(State)
graph.add_node("generate", generate)
graph.set_entry_point("generate") # first node to run
graph.add_edge("generate", END) # exit condition
app = graph.compile()
result = app.invoke({"question": "What is LangGraph?", "answer": ""})
print(result["answer"])
|
3. State Design
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| from typing import Annotated
from langgraph.graph.message import add_messages
import operator
class AgentState(TypedDict):
# Annotated fields use reducers instead of last-write-wins
messages: Annotated[list, add_messages] # append-only
retrieved_docs: Annotated[list, operator.add] # extend list
plan: list[str]
tool_results: list[dict]
current_step: str | None
retry_count: int
needs_approval: bool
error: str | None
final_answer: str | None
|
Reducer Reference
| Reducer | Behavior | Use For |
|---|
add_messages | Appends new messages, deduplicates by ID | Chat message history |
operator.add | Extends list | Accumulating results |
lambda a, b: {**a, **b} | Merges dicts | Metadata accumulation |
| None (default) | Last write wins | Simple scalar fields |
4. Conditional Edges (Branching)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| def router(state: AgentState) -> str:
"""Returns the name of the next node to execute."""
if state.get("error"):
return "error_handler"
if state["retry_count"] >= 3:
return "escalate"
if state["needs_approval"]:
return "human_review"
if state["final_answer"]:
return END
return "reasoning"
graph.add_conditional_edges(
"tool_node", # source node
router, # routing function: state → str
{ # optional: map return value → node name
"error_handler": "error_handler",
"escalate": "escalate",
"human_review": "human_review",
"reasoning": "reasoning",
END: END,
}
)
|
5. Persistence & Memory
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
# In-memory (dev / tests)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
# SQLite (production, single process)
with SqliteSaver.from_conn_string("./state.db") as saver:
app = graph.compile(checkpointer=saver)
# thread_id ties all invocations to the same conversation
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"messages": [{"role": "user", "content": "Hello"}]}, config=config)
app.invoke({"messages": [{"role": "user", "content": "Follow up"}]}, config=config)
# State is automatically restored from checkpoint on the second call
# Inspect stored state
snapshot = app.get_state(config)
print(snapshot.values) # current state dict
print(snapshot.next) # next node(s) to run
print(snapshot.config) # checkpoint metadata
# List all historical checkpoints
for checkpoint in app.get_state_history(config):
print(checkpoint.created_at, checkpoint.values.get("current_step"))
|
6. Human-in-the-Loop
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| # Pause BEFORE specific nodes — execution suspends, state is saved
app = graph.compile(
checkpointer=memory,
interrupt_before=["human_review", "execute_payment"], # list of node names
interrupt_after=["data_extraction"], # pause after instead
)
# Run until an interrupt
state = app.invoke(initial_state, config=config)
# → graph pauses, returns current state
# Inspect what the graph wants to do next
snapshot = app.get_state(config)
print("Next node:", snapshot.next)
print("State:", snapshot.values)
# Optionally modify state before resuming
app.update_state(config, {"needs_approval": False, "retry_count": 0})
# Resume from checkpoint (None = use saved state)
final = app.invoke(None, config=config)
|
7. Streaming Modes
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| # Full state after every node
for state in app.stream(initial, stream_mode="values"):
print(state)
# Only the delta (changed keys) after each node
for update in app.stream(initial, stream_mode="updates"):
print(update)
# Token-level streaming from LLM nodes
for chunk, metadata in app.stream(initial, stream_mode="messages"):
if metadata.get("langgraph_node") == "generate":
print(chunk.content, end="", flush=True)
# All internal events — debugging
for event in app.stream(initial, stream_mode="debug"):
print(event["type"], event.get("step"))
|
stream_mode | Returns | Best For |
|---|
"values" | Full state dict after each node | Observing state evolution |
"updates" | Delta dict after each node | Lightweight monitoring |
"messages" | Token-level chunks + metadata | Streaming UI |
"debug" | All internal events | Debugging graph execution |
8. Subgraphs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| from langgraph.graph import StateGraph
# Build a self-contained subgraph
sub_graph = StateGraph(SubState)
sub_graph.add_node("step_a", step_a)
sub_graph.add_node("step_b", step_b)
sub_graph.set_entry_point("step_a")
sub_graph.add_edge("step_a", "step_b")
sub_graph.add_edge("step_b", END)
sub_app = sub_graph.compile()
# Use the compiled subgraph as a node in the parent graph
def run_sub(parent_state: ParentState) -> ParentState:
result = sub_app.invoke({"query": parent_state["question"]})
parent_state["sub_result"] = result["answer"]
return parent_state
parent = StateGraph(ParentState)
parent.add_node("sub_workflow", run_sub)
|
9. Prebuilt ReAct Agent
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| from langgraph.prebuilt import create_react_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
@tool
def search(query: str) -> str:
"""Search the web for up-to-date information."""
return f"Search results for: {query}"
@tool
def calculator(expression: str) -> str:
"""Evaluate a mathematical expression. e.g. '2 * (3 + 4)'"""
return str(eval(expression)) # use a safe evaluator in production
llm = ChatOpenAI(model="gpt-4o-mini")
agent = create_react_agent(
llm,
tools=[search, calculator],
state_modifier="You are a helpful research assistant.", # system prompt
checkpointer=MemorySaver(), # enable multi-turn memory
)
result = agent.invoke({
"messages": [{"role": "user", "content": "What is 15% of 2340?"}]
})
print(result["messages"][-1].content)
|
10. Error Handling Pattern
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| def safe_tool_node(state: AgentState) -> AgentState:
try:
result = call_external_api(state["params"])
state["tool_results"].append({"status": "ok", "data": result})
state["error"] = None
except TimeoutError:
state["error"] = "tool_timeout"
state["retry_count"] += 1
except PermissionError:
state["error"] = "unauthorized"
state["needs_approval"] = True
except Exception as e:
state["error"] = f"unexpected: {str(e)}"
return state
def retry_router(state: AgentState) -> str:
if state["error"] == "tool_timeout" and state["retry_count"] < 3:
return "tool_node" # retry
if state["error"] == "unauthorized":
return "human_review" # escalate
if state["error"]:
return "error_node" # terminal failure
return "answer_node" # success
graph.add_conditional_edges("tool_node", retry_router)
|
11. Full API Reference
| Method | Signature | Effect |
|---|
add_node | (name, fn) | Register a callable as a graph node |
add_edge | (from, to) | Unconditional transition |
add_conditional_edges | (from, fn, map?) | Routing transition |
set_entry_point | (name) | Designate first node |
set_finish_point | (name) | Mark a node as terminal |
compile | (checkpointer=, interrupt_before=, interrupt_after=) | Build executable graph |
invoke | (state, config) | Synchronous run to completion |
stream | (state, config, stream_mode=) | Streaming run |
aget_state | (config) | Inspect current checkpoint |
update_state | (config, values, as_node=) | Modify paused state |
get_state_history | (config, limit=) | List all past checkpoints |
Conclusion
LangGraph is most useful when the system is no longer a single prompt pipeline and starts behaving like a stateful workflow with branching, retries, memory, and human oversight. This cheat sheet covers the core patterns: state design, conditional edges, persistence, subgraphs, and error handling. Use it to keep control flow explicit and agentic behavior observable rather than hiding orchestration inside opaque loops.
12. Production RAG Agent with LangGraph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
| from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
# Initialize retriever
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
class RAGAgentState(TypedDict):
messages: Annotated[list, add_messages]
retrieved_docs: list[str]
generation_step: int
needs_more_info: bool
@tool
def search_knowledge_base(query: str) -> str:
"""Search the knowledge base for relevant information about the query."""
docs = retriever.invoke(query)
return "\n\n".join(f"[Source {i+1}]: {d.page_content}" for i, d in enumerate(docs))
@tool
def web_search(query: str) -> str:
"""Search the web for current information not in the knowledge base."""
# Integrate with Tavily or SerpAPI
return f"Web results for: {query} (integrate with real search API)"
tools = [search_knowledge_base, web_search]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
tool_node = ToolNode(tools)
SYSTEM_PROMPT = """You are a helpful research assistant.
Use the search_knowledge_base tool to find relevant information.
If the knowledge base doesn't have the answer, use web_search.
Always cite sources in your final answer."""
def agent_node(state: RAGAgentState) -> RAGAgentState:
messages = [{"role": "system", "content": SYSTEM_PROMPT}] + state["messages"]
response = llm.invoke(messages)
return {
"messages": [response],
"generation_step": state["generation_step"] + 1,
}
def should_continue(state: RAGAgentState) -> str:
last_msg = state["messages"][-1]
if state["generation_step"] >= 6: # safety limit
return END
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
return END
graph = StateGraph(RAGAgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "rag-session-001"}}
result = app.invoke({
"messages": [{"role": "user", "content": "What is our remote work policy?"}],
"retrieved_docs": [],
"generation_step": 0,
"needs_more_info": False,
}, config=config)
print(result["messages"][-1].content)
|
13. Multi-Agent Supervisor Pattern
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
| from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Literal
AGENTS = ["researcher", "writer", "reviewer"]
class SupervisorState(TypedDict):
task: str
agent_outputs: dict[str, str]
next_agent: str | None
iteration: int
final_answer: str | None
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def supervisor_node(state: SupervisorState) -> SupervisorState:
"""Decides which agent to run next."""
completed = list(state["agent_outputs"].keys())
prompt = f"""
Task: {state["task"]}
Completed agents: {completed}
Available agents: {AGENTS}
Which agent should run next?
- researcher: gathers facts and information
- writer: produces a draft based on research
- reviewer: reviews and improves the draft
If all agents have completed, return "FINISH".
Return only the agent name or "FINISH".
"""
decision = llm.invoke(prompt).content.strip()
return {"next_agent": None if decision == "FINISH" else decision}
def researcher_node(state: SupervisorState) -> SupervisorState:
result = llm.invoke(f"Research this topic thoroughly: {state['task']}").content
return {"agent_outputs": {**state["agent_outputs"], "researcher": result}}
def writer_node(state: SupervisorState) -> SupervisorState:
research = state["agent_outputs"].get("researcher", "")
result = llm.invoke(f"Write a professional article based on:\n{research}\n\nTopic: {state['task']}").content
return {"agent_outputs": {**state["agent_outputs"], "writer": result}}
def reviewer_node(state: SupervisorState) -> SupervisorState:
draft = state["agent_outputs"].get("writer", "")
result = llm.invoke(f"Review and improve this draft:\n{draft}").content
return {
"agent_outputs": {**state["agent_outputs"], "reviewer": result},
"final_answer": result,
}
def route_supervisor(state: SupervisorState) -> str:
if state["next_agent"] is None or state["iteration"] >= 6:
return END
return state["next_agent"]
graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)
graph.set_entry_point("supervisor")
for agent in AGENTS:
graph.add_edge(agent, "supervisor")
graph.add_conditional_edges("supervisor", route_supervisor, {
"researcher": "researcher",
"writer": "writer",
"reviewer": "reviewer",
END: END,
})
app = graph.compile()
result = app.invoke({
"task": "Explain quantum computing advances in 2024",
"agent_outputs": {},
"next_agent": None,
"iteration": 0,
"final_answer": None,
})
print(result["final_answer"])
|
14. Long-Running Workflows with Checkpointing
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
| from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict
class WorkflowState(TypedDict):
job_id: str
documents: list[str]
processed_count: int
summaries: list[str]
final_report: str | None
def process_batch_node(state: WorkflowState) -> WorkflowState:
"""Process a batch of documents."""
batch_size = 5
start_idx = state["processed_count"]
batch = state["documents"][start_idx : start_idx + batch_size]
new_summaries = []
for doc in batch:
summary = llm.invoke(f"Summarize in 2 sentences: {doc[:500]}").content
new_summaries.append(summary)
return {
"processed_count": state["processed_count"] + len(batch),
"summaries": state["summaries"] + new_summaries,
}
def should_continue_processing(state: WorkflowState) -> str:
if state["processed_count"] >= len(state["documents"]):
return "compile_report"
return "process_batch"
def compile_report_node(state: WorkflowState) -> WorkflowState:
all_summaries = "\n\n".join(state["summaries"])
report = llm.invoke(f"Compile a comprehensive report from:\n{all_summaries}").content
return {"final_report": report}
with SqliteSaver.from_conn_string("./jobs.db") as saver:
graph = StateGraph(WorkflowState)
graph.add_node("process_batch", process_batch_node)
graph.add_node("compile_report", compile_report_node)
graph.set_entry_point("process_batch")
graph.add_conditional_edges("process_batch", should_continue_processing)
graph.add_edge("compile_report", END)
app = graph.compile(checkpointer=saver)
config = {"configurable": {"thread_id": "batch-job-001"}}
# Start job
state = app.invoke({
"job_id": "batch-001",
"documents": ["doc1...", "doc2...", "doc3..."] * 10, # 30 docs
"processed_count": 0,
"summaries": [],
"final_report": None,
}, config=config)
# Job interrupted? Resume from checkpoint:
# state = app.invoke(None, config=config) # picks up where it left off
|
15. Testing LangGraph Workflows
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
| import pytest
from langgraph.checkpoint.memory import MemorySaver
def test_agent_handles_tool_failure():
"""Test that the agent gracefully handles a tool error."""
@tool
def failing_tool(query: str) -> str:
"""A tool that always fails."""
raise RuntimeError("External service unavailable")
memory = MemorySaver()
app = build_agent([failing_tool], memory=memory)
config = {"configurable": {"thread_id": "test-001"}}
result = app.invoke(
{"messages": [{"role": "user", "content": "Use the tool"}], "retry_count": 0},
config=config,
)
# The agent should either retry or gracefully degrade
last_msg = result["messages"][-1].content
assert "unavailable" in last_msg.lower() or "sorry" in last_msg.lower()
assert result["retry_count"] <= 3
def test_human_approval_flow():
"""Test that human-in-the-loop correctly pauses and resumes."""
memory = MemorySaver()
app = sensitive_app.compile(
checkpointer=memory,
interrupt_before=["execute_action"],
)
config = {"configurable": {"thread_id": "test-hitl-001"}}
# First run — should stop before execute_action
state = app.invoke({"request": "delete all Q1 data", "human_approved": None}, config)
snapshot = app.get_state(config)
assert snapshot.next == ("execute_action",)
# Reject and resume
app.update_state(config, {"human_approved": False})
final = app.invoke(None, config)
assert "rejected" in final["final_result"].lower()
|
LangGraph vs LangChain Decision Guide
| Scenario | Use | Reason |
|---|
| Simple prompt → response | LangChain LCEL | No state needed |
| RAG with one retrieval | LangChain | Linear pipeline |
| Multi-turn chat | LangChain + memory | Simple history |
| Retry logic, branching | LangGraph | Conditional edges |
| Human approval required | LangGraph | interrupt_before |
| Long-running job resumption | LangGraph + SqliteSaver | Checkpoint persistence |
| Multi-agent collaboration | LangGraph | State shared across agents |
| Complex agentic loops | LangGraph | Explicit cycle control |
Quick State Design Reference
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator
# Minimal chat state
class ChatState(TypedDict):
messages: Annotated[list, add_messages] # append-only
# Full agent state
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
docs: Annotated[list, operator.add] # extend
plan: list[str] # overwrite
step: int # overwrite
error: str | None # overwrite
approved: bool | None # overwrite
final_answer: str | None # overwrite
# Rule: use Annotated reducers for accumulating data,
# bare types for last-write-wins scalar fields
|
Graph Pattern Quick Reference
| Pattern | Implementation |
|---|
| Linear A→B→C | add_edge(A,B); add_edge(B,C) |
| Branch | add_conditional_edges(A, router_fn) |
| Retry loop | conditional_edges routing back to A |
| Fan-out | Send API in conditional_edges |
| Human pause | interrupt_before=["node_name"] |
| Subgraph | compiled_subgraph as node function |
| Persistent memory | SqliteSaver checkpointer |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
| # Pattern 1: Async node execution for I/O-bound operations
from langgraph.graph import StateGraph, END
import asyncio
async def async_retrieval_node(state):
"""Use async for retrieval to avoid blocking."""
docs = await retriever.ainvoke(state["question"])
return {"retrieved_docs": [d.page_content for d in docs]}
async def async_llm_node(state):
"""Async LLM call for non-blocking generation."""
context = "\n".join(state["retrieved_docs"])
response = await llm.ainvoke(f"Context: {context}\nQuestion: {state['question']}")
return {"answer": response.content}
# Pattern 2: State compression to avoid memory growth
def compress_state_node(state):
"""Trim state to prevent memory bloat in long conversations."""
messages = state.get("messages", [])
if len(messages) > 20:
# Keep first (system) + last 10 messages
compressed = messages[:1] + messages[-10:]
return {"messages": compressed}
return {} # no change needed
# Pattern 3: Batched graph execution
async def run_parallel_graphs(inputs: list[dict], compiled_graph) -> list[dict]:
"""Run the same graph on multiple inputs in parallel."""
tasks = [
compiled_graph.ainvoke(inp, config={"configurable": {"thread_id": str(i)}})
for i, inp in enumerate(inputs)
]
return await asyncio.gather(*tasks)
# Execute 10 independent queries in parallel
results = asyncio.run(run_parallel_graphs(
[{"question": q, "retry_count": 0} for q in questions],
app,
))
|
State Schema Best Practices
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| # DO: Use Annotated reducers for accumulating data
class GoodState(TypedDict):
messages: Annotated[list, add_messages] # LangGraph-aware appender
doc_ids: Annotated[list, operator.add] # simple list extension
score: float # scalar: last write wins
error: str | None # nullable scalar
# DON'T: Mutate state in-place inside a node
def bad_node(state):
state["messages"].append(new_message) # WRONG: mutates in-place
return state # breaks checkpointing
# DO: Return only the changed fields
def good_node(state):
return {"messages": [new_message]} # CORRECT: LangGraph merges this
|
LangGraph Production Principles
- State is the single source of truth: All inter-node communication flows through typed state
- Nodes must be pure functions: Same input always produces same output; no hidden side effects
- Use reducers for accumulation:
Annotated[list, add_messages] prevents subtle bugs from in-place mutation - Name your thread IDs meaningfully: Use user session IDs, not random UUIDs, for debuggability
- Always set max_iterations: An agent with no iteration cap can loop indefinitely and exhaust your budget
- Persist in production:
MemorySaver is for dev; use SqliteSaver or PostgresSaver for production - Test state machine logic separately: Unit-test routing functions independently from LLM calls
- Use
interrupt_before for auditability: Pause before destructive operations for human review
LangGraph shines when the agent needs to be observable, resumable, and safe. The explicit state machine is the price you pay for those properties — and in production, it is worth it.
LangGraph Conclusion
LangGraph is the production-grade foundation for stateful, cyclic LLM applications. Where LangChain LCEL handles linear pipelines, LangGraph handles workflows that need branching, loops, persistence, human-in-the-loop approval, and multi-agent collaboration. The explicit state machine design—with typed states, named nodes, conditional edges, and checkpoint-based persistence—trades initial complexity for long-term observability and reliability. For any LLM system that needs to handle more than a single round-trip with the model, LangGraph’s discipline pays dividends.
LangGraph Resources
LangGraph: Key Takeaways
- State is the single source of truth; all inter-node communication flows through typed state
- Use
Annotated reducers for accumulating data; plain types for scalar overwrite - Never mutate state in-place inside nodes; always return the changed keys
MemorySaver is for dev; use SqliteSaver or PostgresSaver in productioninterrupt_before requires a checkpointer to be effective- Always set a max iteration/step limit to prevent infinite agentic loops
- LangGraph is the right choice when you need branching, cycles, persistence, or human-in-the-loop
16. Parallel Node Execution
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
| from langgraph.graph import StateGraph, END
from typing import TypedDict
import asyncio
class ResearchState(TypedDict):
query: str
web_result: str | None
kb_result: str | None
analysis: str | None
async def web_search_node(state: ResearchState) -> ResearchState:
"""Async web search — runs in parallel with kb_search."""
result = await async_web_search(state["query"])
return {"web_result": result}
async def kb_search_node(state: ResearchState) -> ResearchState:
"""Knowledge base search — runs in parallel with web_search."""
result = await async_kb_search(state["query"])
return {"kb_result": result}
def synthesis_node(state: ResearchState) -> ResearchState:
"""Synthesize results from both search paths."""
context = f"Web:\n{state['web_result']}\n\nKB:\n{state['kb_result']}"
answer = llm.invoke(f"Synthesize: {context}\n\nQuery: {state['query']}").content
return {"analysis": answer}
# Build graph with parallel branches
graph = StateGraph(ResearchState)
graph.add_node("web_search", web_search_node)
graph.add_node("kb_search", kb_search_node)
graph.add_node("synthesize", synthesis_node)
graph.set_entry_point("web_search") # LangGraph can run parallel entry nodes
# Both search nodes point to synthesis
graph.add_edge("web_search", "synthesize")
graph.add_edge("kb_search", "synthesize")
graph.add_edge("synthesize", END)
# For true parallelism in LangGraph, use Send API:
from langgraph.constants import Send
def fan_out(state: ResearchState) -> list[Send]:
return [
Send("web_search", state),
Send("kb_search", state),
]
graph.add_conditional_edges("router", fan_out)
|
17. State Versioning and Migration
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| from langgraph.graph import StateGraph
from typing import TypedDict
from pydantic import BaseModel
# Version 1 state
class StateV1(TypedDict):
messages: list
answer: str | None
# Version 2 state (adds new field)
class StateV2(TypedDict):
messages: list
answer: str | None
confidence: float | None # NEW in v2
source_docs: list # NEW in v2
def migrate_v1_to_v2(old_state: dict) -> StateV2:
"""Migrate persisted v1 state to v2 schema."""
return {
"messages": old_state.get("messages", []),
"answer": old_state.get("answer"),
"confidence": None, # default for new field
"source_docs": [], # default for new field
}
# Load old checkpoint and migrate
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("./state.db") as saver:
app = graph_v2.compile(checkpointer=saver)
config = {"configurable": {"thread_id": "user-123"}}
snapshot = app.get_state(config)
if snapshot and "confidence" not in snapshot.values:
# Old v1 state detected — migrate
migrated = migrate_v1_to_v2(snapshot.values)
app.update_state(config, migrated)
|
18. Observability and Debugging
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| from langsmith import traceable
from langgraph.graph import StateGraph, END
# Method 1: LangSmith tracing (automatic with env vars set)
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_key"
# All graph.compile().invoke() calls are automatically traced
# Method 2: Custom event logging
class LoggingGraph:
def __init__(self, compiled_graph):
self.graph = compiled_graph
def invoke(self, state: dict, config: dict = None) -> dict:
print(f"[GRAPH START] Initial state keys: {list(state.keys())}")
for step in self.graph.stream(state, config, stream_mode="updates"):
for node_name, update in step.items():
print(f"[NODE: {node_name}] Updated keys: {list(update.keys())}")
final = self.graph.invoke(state, config)
print(f"[GRAPH END] Final state keys: {list(final.keys())}")
return final
# Method 3: State inspection at each step
logged_app = LoggingGraph(app)
result = logged_app.invoke(initial_state, config)
# Method 4: Replay a specific checkpoint
snapshot = app.get_state(config)
for step in app.get_state_history(config, limit=10):
print(f"Step {step.config['configurable']['checkpoint_id'][:8]}: "
f"next={step.next}, created={step.created_at}")
|
19. Deployment Patterns
FastAPI + LangGraph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
| from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.checkpoint.memory import MemorySaver
app_api = FastAPI()
memory = MemorySaver()
agent = compiled_graph.compile(checkpointer=memory)
class ChatRequest(BaseModel):
session_id: str
message: str
class ChatResponse(BaseModel):
answer: str
session_id: str
@app_api.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
config = {"configurable": {"thread_id": request.session_id}}
state = {
"messages": [{"role": "user", "content": request.message}],
"retry_count": 0,
}
try:
result = agent.invoke(state, config)
answer = result["messages"][-1].content
return ChatResponse(answer=answer, session_id=request.session_id)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app_api.get("/session/{session_id}")
async def get_session_state(session_id: str):
config = {"configurable": {"thread_id": session_id}}
snapshot = agent.get_state(config)
if not snapshot:
raise HTTPException(status_code=404, detail="Session not found")
return {
"session_id": session_id,
"next": snapshot.next,
"message_count": len(snapshot.values.get("messages", [])),
}
|
20. Common Pitfalls
| Pitfall | Symptom | Fix |
|---|
| Infinite loop | Graph never terminates | Add max_iterations guard in router |
| Missing state key | KeyError in node | Use .get() with defaults; validate state schema |
| Checkpoint grows unbounded | Memory leak in MemorySaver | Use SqliteSaver with retention policy |
| Race condition in parallel nodes | Inconsistent state | Use Annotated reducers; never mutate state in-place |
| Thread ID collision | Users share state | Use UUID-based thread IDs, never static strings |
| Node function not pure | Flaky behavior on retry | Ensure nodes are idempotent pure functions |
interrupt_before ignored | No pause occurs | Must use checkpointer and call invoke(None, config) to resume |
