langgraph-code-review

LangGraph コードレビュー

LangGraph のコードをレビューする際は、次のカテゴリの問題を確認してください。

幻覚の防止（ゲート 0 — 他のすべてのゲートより先に実行）

指摘を行う前に — バグ、アンチパターン、改善点を指摘する場合は — 必ず、このターンで読んだソースから引用して、判断対象となる正確な成果物を示してください。

• コードに関する指摘：file:line と、いま改めて読み直した該当コード。
• レビュー対象のグラフ／状態コード：指摘の根拠となる StateGraph、ノード、エッジ、または状態スキーマのスニペットを、いま読み直したファイルから引用します。

成果物だけが信頼できる唯一の情報源です。ブランチ名、作業ディレクトリ、周辺ファイル、記憶からレビュー対象を推測してはいけません。頭の中のモデルが読み直したソースと異なる場合は、ソースを優先してください。同じターンに対象を引用せずに行った指摘は無効です。まず引用するか、指摘を行わないでください。

このゲートは、コンテキストによるプライミングを受けた LLM が、ファイルに存在しないコードを自信満々に指摘することを防ぐためにあります。以下のゲートより先に実行します。

レビューゲート（順番に実施）

順番どおりに完了してください。次に進む前に、各ステップの客観的な合格条件を満たす必要があります。

1. グラフコードを特定する — レビュー対象の範囲で StateGraph、compile(、invoke、ainvoke、add_node、add_edge、add_conditional_edges を検索します。合格条件： ファイルパスの短い一覧を示すか、検索後に「対象範囲にはありません」と明記します。

2. 状態スキーマを把握する — グラフの各状態型（TypedDict、BaseModel など）について、リスト、辞書、またはメッセージを保持するフィールドと、Annotated およびリデューサー（add_messages、operator.add など）の有無を一覧にします。合格条件： 該当するすべてのフィールドが、以下のリデューサーパターンでカバーされているか、意図的に上書きするものとして明示されています。

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. リストフィールドにリデューサーがない

# 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. チェックポインター使用時にスレッド 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. destinations のない 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. ループの再帰制限を考慮している