LangGraphとは?AIエージェントの実装に最適なフレームワークを図解で解説

「LangGraphという名前は見かけるが、LangChainと何が違うのかよくわからない」「インストールしてみたけど、どこから手をつければいいかわからない」。Pythonは書けるし、OpenAI APIも触ったことがある。でもAIエージェント実装となると、途端に手が止まってしまう。そういう段階の方に向けた記事です。

AIエージェントの基本的な仕組み(ReActループ・Tool-callingの考え方)については「AIエージェントとは?仕組み・生成AIとの違い・活用事例を解説」で詳しく解説しています。この記事では、LangGraph固有の設計思想と、他のフレームワークと差別化される3つの機能(Checkpoint・Streaming・interrupt)を実装コードとともに解説します。

この記事を読み終えると、次のことが理解できます。

  • StateGraph・ノード・エッジという3つの基本概念が何者かを理解できる
  • TavilyのWeb検索ツールを使ったReActエージェントを最初から最後まで実装できる
  • Checkpointで会話を継続させ、Streamingでリアルタイム出力し、interruptで人間が介入するパターンを実装できる

無料相談はこちら: LangGraph・AIエージェント開発に関する疑問や社内実装の相談はリベルクラフトの無料相談フォームからどうぞ。

LangGraphとは?LangChainとの違いを整理する

LangGraphとは、LangChain社が開発したAIエージェント向けのオーケストレーションライブラリです。グラフ構造(ノードとエッジ)を使って、LLMの呼び出し・ツールの実行・条件分岐・ループといった複雑な処理フローを明示的に制御できます。

2024年末以降、LangChain社はLangGraphを「エージェントのオーケストレーション層」として公式に位置づけ、複雑なエージェントシステムを作る際の標準的なフレームワークとしています。2026年5月時点の最新バージョンは 1.2.0(2026年5月11日リリース)です。

LangChainとLangGraphの役割分担

混乱しやすいポイントなので整理します。

ライブラリ 役割 主な機能
LangChain LLMアプリの部品箱 プロンプト管理・ツール統合・LLMラッパー
LangGraph フロー制御のオーケストレーター グラフ構造・状態管理・ループ・条件分岐

関係としては「LangChainが用意した部品をLangGraphが組み上げる」というイメージです。ただし「LangChainなしではLangGraphを使えない」というわけではありません。LangGraphはLangChainのコンポーネントに依存せず単体でも動きます(OpenAIのSDKを直接使った実装も可能です)。「LangGraphはLangChainの置き換え」という誤解がありますが、補完関係にあります。

LangChainは部品箱・LangGraphはオーケストレーターという役割分担を示す図

なぜLangGraphがAIエージェント実装に選ばれるか

AIエージェント実装に使えるフレームワークはいくつかあります(OpenAI Agents SDK・AutoGen・CrewAIなど)。LangGraphが選ばれる理由は、次の3点に絞られます。

  1. ループと分岐を明示的に制御できる:「LLMがToolを呼んだら→Tool実行→結果をLLMに戻す→次の判断」というループを、グラフのエッジとして視覚的に定義できる
  2. Checkpointによる状態永続化:会話の途中状態をDBに保存し、中断・再開・状態の巻き戻しができる(他フレームワークにない強み)
  3. Streaming・Human-in-the-loopが標準サポート:トークン単位のリアルタイム出力と、エージェントの動作を人間が途中で確認・修正する仕組みが組み込まれている

判断基準をひとことで言うと:「シンプルな単発エージェントならOpenAI Agents SDKで十分。複雑な状態管理・複数エージェント連携・本番運用が必要ならLangGraphを選ぶ」。

LangGraphの基本概念:StateGraph・ノード・エッジ

コードを書く前に、LangGraphを構成する4つのコンポーネントを理解しておきます。この4つが分かれば、実装コードがすんなり読めるようになります。

LangGraphのStateGraph・ノード・エッジの4コンポーネント関係図
[グラフの構成要素]

State(状態)   ←  グラフ全体を流れるデータ構造
  │
  ↓
Node(ノード)  ←  処理を実行するPython関数
  │
  ↓
Edge(エッジ)  ←  ノード間の遷移ルール(条件分岐も可能)
  │
  ↓
StateGraph      ←  State + Node + Edge をまとめてコンパイルするクラス

State(状態):グラフ全体を流れるデータ構造

StateはLangGraphの「血液」のような存在です。グラフに入力されて、すべてのノードを通り抜けながら更新されていくデータ構造です。TypedDictまたはPydantic BaseModelで定義します。

最小構成の例を見てみます。

# State定義:add_messages の import 元に注意
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages  # langgraph.graph.message から import

class AgentState(TypedDict):
    # messages フィールドは「追記(append)方式」で結合される
    messages: Annotated[list, add_messages]

Annotated[list, add_messages] の部分がReducerと呼ばれる仕組みです。「複数のノードが同じフィールドに書き込むとき、どう結合するかを決めるルール」と理解してください。add_messages を指定すると、各ノードが返すメッセージが上書きではなく追記されます。

ハマりポイント: Annotated[list, add_messages] を書き忘れると、複数ノードが同時更新するたびに前のメッセージが上書きされ、会話履歴が消えます。症状として「エージェントが毎回『初めまして』から始まる」「Toolの実行結果が消える」ことがあったら、まずここを疑ってください。

Node(ノード):処理を実行するPython関数

ノードは「Stateを受け取り、処理して、更新したいStateの一部を辞書で返す」だけの普通のPython関数です。LLM呼び出し・Tool実行・ルーティング判定など、任意の処理を書けます。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def agent_node(state: AgentState) -> dict:
    """LLMを呼び出して応答をstateに返すノード"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # 変更したいフィールドだけを辞書で返す

ポイントはStateの全フィールドを返す必要はないという点です。変更したいフィールドだけを辞書で返せばOKです。

Edge(エッジ):ノード間の遷移を定義する

エッジには2種類あります。

  • 通常エッジ(add_edge:常に同じノードへ遷移する
  • 条件付きエッジ(add_conditional_edges:Stateの中身を見て遷移先を決める関数を渡す

AIエージェントの「ReActループ」は条件付きエッジで実現されます。「LLMがToolを呼びたいと言ったら→toolノードへ。そうでなければ→終了」というロジックです。

from langgraph.graph import END

def should_continue(state: AgentState) -> str:
    """最後のメッセージにtool_callsがあればtools へ、なければ終了"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 条件付きエッジの登録
graph.add_conditional_edges("agent", should_continue)

STARTEND はLangGraphが用意している特殊ノードで、グラフの開始点と終了点を表します。

StateGraph:グラフの組み立てとコンパイル

StateGraph はノードとエッジを組み合わせてグラフを「設計図」として定義し、compile() で実行可能な CompiledStateGraph に変換するクラスです。

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

app = graph.compile()  # ここで実行可能オブジェクトに変換

compile() が完了するまでは実行できません。ノード・エッジの定義ミスはここでエラーとして検出されます。コンパイル後は invokestreamainvokeastream の4メソッドが使えます。

LangGraphでAIエージェントを実装するチュートリアル

概念を理解したので、実際に動くコードを書いていきます。題材はTavily Search APIを使ったWebエージェントです。LangGraph公式チュートリアルでも採用されており、Tavily APIには無料枠(月1,000リクエスト)があるため試しやすい構成です。

前提: Python 3.10以上(LangGraph 1.x はPython 3.9以下非対応)

Step 1:インストールと環境設定

必要なパッケージをインストールします。

# 基本パッケージのインストール
pip install -U langgraph langchain-openai langchain-community tavily-python python-dotenv

-U フラグで最新版に更新されます。Checkpointer(後述)を使う場合は追加インストールが必要ですが、まずは最小構成で動かします。

.env ファイルにAPIキーを記載します。

# .env ファイル
OPENAI_API_KEY=sk-...
TAVILY_API_KEY=tvly-...
# 環境変数の読み込み(スクリプト冒頭に記載)
import os
from dotenv import load_dotenv

load_dotenv()
# os.environ["OPENAI_API_KEY"] と os.environ["TAVILY_API_KEY"] が自動設定される

Step 2:StateとToolの定義

グラフに流れるデータ構造(State)と、エージェントが使うツールを定義します。ここが設計の出発点です。

from typing import Annotated
from typing_extensions import TypedDict
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph.message import add_messages

# ─── State定義 ───────────────────────────────────────────────────
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # Annotated[list, add_messages] にすることで、各ノードが返す
    # メッセージが上書きではなく追記される

# ─── ツール定義 ──────────────────────────────────────────────────
search_tool = TavilySearchResults(max_results=3)  # 最大3件の検索結果を取得
tools = [search_tool]

TavilySearchResults はWeb検索を行うツールです。max_results=3 で取得件数を絞ることで、LLMへの入力トークン数を抑えられます。

Step 3:ノードの実装(agentノードとtoolノード)

2つのノードを実装します。tool_node はLangGraphの prebuilt として提供されているため、自前実装は不要です。

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode  # langgraph.prebuilt から import

# ─── LLMにツールをバインド ────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)  # LLMにツールリストを認識させる

# ─── agentノード:LLMを呼び出す ───────────────────────────────────
def agent_node(state: AgentState) -> dict:
    """
    state["messages"] を受け取り、LLMに渡して応答を返す。
    LLMがツールを呼びたい場合は tool_calls を含む AIMessage が返ってくる。
    """
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# ─── toolノード:Toolを実際に実行する(prebuilt)─────────────────
tool_node = ToolNode(tools)
# ToolNode は state["messages"] の末尾にある tool_calls を自動検出して実行する
# 自前でレスポンスをパースする処理は不要

ToolNode の便利な点は、LLMが返した tool_calls を自動的に検出して実行してくれることです。ツールの実行結果は ToolMessage としてStateに追記されます。

Step 4:グラフの組み立て・コンパイル・実行

StateGraphにノードとエッジを登録し、コンパイルして実際に動かします。

from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage

# ─── 条件分岐:Toolを呼ぶか終了するか ────────────────────────────
def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if last.tool_calls:      # LLMがToolを呼びたいと判断した場合
        return "tools"
    return END               # Tool不要 → 終了

# ─── グラフの組み立て ─────────────────────────────────────────────
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")              # 開始 → agent
graph.add_conditional_edges("agent", should_continue)  # agent → tools or END
graph.add_edge("tools", "agent")           # tools の後は必ず agent に戻る

app = graph.compile()

# ─── 実行 ─────────────────────────────────────────────────────────
result = app.invoke({
    "messages": [HumanMessage(content="LangGraph 1.2.0 の主な新機能を調べて要約して")]
})
print(result["messages"][-1].content)

グラフの流れをテキストで図示すると次のようになります。

START
  │
  ▼
[agent] ─── tool_calls あり ──→ [tools]
  ▲                                │
  └────────────────────────────────┘
  │
  └─── tool_calls なし ──→ END

このReActループが止まるのは「LLMがToolを呼ばなくなった(回答が出た)」タイミングです。Toolを何度呼ぶかはLLMが自律的に判断します。

create_react_agentで最小構成を素早く作る

LangGraphには、StateGraph・ノード・エッジを手組みしなくても、よく使われるReActエージェントのパターンを1行で作れる create_react_agent というショートカット関数があります。

from langgraph.prebuilt import create_react_agent

# ツールを渡すだけで動くReActエージェントを生成
app_quick = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[search_tool]
)

result = app_quick.invoke({
    "messages": [HumanMessage(content="東京の今日の天気を調べて教えて")]
})
print(result["messages"][-1].content)

create_react_agent が内部でStateGraph・ToolNode・条件付きエッジを自動生成します。プロトタイプや動作確認にはこちらが便利です。一方、Checkpointのカスタマイズ・複雑な条件分岐・マルチエージェント構成が必要になったら、Step 1〜4で示した手組みのStateGraphに移行します。

Craft College Advanced では、LangGraphを含むAIエージェント実装をハンズオン形式で学べます。「授業ついていけるか不安」という方はまず無料相談からどうぞ。

LangGraph固有の重要機能:Checkpoint・Streaming・interrupt

ここからが本記事の核心です。LangGraphが他のフレームワークと差別化される3つの機能を実装コードとともに解説します。この3機能を使いこなせることが、エージェントを「本番で使える状態」に引き上げます。

Checkpointで会話履歴と状態を永続化する

Checkpointerを使うと、グラフの各ステップの状態がスナップショットとして保存されます。これによって以下の3つが可能になります。

  • 会話の継続(マルチターン対話)thread_id を使って同一会話を複数ターンにわたって継続できる
  • 中断・再開:アプリが途中でクラッシュしても、再起動後に同じ状態から再開できる
  • 状態の巻き戻し(タイムトラベル):過去の特定のステップに戻って実行し直せる

LangGraphが提供するCheckpointerは3種類あります。

Checkpointer パッケージ 適用場面 永続性
MemorySaver langgraph組み込み テスト・プロトタイプ なし(プロセス終了で消える)
SqliteSaver langgraph-checkpoint-sqlite 開発・シングルサーバー ファイル(SQLite DB)
PostgresSaver langgraph-checkpoint-postgres 本番・分散環境 PostgreSQL(高可用性)
MemorySaver・SqliteSaver・PostgresSaverの用途別使い分けを示す比較図

まずは MemorySaver を使って動作を確認します。

from langgraph.checkpoint.memory import MemorySaver  # langgraph.checkpoint.memory から import
from langchain_core.messages import HumanMessage

# ─── Checkpointerを使ってコンパイル ─────────────────────────────
memory = MemorySaver()
app_with_memory = graph.compile(checkpointer=memory)

# ─── thread_id でセッションを識別 ────────────────────────────────
config = {"configurable": {"thread_id": "user-001"}}
# thread_id が同じなら同一スレッド扱い(会話が引き継がれる)
# 別ユーザーには別の thread_id を割り当てる

# ─── 1回目:最初の質問 ───────────────────────────────────────────
app_with_memory.invoke(
    {"messages": [HumanMessage(content="LangGraphとは何ですか?")]},
    config=config
)

# ─── 2回目:前の会話履歴が自動的に引き継がれる ─────────────────
result = app_with_memory.invoke(
    {"messages": [HumanMessage(content="さっきの回答をもっと詳しく教えて")]},
    config=config
)
print(result["messages"][-1].content)

# ─── 現在の状態を確認 ────────────────────────────────────────────
snapshot = app_with_memory.get_state(config)
print(snapshot.next)  # 次に実行されるノード(空のタプルなら完了)

Checkpointerをつけると「さっきの回答」「それについてもっと詳しく」という自然な会話が成立します。thread_id がセッションの識別子です。本番ではユーザーIDやセッションIDを文字列として渡すと管理しやすくなります。

本番環境でSqliteSaverを使う場合:

pip install langgraph-checkpoint-sqlite  # 別途インストールが必要
from langgraph.checkpoint.sqlite import SqliteSaver  # langgraph.checkpoint.sqlite から import

with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer:
    app_sqlite = graph.compile(checkpointer=checkpointer)
    # 以降の使い方は MemorySaver と同じ

MemorySaver はプロセス終了で消えるため、本番利用には SqliteSaver(シングルサーバー)または PostgresSaver(分散環境)を使うことを公式が推奨しています。

Streamingでリアルタイム出力を実装する

LangGraphは stream() / astream() によってグラフの各ステップの出力をリアルタイムに受け取れます。チャットUIやAPIレスポンスの体験向上に直結します。

stream_mode パラメータで出力の種類を選択します。

stream_mode 出力内容 主な用途
"values" 各ステップ後の全State デバッグ・状態確認
"updates" 各ステップの差分のみ 本番・効率重視
"messages" LLMトークンをリアルタイム チャットUI
"debug" checkpoints + tasks の詳細情報 LangGraph Studio連携
"custom" ノード内で StreamWriter を使って任意データを送出 独自イベント

最もよく使う "messages" モードと "updates" モードの実装例を示します。

config = {"configurable": {"thread_id": "stream-demo"}}

# ─── stream_mode="messages":トークン単位のリアルタイム出力 ────────
# チャットUIで「文字が流れていく」体験を実現するためのモード
for chunk, metadata in app_with_memory.stream(
    {"messages": [HumanMessage(content="AIエージェントとは何か3行で説明して")]},
    config=config,
    stream_mode="messages"
):
    if hasattr(chunk, "content") and chunk.content:
        print(chunk.content, end="", flush=True)  # トークンが出るたびに逐次表示
# ─── stream_mode="updates":ノードごとの更新差分を確認 ────────────
# どのノードが何を返したかを追跡するのに使う
for event in app_with_memory.stream(
    {"messages": [HumanMessage(content="今日の気温を調べて")]},
    config=config,
    stream_mode="updates"
):
    # {"agent": {"messages": [...]}} のような辞書が流れてくる
    print(event)

複数の stream_mode をリスト形式で同時指定することもできます(例:stream_mode=["updates", "messages"])。

interrupt()でHuman-in-the-loopを実装する

LangGraphの最も強力な差別化機能の一つです。interrupt() を使うと、グラフをノード内の任意の地点で一時停止し、人間の確認・修正・承認を受けてから再開できます。

発注・送信・削除など、不可逆操作を伴うエージェントに必須のパターンです。「エージェントが自動実行する前に内容を確認したい」というニーズに応えます。

from langgraph.types import interrupt, Command  # langgraph.types から import

def approval_node(state: AgentState) -> dict:
    """
    エージェントの実行を一時停止し、人間の確認を求める。
    interrupt() を呼ぶとここでグラフが一時停止し、
    GraphInterrupt 例外が発生して呼び出し元に値が返される。
    """
    decision = interrupt({
        "message": "以下のツール呼び出しを実行してよいですか?",
        "pending_action": state["messages"][-1].tool_calls
    })
    if decision.get("approved"):
        return {"messages": state["messages"]}
    else:
        # キャンセルされた場合
        from langchain_core.messages import AIMessage
        return {"messages": [AIMessage(content="操作をキャンセルしました。")]}

より簡潔な方法として、compile()interrupt_before を渡す方法もあります。特定ノードの直前で自動的に一時停止させる設定です。

# ─── interrupt_before:toolsノードの手前に自動ブレークポイントを設置 ────
# checkpointer は interrupt を使う際に必須
app_with_interrupt = graph.compile(
    checkpointer=memory,
    interrupt_before=["tools"]  # tools ノードの直前で必ず一時停止する
)

# ─── 実行(tools の手前で自動停止)────────────────────────────────
config = {"configurable": {"thread_id": "interrupt-demo"}}
app_with_interrupt.invoke(
    {"messages": [HumanMessage(content="Pythonの最新バージョンを調べて教えて")]},
    config=config
)
# → tools ノードの直前で停止。この時点で人間が内容を確認できる

# ─── 承認して再開(Command(resume=...) を使う)────────────────────
app_with_interrupt.invoke(
    Command(resume={"approved": True}),
    config=config
)

注意点: interrupt() を使うには compile()checkpointer を必ず渡す必要があります。Checkpointerがないと状態を保存できないため、interrupt後の再開ができません。

コラム:Graph API vs Functional API

LangGraph 1.x では、StateGraph を使う従来のGraph APIに加えて、より関数型のスタイルで書けるFunctional API@task@entrypoint デコレータ)が導入されています。

スタイル 特徴 向いているケース
Graph API(StateGraph) ノード・エッジを明示的に定義。グラフ構造が視覚的に分かりやすい チュートリアル・グラフの全体像を把握したい場合
Functional API(@task/@entrypoint) Pythonの関数として自然に書ける。既存コードへの組み込みが容易 既存Pythonコードをベースにエージェント化する場合

本記事のチュートリアルはGraph APIで書いています。Functional APIは「既存のPythonコードをなるべく書き換えずにLangGraphに乗せたい」というケースで有効です。公式ドキュメントの Functional API セクションを参照してください。

よくあるエラーと対処法

LangGraph初心者が実装時にハマりやすいパターンを3つに絞って紹介します。

RecursionError:グラフが止まらない

条件付きエッジの分岐関数にバグがあると、ノード間を永遠にループしてLangGraphの再帰制限(recursion_limit)に達します。

対処法:

# compile時 または invoke時に recursion_limit を指定
result = app.invoke(
    {"messages": [HumanMessage(content="...")]},
    config={"recursion_limit": 25}  # デフォルトは25。処理が複雑なら増やす
)

デバッグの手順として:should_continue 関数の return 値が add_conditional_edges に渡したノード名またはENDと一致しているかを最初に確認してください。「tools」を返しているのにノード名を「tool」(複数形なし)で登録していた、というケースがよくあります。

Reducerの設定忘れ

messages フィールドに Annotated[list, add_messages] を付け忘れると、複数ノードが書き込むたびに前のメッセージが上書きされます。

# NG:Reducerなし → 各ノードが書き込むたびに上書きされる
class BadState(TypedDict):
    messages: list

# OK:add_messages を指定 → 追記方式になる
class GoodState(TypedDict):
    messages: Annotated[list, add_messages]

症状: エージェントが毎回「初めまして」から始まる・Tool実行結果がStateから消える → まずReducerを疑ってください。

thread_idを指定せずにCheckpointerを使う

Checkpointerを compile() に渡したのに、invoke() / stream()config を渡し忘れると RuntimeError が発生します。

# NG:config なし → RuntimeError(thread_id が必要)
app_with_memory.invoke({"messages": [HumanMessage(content="...")]})

# OK:thread_id を必ず指定
config = {"configurable": {"thread_id": "session-001"}}
app_with_memory.invoke(
    {"messages": [HumanMessage(content="...")]},
    config=config
)

本番では thread_id にユーザーIDやセッションIDを文字列として渡すと、複数ユーザーの会話を独立して管理できます。

LangGraphからマルチエージェントへ

シングルエージェントを実装できた方の「次の一手」として、LangGraphがマルチエージェントシステムにどう拡張されるかを概念レベルで紹介します。

Supervisorパターンの概要

Supervisor(指示役エージェント)が複数のWorker(専門エージェント)に仕事を割り振るパターンです。

user
  │
  ▼
[Supervisor] ─── Web検索タスク ──→ [Web検索エージェント]
     ▲           コード実行タスク ──→ [コード実行エージェント]
     │           文書要約タスク ──→ [文書要約エージェント]
     │
     └────── 各Workerからの結果を統合してuserに回答

LangGraphでは、Supervisorノードが Command オブジェクトを使って次に実行するWorkerノードを動的に指定します。

from langgraph.types import Command

def supervisor_node(state: AgentState) -> Command:
    """
    SupervisorはLLMでどのWorkerに仕事を振るかを決定する。
    Command(goto=...) で次のノードを明示的に指定できる。
    """
    # LLMの判断に基づいて次のWorkerを決定
    next_worker = decide_next_worker(state)  # "web_search" or "code_exec" or END
    return Command(goto=next_worker, update={"messages": state["messages"]})

Commandlanggraph.types から import します。goto で次のノードを、update で状態の更新内容を同時に指定できる点が特徴です。

LangGraph Studio・LangGraph Cloudについて

LangGraph Studioはグラフの可視化・ステップ実行・デバッグができる開発ツールです。2025年5月にv2がリリースされ、以下の機能が使えます。

  • グラフ構造の可視化:コンパイルされたStateGraphをノード・エッジとして視覚的に確認できる
  • ステップ実行:各ノードの実行をステップごとに追跡し、Stateの変化をリアルタイムで確認できる
  • 本番トレースの再現:本番環境で発生した実行フローをローカルで再現してデバッグできる
  • interrupt の動作確認:Human-in-the-loopの一時停止・再開の動作をGUIで確認できる

コードベースのみでグラフ構造を確認したい場合は以下の方法があります。

# Mermaid形式でグラフ構造をテキスト出力
print(app.get_graph().draw_mermaid())

# PNG画像として出力(IPython・Jupyter環境)
from IPython.display import Image
Image(app.get_graph().draw_mermaid_png())

LangGraph Studio は Mac・Windows・Linuxで動き、langgraph dev コマンドで起動します(別途 langgraph-cli のインストールが必要)。

LangGraph Platform(旧称 LangGraph Cloud、現在は「LangSmith Deployment」と一部機能が統合されつつある)は、エージェントをサーバーレスでホスティングする有料サービスです。Persistence・Streaming・Human-in-the-loopがマネージドで使えます。本番運用を視野に入れる場合は検討の価値があります。

LangGraphとMCPの組み合わせ

MCP(Model Context Protocol)はAnthropicが提唱するオープンプロトコルで、LLMが外部ツール・データソースと標準化されたインターフェースで連携するための仕様です。

LangGraphでは、MCP対応のサーバーをToolとして組み込む langchain-mcp-adapters ライブラリ(LangChain社提供)が利用できます。これにより、MCPサーバーとして公開されているツール群(ファイル操作・データベース・外部API等)をLangGraphのToolNodeに組み込めます。

pip install langchain-mcp-adapters

まとめ

LangGraphの基本概念と主要機能をまとめます。

  • LangGraphの立ち位置:LangChainの上に乗るグラフ型オーケストレーションライブラリ。ループ・条件分岐・状態管理が必要なエージェントに最適
  • 3つの基本コンポーネント:StateGraph(設計図)・ノード(処理関数)・エッジ(遷移ルール)で構成。compile() で実行可能オブジェクトに変換
  • CheckpointMemorySaver(開発用)や SqliteSaver(本番用)でグラフの状態を永続化。thread_id でセッション管理
  • Streamingstream_mode="messages" でトークン単位のリアルタイム出力、"updates" でノードごとの差分確認
  • interrupt()from langgraph.types import interrupt で実装。不可逆操作の前に人間の確認を挟む Human-in-the-loop パターンに必須

実装の進め方としては、まず MemorySaver を使った最小構成(Tool 1個)で動かす → Checkpointとinterruptを加えて本番仕様に育てる という段階的アプローチが現実的です。最初から全機能を詰め込もうとせず、動くものを作ってから機能を追加していく進め方がPoC止まりを防ぐコツです。

LangGraph実装をもっと体系的に学ぶなら

Craft College Advanced では、LangGraphを含むAIエージェント実装を週次のハンズオン形式で学べます。カリキュラムでは Week 7でLangGraph基礎・Week 8でマルチエージェント・Week 4でAgentic RAG を扱います。「記事は読めたけど、実際のコードでつまずいている」「本番環境への実装まで伴走してほしい」という方に向いています。

詳細・受講相談はCraft College Advancedのページからどうぞ。

LangGraph・AIエージェント開発の相談は「リベルクラフト」へ

ここまで、LangGraphの基本概念からチュートリアル実装、Checkpoint・Streaming・interruptの使い方、マルチエージェントへの拡張までを解説してきました。

  • 記事の実装は理解できたが、自社の業務データ・システムと接続する部分でつまずいている
  • PoCから本番運用まで、設計・実装を伴走してほしい

という方は、リベルクラフトへお気軽にご相談ください。

リベルクラフトでは、LangGraphを使ったAIエージェントの設計・実装支援から、社内データを参照できる仕組みの構築、本番運用を見据えたCheckpoint・Human-in-the-loop設計までを一貫して伴走支援しています。

リベルクラフト公式サイト

⇨リベルクラフトへの無料相談はこちら

この記事を書いた人

慶應義塾大学で金融工学を専攻。 卒業後はスタートアップのデータサイエンティストとして、AI・データ活用コンサルティング事業などに従事。 その後、株式会社セブン&アイ・ホールディングスにて、小売・物流事業におけるAI・データ活用の推進に貢献。 株式会社リベルクラフトを設立し、AIやデータサイエンスなどデータ活用領域に関する受託開発・コンサルティングや法人向けトレーニング、教育事業を展開。

関連記事

無料相談