Langgraph Code Review

# BAD - mutates state directlydef my_node(state: State) -> None:    state["messages"].append(new_message)  # Mutation!
# GOOD - returns partial updatedef my_node(state: State) -> dict:    return {"messages": [new_message]}  # Let reducer handle it

# BAD - no reducer, each node overwritesclass State(TypedDict):    messages: list  # Will be overwritten, not appended!
# GOOD - reducer appendsclass State(TypedDict):    messages: Annotated[list, operator.add]    # Or use add_messages for chat:    messages: Annotated[list, add_messages]

# BAD - returns invalid node namedef router(state) -> str:    return "nonexistent_node"  # Runtime error!
# GOOD - use Literal type hint for safetydef router(state) -> Literal["agent", "tools", "__end__"]:    if condition:        return "agent"    return END  # Use constant, not string

# BAD - interrupt without checkpointerdef my_node(state):    answer = interrupt("question")  # Will fail!    return {"answer": answer}
graph = builder.compile()  # No checkpointer!
# GOOD - checkpointer required for interruptsgraph = builder.compile(checkpointer=InMemorySaver())

# BAD - no thread_idgraph.invoke({"messages": [...]})  # Error with checkpointer!
# GOOD - always provide thread_idconfig = {"configurable": {"thread_id": "user-123"}}graph.invoke({"messages": [...]}, config)

# BAD - add_messages expects message-like objectsclass State(TypedDict):    messages: Annotated[list, add_messages]
def node(state):    return {"messages": ["plain string"]}  # May fail!
# GOOD - use proper message types or tuplesdef node(state):    return {"messages": [("assistant", "response")]}    # Or: [AIMessage(content="response")]

# 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 fieldsdef my_node(state: State) -> dict:    return {"counter": state["counter"] + 1}

# BAD - Pydantic model without reducer loses append behaviorclass State(BaseModel):    messages: list  # No reducer!
# GOOD - use Annotated even with Pydanticclass State(BaseModel):    messages: Annotated[list, add_messages]

# BAD - no edge from STARTbuilder.add_node("process", process_fn)builder.add_edge("process", END)graph = builder.compile()  # Error: no entrypoint!
# GOOD - connect STARTbuilder.add_edge(START, "process")

# BAD - orphan nodebuilder.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 visualizationprint(graph.get_graph().draw_mermaid())

# BAD - missing path in conditionaldef router(state) -> Literal["a", "b", "c"]:    ...
builder.add_conditional_edges("node", router, {"a": "a", "b": "b"})# "c" path missing!
# GOOD - include all possible returnsbuilder.add_conditional_edges("node", router, {"a": "a", "b": "b", "c": "c"})# Or omit path_map to use return values as node names

# 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 destinationsbuilder.add_node("dynamic", dynamic, destinations=["next", END])

# BAD - async node called with sync invokeasync def my_node(state):    result = await async_operation()    return {"result": result}
graph.invoke(input)  # May not await properly!
# GOOD - use ainvoke for async graphsawait graph.ainvoke(input)# Or provide both sync and async versions

# BAD - blocking call in async nodeasync def my_node(state):    result = requests.get(url)  # Blocks event loop!    return {"result": result}
# GOOD - use async HTTP clientasync def my_node(state):    async with httpx.AsyncClient() as client:        result = await client.get(url)    return {"result": result}

# BAD - AI message with tool_calls but no tool executionmessages = [    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 ToolMessagemessages = [    HumanMessage(content="search for X"),    AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]),    ToolMessage(content="results", tool_call_id="1")]

# BAD - model may call multiple tools including interruptmodel = ChatOpenAI().bind_tools([interrupt_tool, other_tool])# If both called in parallel, interrupt behavior is undefined
# GOOD - disable parallel tool calls before interruptmodel = ChatOpenAI().bind_tools(    [interrupt_tool, other_tool],    parallel_tool_calls=False)

# BAD - in-memory checkpointer loses state on restartgraph = builder.compile(checkpointer=InMemorySaver())  # Testing only!
# GOOD - use persistent storage in productionfrom langgraph.checkpoint.postgres import PostgresSavercheckpointer = PostgresSaver.from_conn_string(conn_string)graph = builder.compile(checkpointer=checkpointer)

# BAD - subgraph with explicit False prevents persistencesubgraph = sub_builder.compile(checkpointer=False)
# GOOD - use None to inherit parent's checkpointersubgraph = sub_builder.compile(checkpointer=None)  # Inherits from parent# Or True for independent checkpointingsubgraph = sub_builder.compile(checkpointer=True)

# BAD - returning large data in every nodedef node(state):    large_data = fetch_large_data()    return {"large_field": large_data}  # Checkpointed every step!
# GOOD - use references or storefrom langgraph.store.memory import InMemoryStore
def node(state, *, store: BaseStore):    store.put(namespace, key, large_data)    return {"data_ref": f"{namespace}/{key}"}

# BAD - no protection against infinite loopsdef router(state):    return "agent"  # Always loops!
# GOOD - check remaining steps or use RemainingStepsfrom 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"