AIエージェントを「とりあえず動かす」段階を超えると、必ずぶつかるのがループ・分岐・人間の承認・障害からの復旧をどう設計するかという問題です。LangGraph入門として本記事では、LLMエージェントをノード・エッジ・状態からなるステートマシンとして組むこのOSSフレームワークを、最小コードから本番運用まで一次ソースで整理します。
そもそもAIエージェントとは何かをまず俯瞰したい方は、姉妹記事の AIエージェントとは?仕組み・種類・代表的OSSフレームワークを初心者向けに解説【2026年版】 を先に読むと、本記事の位置づけがつかみやすくなります。
30秒で理解する
- ・LangGraphは「長時間動くステートフルなエージェント」を構築・運用するための低レベルなオーケストレーション基盤。GitHubスター約34K、最新版は1.2.4(2026年6月2日)。
- ・LangChainが高レベルの抽象、LangGraphがその下の実行ランタイム。LangChainなしでも単体で使える。
- ・状態(State)・ノード・エッジ・条件分岐の4概念でグラフを書く。ループや分岐を「明示的に」記述できるのが最大の特徴。
- ・Checkpointで状態を自動保存し、会話メモリ・障害復旧・タイムトラベル・human-in-the-loopを実現する。
- ・CrewAIは手早いプロトタイプ、LangGraphは制御フローを厳密に支配したい本番向き、TypeScript単独ならMastra、という棲み分け。
LangGraphは「魔法の自律エージェント」を約束するフレームワークではありません。むしろ逆で、エージェントの挙動を状態遷移図として開発者が支配するための道具です。LLMの非決定性を、グラフという決定的な骨格の中に閉じ込める。この発想が、本番投入できるエージェントとデモ止まりのエージェントを分けます。
LangGraphとは(LangChainとの違い、エージェントOS的位置づけ)
LangGraphは公式に「長時間動くステートフルなエージェントを構築・管理・デプロイするための低レベルなオーケストレーションフレームワークおよびランタイム」と説明されています。KlarnaやReplit、Uber、Elastic、J.P. Morganといった企業が本番で採用していると公表されており、デモではなく実運用を前提に設計されている点が特徴です。
よくある混乱が「LangChainとLangGraphはどう違うのか」です。両者は競合ではなく階層が異なります。整理すると次のとおりです。
・LangChain:モデル・ツール・エージェントループの抽象化と、各種サービスへの連携(インテグレーション)を提供する高レベルのフレームワーク
・LangGraph:その下で動く低レベルの実行ランタイム。耐久的な実行(durable execution)・ストリーミング・human-in-the-loop・永続化を担う
・両者の関係:LangChainの新しいエージェント抽象はLangGraphの上に構築されている。ただし公式は「LangGraphを使うのにLangChainは必須ではない」と明言しており、LangGraphは単体でも利用できる
つまりLangGraphは、特定のモデルや上位フレームワークに縛られない「エージェントの実行基盤」として位置づけられます。Microsoftがコンテキスト層として提唱する構想を読み解いた Microsoft IQとは?Build 2026で示されたコンテキスト層の全体像 と並べると、「モデルの上に何を積めばエージェントが本番で動くか」という2026年共通の問いに、ベンダー横断のOSS側から答えているのがLangGraphだと理解できます。
LangGraphが解くのは、素のLLM呼び出しでは扱いにくい4つの難所です。
・ループ:ツールを呼んで結果を見て、また考える、を条件が満たされるまで繰り返す
・分岐:LLMの出力に応じて次に進む先を動的に切り替える
・耐久性:途中でノードが失敗しても、最後に成功したステップから再開する
・介在:危険な操作の前で実行を止め、人間の承認を待ってから続行する
これらを「フレームワークの隠れた挙動」ではなく、開発者が書いたグラフの構造として表現できる。これがLangGraphの設計思想です。
アーキテクチャ(ステートマシン+グラフ)
LangGraphのアーキテクチャは、4つの構成要素に分解できます。
・State(状態):グラフ全体で共有される型付きのデータ。PythonではTypedDict、TypeScriptではAnnotationで定義する
・Node(ノード):状態を受け取り、状態の更新(差分)を返す関数。LLM呼び出しやツール実行を担う
・Edge(エッジ):ノード間の遷移。固定のエッジと、状態に応じて行き先を変える条件付きエッジ(Conditional Edges)がある
・Reducer(リデューサ):状態を更新するときの合成方法。たとえばメッセージ履歴は「上書き」ではなく「追記」したいのでoperator.addを指定する
実行は「スーパーステップ」という単位で進みます。1スーパーステップは、その時点でスケジュールされた全ノードを(並列に動くこともある)一度に実行する1ティックです。各スーパーステップの境界でチェックポイント(状態のスナップショット)が保存され、ここが障害復旧やタイムトラベルの起点になります。
最小のReActループ(LLMが考え→ツールを呼び→結果を見て→また考える)をグラフで表すと、次の形になります。
LLMが次の行動を決める] A -->|tool_calls あり| B[tool_node
ツールを実行する] A -->|tool_calls なし| E([END]) B --> A
ポイントは、llm_callからtool_nodeへ進むか終了するかを条件付きエッジが決め、tool_nodeからは無条件でllm_callに戻る点です。この「戻り」のエッジがループを作ります。素のコードでwhileを書く代わりに、グラフの形そのものが制御フローになっているのが分かります。
最小コード例
まずはインストールから。LangGraphはPython版とTypeScript版(LangGraph.js)の両方が提供されています。
# Python
pip install -U langgraph langchain "langchain[anthropic]"
# TypeScript / JavaScript
npm install @langchain/langgraph @langchain/core @langchain/anthropic
最短で動かすなら、公式のprebuilt create_react_agent を使うのが入門の近道です。状態もエッジも自分で書かずに、ツールを渡すだけでReActエージェントが立ち上がります。
from langgraph.prebuilt import create_react_agent
def get_weather(city: str) -> str:
"""指定した都市の天気を返す。"""
return f"{city} は晴れ、22度です。"
agent = create_react_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "東京の天気は?"}]}
)
print(result["messages"][-1].content)
制御フローを自分で支配したくなったら、StateGraphに降ります。次は同じReActループを、状態・ノード・条件付きエッジを明示的に書いて組んだ例です。
from typing import Literal
from typing_extensions import TypedDict, Annotated
import operator
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:
"""a と b を足す。"""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""a と b を掛ける。"""
return a * b
tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)
# 状態: messages は「追記」したいので operator.add をリデューサに指定
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
llm_calls: int
def llm_call(state: State):
return {
"messages": [model_with_tools.invoke(
[SystemMessage(content="あなたは算術を行うアシスタントです。")]
+ state["messages"]
)],
"llm_calls": state.get("llm_calls", 0) + 1,
}
def tool_node(state: State):
result = []
for call in state["messages"][-1].tool_calls:
obs = tools_by_name[call["name"]].invoke(call["args"])
result.append(ToolMessage(content=obs, tool_call_id=call["id"]))
return {"messages": result}
# 条件付きエッジ: ツール呼び出しが残っていれば tool_node、なければ END
def should_continue(state: State) -> Literal["tool_node", "__end__"]:
return "tool_node" if state["messages"][-1].tool_calls else END
builder = StateGraph(State)
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()
out = agent.invoke({"messages": [HumanMessage(content="3 と 4 を足して。")]})
for m in out["messages"]:
m.pretty_print()
TypeScript(LangGraph.js)でも、ほぼ同じ概念でグラフを書けます。状態の合成方法をスキーマ側で宣言する点が異なります。
import { ChatAnthropic } from "@langchain/anthropic";
import { tool } from "@langchain/core/tools";
import { StateGraph, START, END } from "@langchain/langgraph";
import { SystemMessage, HumanMessage, AIMessage } from "@langchain/core/messages";
import * as z from "zod";
const model = new ChatAnthropic({ model: "claude-sonnet-4-6", temperature: 0 });
const add = tool(({ a, b }) => a + b, {
name: "add",
description: "2つの数を足す",
schema: z.object({ a: z.number(), b: z.number() }),
});
const toolsByName = { [add.name]: add };
const modelWithTools = model.bindTools([add]);
const llmCall = async (state) => ({
messages: [
await modelWithTools.invoke([
new SystemMessage("あなたは算術アシスタントです。"),
...state.messages,
]),
],
});
const toolNode = async (state) => {
const last = state.messages.at(-1);
const result = [];
for (const call of last?.tool_calls ?? []) {
result.push(await toolsByName[call.name].invoke(call));
}
return { messages: result };
};
const shouldContinue = (state) =>
state.messages.at(-1)?.tool_calls?.length ? "toolNode" : END;
主要な機能
LangGraphを「ただのグラフ実行器」から「本番エージェント基盤」に押し上げているのが、次の4機能です。
StateGraph と Reducer:状態を型で定義し、更新の合成方法(リデューサ)を指定できます。メッセージ履歴は追記、カウンタは加算、といった更新ルールを状態スキーマに宣言しておくことで、各ノードは「差分」を返すだけで済みます。これにより並列ノードの結果も衝突なくマージできます。
Conditional Edges(条件付きエッジ):LLMの出力や状態を見て次の行き先を動的に決めます。add_conditional_edgesに判定関数を渡すだけで、分岐・ループ・早期終了を表現できます。ここがエージェントの「自律性」を安全に閉じ込める要です。
Checkpoint(永続化):compile(checkpointer=...)でチェックポインタを差し込むと、各ステップの状態がthread_idごとに自動保存されます。会話メモリ、障害からの再開、過去状態への巻き戻し(タイムトラベル)がこれ一つで実現します。
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver() # 本番は PostgresSaver / ローカルは SqliteSaver
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-123"}}
graph.invoke({"messages": [HumanMessage(content="3 と 4 を足して")]}, config)
# 同じ thread_id で続けると、前回の状態が引き継がれる
graph.invoke({"messages": [HumanMessage(content="その結果に 10 を掛けて")]}, config)
# 現在の状態と履歴を取得(デバッグ・タイムトラベルに使う)
snapshot = graph.get_state(config)
history = list(graph.get_state_history(config))
開発用のInMemorySaver、ローカル用のSqliteSaver、本番用のPostgresSaverが用意されており、いずれもBaseCheckpointSaverという共通インターフェースに従います。スレッドをまたいで情報を共有したい場合は、別途Storeインターフェースで名前空間付きのメモリや意味検索を扱えます。
Human-in-the-loop(人間の介在):interrupt()で実行を止め、Command(resume=...)で再開します。危険な操作の前に人間の承認を挟む、典型的なゲートが数行で書けます。
from langgraph.types import interrupt, Command
def human_approval(state: State):
# ここで実行が一時停止し、状態は thread_id に保存される
decision = interrupt({"question": "この操作を承認しますか?", "draft": state["messages"][-1].content})
return {"approved": decision}
# 停止後、別プロセス・別時刻からでも安全に再開できる
graph.invoke(Command(resume="approve"), config={"configurable": {"thread_id": "user-123"}})
Streaming(ストリーミング):エージェントは応答までに何度もLLMとツールを往復するため、ユーザー体験を保つにはストリーミングが欠かせません。LangGraphはgraph.stream(...)で実行中の状態や出力を逐次受け取れます。stream_modeで粒度を切り替えられ、"values"なら各ステップ後の状態全体、"updates"なら各ノードが返した差分だけ、"messages"ならLLMのトークン単位の出力を流せます。
# 各ノードが返した「差分」だけを逐次受け取る
for chunk in graph.stream(
{"messages": [HumanMessage(content="3 と 4 を足して 10 を掛けて")]},
config={"configurable": {"thread_id": "user-123"}},
stream_mode="updates",
):
print(chunk)
これにより、ツール実行中に「今どのノードを処理しているか」を画面に出したり、最終回答をトークン単位でタイプライター表示したりできます。チェックポイントと組み合わせれば、途中で止めて再開しても続きからストリーミングを再開できます。
実践Tips
入門の次に効く、運用寄りの勘所をまとめます。
・最初はcreate_react_agentで動かし、制御が足りなくなってからStateGraphに降りる。最初から低レベルAPIで書くと挫折しやすい
・状態は小さく保つ。グラフ全体で共有される状態に何でも詰め込むと、ノード間の依存が見えなくなる。ノード固有の作業データはローカルに留める
・リデューサを意識する。メッセージはoperator.addで追記、それ以外のキーはデフォルトで上書き。並列ノードがある場合は特に「合成方法」を先に決める
・再帰制限を設定する。条件付きエッジのバグでループが止まらない事故を防ぐため、config={"recursion_limit": N}で上限を決めておく
・チェックポインタは開発の最初から入れる。後付けすると会話メモリやデバッグの設計をやり直すことになる
・グラフを可視化する。graph.get_graph().draw_mermaid()でMermaid図を出力でき、レビューやドキュメント化に役立つ
CrewAI / AutoGen / Mastra / OpenAI Agents SDK との比較表
「どれを選ぶか」は、制御の厳密さ・言語・モデルの3軸でほぼ決まります。2026年時点の主要フレームワークを整理します。
| フレームワーク | 言語 | 抽象モデル | 制御フロー | 永続化・HITL | 向いている場面 |
|---|---|---|---|---|---|
| LangGraph | Python / TS | ノード・エッジ・状態のステートマシン | 明示的(ループ・分岐・条件を自分で書く) | チェックポイント/interruptで標準対応 | 本番品質、複雑な制御フロー、人間承認 |
| CrewAI | Python | 役割を持つエージェントのチーム+タスク | 宣言的(役割とタスクを記述) | 限定的 | 役割分担が自然な業務、高速プロトタイプ |
| AutoGen / AG2 | Python | エージェント同士の会話 | 会話駆動 | 限定的 | 研究・実験。本番採用は減少傾向 |
| Mastra | TypeScript | エージェント+ワークフロー+メモリ+評価 | ワークフロー記述 | TSスタックで一体提供 | TS単独で完結させたいプロダクト |
| OpenAI Agents SDK | Python | ハンドオフ中心の軽量SDK | ハンドオフ駆動 | 標準的 | OpenAIモデル中心の構成 |
選び方の目安を言葉にすると次のとおりです。
・作業が自然に「役割」に分かれ、午後いっぱいで動くプロトタイプが欲しい → CrewAI
・1つのワークフローにループ・分岐・リトライ・耐久的チェックポイント・人間承認が必要 → LangGraph
・バックエンドがTypeScriptで、エージェント・ツール・メモリ・RAG・評価を1スタックで完結させたい → Mastra
・OpenAIモデル中心で軽量に組みたい → OpenAI Agents SDK
なおAutoGenはMicrosoftのSemantic Kernelと統合後、独立プロジェクトAG2として分岐しました。更新は続いていますが本番採用は減少傾向で、多くのチームがLangGraphやCrewAIへ移行したと報じられています。9種をStar数と実コードで突き合わせた詳細は AIエージェントフレームワーク比較2026|LangGraph・CrewAI・Dify等9種をStar数・実コードで検証 にまとめています。
本番運用(LangSmith観測、デプロイ、スケーリング)
エージェントはループと分岐で挙動が見えづらく、「なぜこの出力になったか」を後から追えないと運用が破綻します。LangGraphの本番運用では、観測と永続化を最初から組み込むのが定石です。
・観測:LangSmithでトレースを取り、どのノードで何回LLMを呼んだか・どこで失敗したかを可視化する。実行パスをグラフで追えるため、ループの暴走やツール選択ミスの原因特定が速い
・永続化:本番ではPostgresSaverでチェックポイントを外部DBに保存する。プロセスが落ちてもthread_idから再開でき、会話メモリも維持される
・デプロイ:チェックポインタを外部ストレージにすればステートレスなワーカーとしてスケールできる。状態はDB側に持たせ、アプリは水平にスケールアウトする設計が基本
・コスト管理:再帰制限とLLM呼び出し回数の監視を初期から入れる。条件付きエッジのバグはそのままAPI課金の暴走になる
LangGraphは約34.5M回/月のPyPIダウンロードを記録しており、エコシステムと運用ノウハウの蓄積という点でも厚みがあります。トークン消費そのものを削る設計は AIエージェントのトークン最適化ガイド2026 を、本番品質へ引き上げる設計原則は 12-Factor Agents完全解説 を併せて参照すると、観測・永続化・コストの3点が一本につながります。
よくある落とし穴
入門者がつまずきやすいポイントを先回りで挙げます。
・状態を上書きしてしまう:messagesにoperator.add(追記)のリデューサを付け忘れると、毎ステップで履歴が消える。「会話が記憶されない」の典型原因
・ループが止まらない:条件付きエッジの終了条件を間違えると無限ループになる。recursion_limitを必ず設定し、開発中はトレースで遷移を確認する
・チェックポインタなしでHITLを使う:interrupt()はチェックポインタが有効でないと再開できない。compile(checkpointer=...)を忘れない
・thread_idの管理漏れ:会話やユーザーごとにthread_idを分けないと、別ユーザーの状態が混ざる。逆に毎回新しいIDを振るとメモリが効かない
・LangChainを「必須」と誤解する:LangGraphは単体で動く。必要な部品だけを選び、上位フレームワークに引きずられない
・状態に巨大データを詰める:状態は毎ステップ保存・合成される。大きなファイルやベクトルは状態ではなく外部ストアに置き、参照だけを持つ
外部入力を扱うエージェントでは、プロンプトインジェクション対策も忘れてはいけません。危険なツールの前に前述のinterrupt()で人間承認を挟む、外部テキストとシステム指示を分離する、といった多層防御が前提です。
参考リンク
・LangGraph 公式ドキュメント(docs.langchain.com):https://docs.langchain.com/oss/python/langgraph/overview
・LangGraph GitHubリポジトリ(langchain-ai/langgraph):https://github.com/langchain-ai/langgraph
・LangGraph クイックスタート:https://docs.langchain.com/oss/python/langgraph/quickstart
・LangGraph 永続化(Persistence)ガイド:https://docs.langchain.com/oss/python/langgraph/persistence