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 从公开仓库中收录这些内容。

举报或申请下架