LangGraph Code Review
When reviewing LangGraph code, check for these categories of issues.
Review gates (sequenced)
Complete in order. Each step has an objective pass condition before moving on.
-
Locate graph code — Search the review scope for
StateGraph,compile(,invoke,ainvoke,add_node,add_edge,add_conditional_edges. Pass: a short list of file paths (or explicit “none in scope” after searching). -
Map state schema — For each graph state type (
TypedDict,BaseModel, etc.), list fields that hold lists, dicts, or messages and whetherAnnotated+ reducers (add_messages,operator.add, …) are present. Pass: every such field is either covered by a reducer pattern below or explicitly flagged as intentional overwrite. -
Trace persistence — If interrupts,
thread_id, or checkpoint APIs appear, follow them tocompile(..., checkpointer=...)and invocationconfig. Pass: behavior matches the interrupt/checkpointer/thread_id guidance below—or you document a concrete mismatch with file:line. -
Report with evidence — For each finding you will deliver, record file path and line number(s) (or a minimal quoted snippet). Pass: no critical or high-severity issue is stated without that citation.
-
Run the checklist — Use the checklist at the end of this skill; each item is satisfied, not applicable (with reason), or open with evidence. Pass: no item left silently unchecked.
Critical Issues
1. State Mutation Instead of Return
# BAD - mutates state directly
def my_node(state: State) -> None:
state["messages"].append(new_message) # Mutation!
# GOOD - returns partial update
def my_node(state: State) -> dict:
return {"messages": [new_message]} # Let reducer handle it
2. Missing Reducer for List Fields
# BAD - no reducer, each node overwrites
class State(TypedDict):
messages: list # Will be overwritten, not appended!
# GOOD - reducer appends
class State(TypedDict):
messages: Annotated[list, operator.add]
# Or use add_messages for chat:
messages: Annotated[list, add_messages]
3. Wrong Return Type from Conditional Edge
# BAD - returns invalid node name
def router(state) -> str:
return "nonexistent_node" # Runtime error!
# GOOD - use Literal type hint for safety
def router(state) -> Literal["agent", "tools", "__end__"]:
if condition:
return "agent"
return END # Use constant, not string
4. Missing Checkpointer for Interrupts
# BAD - interrupt without checkpointer
def my_node(state):
answer = interrupt("question") # Will fail!
return {"answer": answer}
graph = builder.compile() # No checkpointer!
# GOOD - checkpointer required for interrupts
graph = builder.compile(checkpointer=InMemorySaver())
5. Forgetting Thread ID with Checkpointer
# BAD - no thread_id
graph.invoke({"messages": [...]}) # Error with checkpointer!
# GOOD - always provide thread_id
config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [...]}, config)
State Schema Issues
6. Using add_messages Without Message Types
# BAD - add_messages expects message-like objects
class State(TypedDict):
messages: Annotated[list, add_messages]
def node(state):
return {"messages": ["plain string"]} # May fail!
# GOOD - use proper message types or tuples
def node(state):
return {"messages": [("assistant", "response")]}
# Or: [AIMessage(content="response")]
7. Returning Full State Instead of Partial
# BAD - returns entire state (may reset other fields)
def my_node(state: State) -> State:
return {
"counter": state["counter"] + 1,
"messages": state["messages"], # Unnecessary!
"other": state["other"] # Unnecessary!
}
# GOOD - return only changed fields
def my_node(state: State) -> dict:
return {"counter": state["counter"] + 1}
8. Pydantic State Without Annotations
# BAD - Pydantic model without reducer loses append behavior
class State(BaseModel):
messages: list # No reducer!
# GOOD - use Annotated even with Pydantic
class State(BaseModel):
messages: Annotated[list, add_messages]
Graph Structure Issues
9. Missing Entry Point
# BAD - no edge from START
builder.add_node("process", process_fn)
builder.add_edge("process", END)
graph = builder.compile() # Error: no entrypoint!
# GOOD - connect START
builder.add_edge(START, "process")
10. Unreachable Nodes
# BAD - orphan node
builder.add_node("main", main_fn)
builder.add_node("orphan", orphan_fn) # Never reached!
builder.add_edge(START, "main")
builder.add_edge("main", END)
# Check with visualization
print(graph.get_graph().draw_mermaid())
11. Conditional Edge Without All Paths
# BAD - missing path in conditional
def router(state) -> Literal["a", "b", "c"]:
...
builder.add_conditional_edges("node", router, {"a": "a", "b": "b"})
# "c" path missing!
# GOOD - include all possible returns
builder.add_conditional_edges("node", router, {"a": "a", "b": "b", "c": "c"})
# Or omit path_map to use return values as node names
12. Command Without destinations
# BAD - Command return without destinations (breaks visualization)
def dynamic(state) -> Command[Literal["next", "__end__"]]:
return Command(goto="next")
builder.add_node("dynamic", dynamic) # Graph viz won't show edges
# GOOD - declare destinations
builder.add_node("dynamic", dynamic, destinations=["next", END])
Async Issues
13. Mixing Sync/Async Incorrectly
# BAD - async node called with sync invoke
async def my_node(state):
result = await async_operation()
return {"result": result}
graph.invoke(input) # May not await properly!
# GOOD - use ainvoke for async graphs
await graph.ainvoke(input)
# Or provide both sync and async versions
14. Blocking Calls in Async Context
# BAD - blocking call in async node
async def my_node(state):
result = requests.get(url) # Blocks event loop!
return {"result": result}
# GOOD - use async HTTP client
async def my_node(state):
async with httpx.AsyncClient() as client:
result = await client.get(url)
return {"result": result}
Tool Integration Issues
15. Tool Calls Without Corresponding ToolMessage
# BAD - AI message with tool_calls but no tool execution
messages = [
HumanMessage(content="search for X"),
AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}])
# Missing ToolMessage! Next LLM call will fail
]
# GOOD - always pair tool_calls with ToolMessage
messages = [
HumanMessage(content="search for X"),
AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]),
ToolMessage(content="results", tool_call_id="1")
]
16. Parallel Tool Calls Before Interrupt
# BAD - model may call multiple tools including interrupt
model = ChatOpenAI().bind_tools([interrupt_tool, other_tool])
# If both called in parallel, interrupt behavior is undefined
# GOOD - disable parallel tool calls before interrupt
model = ChatOpenAI().bind_tools(
[interrupt_tool, other_tool],
parallel_tool_calls=False
)
Checkpointing Issues
17. InMemorySaver in Production
# BAD - in-memory checkpointer loses state on restart
graph = builder.compile(checkpointer=InMemorySaver()) # Testing only!
# GOOD - use persistent storage in production
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(conn_string)
graph = builder.compile(checkpointer=checkpointer)
18. Subgraph Checkpointer Confusion
# BAD - subgraph with explicit False prevents persistence
subgraph = sub_builder.compile(checkpointer=False)
# GOOD - use None to inherit parent's checkpointer
subgraph = sub