公開2026.09.06更新2026.09.27 AIエージェント・MCP

LangGraphでAIエージェントを実装する手順：状態管理から承認フローまで

LangGraph は、LLMを使った処理を「状態」と「グラフ」で書くためのPythonライブラリです。PyPIのlanggraph パッケージを見ると、2026年9月時点の最新は 1.2.11（2026年8月11日公開）、対応Pythonは3.10以上。GitHubのリポジトリはMITライセンスで、スター数は41.1kです。

使いどころは、条件分岐とループ、そして途中停止を明示的に書きたい場面に限られるものです。この記事では StateGraph の最小構成から、チェックポインタによる状態の保存、interrupt を挟んだ人の承認待ち、Functional API での後付け導入までを、公式ドキュメントの記載を根拠にコード付きで追いかけます。LangChain との関係整理はLangChainとLangGraphの違いで扱っているため、本稿は手を動かす部分に絞ります。

まとめ：LangGraph導入で先に決める5つの設計判断

先に結論を置きます。実装で立ち止まりやすい箇所と、その決め方は次の通りです。

状態のスキーマを先に固める：ノードは状態の部分更新を返すだけの関数です。どのキーを追記（reducer あり）にし、どのキーを上書きにするかを決めてから書き始めると、後段の分岐で迷いません。
ループは条件付きエッジで表現する：Python の while でループを回すと、途中停止も再開もできなくなります。add_conditional_edges でLLM呼び出しとツール実行を往復させる形にします。
チェックポインタは最初から入れる：後から足すと thread_id の設計をやり直すことになります。開発は InMemorySaver、本番は PostgresSaver という置き換え前提で組みます。
人の承認は interrupt で挟む：ただし再開時はノードの先頭から再実行されます。interrupt より前に置いた副作用は、必ず二重実行されても壊れない形にします。
グラフを描く必然性が無いなら使わない：単発のプロンプト実行や、分岐の無い直列処理に StateGraph を持ち込むと、コード量だけが増えます。判断基準は後段の採用条件の章で条件付きに示します。

以下、コードと公式ドキュメントの記載に沿って順に見ていきます。

State・Node・Edgeで処理フローを組み立てるLangGraphの基本構造

LangGraph の登場人物は4つだけです。全体を保持する State、処理の単位である Node、遷移を決める Edge、そしてそれらを束ねてビルドする StateGraph。公式のLangGraph overview は、この仕組みを「手で書いた確定的なロジックと、LLMが決める処理を単一のグラフ内で混在させられる」ものと説明しています。

状態をTypedDictとreducerで定義してノード間で受け渡す

State はグラフ全体で共有される辞書です。TypedDict でキーと型を宣言し、蓄積したいキーには Annotated で reducer を付けます。operator.add を指定したキーは、ノードが返した値が既存のリストへ連結されます。指定しないキーは単純な上書きです。

この違いが後々効いてきます。会話履歴のように積み上げたいものは reducer あり。ループ回数やフラグのように毎回置き換えたいものは reducer なし。両者を取り違えると、履歴が消える、あるいはカウンタがリストになる、といった形で壊れます。

ノード側は状態全体を返す必要がありません。更新したいキーだけを含む辞書を返せば、reducer の規則に沿ってマージされます。戻り値を最小限にしておくと、ノード単体のテストが書きやすくなります。

add_nodeとadd_edgeで処理順を宣言しcompileで実行形にする

グラフの組み立ては宣言的です。StateGraph(State) でビルダーを作り、add_node で関数を登録し、add_edge で遷移を張ります。開始点は START、終了点は END という特殊なノード名で表します。

ビルダーはそのままでは実行できません。compile() を呼んで初めて invoke や stream を持つ実行可能なオブジェクトに変換される仕組みです。この2段階になっているおかげで、compile 時にチェックポインタや割り込み設定を差し替えられます。同じグラフ定義を、開発ではメモリ保存、本番ではDB保存で動かす、という切り替えがここで効きます。

compile 時にはグラフの整合性も検査されます。どこからも到達できないノードや、存在しないノード名へのエッジは、この時点でエラーになります。実行してから気付くのではなく、組み立て時に落ちる設計です。

add_conditional_edgesで分岐とループを表現する制御の書き方

分岐は、状態を受け取って次のノード名を返す関数で書きます。add_conditional_edges("llm_call", should_continue, ["tool_node", END]) のように、判定関数と遷移先の候補を渡す形です。返り値がそのまま次のノードになります。

ループも同じ仕組みで作る構成です。ツール実行ノードから再びLLM呼び出しノードへエッジを張れば、ツール呼び出しが無くなるまで往復し続ける構造になります。これがエージェントの中核をなす制御で、終了条件の設計そのものについてはエージェントループの仕組みと終了条件の設計で詳しく整理しています。

無限ループの防止は自前で用意します。反復回数を状態に持たせ、上限を超えたら END へ倒す判定を判定関数に加えるのが基本形です。LangGraph 側にも再帰上限がありますが、それは最後の安全網であって、業務ロジックとしての打ち切り条件は自分で書きます。

langgraph 1.2系をインストールして最小エージェントを動かす手順

ここからは実際に動かします。公式のQuickstart が示す構成を、算術ツールを呼ぶ最小のエージェントとして組み直したものです。

pip install langgraphで入る1.2系のバージョンと依存の確認

インストールは1行です。GitHubのREADMEも pip install -U langgraph を案内しています。モデル呼び出しには別途プロバイダのパッケージが必要になるため、ここでは Anthropic のモデルを使う構成で入れます。

pip install -U langgraph langchain langchain-anthropic
python -c "import langgraph; print(langgraph.__version__)"
# 1.2.11  （2026年9月時点のPyPI最新）

export ANTHROPIC_API_KEY="your-api-key"

Python は3.10以上が条件です。3.9以下の環境では Annotated と型構文の扱いが変わるため、先にランタイムを上げてください。LangGraph 単体は LangChain に依存せずに使えますが、モデルやツールの抽象化を借りるなら langchain も併せて入れる構成が扱いやすくなります。

ツール呼び出しループを持つエージェント一式の実装コードと実行結果

次のコードは、そのままファイルに貼って実行できる全文です。LLMがツールを呼びたがる限りツール実行へ回し、呼ばなくなったら終了する、という往復を条件付きエッジで表現しています。

import operator
from typing import Literal
from typing_extensions import TypedDict, Annotated

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain.messages import AnyMessage, SystemMessage, ToolMessage, HumanMessage
from langgraph.graph import StateGraph, START, END

model = init_chat_model("claude-sonnet-4-6", temperature=0)

@tool
def add(a: int, b: int) -> int:
    """2つの整数を足す"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """2つの整数を掛ける"""
    return a * b

tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

def llm_call(state: MessagesState):
    return {
        "messages": [model_with_tools.invoke(
            [SystemMessage(content="算術計算を実行してください")] + state["messages"]
        )],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

def tool_node(state: MessagesState):
    result = []
    for call in state["messages"][-1].tool_calls:
        observation = tools_by_name[call["name"]].invoke(call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=call["id"]))
    return {"messages": result}

def should_continue(state: MessagesState) -> Literal["tool_node", "__end__"]:
    return "tool_node" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
builder.add_edge("tool_node", "llm_call")
agent = builder.compile()

result = agent.invoke({"messages": [HumanMessage(content="3と4を足して、その結果を5倍して")]})
for m in result["messages"]:
    m.pretty_print()

実行すると、人間のメッセージ、ツール呼び出しを含むAIメッセージ、ツールの戻り値、最終回答、という順にメッセージが積み上がります。llm_calls の値を見れば、何回モデルを呼んだかもそのまま分かります。

グラフ構造の可視化とstream実行で処理経路を確認する具体手順

組んだグラフが意図通りかを確かめる手段は2つです。1つは構造の可視化で、コンパイル済みオブジェクトから get_graph() を取り、Mermaid記法やPNGとして出力します。ノードとエッジの過不足は、絵にすると一目で分かります。

もう1つは invoke を stream に置き換える方法です。ノードが1つ終わるたびに、そのノード名と更新後の状態が流れてくるため、どこで分岐がおかしくなったかを追跡できます。ステップ単位の粒度で見えるので、判定関数の戻り値が想定と違う場面をすぐ特定できます。

本番運用ではトレースを外部に飛ばす選択肢もあります。overview は LangSmith 連携によるデバッグとトレーシングを機能の1つとして挙げており、ローカルのprint出力から移行する先として案内されています。

チェックポインタとinterruptで実装する永続化と人による承認フロー

グラフを組めた次に来るのが、状態をどこに残すかという話です。公式のPersistenceのページは、チェックポインタがスレッド単位でグラフ状態のスナップショットを永続化する仕組みだと説明しています。

InMemorySaverからPostgresSaverまでチェックポインタの使い分け

ドキュメントに挙がっている保存先は3種類です。用途がはっきり分かれます。

保存先	保存の実体	想定する用途
InMemorySaver	プロセスのメモリ	開発・単体テスト
SqliteSaver	ローカルファイル	単一ホストの検証環境
PostgresSaver	PostgreSQL	本番・非同期対応

InMemorySaver はプロセスが落ちた時点で消えます。検証中は手軽ですが、これを本番に持ち込むと再起動で会話が飛びます。PostgresSaver には制限が明記されていて、thread_id は255文字以下です。ユーザーIDとセッションIDを連結したキーを使う設計なら、この上限を先に確認しておきます。

thread_idを指定して会話状態を保存し途中から再開する手順

チェックポインタは compile 時に渡し、実行時には config で thread_id を指定します。同じ thread_id で2回目を呼ぶと、前回の状態に続けて処理が進みます。

from langgraph.checkpoint.memory import InMemorySaver

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

config = {"configurable": {"thread_id": "user-42"}}
agent.invoke({"messages": [HumanMessage(content="3と4を足して")]}, config)
agent.invoke({"messages": [HumanMessage(content="その結果に10を足して")]}, config)

snapshot = agent.get_state(config)
print(snapshot.values["llm_calls"])
print(snapshot.next)

2回目の呼び出しで「その結果」が通じるのは、履歴が状態として残っているからです。get_state で取れるスナップショットには、現在の値に加えて次に実行されるノードが入ります。処理が途中で止まっているスレッドを一覧して再開する運用は、この値を見て組み立てます。

overview が挙げる「障害をまたいで長時間動き続けるエージェント」も、この保存の上に成り立っています。プロセスが落ちても、状態はチェックポインタ側に残っているためです。

interruptとCommandで承認待ちを挟む実装と再実行の副作用対策

Interruptsのページによれば、人の判断を挟むには3点が必要です。チェックポインタ、config に指定する thread_id、そしてノード内の interrupt() 呼び出し。再開は Command(resume=...) で行い、渡した値が interrupt() の戻り値になります。

from langgraph.types import interrupt, Command

def approval_node(state: MessagesState):
    decision = interrupt({
        "question": "この操作を実行してよいか",
        "calls": [c["name"] for c in state["messages"][-1].tool_calls],
    })
    if decision is True:
        return {"messages": []}
    return {"messages": [ToolMessage(content="実行は却下されました", tool_call_id="stop")]}

config = {"configurable": {"thread_id": "approval-123"}}
state = agent.invoke({"messages": [HumanMessage(content="本番DBの在庫を更新して")]}, config)
print(state["__interrupt__"])

agent.invoke(Command(resume=True), config)

ここで踏みやすい地雷が1つあります。再開時に走るのは「interrupt を呼んだ行の続き」ではなく、そのノードの先頭からの再実行です。ドキュメントもこの挙動を明示したうえで、条件分岐で interrupt をスキップしないこと、while True の中で複数回呼ばないこと、bare な try/except で包まないことを禁止事項として挙げています。

実務で効いてくるのは副作用の扱いです。interrupt より前でメール送信やAPI課金を行うと、承認のたびに二重で走ります。外部への書き込みは interrupt の後ろに置くか、冪等キーを付けて重複を吸収する。この2択で設計します。

Graph APIとFunctional APIの書き分けと既存コードへの組み込み方

ここまでは StateGraph を使う Graph API の話でした。Functional API のページは、既存のPythonコードに最小の変更で永続化・メモリ・人の介入・ストリーミングを足す経路として、デコレータ方式を案内しています。

@entrypointと@taskで既存の関数に永続化を後付けする書き方

書き方はシンプルです。ワークフローの入口に @entrypoint、その中で呼ぶ処理単位に @task を付けます。task は並列実行が可能で、リトライポリシー・タイムアウト・キャッシュの指定にも対応します。

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt

@task
def summarize(text: str) -> str:
    return model.invoke("200字で要約してください: " + text).content

@entrypoint(checkpointer=InMemorySaver())
def workflow(inputs: dict) -> str:
    draft = summarize(inputs["text"]).result()
    approved = interrupt({"draft": draft})
    return draft if approved else "却下されました"

task を呼ぶと即座に結果は返らず、.result() で待ち合わせる仕様です。この形にしておけば、複数の task を先に起動してから順に結果を回収する並列処理へそのまま伸ばせます。entrypoint に checkpointer を渡している点も見どころで、グラフを描かないまま interrupt が使えます。

Graph APIとFunctional APIを選ぶ判断基準と併用時の注意

選び分けの軸は「制御構造を外に出したいか」の一点です。分岐やループの形そのものをレビュー対象にしたい、図として共有したいなら Graph API。既存の関数群がすでにあり、そこに保存と再開だけを足したいなら Functional API を選びます。

観点	Graph API	Functional API
書き方	StateGraphクラス	デコレータ
制御フロー	エッジで宣言	Pythonの制御構文
可視化	グラフとして出力可	コードを読む
向く場面	新規の複雑な分岐	既存コードへの追加

両者は同じ実行基盤を共有しており、同一アプリケーション内での併用も、相互の呼び出しも可能です。ただし混在させると、状態がどこで更新されたかを追う経路が2系統になります。チームで触るコードなら、モジュール単位でどちらかに寄せる運用のほうが事故が減ります。

LangGraphを採用すべき条件と過剰投資になる場面の切り分け方

ここからは判断の話です。フレームワークを入れる決定は、動くかどうかではなく、運用に見合うかで決めます。

LangGraphを採用してよい3条件と単純なチェーンで足りる場面

採用が正解になるのは、次の3つのうち2つ以上に当てはまるときです。1つ目、処理が分岐しループする。2つ目、処理の途中で人の承認や差し戻しが入る。3つ目、1回の処理が長く、途中で落ちた場合に続きから再開する必要がある。

逆に、入力を受けてプロンプトを1回投げて返すだけの機能に StateGraph を持ち込むのは過剰です。状態クラスとノード関数とエッジ定義を書いた結果、素直に書けば20行で済むものが80行に膨らむ計算です。この場合はLangChainのチェーンで足ります。判断の前提はLangChainの仕組みと採用判断で整理しています。

迷いやすいのが「将来ループを足すかもしれない」というケースです。分岐の要件がまだ言語化できていない段階でグラフを組むと、状態のスキーマが二転三転します。要件が固まるまでは直列で書き、分岐が2つ以上になった時点で移すほうが手戻りは小さくなります。

セルフホストとLangGraph Platformの分界点を運用体制で決める

ライブラリ本体はMITライセンスで、自前のサーバーやコンテナ基盤に載せて動かせます。一方、デプロイと運用管理を任せる LangGraph Platform を使う選択肢もあります。LangChain社のプラン表では、2026年9月時点で Developer が1席0ドル・月5,000ベーストレース、Plus が1席39ドル・月10,000ベーストレースに小規模なサーバーレスデプロイ1つが付く構成です。セルフホストとハイブリッドは Enterprise プランのみで提供されます。

分界点は技術ではなく体制で引くのが妥当です。コンテナ基盤とDBの運用担当が社内にいて、トレースも自前の基盤に集約しているなら、セルフホストで足りる条件です。逆に、AI機能の担当が1〜2名で、インフラ運用の当番を回せないなら、月額を払ってPlatform側に寄せたほうが総コストは下がります。デプロイ手順とStudioの使い勝手はLangGraph Platformのデプロイ手順と料金で実測しています。

受託開発の現場でLangGraph導入が破綻する3つの典型パターン

破綻の形はだいたい決まっています。1つ目は、状態に何でも詰め込んだ結果、1つのノードを直すと関係ない機能が壊れるパターン。State は共有のグローバル変数と同じ性質を持つため、キーの持ち主を決めずに増やすと保守できなくなります。

2つ目は、エージェントを増やせば賢くなると考えて、役割ごとにノードを量産するパターンです。呼び出し回数に比例してレイテンシとトークン費用が伸びるだけで、精度は上がりません。分割の判断基準はマルチエージェントの構成パターンと採用判断に整理があります。3つ目は、interrupt を入れたのに承認画面の運用設計が無く、止まったスレッドが放置される形です。

いずれも、フレームワークの機能ではなく設計と運用の側で起きます。エージェントの構成設計から本番運用までを含めて相談したい場合は、AIエージェント開発で受託の範囲と進め方を公開しています。

よくある質問

LangGraph の実装で問い合わせの多い論点を5つ挙げます。

LangGraphはLangChainが無いと使えませんか？

単体利用も可能です。公式のoverviewは、LangChainのコンポーネントは各種サービスとの連携に使うものであり、LangGraph は LangChain 無しでも利用できると明記しています。実際、モデル呼び出しを自前のSDKで書き、グラフの制御だけをLangGraphに任せる構成も成立します。ただしツール定義やメッセージ型の抽象化は LangChain 側が持っているため、ゼロから書き直す手間と天秤にかけて決めるのが現実的です。

チェックポインタを付けないとどうなりますか？

1回の invoke が終わった時点で状態は消えます。会話の続きも、途中からの再開もできません。加えて interrupt による停止と再開はチェックポインタが前提なので、人の承認を挟む処理は動かせなくなります。開発初期は InMemorySaver で始め、本番移行時に PostgresSaver へ差し替える形が扱いやすい進め方です。compile の引数を変えるだけで置き換えられます。

interruptで止めた処理はどこから再開されますか？

interrupt を呼んだ行の続きからではなく、そのノードの先頭から再実行される仕様です。ドキュメントもこの点を明示しており、interrupt の呼び出し順を実行ごとに変えないこと、条件によってスキップしないことを求めています。したがって、interrupt より手前に外部への書き込みを置くと二重実行になります。副作用は interrupt の後ろへ移すか、冪等な形に直してください。

LangGraphの利用に費用はかかりますか？

ライブラリ本体はMITライセンスで無償です。費用が発生するのは、呼び出すLLMのAPI利用料と、LangGraph Platform やトレース基盤を使う場合のプラン料金です。プラン表では Developer が1席0ドルで月5,000ベーストレース、Plus が1席39ドルで月10,000ベーストレースという構成になっています（2026年9月時点）。自前のサーバーで動かし、トレースも自社基盤に送るなら、LangChain社への支払いは発生しません。

CrewAIやAutoGenとはどう使い分けますか？

役割分担を宣言的に書きたいなら CrewAI、処理の遷移を細かく制御したいなら LangGraph という使い分けが判断基準です。LangGraph は状態遷移を自分で書く前提のため記述量は増えますが、分岐・ループ・途中停止を思い通りに設計できます。フレームワーク同士の比較はCrewAIとAutoGen・LangGraphの違いで条件別に整理しています。