LangGraph Code Review

LangGraphコードをレビューする際は、以下のカテゴリーの問題をチェックしてください。

レビューゲート（順序付き）

順番に完了してください。各ステップには、次のステップに進む前の目標合格条件があります。

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 ガイダンスと一致するか、file:line で具体的なミスマッチを文書化しています。

4. 証拠と共にレポート — 報告する各結果について、ファイルパスと行番号（または最小限の引用されたスニペット）を記録してください。合格： 重大または高重大度の問題が、その引用なしに述べられていません。

5. チェックリストを実行 — このスキルの最後のチェックリストを使用します。各項目は 満たされている、該当なし（理由付き）、または 未解決（証拠付き）です。合格： 静かにチェックされていない項目がありません。

重大な問題

1. 状態の変異の代わりに戻り値を返す

# BAD - 状態を直接変異
def my_node(state: State) -> None:
    state["messages"].append(new_message)  # Mutation!

# GOOD - 部分更新を返す
def my_node(state: State) -> dict:
    return {"messages": [new_message]}  # レデューサーに任せる

2. リストフィールドの レデューサーを忘れる

# BAD - レデューサーなし。各ノードが上書き
class State(TypedDict):
    messages: list  # 追加されず、上書きされます！

# GOOD - レデューサーが追加
class State(TypedDict):
    messages: Annotated[list, operator.add]
    # または チャット用の add_messages を使用:
    messages: Annotated[list, add_messages]

3. 条件付きエッジから間違った戻り値型

# BAD - 無効なノード名を返す
def router(state) -> str:
    return "nonexistent_node"  # ランタイムエラー！

# GOOD - 安全性のために Literal 型ヒント を使用
def router(state) -> Literal["agent", "tools", "__end__"]:
    if condition:
        return "agent"
    return END  # 文字列ではなく定数を使用

4. 割り込みに対するチェックポイントを忘れる

# BAD - チェックポイントなしの割り込み
def my_node(state):
    answer = interrupt("question")  # 失敗します！
    return {"answer": answer}

graph = builder.compile()  # チェックポイントなし！

# GOOD - 割り込みにはチェックポイントが必須
graph = builder.compile(checkpointer=InMemorySaver())

5. チェックポイント付きの Thread ID を忘れる

# BAD - thread_id がない
graph.invoke({"messages": [...]})  # チェックポイント付きでエラー！

# GOOD - 常に thread_id を提供
config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [...]}, config)

状態スキーマの問題

6. メッセージ型なしで add_messages を使用

# BAD - add_messages はメッセージのようなオブジェクトを期待
class State(TypedDict):
    messages: Annotated[list, add_messages]

def node(state):
    return {"messages": ["plain string"]}  # 失敗する可能性があります！

# GOOD - 適切なメッセージ型またはタプルを使用
def node(state):
    return {"messages": [("assistant", "response")]}
    # または: [AIMessage(content="response")]

7. 部分的ではなく完全な状態を返す

# BAD - 完全な状態を返す（他のフィールドをリセットする可能性がある）
def my_node(state: State) -> State:
    return {
        "counter": state["counter"] + 1,
        "messages": state["messages"],  # 不要！
        "other": state["other"]          # 不要！
    }

# GOOD - 変更したフィールドのみを返す
def my_node(state: State) -> dict:
    return {"counter": state["counter"] + 1}

8. アノテーションなしの Pydantic 状態

# BAD - レデューサーなしの Pydantic モデルは追加動作を失う
class State(BaseModel):
    messages: list  # レデューサーなし！

# GOOD - Pydantic でもアノテーションを使用
class State(BaseModel):
    messages: Annotated[list, add_messages]

グラフ構造の問題

9. エントリーポイントを忘れる

# BAD - START からのエッジがない
builder.add_node("process", process_fn)
builder.add_edge("process", END)
graph = builder.compile()  # エラー: エントリーポイントなし！

# GOOD - START に接続
builder.add_edge(START, "process")

10. 到達不可能なノード

# BAD - 孤立したノード
builder.add_node("main", main_fn)
builder.add_node("orphan", orphan_fn)  # 到達できません！
builder.add_edge(START, "main")
builder.add_edge("main", END)

# ビジュアライゼーション で確認
print(graph.get_graph().draw_mermaid())

11. すべてのパスのない条件付きエッジ

# BAD - 条件付きで パスが不足
def router(state) -> Literal["a", "b", "c"]:
    ...

builder.add_conditional_edges("node", router, {"a": "a", "b": "b"})
# "c" パスが不足！

# GOOD - すべての可能な戻り値を含める
builder.add_conditional_edges("node", router, {"a": "a", "b": "b", "c": "c"})
# またはパスマップを省略して戻り値をノード名として使用

12. 宛先のない Command

# BAD - 宛先のないCommand戻り値（グラフビジュアライゼーション を壊す）
def dynamic(state) -> Command[Literal["next", "__end__"]]:
    return Command(goto="next")

builder.add_node("dynamic", dynamic)  # グラフビジュアライゼーション はエッジを表示しません

# GOOD - 宛先を宣言
builder.add_node("dynamic", dynamic, destinations=["next", END])

非同期の問題

13. 同期/非同期の混在

# BAD - 非同期ノードが同期 invoke で呼ばれる
async def my_node(state):
    result = await async_operation()
    return {"result": result}

graph.invoke(input)  # 正しく await されない可能性があります！

# GOOD - 非同期グラフには ainvoke を使用
await graph.ainvoke(input)
# または同期版と非同期版の両方を提供

14. 非同期コンテキストでのブロッキング呼び出し

# BAD - 非同期ノード内のブロッキング呼び出し
async def my_node(state):
    result = requests.get(url)  # イベントループをブロック！
    return {"result": result}

# GOOD - 非同期 HTTP クライアントを使用
async def my_node(state):
    async with httpx.AsyncClient() as client:
        result = await client.get(url)
    return {"result": result}

ツール統合の問題

15. 対応する ToolMessage のないツール呼び出し

# BAD - tool_calls がある AI メッセージだが、ツール実行がない
messages = [
    HumanMessage(content="search for X"),
    AIMessage(content="", tool_calls=[{"id": "1", "name": "search", ...}])
    # ToolMessage が不足！次の LLM 呼び出しは失敗します
]

# GOOD - 常に tool_calls を 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 = ChatOpenAI().bind_tools([interrupt_tool, other_tool])
# 両方が並列で呼ばれた場合、割り込み動作は未定義

# GOOD - 割り込みの前に並列ツール呼び出しを無効化
model = ChatOpenAI().bind_tools(
    [interrupt_tool, other_tool],
    parallel_tool_calls=False
)

チェックポイントの問題

17. 本番環境での InMemorySaver

# BAD - メモリ内チェックポイント はリスタート時に状態を失う
graph = builder.compile(checkpointer=InMemorySaver())  # テストのみ！

# GOOD - 本番環境では永続ストレージを使用
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(conn_string)
graph = builder.compile(checkpointer=checkpointer)

18. サブグラフチェックポイント の混乱

# BAD - 明示的な False を持つサブグラフは永続化を防ぐ
subgraph = sub_builder.compile(checkpointer=False)

# GOOD - None を使用して親のチェックポイント を継承
subgraph = sub_builder.compile(checkpointer=None)  # 親から継承
# または独立したチェックポイント の場合は True
subgraph = sub_builder.compile(checkpointer=True)

パフォーマンスの問題

19. すべての更新で大きな状態

# BAD - 各ノードで大きなデータを返す
def node(state):
    large_data = fetch_large_data()
    return {"large_field": large_data}  # 各ステップでチェックポイント！

# GOOD - 参照を使用またはストアに保存
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 - 無限ループから保護なし
def router(state):
    return "agent"  # 常にループ！

# GOOD - 残りステップを確認するか 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. チェックポイント を使用する場合、config で Thread ID が提供される
7. ツール呼び出しが ToolMessage と組み合わせられている
8. 非同期ノードが非同期操作を使用
9. 本番環境が永続チェックポイント を使用
10. ループに対して再帰制限が考慮されている