langgraph-code-review

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

LangGraph Code Review

LangGraph代码审查

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

审查LangGraph代码时，需检查以下几类问题。

Critical Issues

严重问题

1. State Mutation Instead of Return

1. 直接修改状态而非返回更新

python

undefined

python

undefined

BAD - mutates state directly

错误示例 - 直接修改状态

def my_node(state: State) -> None: state["messages"].append(new_message) # Mutation!

def my_node(state: State) -> None: state["messages"].append(new_message) # 直接修改！

GOOD - returns partial update

正确示例 - 返回部分更新

def my_node(state: State) -> dict: return {"messages": [new_message]} # Let reducer handle it

undefined

def my_node(state: State) -> dict: return {"messages": [new_message]} # 交由reducer处理

undefined

2. Missing Reducer for List Fields

2. 列表字段缺少Reducer

python

undefined

python

undefined

BAD - no reducer, each node overwrites

错误示例 - 无reducer，每个节点会覆盖值

class State(TypedDict): messages: list # Will be overwritten, not appended!

class State(TypedDict): messages: list # 会被覆盖，而非追加！

GOOD - reducer appends

正确示例 - 使用reducer追加

class State(TypedDict): messages: Annotated[list, operator.add] # Or use add_messages for chat: messages: Annotated[list, add_messages]

undefined

class State(TypedDict): messages: Annotated[list, operator.add] # 或者对聊天场景使用add_messages: messages: Annotated[list, add_messages]

undefined

3. Wrong Return Type from Conditional Edge

3. 条件边返回错误类型

python

undefined

python

undefined

BAD - returns invalid node name

错误示例 - 返回不存在的节点名称

def router(state) -> str: return "nonexistent_node" # Runtime error!

def router(state) -> str: return "nonexistent_node" # 运行时错误！

GOOD - use Literal type hint for safety

正确示例 - 使用Literal类型提示保证安全

def router(state) -> Literal["agent", "tools", "end"]: if condition: return "agent" return END # Use constant, not string

undefined

def router(state) -> Literal["agent", "tools", "end"]: if condition: return "agent" return END # 使用常量，而非字符串

undefined

4. Missing Checkpointer for Interrupts

4. 使用中断但未配置检查点

python

undefined

python

undefined

BAD - interrupt without checkpointer

错误示例 - 未配置检查点就使用中断

def my_node(state): answer = interrupt("question") # Will fail! return {"answer": answer}

graph = builder.compile() # No checkpointer!

def my_node(state): answer = interrupt("question") # 执行失败！ return {"answer": answer}

graph = builder.compile() # 无检查点！

GOOD - checkpointer required for interrupts

正确示例 - 使用中断必须配置检查点

graph = builder.compile(checkpointer=InMemorySaver())

undefined

graph = builder.compile(checkpointer=InMemorySaver())

undefined

5. Forgetting Thread ID with Checkpointer

5. 使用检查点但未提供线程ID

python

undefined

python

undefined

BAD - no thread_id

错误示例 - 未提供thread_id

graph.invoke({"messages": [...]}) # Error with checkpointer!

graph.invoke({"messages": [...]}) # 配置检查点后会报错！

GOOD - always provide thread_id

正确示例 - 始终提供thread_id

config = {"configurable": {"thread_id": "user-123"}} graph.invoke({"messages": [...]}, config)

undefined

config = {"configurable": {"thread_id": "user-123"}} graph.invoke({"messages": [...]}, config)

undefined

State Schema Issues

状态模式问题

6. Using add_messages Without Message Types

6. 使用add_messages但未使用消息类型

python

undefined

python

undefined

BAD - add_messages expects message-like objects

错误示例 - add_messages需要类消息对象

class State(TypedDict): messages: Annotated[list, add_messages]

def node(state): return {"messages": ["plain string"]} # May fail!

class State(TypedDict): messages: Annotated[list, add_messages]

def node(state): return {"messages": ["plain string"]} # 可能执行失败！

GOOD - use proper message types or tuples

正确示例 - 使用正确的消息类型或元组

def node(state): return {"messages": [("assistant", "response")]} # Or: [AIMessage(content="response")]

undefined

def node(state): return {"messages": [("assistant", "response")]} # 或者：[AIMessage(content="response")]

undefined

7. Returning Full State Instead of Partial

7. 返回完整状态而非部分更新

python

undefined

python

undefined

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

def my_node(state: State) -> State: return { "counter": state["counter"] + 1, "messages": state["messages"], # 不必要！ "other": state["other"] # 不必要！ }

GOOD - return only changed fields

正确示例 - 仅返回修改的字段

def my_node(state: State) -> dict: return {"counter": state["counter"] + 1}

undefined

def my_node(state: State) -> dict: return {"counter": state["counter"] + 1}

undefined

8. Pydantic State Without Annotations

8. Pydantic状态未使用注解

python

undefined

python

undefined

BAD - Pydantic model without reducer loses append behavior

错误示例 - Pydantic模型无reducer会丢失追加行为

class State(BaseModel): messages: list # No reducer!

class State(BaseModel): messages: list # 无reducer！

GOOD - use Annotated even with Pydantic

正确示例 - 即使使用Pydantic也要用Annotated

class State(BaseModel): messages: Annotated[list, add_messages]

undefined

class State(BaseModel): messages: Annotated[list, add_messages]

undefined

Graph Structure Issues

图结构问题

9. Missing Entry Point

9. 缺少入口点

python

undefined

python

undefined

BAD - no edge from START

错误示例 - 无从START出发的边

builder.add_node("process", process_fn) builder.add_edge("process", END) graph = builder.compile() # Error: no entrypoint!

builder.add_node("process", process_fn) builder.add_edge("process", END) graph = builder.compile() # 错误：无入口点！

GOOD - connect START

正确示例 - 连接START

builder.add_edge(START, "process")

undefined

builder.add_edge(START, "process")

undefined

10. Unreachable Nodes

10. 不可达节点

python

undefined

python

undefined

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)

builder.add_node("main", main_fn) builder.add_node("orphan", orphan_fn) # 永远无法被访问！ builder.add_edge(START, "main") builder.add_edge("main", END)

Check with visualization

通过可视化检查

print(graph.get_graph().draw_mermaid())

undefined

print(graph.get_graph().draw_mermaid())

undefined

11. Conditional Edge Without All Paths

11. 条件边未覆盖所有路径

python

undefined

python

undefined

BAD - missing path in conditional

错误示例 - 条件中缺少路径

def router(state) -> Literal["a", "b", "c"]: ...

builder.add_conditional_edges("node", router, {"a": "a", "b": "b"})

def router(state) -> Literal["a", "b", "c"]: ...

builder.add_conditional_edges("node", router, {"a": "a", "b": "b"})

"c" path missing!

缺少"c"路径！

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

或者省略path_map，直接使用返回值作为节点名称

undefined

undefined

12. Command Without destinations

12. Command未指定目标节点

python

undefined

python

undefined

BAD - Command return without destinations (breaks visualization)

错误示例 - 返回Command但未指定目标节点（会破坏可视化）

def dynamic(state) -> Command[Literal["next", "end"]]: return Command(goto="next")

builder.add_node("dynamic", dynamic) # Graph viz won't show edges

def dynamic(state) -> Command[Literal["next", "end"]]: return Command(goto="next")

builder.add_node("dynamic", dynamic) # 图可视化无法显示边

GOOD - declare destinations

正确示例 - 声明目标节点

builder.add_node("dynamic", dynamic, destinations=["next", END])

undefined

builder.add_node("dynamic", dynamic, destinations=["next", END])

undefined

Async Issues

异步问题

13. Mixing Sync/Async Incorrectly

13. 同步/异步混用错误

python

undefined

python

undefined

BAD - async node called with sync invoke

错误示例 - 异步节点使用同步invoke调用

async def my_node(state): result = await async_operation() return {"result": result}

graph.invoke(input) # May not await properly!

async def my_node(state): result = await async_operation() return {"result": result}

graph.invoke(input) # 可能无法正确等待！

GOOD - use ainvoke for async graphs

正确示例 - 异步图使用ainvoke

await graph.ainvoke(input)

Or provide both sync and async versions

或者同时提供同步和异步版本

undefined

undefined

14. Blocking Calls in Async Context

14. 异步上下文使用阻塞调用

python

undefined

python

undefined

BAD - blocking call in async node

错误示例 - 异步节点中使用阻塞调用

async def my_node(state): result = requests.get(url) # Blocks event loop! return {"result": result}

async def my_node(state): result = requests.get(url) # 会阻塞事件循环！ return {"result": result}

GOOD - use async HTTP client

正确示例 - 使用异步HTTP客户端

async def my_node(state): async with httpx.AsyncClient() as client: result = await client.get(url) return {"result": result}

undefined

async def my_node(state): async with httpx.AsyncClient() as client: result = await client.get(url) return {"result": result}

undefined

Tool Integration Issues

工具集成问题

15. Tool Calls Without Corresponding ToolMessage

15. 工具调用未对应ToolMessage

python

undefined

python

undefined

BAD - AI message with tool_calls but no tool execution

错误示例 - AI消息包含tool_calls但未执行工具

messages = [ HumanMessage(content="search for X"), AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]) # Missing ToolMessage! Next LLM call will fail ]

messages = [ HumanMessage(content="search for X"), AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]) # 缺少ToolMessage！下一次LLM调用会失败 ]

GOOD - always pair tool_calls with ToolMessage

正确示例 - 始终为tool_calls配对ToolMessage

messages = [ HumanMessage(content="search for X"), AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]), ToolMessage(content="results", tool_call_id="1") ]

undefined

messages = [ HumanMessage(content="search for X"), AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}]), ToolMessage(content="results", tool_call_id="1") ]

undefined

16. Parallel Tool Calls Before Interrupt

16. 中断前使用并行工具调用

python

undefined

python

undefined

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 )

undefined

model = ChatOpenAI().bind_tools( [interrupt_tool, other_tool], parallel_tool_calls=False )

undefined

Checkpointing Issues

检查点问题

17. InMemorySaver in Production

17. 生产环境使用InMemorySaver

python

undefined

python

undefined

BAD - in-memory checkpointer loses state on restart

错误示例 - 内存检查点会在重启后丢失状态

graph = builder.compile(checkpointer=InMemorySaver()) # Testing only!

graph = builder.compile(checkpointer=InMemorySaver()) # 仅用于测试！

GOOD - use persistent storage in production

正确示例 - 生产环境使用持久化存储

from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string(conn_string) graph = builder.compile(checkpointer=checkpointer)

undefined

from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string(conn_string) graph = builder.compile(checkpointer=checkpointer)

undefined

18. Subgraph Checkpointer Confusion

18. 子图检查点配置混乱

python

undefined

python

undefined

BAD - subgraph with explicit False prevents persistence

错误示例 - 子图显式设置False会阻止持久化

subgraph = sub_builder.compile(checkpointer=False)

GOOD - use None to inherit parent's checkpointer

正确示例 - 使用None继承父图的检查点

subgraph = sub_builder.compile(checkpointer=None) # Inherits from parent

subgraph = sub_builder.compile(checkpointer=None) # 继承父图配置

Or True for independent checkpointing

或者设为True使用独立检查点

subgraph = sub_builder.compile(checkpointer=True)

undefined

subgraph = sub_builder.compile(checkpointer=True)

undefined

Performance Issues

性能问题

19. Large State in Every Update

19. 每次更新都返回大状态数据

python

undefined

python

undefined

BAD - returning large data in every node

错误示例 - 每个节点都返回大型数据

def node(state): large_data = fetch_large_data() return {"large_field": large_data} # Checkpointed every step!

def node(state): large_data = fetch_large_data() return {"large_field": large_data} # 每一步都会被检查点存储！

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

undefined

from langgraph.store.memory import InMemoryStore

def node(state, *, store: BaseStore): store.put(namespace, key, large_data) return {"data_ref": f"{namespace}/{key}"}

undefined

20. Missing Recursion Limit Handling

20. 未处理递归限制

python

undefined

python

undefined

BAD - no protection against infinite loops

错误示例 - 无无限循环防护

def router(state): return "agent" # Always loops!

def router(state): return "agent" # 始终循环！

GOOD - check remaining steps or use RemainingSteps

正确示例 - 检查剩余步骤或使用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"

undefined

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"

undefined