tomevault-io LangGraph Code Review
When reviewing LangGraph code, check for these categories of issues.
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_builder.compile(checkpointer=None) # Inherits from parent
# Or True for independent checkpointing
subgraph = sub_builder.compile(checkpointer=True)
Performance Issues
19. Large State in Every Update
# BAD - returning large data in every node
def node(state):
large_data = fetch_large_data()
return {"large_field": large_data} # Checkpointed every step!
# GOOD - use references or store
from langgraph.store.memory import InMemoryStore
def node(state, *, store: BaseStore):
store.put(namespace, key, large_data)
return {"data_ref": f"{namespace}/{key}"}
20. Missing Recursion Limit Handling
# BAD - no protection against infinite loops
def router(state):
return "agent" # Always loops!
# GOOD - check remaining steps or use RemainingSteps
from langgraph.managed import RemainingSteps
class State(TypedDict):
messages: Annotated[list, add_messages]
remaining_steps: RemainingSteps
def check_limit(state):
if state["remaining_steps"] < 2:
return END
return "continue"
Code Review Checklist
- State schema uses Annotated with reducers for collections
- Nodes return partial state updates, not mutations
- Conditional edges return valid node names or END
- Graph has path from START to all nodes
- Checkpointer provided if using interrupts
- Thread ID provided in config when using checkpointer
- Tool calls paired with ToolMessages
- Async nodes use async operations
- Production uses persistent checkpointer
- Recursion limits considered for loops
Converted and distributed by TomeVault — claim your Tome and manage your conversions.