Langgraph Code Review

existential-birds/beagle/plugins/beagle-ai/skills/langgraph-code-review

作者 existential-birdsd1a74899fbfec74974d1818e4cac7c3d54d44b65無授權條款83 個星標收錄於 2026年10月9日更新於 2026年10月9日儲存庫3 週前更新

Reviews LangGraph code for bugs, anti-patterns, and improvements. Use when reviewing code that uses StateGraph, nodes, edges, checkpointing, or other LangGraph features. Catches common mistakes in state management, graph structure, and async patterns.

AI 產生的概覽

審查 LangGraph 程式碼中的缺陷、反模式與改進點,涵蓋狀態、圖結構、非同步與檢查點。

功能
為 LangGraph 程式碼提供一套依序執行的審查流程,首先要求在提出任何結論前引用正在評判的確切程式碼。接著依序定位圖程式碼、梳理狀態結構、追蹤持久化、附上檔案與行號引用報告發現,並執行檢查清單。它歸納了二十類常見問題,涉及狀態變更、reducer、條件邊、檢查點、非同步模式、工具整合與效能,每類都附有錯誤與正確程式碼範例。產出的是審查發現與已完成的檢查清單,而非修改程式碼。
適用情境
適用於審查使用 StateGraph、節點、邊、檢查點或其他 LangGraph 功能的程式碼。其目標是找出狀態管理、圖結構與非同步模式中的常見錯誤。
執行需求
無需腳本或工具,僅為指令型技能。審查者需要能存取被審查的 LangGraph 原始檔。

LangGraph Code Review

When reviewing LangGraph code, check for these categories of issues.

Anti-confabulation (gate 0 — runs before every other gate)

Before issuing any finding — flag a bug, anti-pattern, or improvement — you MUST echo the exact artifact you are judging, quoted from a source you read in this turn:

  • The code finding: its file:line plus the cited code, read freshly now.
  • The graph/state code under review: the StateGraph, node, edge, or state-schema snippet your finding depends on, quoted from the file you just read.

The artifact is the only source of truth. Never infer what you are reviewing from the branch name, the working directory, surrounding files, or recollection. If your mental model differs from the freshly read source, the source wins. A finding issued without a same-turn echo of its target is invalid — emit the echo first, or do not emit the finding.

This gate exists because an LLM under contextual priming will confidently flag code that is not in the file. It runs before the gates below.

Review gates (sequenced)

Complete in order. Each step has an objective pass condition before moving on.

  1. 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).

  2. Map state schema — For each graph state type (TypedDict, BaseModel, etc.), list fields that hold lists, dicts, or messages and whether Annotated + 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.

  3. Trace persistence — If interrupts, thread_id, or checkpoint APIs appear, follow them to compile(..., checkpointer=...) and invocation config. Pass: behavior matches the interrupt/checkpointer/thread_id guidance below—or you document a concrete mismatch with file:line.

  4. 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.

  5. 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

python
# 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

2. Missing Reducer for List Fields

python
# 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]

3. Wrong Return Type from Conditional Edge

python
# 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

4. Missing Checkpointer for Interrupts

python
# 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())

5. Forgetting Thread ID with Checkpointer

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

State Schema Issues

6. Using add_messages Without Message Types

python
# 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")]

7. Returning Full State Instead of Partial

python
# 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}

8. Pydantic State Without Annotations

python
# 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]

Graph Structure Issues

9. Missing Entry Point

python
# 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")

10. Unreachable Nodes

python
# 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())

11. Conditional Edge Without All Paths

python
# 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

12. Command Without destinations

python
# 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])

Async Issues

13. Mixing Sync/Async Incorrectly

python
# 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

14. Blocking Calls in Async Context

python
# 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}

Tool Integration Issues

15. Tool Calls Without Corresponding ToolMessage

python
# 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")]

16. Parallel Tool Calls Before Interrupt

python
# 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)

Checkpointing Issues

17. InMemorySaver in Production

python
# 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)

18. Subgraph Checkpointer Confusion

python
# 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)

Performance Issues

19. Large State in Every Update

python
# 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}"}

20. Missing Recursion Limit Handling

python
# 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"

Code Review Checklist

  1. State schema uses Annotated with reducers for collections
  2. Nodes return partial state updates, not mutations
  3. Conditional edges return valid node names or END
  4. Graph has path from START to all nodes
  5. Checkpointer provided if using interrupts
  6. Thread ID provided in config when using checkpointer
  7. Tool calls paired with ToolMessages
  8. Async nodes use async operations
  9. Production uses persistent checkpointer
  10. Recursion limits considered for loops

來源與署名

來源:existential-birds/beagle位於plugins/beagle-ai/skills/langgraph-code-review提交d1a7489

授權條款: 無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架