langgraph-code-review

LangGraph 代码审查

审查 LangGraph 代码时，请检查以下类别的问题。

防止虚构（第 0 道关卡——先于其他所有关卡运行）

在提出任何发现——标记 Bug、反模式或改进项——之前，必须从你在本轮中阅读的来源中引用并逐字复述你正在评判的确切对象：

• 代码发现：其 file:line 和引用的代码，必须现在重新读取。
• 正在审查的图/状态代码：你所提出的发现所依据的 StateGraph、节点、边或状态架构片段，必须引用自你刚刚读取的文件。

该对象是唯一的事实来源。切勿根据分支名称、工作目录、周边文件或记忆推断你正在审查的内容。如果你的思维模型与本轮刚读取的源代码不符，**以源代码为准。**提出发现时若未在同一轮复述其目标，该发现即无效——先复述，或者不要提出该发现。

设置此关卡是因为 LLM 在上下文诱导下会自信地指出文件中并不存在的代码。它先于下方的关卡运行。

审查关卡（按顺序执行）

按顺序完成。每一步都有明确的通过条件，通过后才能继续。

1. 定位图代码 — 在审查范围内搜索 StateGraph、compile(、invoke、ainvoke、add_node、add_edge、add_conditional_edges。**通过条件：**列出简短的文件路径清单（或搜索后明确写出“范围内没有”）。

2. 梳理状态架构 — 对每种图状态类型（TypedDict、BaseModel 等），列出存放列表、字典或消息的字段，并注明是否存在 Annotated 和 reducer（add_messages、operator.add 等）。**通过条件：**每个此类字段都由下方的 reducer 模式覆盖，或明确标记为有意覆盖。

3. 追踪持久化 — 如果出现中断、thread_id 或检查点 API，请追踪到 compile(..., checkpointer=...) 和调用时的 config。**通过条件：**行为符合下方关于中断、检查点和 thread_id 的指导——或者记录包含文件和行号的具体不匹配之处。

4. 提供有依据的报告 — 对每个要提交的发现，记录文件路径和行号（或最小的引用片段）。**通过条件：**没有任何严重或高严重性问题是在缺少该引用的情况下提出的。

5. 执行检查清单 — 使用本技能末尾的检查清单；每项都必须标记为已满足、不适用（附理由）或待处理（附依据）。**通过条件：**没有任何检查项被默默跳过。

严重问题

1. 修改状态而非返回状态

# 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. 列表字段缺少 Reducer

# 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. 条件边的返回类型错误

# 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. 中断时缺少检查点保存器

# 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. 使用检查点保存器时忘记 Thread ID

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

状态架构问题

6. 使用 add_messages 时未提供消息类型

# 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. 返回完整状态而非部分状态

# 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 状态缺少注解

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

图结构问题

9. 缺少入口点

# 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. 存在不可达节点

# 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. 条件边未覆盖所有路径

# 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 未指定目标

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

异步问题

13. 同步与异步使用不当

# 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. 异步上下文中的阻塞调用

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

工具集成问题

15. 工具调用没有对应的 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. 中断前并行调用工具

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

检查点问题

17. 在生产环境中使用 InMemorySaver

# 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. 子图检查点保存器使用混淆

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

性能问题

19. 每次更新都携带大量状态

# 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. 缺少递归限制处理

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

代码审查清单

1. 状态架构使用带有集合归约器的 Annotated
2. 节点返回部分状态更新，而非直接修改状态
3. 条件边返回有效的节点名称或 END
4. 图中从 START 到所有节点都有路径
5. 使用中断时提供检查点保存器
6. 使用检查点保存器时，在配置中提供线程 ID
7. 工具调用与 ToolMessages 成对出现
8. 异步节点使用异步操作
9. 生产环境使用持久化检查点保存器
10. 考虑循环的递归限制