【状態管理型AI】LangGraph超入門！Pythonで複雑なループや分岐を持つ自律型マルチエージェントを構築する実践ガイド

LLM（大規模言語モデル）を活用したAIエージェント開発において、単一のプロンプトや単純なシーケンス（一本道）の処理から、「ユーザーの意図に応じて自律的に推論・処理を繰り返し、結果をフィードバックするループ構造」へとパラダイムがシフトしています。

従来のLangChainや単純なパイプライン構成では、複雑なループや条件分岐、複数のAIエージェント間の協調動作を安全かつ柔軟に記述することが困難でした。この課題をクリアするために登場したのが、「LangGraph」です。

本記事では、LangGraphの基本的な仕組みから、状態（State）を用いたプログラムの設計方法、そして実際に動作するマルチエージェント（リサーチ＆レビュー）の構築手順までを、手を動かしながら学べる形で徹底解説します。

---

LangGraphとは？ 状態管理型エージェントの革新性

LangGraphは、LangChainファミリーの一員として開発されている、グラフ構造（Graph）に基づいたAIエージェント開発フレームワークです。

最大の特徴は、「State（状態）」を中心に据えたアーキテクチャにあります。従来のLangChainが「入力から出力への一方向の流れ」を得意としていたのに対し、LangGraphは「処理のノード（Node）間で共有される状態を更新しつつ、条件分岐（Edge）によって処理をループさせる」という制御構造を極めて直感的に実装できます。

なぜLangGraphが必要なのか？

1. ループ構造のネイティブサポート: LLMが出力した結果をセルフチェックし、NGであれば再度プロンプトを修正して再実行する、といった循環型のフロー（人間の推論ステップに近い動き）をシンプルに記述できます。
2. 堅牢な状態（State）管理: エージェントが実行された履歴やコンテキスト、取得したツール実行結果などを一元管理できるため、複雑なマルチエージェント間でもデータの整合性を保てます。
3. 人間による介入（Human-in-the-loop）: 処理の途中で一時停止し、人間の承認を得てから再開するような、実務で必須となるワークフローを容易に統合できます。

オープンソースとして開発が進められており、ソースコードは LangGraph GitHub リポジトリ で公開されています。

類似のエージェントフレームワークとしては、型安全性を重視した Pydantic AIの実践ガイド や、Hugging Face製の軽量フレームワークである smolagentsの入門記事 も存在します。開発するシステムの特性や要件に合わせて選択すると良いでしょう。

---

LangGraphを構成する3つの重要コンセプト

LangGraphを理解する上で、以下の3つの概念だけは確実に押さえておく必要があります。

コンセプト	役割
State（状態）	グラフ全体で共有されるデータ構造。各ノードはこのStateを読み込み、書き換える（更新する）ことで協調動作する。
Nodes（ノード）	具体的な「処理」を実行する関数やLLMの呼び出し処理。Stateを受け取り、更新されたStateを返却する。
Edges（エッジ）	ノードとノードを繋ぐ「制御の経路」。通常の遷移だけでなく、条件によって次の遷移先を決定する「Conditional Edge」もある。

この「State」「Node」「Edge」を定義し、最後にひとつのグラフとしてコンパイルすることで、自律的なエージェントが完成します。

---

実装準備：環境構築とパッケージのインストール

まずはPythonの仮想環境を作成し、必要なライブラリをインストールしましょう。今回はLangGraphに加えて、LLMとしてOpenAIのAPIを使用するため、langchain-openai も併せてインストールします。

pip install langgraph langchain-openai

次に、OpenAIのAPIキーを環境変数に設定します。

export OPENAI_API_KEY="your-openai-api-key-here"

これで準備は完了です。さっそく、実践的なエージェント構築に移りましょう。

---

【実践】リサーチ＆レビューを行うマルチエージェントの構築

今回は、実務でもよくある「ユーザーが指定したテーマについてAIが下書きを作成し、レビュアーAIがチェックを行い、合格基準に達するまで修正ループを繰り返す」という自律型エージェントを構築します。

1. State（状態）の定義

まずはグラフ内で引き回す状態を定義します。Pythonの TypedDict を用いて、どのような値（キー）を管理するかを決めます。

from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    # ユーザーからの命令やテーマ
    task: str
    # 下書きライターの出力
    draft: str
    # レビュアーのフィードバック
    feedback: str
    # レビューの合否判定 (True / False)
    approved: bool
    # 処理のステップ数（無限ループ防止用）
    step_count: int

2. Nodes（ノード）の作成

次に、具体的な処理を行うノードを定義します。今回は「ライター（Writer）」と「レビュアー（Reviewer）」の2つの役割を定義します。

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

# LLMの初期化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# ノード1: ライター（下書きを作成、または修正する）
def writer_node(state: AgentState) -> dict:
    print(f"--- [Writer] 下書き作成中 (ステップ: {state.get('step_count', 0) + 1}) ---")
    
    task = state["task"]
    draft = state.get("draft", "")
    feedback = state.get("feedback", "")
    
    if not draft:
        # 初回作成時
        prompt = f"お題: {task} に関する簡単な解説文を200文字程度で作成してください。"
    else:
        # フィードバックがある場合（修正）
        prompt = f"元の下書き:\n{draft}\n\nフィードバック:\n{feedback}\n\n指示: 上記のフィードバックに従って下書きを修正・洗練させてください。"
        
    response = llm.invoke([
        SystemMessage(content="あなたはプロのライターです。"),
        HumanMessage(content=prompt)
    ])
    
    current_step = state.get("step_count", 0) + 1
    return {"draft": response.content, "step_count": current_step}

# ノード2: レビュアー（下書きをチェックし、合否判定とフィードバックを行う）
def reviewer_node(state: AgentState) -> dict:
    print("--- [Reviewer] 下書きを評価中 ---")
    
    draft = state["draft"]
    
    prompt = (
        f"以下の文章を厳しくチェックしてください。\n\n"
        f"文章:\n{draft}\n\n"
        "評価基準:\n"
        "1. 具体的で分かりやすいか。\n"
        "2. 誤解を招く表現はないか。\n\n"
        "出力フォーマット:\n"
        "1行目に必ず 'APPROVED' または 'REJECTED' とだけ書いてください。\n"
        "2行目以降に、具体的なフィードバックや修正点を記述してください。"
    )
    
    response = llm.invoke([
        SystemMessage(content="あなたは編集部のチーフエディターです。厳格にチェックしてください。"),
        HumanMessage(content=prompt)
    ])
    
    result_text = response.content.strip()
    lines = result_text.split("\n")
    status = lines[0].upper()
    feedback = "\n".join(lines[1:]) if len(lines) > 1 else ""
    
    approved = "APPROVED" in status
    
    print(f"   [判定]: {'合格' if approved else '不合格'}")
    if not approved:
        print(f"   [指摘]: {feedback}")
        
    return {"feedback": feedback, "approved": approved}

3. Edges（エッジ）と遷移ロジックの定義

レビュアーの判定結果（approved）をもとに、処理を終了するか、それともライターに戻して再度修正させるかを判断する「条件付きエッジ（Conditional Edge）」を定義します。

def should_continue(state: AgentState) -> str:
    # 無限ループを避けるため、最大ステップ数を設定（ここでは3回）
    if state.get("step_count", 0) >= 3:
        print("--- [System] 最大ステップ数に達したため、承認をスキップして終了します ---")
        return "end"
        
    if state["approved"]:
        return "end"
    else:
        return "re_write"

4. グラフの構築とコンパイル

定義したノードとエッジを組み合わせ、実行可能なグラフを定義します。LangGraphには StateGraph というクラスが用意されています。

from langgraph.graph import StateGraph, END

# 1. 状態スキーマを渡してグラフを初期化
workflow = StateGraph(AgentState)

# 2. ノードの登録
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node)

# 3. エントリーポイント（開始ノード）の設定
workflow.set_entry_point("writer")

# 4. 通常のエッジ（writer実行後は必ずreviewerを実行）
workflow.add_edge("writer", "reviewer")

# 5. 条件付きエッジの登録
workflow.add_conditional_edges(
    "reviewer",
    should_continue,
    {
        "re_write": "writer",
        "end": END
    }
)

# 6. コンパイルして実行可能なアプリケーション（Runnable）に変換
app = workflow.compile()

5. 実行する

作成したグラフを実行します。初期状態として task と初期の step_count をインプットとして渡します。

inputs = {
    "task": "「LangGraph」というAIエージェント開発用のライブラリを初心者向けに解説する",
    "step_count": 0
}

# グラフの実行
config = {"recursion_limit": 20}  # LangGraph全体の最大再帰制限
final_state = app.invoke(inputs, config)

print("\n=== 最終成果物 ===")
print(final_state["draft"])

---

よくあるエラーとトラブルシューティング

LangGraphを使用した開発では、その特殊な状態管理システムが原因で、開発初期にいくつか特有のエラーに直面することがあります。以下に、代表的なエラーとその原因・対策をまとめました。

1. InvalidUpdateError (Stateの更新エラー)

• 原因: ノードが TypedDict で定義された状態（State）に対して、無効な更新を行った場合に発生します。例えば、Stateのキーに対して、許可されていない型（リストへの単純な文字列の追加など）を代入しようとした場合です。
• 解決方法: 複数のノードが同じリスト型のキーに対して要素を追加していく場合は、以下のように Annotated と operator.add を組み合わせて「Reducer（追加処理）」を指定してください。

  # リストを上書きせず、追記していくためのReducer設定
  class AgentState(TypedDict):
      messages: Annotated[list, operator.add]
  

2. 無限ループによるAPIトークンの急激な消費

• 原因: 条件付きエッジ（Conditional Edge）の評価ロジックにバグがあり、終了ノード（END）に到達するルートが遮断されたり、終了条件が永遠に満たされない場合に発生します。
• 解決方法:
  • AgentState 内に必ず step_count のようなカウンターを保持させ、エッジ判定関数（should_continue 等）で上限（例: 3回〜5回）に達したら強制的に END に遷移させるロジックを挟む。
  • app.invoke() を呼び出す際の config オブジェクトに、安全策として `{