LangGraphとは・使い方入門｜APIキーなしで動かす実行例つき

LangGraph のプログラムは、「状態（State）を持った箱」をノードからノードへ回していく形をしています。エージェントの典型は、モデルを呼ぶノードとツールを実行するノードの間をループする形です。

flowchart TD
    S([START]) --> A[agent ノード<br>モデルを呼ぶ]
    A -->|tool_calls あり| T[tools ノード<br>ツールを実行]
    T -->|結果を追加| A
    A -->|なし| E([END])
    A -.->|保存| C[(チェックポイント)]

よく調べられることの答えを先にまとめます。

確認環境：Ubuntu 24.04.5 LTS 上の Python 3.13.18 で、作業用のフォルダーに venv を作り、langgraph 1.2.18（langgraph-checkpoint 4.3.1・langgraph-prebuilt 1.1.1・langgraph-checkpoint-sqlite 3.1.2）、langchain-core 1.7.1、langchain 1.4.7 を入れて動かしました。OpenAI などの外部の LLM API と LangSmith には一切つないでいません。API キーは設定せず、LangSmith の送信は LANGSMITH_TRACING=false などで無効にし、すべてのスクリプトをネットワークの無い名前空間（unshare -n）の中で実行しました。モデルは、決まった応答を順に返す偽のモデル（langchain-core の GenericFakeChatModel・FakeListChatModel）です。strace で通信のシステムコールを見ると、エージェントの例で connect() は1回も呼ばれていませんでした。LangGraph Studio・LangSmith Deployment・料金は公式ドキュメントと公式サイトで確かめた範囲だけを書いています。図は mermaid 11.16.0 で描画を確かめました（確認日 2026年10月29日）。

LangGraphとは（何ができるライブラリか）

この節では、LangGraph の正体と、ほかのエージェントの作り方と比べて何がうれしいのかが分かります。

LangGraph は、公式ドキュメントの言葉では「長時間動き、状態を持つエージェントを作り、管理し、デプロイするための低水準のオーケストレーション（処理の段取り）のフレームワークと実行環境」です。GitHub の README は「状態を持つエージェントを作るための低水準のオーケストレーション・フレームワーク」と1行で説明し、ライセンスは MIT です。読み方は「ラングラフ」です。

大事なのは「低水準」という点です。LangGraph はプロンプトの書き方やモデルの選び方を決めません。決めるのは「どの順番で何を実行し、途中の状態をどこに持ち、どこで止めて、どこから再開するか」です。ノードの中身はふつうの Python の関数なので、LLM を呼ばない処理も同じグラフに並べられます。README には、Pregel（Google の大規模グラフ処理の仕組み）と Apache Beam から着想を得たと書かれています。

構成要素の早見表

エージェントそのものの考え方（ワークフローとの違い、どこまでを AI に任せるか）は、関連記事「AIエージェントとは｜仕組みとワークフローとの違い・作り方入門」で扱っています。

LangChainとの違いと使い分け（判断表）

この節では、LangChain と LangGraph の役割の違いと、どちらから書き始めるかの判断が分かります。

LangChain はモデル・プロンプト・ツール・検索などの「部品」と、すぐ使える高水準のエージェント（create_agent）を持つフレームワークです。LangGraph は、その下で動く「処理の流れと状態」を受け持つ実行エンジンです。公式ドキュメントは、LangChain のエージェントは LangGraph の上に作られていること、LangGraph は LangChain が無くても使えること、よくある「モデルとツールのループ」なら LangChain のエージェントを勧めることを書いています。

手元でも、langchain.agents.create_agent が返したものの型は LangGraph の CompiledStateGraph でした。つまり LangChain のエージェントは、出来上がった LangGraph のグラフです。

同じ偽のモデルで比べました。LCEL は1本道のつなぎで、型は RunnableSequence です。そこに「短すぎたら書き直す」ループを足したいときは、LCEL のチェーンを LangGraph のノードの中で呼びます。

chain = ChatPromptTemplate.from_template("{topic}を説明して") | llm | StrOutputParser()

def gen(s):      # LCEL のチェーンを、そのままノードの中で呼ぶ
    return {"text": chain.invoke({"topic": s["topic"]}), "tries": s.get("tries", 0) + 1}
def check(s):    # 20文字以上になるか、3回試したら終わり
    return END if len(s["text"]) >= 20 or s["tries"] >= 3 else "gen"

b = StateGraph(S); b.add_node("gen", gen)
b.add_edge(START, "gen"); b.add_conditional_edges("gen", check, ["gen", END])

LCEL: LangGraph は状態つきのグラフでエージェントを組むライブラリです。
LCEL の型: RunnableSequence
LangGraph: {'topic': 'LangGraph', 'text': 'LangGraph は、ノードとエッジと State でエージェントの処理の流れを組むライブラリです。', 'tries': 2}

1回目の「短い説明」は条件を満たさず、2回目で終わっています（tries: 2）。

どちらで書くかの判断表

LangChain の基本（モデルの呼び出し・LCEL・RAG・create_agent）は、関連記事「LangChain入門｜PythonでLLMアプリを作る基本とコピペ集」で扱っています。この記事は、その土台の LangGraph を直接書く方法を受け持ちます。

LangGraphのインストールと、外部APIにつながない準備

この節では、LangGraph の入れ方、入る依存パッケージ、API キーなしで試すための設定が分かります。

インストールは pip install -U langgraph（uv なら uv add langgraph）です。PyPI によると 2026年10月29日時点の最新は 1.2.18（2026年10月26日公開）で、Python 3.10 以上が必要です。JavaScript／TypeScript 版もあり、npm では @langchain/langgraph と @langchain/core を入れます（公式ドキュメント）。

python3 -m venv .venv && . .venv/bin/activate
pip install -U langgraph                       # 本体
pip install -U langgraph-checkpoint-sqlite     # SQLite に記憶を残すなら
pip install -U langchain                       # create_agent を使うなら

# この記事の確認では、外部へ何も送らないように先に設定した
export LANGSMITH_TRACING=false LANGCHAIN_TRACING_V2=false
unset OPENAI_API_KEY ANTHROPIC_API_KEY LANGSMITH_API_KEY

入ったパッケージと大きさです。langgraph 本体は小さく、langchain-core とその依存（langsmith・pydantic）が大半を占めます。

API キーなしで動かす：偽のモデル

LangGraph の仕組みを覚えるだけなら、本物の LLM は要りません。langchain-core には、決まった応答を順に返す偽のチャットモデルが入っています。ツール呼び出しの応答も作れるように、bind_tools を受け流す小さな派生クラスを作りました。この記事の例はすべてこれで動かしています。

# fake.py：決まった応答を順に返す偽のチャットモデル（外部の API には一切つながない）
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage

class ScriptedModel(GenericFakeChatModel):
    def bind_tools(self, tools, **kwargs):
        return self          # ツールの定義は受け取るだけ。応答は台本どおり

def scripted(*replies):
    return ScriptedModel(messages=iter(list(replies)))

def call(name, args, id):    # 「このツールを呼びたい」という応答を作る
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": id}])

本物のモデルに替えるときは、各社の連携パッケージ（langchain-openai など）のモデルを渡すだけで、グラフの側は変わりません。テストでも、API の料金をかけずに分岐やループを確かめられます。

LangGraphの使い方：最小のグラフ（State・ノード・エッジ・compile）

この節では、LangGraph の基本の5手順と、ノードが State をどう受け取り、どう返すのかが分かります。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str
    length: int

def upper(state: State) -> dict:
    print("  upper が受け取った:", state)
    return {"text": state["text"].upper()}      # 変えたいキーだけ返す

def count(state: State) -> dict:
    print("  count が受け取った:", state)
    return {"length": len(state["text"])}

builder = StateGraph(State)
builder.add_node("upper", upper)
builder.add_node("count", count)
builder.add_edge(START, "upper")
builder.add_edge("upper", "count")
builder.add_edge("count", END)
graph = builder.compile()

print(type(graph).__name__)
print("結果:", graph.invoke({"text": "hello langgraph"}))

CompiledStateGraph
  upper が受け取った: {'text': 'hello langgraph'}
  count が受け取った: {'text': 'HELLO LANGGRAPH'}
結果: {'text': 'HELLO LANGGRAPH', 'length': 15}

グラフを図にする（draw_mermaid）

graph.get_graph().draw_mermaid() は、グラフを Mermaid 記法の文字列で返します。手元では次の文字列が出ました。これを Mermaid に対応したエディタや Markdown に貼ると図になります。

---
config:
  flowchart:
    curve: linear
---
graph TD;
	__start__([<p>__start__</p>]):::first
	upper(upper)
	count(count)
	__end__([<p>__end__</p>]):::last
	__start__ --> upper;
	upper --> count;
	count --> __end__;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

固定のエッジは実線（-->）、条件付きのエッジは点線（-.->）で出力されます。

PNG にする draw_mermaid_png() は、既定ではグラフの定義を外部のサービス（mermaid.ink）に送って画像を作ります。ネットワークの無い環境で呼ぶと、次のエラーになりました。社外に出したくない処理の流れを図にするときは、draw_mermaid() の文字列を手元で描画するか、エラー文が案内しているとおり手元のブラウザで描く方法（MermaidDrawMethod.PYPPETEER）を選びます。

ValueError : Failed to reach https://mermaid.ink API while trying to render your graph after 1 retries. To resolve this issue:
1. Check your internet connection and try again
2. Try with higher retry settings: `draw_mermaid_png(..., max_retries=5, retry_delay=2.0)`
3. Use the Pyppeteer rendering method which will render your graph locally in a browser: `draw_mermaid_png(..., draw_method=MermaidDrawMethod.PYPPETEER)`

端末の中で見たいときの draw_ascii() は、grandalf を入れていないと ImportError : Install grandalf to draw graphs: `pip install grandalf`. になりました。入れると文字で描けますが、手元のエージェントの例では tools から agent へ戻る線が描かれなかったため、ループの確認には Mermaid のほうが確実です。状態と遷移を図や表で整理する方法は、関連記事「状態遷移図と状態遷移表の書き方｜テスト設計とMermaid・PlantUML」で扱っています。

Stateとリデューサー（上書き・追記・add_messages）

この節では、ノードの返した値が State にどう合わさるのかと、会話のメッセージの扱いが分かります。

何も指定しないキーは「上書き」、Annotated[型, 関数] でリデューサーを付けたキーは「その関数で合成」になります。同じグラフ（a→b の順に、それぞれ ["a"]・["b"] を返す）を、2通りの State で動かしました。

import operator
from typing import Annotated, TypedDict

class Overwrite(TypedDict):
    logs: list                              # リデューサーなし：上書き

class Append(TypedDict):
    logs: Annotated[list, operator.add]     # リストを足し算：追記

上書き   : {'logs': ['b']}
operator.add: {'logs': ['start', 'a', 'b']}

会話のメッセージには、専用のリデューサー add_messages を使います。メッセージを追記するだけでなく、同じ id のメッセージが来たら置き換えます。messages キーだけを持つ State は MessagesState として用意されています。

class Chat(TypedDict):
    messages: Annotated[list, add_messages]

def bot(s): return {"messages": [AIMessage(content="こんにちは", id="ai-1")]}
def fix(s): return {"messages": [AIMessage(content="こんにちは（修正）", id="ai-1")]}   # 同じ id

  HumanMessage  id=aa17c547-2c62-4576-8b1f-465027dbbd98   やあ
  AIMessage     id=ai-1                                   こんにちは（修正）

入力に渡した ("user", "やあ") のタプルは HumanMessage に変換され、id が自動で付きました。AIMessage は2つにならず、後から来た「修正」で置き換わっています。

LangGraphの条件分岐とループ（add_conditional_edges）

この節では、State を見て次のノードを選ぶ書き方と、ループが止まらないときの安全装置が分かります。

条件付きエッジは「State を受け取り、次のノード名を返す関数」です。終わりたいときは END を返します。「書く→採点→80点未満なら書き直す」というループを、決まった点数（40・65・85）を返す偽の採点役で動かしました。

from typing import Literal

SCORES = [40, 65, 85]   # 決まった採点（偽の評価役）

def write(s):
    t = s.get("tries", 0) + 1
    return {"draft": f"下書き v{t}", "tries": t}

def review(s):
    return {"score": SCORES[min(s["tries"], len(SCORES)) - 1]}

def route(s) -> Literal["write", "__end__"]:      # 戻り値の型で行き先を図に出せる
    print(f"  route: {s['draft']} score={s['score']}")
    return END if s["score"] >= 80 else "write"

b = StateGraph(State)
b.add_node("write", write); b.add_node("review", review)
b.add_edge(START, "write"); b.add_edge("write", "review")
b.add_conditional_edges("review", route)
g = b.compile()
print("結果:", g.invoke({"tries": 0}))

  route: 下書き v1 score=40
  route: 下書き v2 score=65
  route: 下書き v3 score=85
結果: {'draft': '下書き v3', 'score': 85, 'tries': 3}

ルーターの戻り値に Literal[...] を書いておくと、draw_mermaid() の図にも review -.-> write と review -.-> __end__ の点線が出ました。書かないと行き先が分からないため、第3引数に ["write", END] のような候補の一覧か、{"戻り値": "ノード名"} の対応表を渡します。

ループが止まらないとき：recursion_limit

採点を常に10点にして、終わらないループを作りました。recursion_limit に6を渡すと、6ステップで止まり GraphRecursionError になります。

GraphRecursionError : Recursion limit of 6 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT

何も指定しないときの上限は、手元の 1.2.18 では 10007 ステップでした。止まらないグラフを実行すると「Recursion limit of 10007 reached」で止まり、ソースコードでは環境変数 LANGGRAPH_DEFAULT_RECURSION_LIMIT で変えられる既定値として書かれていました。公式のエラーのページは具体的な数字を書いておらず、上げたいときは {"recursion_limit": 1000} のように config で渡す例を載せています。1ステップごとに LLM を呼ぶエージェントで1万回近く回ると料金がかさむため、本物のモデルを使うときは自分で小さめの上限を渡し、ループの中に「回数」や「点数」などの終了条件を必ず置きます。

LangGraphでAIエージェントを作る（ToolNode・create_agent）

この節では、モデルとツールを往復するエージェントを、LangGraph で一から組む方法と、create_agent との関係が分かります。

エージェントの基本形は「モデルのノード」と「ツールのノード」の往復です。モデルがツールを呼びたいと答えたら（tool_calls がある）ツールのノードへ進み、結果を持ってモデルへ戻ります。ツールを呼ばない答えが出たら終わります。この分岐は tools_condition として用意されています。

from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from fake import scripted, call

@tool
def get_stock(item: str) -> int:
    """商品の在庫数を返す"""
    return {"ノートPC": 3, "マウス": 0}.get(item, -1)

model = scripted(                                   # 偽のモデルの台本
    call("get_stock", {"item": "ノートPC"}, "c1"),
    call("get_stock", {"item": "マウス"}, "c2"),
    AIMessage(content="ノートPCは3台、マウスは在庫切れです。"),
).bind_tools([get_stock])

def agent(state: MessagesState):
    return {"messages": [model.invoke(state["messages"])]}

b = StateGraph(MessagesState)
b.add_node("agent", agent)
b.add_node("tools", ToolNode([get_stock]))
b.add_edge(START, "agent")
b.add_conditional_edges("agent", tools_condition)   # tool_calls があれば tools、無ければ終了
b.add_edge("tools", "agent")
g = b.compile()

stream_mode="updates" で、どのノードが何を返したかを順に出しました。

[agent] AIMessage: '' [{'name': 'get_stock', 'args': {'item': 'ノートPC'}, 'id': 'c1', 'type': 'tool_call'}]
[tools] ToolMessage: '3'
[agent] AIMessage: '' [{'name': 'get_stock', 'args': {'item': 'マウス'}, 'id': 'c2', 'type': 'tool_call'}]
[tools] ToolMessage: '0'
[agent] AIMessage: 'ノートPCは3台、マウスは在庫切れです。' 

ToolNode は、モデルが出した引数で get_stock を呼び、戻り値を ToolMessage にして State に追加しています。本物のモデルでも、その後の流れは同じです。ツール呼び出しの仕組みそのものは、関連記事「Function Calling（tool use）の仕組みとPython実装」で扱っています。

同じものを create_agent で書くと

from langchain.agents import create_agent
agent = create_agent(model=m, tools=[get_stock], system_prompt="在庫の質問に答える")
out = agent.invoke({"messages": [("user", "ノートPCの在庫は？")]})

CompiledStateGraph
  HumanMessage 'ノートPCの在庫は？'
  AIMessage    ''
  ToolMessage  '3'
  AIMessage    'ノートPCは3台あります。' 

戻ってきたのは LangGraph の CompiledStateGraph で、図にすると model と tools の2ノードが往復する、上の手書きのグラフと同じ形でした。以前の版で使われた langgraph.prebuilt.create_react_agent は、1.2.18 でも動きますが、呼ぶと次の警告が出ました。

警告: LangGraphDeprecatedSinceV10 create_react_agent has been moved to `langchain.agents`. Please update your import to `from langchain.agents import create_agent`. Deprecated in LangGraph V1.0 to be removed in V2.0.

複数のエージェント：司会役が仕事を回す

司会役（supervisor）が「次は誰か」を決め、調査役・執筆役に回す形も、ノードを足すだけで作れます。ノードから Command(goto=..., update=...) を返すと、「State の更新」と「次の行き先」を1回で指定できます。3役とも決まった応答の偽モデルにし、非同期の ainvoke で動かしました。

async def supervisor(s) -> Command[Literal["researcher", "writer", "__end__"]]:
    nxt = (await supervisor_llm.ainvoke(f"次は誰？ これまで: {s['log']}")).content
    return Command(goto=END if nxt == "FINISH" else nxt, update={"log": [f"supervisor:{nxt}"]})

async def researcher(s) -> Command[Literal["supervisor"]]:
    r = (await researcher_llm.ainvoke(s["task"])).content
    return Command(goto="supervisor", update={"log": [r]})

b = StateGraph(S)
b.add_node("supervisor", supervisor); b.add_node("researcher", researcher); b.add_node("writer", writer)
b.add_edge(START, "supervisor")          # 残りのつなぎは Command が決める
out = asyncio.run(b.compile().ainvoke({"task": "LangGraph のライセンスを調べて記事にする", "log": []}))

  supervisor → researcher
  supervisor → writer
  supervisor → FINISH
   supervisor:researcher
   調査メモ：LangGraph 1.2.18、MIT ライセンス
   supervisor:writer
   記事案：LangGraph は MIT ライセンスで公開されています
   supervisor:FINISH

エッジは START→supervisor の1本しか書いていませんが、戻り値の型 Command[Literal[...]] から、図には supervisor から3方向への点線が出ました。司会役・分担・評価役などの型の選び方は、関連記事「AIエージェントの設計パターン一覧｜5つの型の違いと選び方」で扱っています。

チェックポイントで会話を記憶する（MemorySaver・SqliteSaver・Store）

この節では、会話の続きを覚えさせる方法、保存された履歴の見方、過去の地点からのやり直しが分かります。

compile のときにチェックポインターを渡し、実行のたびに {"configurable": {"thread_id": "..."}} を渡すと、その thread_id ごとに State が残ります。公式ドキュメントでは、thread は保存の単位、チェックポイントはその thread の State のスナップショットと説明されています。MemorySaver は InMemorySaver の別名で、手元でも MemorySaver is InMemorySaver は True でした。

def bot(state: MessagesState):
    n = sum(1 for m in state["messages"] if m.type == "human")
    last = state["messages"][-1].content
    return {"messages": [AIMessage(content=f"{n}回目の発言『{last}』を受け取りました（履歴 {len(state['messages'])} 件）")]}

g = build_graph().compile(checkpointer=InMemorySaver())
for tid, text in [("A", "こんにちは"), ("A", "在庫を教えて"), ("B", "はじめまして")]:
    out = g.invoke({"messages": [("user", text)]}, {"configurable": {"thread_id": tid}})

thread=A: 1回目の発言『こんにちは』を受け取りました（履歴 1 件）
thread=A: 2回目の発言『在庫を教えて』を受け取りました（履歴 3 件）
thread=B: 1回目の発言『はじめまして』を受け取りました（履歴 1 件）
get_state: next= () step= 4 件数= 4
履歴:
  step= 4 source=loop   next=() msgs=4 id=1f1d3b6c-29ee…
  step= 3 source=loop   next=('bot',) msgs=3 id=1f1d3b6c-29ec…
  step= 2 source=input  next=('__start__',) msgs=2 id=1f1d3b6c-29eb…
  step= 1 source=loop   next=() msgs=2 id=1f1d3b6c-29e3…
  step= 0 source=loop   next=('bot',) msgs=1 id=1f1d3b6c-29e0…
  step=-1 source=input  next=('__start__',) msgs=0 id=1f1d3b6c-29d5…

thread A の2回目は、1回目の会話（2件）に新しい発言を足した3件を受け取っています。thread B は別の会話として1件から始まりました。get_state_history を見ると、2回の invoke で6個のチェックポイント（入力を受けた時点・ノードの前・ノードの後）が残っていました。next は「次に実行するノード」で、空なら終わった状態です。

プロセスを終えても残す：SqliteSaver

InMemorySaver はメモリーに置くだけなので、プロセスが終わると消えます。公式ドキュメントも本番では永続のチェックポインターを使うよう書いており、PostgreSQL の PostgresSaver（langgraph-checkpoint-postgres）と、開発向けの SqliteSaver を挙げています。SqliteSaver で、別々の3つのプロセスから同じ thread_id に話しかけました。

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
conn = sqlite3.connect("chk.sqlite", check_same_thread=False)
g = build_graph().compile(checkpointer=SqliteSaver(conn))
out = g.invoke({"messages": [("user", sys.argv[1])]}, {"configurable": {"thread_id": "user-42"}})

$ python ex05_memory.py 1つ目
1回目の発言『1つ目』を受け取りました（履歴 1 件）
$ python ex05_memory.py 2つ目
2回目の発言『2つ目』を受け取りました（履歴 3 件）
$ python ex05_memory.py 3つ目
3回目の発言『3つ目』を受け取りました（履歴 5 件）

ファイル（36KB）の中には checkpoints と writes の2つの表ができ、thread_id user-42 のチェックポイントが9行残っていました。checkpoints の列は thread_id・checkpoint_ns・checkpoint_id・parent_checkpoint_id・type・checkpoint・metadata です。

過去の地点からやり直す（タイムトラベル）

保存されたチェックポイントを選び、update_state で値を書き換えてから続きを実行すると、そこから分岐した別の結果を作れます。「計画を立てる→書く」のグラフで、計画が終わった地点に戻り、計画だけを差し替えました。

past = next(s for s in g.get_state_history(cfg) if s.next == ("write",))
forked = g.update_state(past.config, {"plan": "障害報告を5章で書く"})
print("分岐後:", g.invoke(None, forked)["report"])     # 入力に None を渡すと続きから

1回目: 【報告】障害報告を3章で書く（完成）
戻る地点: {'topic': '障害報告', 'plan': '障害報告を3章で書く'} next= ('write',)
分岐後: 【報告】障害報告を5章で書く（完成）

スレッドをまたぐ記憶：Store

チェックポイントは thread の中だけの記憶です。利用者の好みのように会話をまたいで残したいものは Store（公式ドキュメントの言う長期記憶）に置きます。ノードの引数に runtime: Runtime を書くと runtime.store から読み書きできました。

前回の記録: なし
前回の記録: {'value': 'Python'} （別スレッド）

中断と再開（interrupt・Command・Human-in-the-loop）

この節では、人の承認を待つ処理の書き方と、再開したときにノードがどこから動き直すのかが分かります。

ノードの中で interrupt(値) を呼ぶとグラフがそこで止まり、同じ thread_id で Command(resume=返事) を渡すと、その返事が interrupt の戻り値になって続きが動きます。メールの下書きを作り、人が「送ってよい」と答えたら送信（ここではモック）する流れです。

from langgraph.types import interrupt, Command

def approve(s):
    calls["approve"] += 1
    print(f"  approve 開始（{calls['approve']}回目）")
    answer = interrupt({"question": "この文面で送信してよいですか？", "mail": s["mail"]})
    print("  再開時に受け取った値:", answer)
    return {"status": "送信済み（モック）" if answer == "yes" else "差し戻し"}

g = b.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "mail-1"}}
out = g.invoke({"mail": "", "status": "下書き"}, cfg)     # ここで止まる
out = g.invoke(Command(resume="yes"), cfg)               # 返事を渡して再開

  approve 開始（1回目）
1回目の戻り値: {'mail': 'お見積りを送付します。ご確認をお願いします。', 'status': '下書き', '__interrupt__': [Interrupt(value={'question': 'この文面で送信してよいですか？', 'mail': 'お見積りを送付します。ご確認をお願いします。'}, id='327e3087ddb1219a37fa57bbe9cfc989', response_schema=None)]}
next: ('approve',) / interrupts: [{'question': 'この文面で送信してよいですか？', 'mail': 'お見積りを送付します。ご確認をお願いします。'}]
  approve 開始（2回目）
  再開時に受け取った値: yes
再開後: {'mail': 'お見積りを送付します。ご確認をお願いします。', 'status': '送信済み（モック）'}
呼ばれた回数: {'draft': 1, 'approve': 2}

止まったときの戻り値には __interrupt__ が入り、画面に出す質問と文面を取り出せます。再開すると approve ノードは「最初から」もう一度動きました（「approve 開始」が2回）。公式ドキュメントも、再開時はノード全体が最初から実行され、interrupt より前のコードはもう一度動くと書いています。前のノード draft は1回しか動いていません。

チェックポインターを付けずに compile したグラフでも、interrupt で止まって __interrupt__ は返りました。しかし再開しようとすると、次のエラーになりました。承認待ちを作るなら、チェックポインターと thread_id は必須です。

チェックポインターなしで再開: RuntimeError : Cannot use Command(resume=...) without checkpointer

関数で書く Functional API（@entrypoint・@task）でも interrupt は使えます。2つの task の結果を確認待ちにし、Command(resume=True) で {'results': ['こんにちは', 'さようなら'], 'approved': True} が返りました。

ストリーミング（stream_mode の違いを実行結果で）

この節では、途中経過を受け取る stream_mode の種類と、それぞれ何が流れてくるのかが分かります。

graph.stream() は、stream_mode で「何を流すか」を選びます。「検索→回答」の2ノードのグラフを、モードを変えて実行しました。検索のノードは get_stream_writer() で進み具合を3回出しています。

--- stream_mode='values'
   {'q': 'LangGraph'}
   {'q': 'LangGraph', 'a': '見つかった資料: 3件'}
   {'q': 'LangGraph', 'a': '見つかった資料: 3件 → 要約しました'}
--- stream_mode='updates'
   {'search': {'a': '見つかった資料: 3件'}}
   {'answer': {'a': '見つかった資料: 3件 → 要約しました'}}
--- stream_mode='custom'
   {'progress': '検索中 1/3'}
   {'progress': '検索中 2/3'}
   {'progress': '検索中 3/3'}
--- 複数指定 ['updates','custom']
   ('custom', {'progress': '検索中 1/3'})
   ...
   ('updates', {'answer': {'a': '見つかった資料: 3件 → 要約しました'}})
--- stream_mode='messages'
   'LangGraph'  node=chat
   ' '          node=chat
   'は'          node=chat
   ...
   'です'         node=chat

表は公式ドキュメントの7つのモードの説明を、手元の出力で確かめたものです。messages モードでは応答が1語ずつ届き、送り元のノードが langgraph_node で分かりました。

1.1 以降は version="v2" を渡すと、出力の形がそろいます。手元では、どのモードでも {'type': 'updates', 'ns': (), 'data': {...}} のような辞書になり、invoke(..., version="v2") は GraphOutput（.value と .interrupts を持つ）を返しました。

並列実行・Send（map-reduce）・リトライ・サブグラフ

この節では、ノードを同時に動かす書き方、件数が実行時に決まる並列、失敗したときの再試行、グラフの入れ子と速さが分かります。

1つのノードから複数のノードへエッジを出すと、それらは同じステップで並列に動きます。0.5秒待つノード a・b・c を、直列と並列でつないで時間を測りました。

for n in "abc": b.add_edge(START, n)      # START から3本：並列
b.add_edge(["a", "b", "c"], "join")       # 3つ全部が終わってから join

並列のノードが同じキーに書くときは、リデューサー（ここでは operator.add）が必要です。無いとエラーになります（よくあるエラーの節）。

件数が実行時に決まる並列：Send

「文書の数だけ要約ノードを動かす」のように、何本に分けるかが入力で変わるときは、条件付きエッジから Send("ノード名", そのノードに渡す State) の一覧を返します。

def fan_out(s):
    return [Send("summarize", {"doc": d}) for d in s["docs"]]
b.add_conditional_edges(START, fan_out, ["summarize"])

Send 5件: 0.31 秒 ['議事録そ…の要約', '仕様書そ…の要約', '日報その…の要約', '手順書そ…の要約', '障害報告…の要約']

失敗したノードを自動で再試行：RetryPolicy

外部 API の一時的な失敗に備えて、ノードごとに retry_policy を付けられます。2回失敗して3回目で成功する偽の API で確かめました。

b.add_node("flaky", flaky,
           retry_policy=RetryPolicy(max_attempts=3, initial_interval=0.2, jitter=False))

  flaky 1回目 0.00s
  flaky 2回目 0.20s
  flaky 3回目 0.60s
RetryPolicy: {'results': ['成功']}
（max_attempts=2 にすると）
上限に達した: ConnectionError 一時的な接続エラー（模擬）

待ち時間は 0.2秒→0.4秒と倍になり、上限を超えると元の例外（ここでは ConnectionError）がそのまま上がりました。本物の API ではジッター（待ち時間のばらつき）を有効のままにしておくと、一斉の再送を避けられます。

グラフを入れ子にする：サブグラフ

コンパイル済みのグラフは、そのまま別のグラフのノードとして add_node できます。State のキーが同じなら、そのまま受け渡されます。前処理だけを別のグラフにして、親のグラフに入れました。

sub = StateGraph(S); sub.add_node("clean", lambda s: {"text": s["text"].strip()})
sub.add_edge(START, "clean"); sub = sub.compile()

parent = StateGraph(S)
parent.add_node("pre", sub)                     # グラフをノードとして入れる
parent.add_node("shout", lambda s: {"text": s["text"] + "!"})
parent.add_edge(START, "pre"); parent.add_edge("pre", "shout"); parent.add_edge("shout", END)

サブグラフ: {'text': '整形して!'}

get_graph(xray=True) で図にすると、Mermaid の subgraph pre の中に clean が入った形で出力されました。

LangGraph を通すと何ミリ秒かかるか

最初の2ノードのグラフを1,000回実行し、1回あたりの時間を測りました（2コアの仮想マシン、3回のうち代表値）。

LLM の呼び出しは秒の単位なので、数ミリ秒の差はエージェントでは気になりません。LLM を使わない細かい処理を何万回も回す用途には向きません。

LangGraphのよくあるエラーと直し方（原文）

この節では、LangGraph でつまずきやすいエラーの原文と、原因・直し方が分かります。

手元でわざと起こしたエラーの原文です。トラブルシューティングの URL が付くものは、公式のエラーのページに説明があります。

エラーにならずに困るものもありました。

• 1State に無いキーを返しても黙って捨てられる{"y": 1} を返したノードは、エラーも警告も無く結果に y が入らなかった。キーの綴り違いに気づきにくい
• 2条件付きエッジの対応表に無い値は実行時に KeyError{"b": "b"} の表に "c" を返すと KeyError: 'c'。compile では見つからない
• 3compile 前の builder は実行できないAttributeError: 'StateGraph' object has no attribute 'invoke'
• 4draw_mermaid_png は外部に送るネットワークが無いと ValueError: Failed to reach https://mermaid.ink API。社外秘の流れは draw_mermaid の文字列で
• 5interrupt の前の処理は再開で2回動く送信や課金などの処理は interrupt の後ろか、別のノードに置く

LangGraph Studio・LangSmith Deployment（旧 LangGraph Platform）と料金

この節では、グラフを画面で確かめる Studio、本番に置くための LangSmith Deployment、それぞれの料金の考え方が分かります。

この節は、外部のサービスにつなぐ必要があるため手元では動かさず、公式ドキュメントと公式サイトで確かめた範囲だけを書きます（2026年10月29日時点）。

• LangGraph Studio：エージェントの各ステップ（モデルに送ったプロンプト、ツールの呼び出しと結果、途中の State）を画面で確かめ、任意のステップから会話をやり直せる開発用の画面です。pip install "langgraph-cli[inmem]"（Python 3.11 以上）で入れ、langgraph.json にグラフの場所を書いて langgraph dev で起動すると、手元の http://127.0.0.1:2024 にサーバーが立ちます。画面は smith.langchain.com 側にあり、公式ドキュメントは LangSmith のアカウントと API キーを前提条件に挙げています。トレースは LANGSMITH_TRACING=false で切ることができ、その場合「データは手元のサーバーから出ない」と書かれています。
• LangSmith Deployment：長時間動く、状態を持つエージェントを本番で動かすための実行環境です。公式サイトの FAQ によると、LangGraph Platform は 2025年10月に LangSmith Deployment へ名前が変わりました。置き方は、LangChain 社が AWS・GCP 上で管理する Cloud（Plus 以上）、制御部分は LangChain 社・実行部分は自社に置く Hybrid、自社の Kubernetes に全部置く Self-hosted（Enterprise）、制御部分なしで Agent Server を Docker などで動かす Standalone server の4つがあります。
• トレース（LangSmith）：LANGSMITH_TRACING=true と LANGSMITH_API_KEY を設定すると、実行の経路や State の移り変わりが LangSmith に送られます。設定しなければ送られません。自分でホストできる計測の道具としては Langfuse もあり、関連記事「Langfuseとは｜使い方・料金・セルフホストとSDKの送信を実例で解説」で扱っています。

超えた分は従量課金で、デプロイは使った CPU・メモリーの時間で計算されます（例：実行の CPU 1時間あたり 0.0675 LSU、1 LSU＝1ドル）。LangGraph のライブラリ自体は MIT ライセンスのオープンソースで、手元や自社のサーバーで動かすだけなら料金はかかりません。かかるのは、使う LLM の API 料金と、LangSmith のサービスを使う場合の料金です。

ほかのエージェントのフレームワークとして、IBM の解説記事は CrewAI・MetaGPT・AutoGen を挙げています。LangGraph の README には、Klarna・Replit・Elastic などが使っていると書かれています。

よくある質問

Q. LangGraphとは何ですか？ A. AI エージェントの処理の流れを、ノード・エッジ・State のグラフとして組む Python（と JavaScript）のライブラリです。LangChain 社が開発し、MIT ライセンスで公開されています。分岐・ループ・記憶・中断と再開を決まった書き方で作れます。

Q. LangGraphは何と読みますか？ A. 「ラングラフ」です。Lang（言語）と Graph（グラフ）を合わせた名前です。

Q. LangGraphは無料で使えますか？ A. ライブラリは無料です（MIT ライセンス）。料金がかかるのは、呼び出す LLM の API と、LangSmith のトレースやデプロイのサービスを使う場合です。

Q. LangChainを使わずにLangGraphだけで使えますか？ A. 使えます。公式ドキュメントも LangChain なしで使えると書いています。pip install langgraph で langchain-core は入りますが、ノードの中で OpenAI などの SDK を直接呼ぶ作りにもできます。この記事の例も、langchain 本体は create_agent の節でしか使っていません。

Q. Python以外でも使えますか？ A. JavaScript／TypeScript 版があります。npm で @langchain/langgraph と @langchain/core を入れます。この記事では Python 版だけを動かしています。

Q. create_react_agentはまだ使えますか？ A. 1.2.18 でも動きますが、呼ぶと LangGraphDeprecatedSinceV10 の警告が出て、langchain.agents の create_agent へ移るよう案内されます。警告には V2.0 で削除予定と書かれています。

Q. MemorySaverとInMemorySaverは違いますか？ A. 同じものです。手元で MemorySaver is InMemorySaver は True でした。どちらもメモリーに置くだけなので、プロセスが終わると消えます。残したいときは SqliteSaver か PostgresSaver を使います。

Q. APIキーなしでLangGraphを試せますか？ A. 試せます。langchain-core の GenericFakeChatModel や FakeListChatModel を使うと、決まった応答を返すモデルとして動きます。この記事の例はすべて、ネットワークの無い環境で動かしました。

Q. GraphRecursionErrorが出たらどうしますか？ A. まずループの終了条件（回数・点数・ツールを呼ばない答え）が正しく END を返すかを確かめます。そのうえで、長い処理なら config の recursion_limit を上げます。上限を上げるだけでは、終わらないループの料金が増えるだけになりがちです。

まとめ

• LangGraph は、エージェントの処理を State・ノード・エッジのグラフで組む MIT ライセンスのライブラリ。LangChain の create_agent も中身は LangGraph のグラフだった。
• 使い方は「State を決める→add_node→add_edge→compile→invoke」。ノードは変えたいキーだけを返し、リデューサーで上書きか追記かが決まる。
• 分岐とループは add_conditional_edges。1.2.18 の既定の recursion_limit は 10007 だったので、本物のモデルでは自分で小さめの上限と終了条件を置く。
• 記憶はチェックポインター＋thread_id、承認待ちは interrupt＋Command(resume)。再開するとノードは最初から動き直すので、interrupt の前に送信などの処理を置かない。
• 偽のモデルを使えば、API キーもネットワークもなしで分岐・ループ・記憶・ストリーミングまで確かめられる。draw_mermaid_png だけは外部に送るので注意する。

参考・出典

確認日はいずれも 2026年10月29日です。

• LangChain Docs「LangGraph overview」：https://docs.langchain.com/oss/python/langgraph/overview
• LangChain Docs「Interrupts」：https://docs.langchain.com/oss/python/langgraph/interrupts
• LangChain Docs「Persistence」：https://docs.langchain.com/oss/python/langgraph/persistence
• LangChain Docs「Streaming」：https://docs.langchain.com/oss/python/langgraph/streaming
• LangChain Docs「GRAPH_RECURSION_LIMIT」：https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT
• LangChain Docs「LangSmith Studio」：https://docs.langchain.com/oss/python/langgraph/studio
• LangChain Docs「Observability」：https://docs.langchain.com/oss/python/langgraph/observability
• LangChain Docs「LangGraph overview（JavaScript）」：https://docs.langchain.com/oss/javascript/langgraph/overview
• LangChain Docs「Deployments」：https://docs.langchain.com/langsmith/deployments
• LangChain「Pricing」：https://www.langchain.com/pricing
• LangChain「LangGraph Platform」（名称変更の FAQ）：https://www.langchain.com/langgraph-platform
• PyPI「langgraph」：https://pypi.org/project/langgraph/
• GitHub「langchain-ai/langgraph」：https://github.com/langchain-ai/langgraph
• IBM「What is LangGraph?」：https://www.ibm.com/think/topics/langgraph

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。