※この記事にはプロモーションが含まれます
Lambda で AWS の日次コストを集計して Web ページに出すところまで作ったものの、表示部分を3か月くらい放置しています。この前ふと「集計結果をLLMに読ませて、異常があればコメントを付けて、場合によっては通知もする…みたいな流れってどう組むのがいいんだろう」と考えました。そこで名前だけはよく見かけていた LangGraph を、Pythonでの基本的な使い方から調べてみました。
自分用のメモに近い内容ですが、同じように「LangGraphって結局なに？」と思っている人の参考になるように、なるべく順番に整理していきます。
この記事でわかること
- LangGraph とは何か、LangChain とどういう関係なのか
- State・Node・Edge を使った最小のグラフの書き方
- 条件分岐と reducer（状態のマージ方法）の考え方
- Claude API をつないでツール呼び出しをさせる構成
- checkpointer で会話を覚えさせる方法と、interrupt で人間の承認を挟む方法
LangGraphとは？ざっくり整理
LangGraph は、LLMを使った処理の流れを「グラフ（ノードとエッジ）」として組むための Python ライブラリです。LangChain を作っているチームが開発していて、エージェントのように「LLMが判断して、ツールを呼んで、結果を見てまた判断して…」というループ構造を書くのが得意です。
普通に Claude API を叩くコードを書くと、だいたいこんな感じになりがちです。
- while ループの中で API を呼ぶ
- tool_use が返ってきたら if で分岐して関数を実行
- 会話履歴のリストを自分で管理
これでも動くんですが、分岐が増えてくると「今どの状態なんだっけ」がどんどん追いにくくなります（自分の書いたコードがまさにそれでした…）。LangGraph はこの「状態」と「流れ」を明示的に定義させることで、見通しを良くしてくれる、という立ち位置のようです。
執筆時点（2026年10月）の PyPI では 1.2.16（2026年10月16日リリース）が公開されていて、その後も 1.2.17 など新しいバージョンが出ています。1.0 系のリリース以降は更新が頻繁に続いているので、ネット上の古い記事（0.x 時代のもの）がまだ多く見つかります。検索して出てきたコードがそのまま動かないときはバージョン差を疑うのが良さそうです。
余談ですが、PyPI のページのタイトルに付いている 🦜🕸️（オウムとクモの巣）の絵文字、LangChain のオウムにグラフのクモの巣を足したんだろうなと思うと、ちょっとかわいいです。
インストールと最小のグラフ
インストールは pip でOKです。Claude を使うので langchain-anthropic も一緒に入れておきます。
pip install -U langgraph langchain-anthropicまずは LLM を使わずに、グラフの骨組みだけを理解するのが近道だと思います。LangGraph の登場人物は基本的にこの3つです。
State・Node・Edge
- State：グラフ全体で共有するデータ。TypedDict で定義するのが一般的
- Node：State を受け取って、更新したい部分だけを dict で返す関数
- Edge：ノード同士をつなぐ矢印。START と END という特別なノードもある
文字列を大文字にして、文字数を数えるだけのグラフを書くとこうなります。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
text: str
length: int
def to_upper(state: State) -> dict:
return {"text": state["text"].upper()}
def count(state: State) -> dict:
return {"length": len(state["text"])}
builder = StateGraph(State)
builder.add_node("to_upper", to_upper)
builder.add_node("count", count)
builder.add_edge(START, "to_upper")
builder.add_edge("to_upper", "count")
builder.add_edge("count", END)
graph = builder.compile()
print(graph.invoke({"text": "hello langgraph"}))
# {'text': 'HELLO LANGGRAPH', 'length': 15}ポイントは、ノードがState 全体を返すのではなく「変えたいキーだけ」を返すところです。to_upper は text しか返していませんが、length はそのまま残ります。最初これを知らなくて、「State 全部コピーして返さなきゃいけないのかな」と思っていました。
あと、compile() を忘れると invoke できないので注意です。builder はあくまで設計図で、compile() して初めて実行できるグラフになる、というイメージです。
ちなみに、組んだグラフは Mermaid 形式で出力できます。
print(graph.get_graph().draw_mermaid())ノードが増えてくると、図で見られるのはかなりありがたいです。
条件分岐とreducerの考え方
グラフっぽさが出てくるのが条件分岐です。add_conditional_edges に「次にどこへ行くか」を返す関数を渡します。
from typing import Literal
def route(state: State) -> Literal["short", "long"]:
return "long" if state["length"] > 10 else "short"
builder.add_conditional_edges(
"count",
route,
{"short": "short_node", "long": "long_node"},
)（上のコードは count から END への固定のエッジを外して、short_node と long_node を別途 add_node している前提です）
LangGraph で一番「なるほど」と思ったのが reducer という仕組みです。デフォルトでは、ノードが返した値は State のキーを上書きします。でも会話履歴みたいに「追記していきたい」データもありますよね。そういうときは Annotated で「どうマージするか」を指定します。
import operator
from typing import Annotated, TypedDict
class State(TypedDict):
messages: Annotated[list, operator.add] # 追記される
status: str # 上書きされる仕組みを理解するために、LangGraph を使わずに「reducer があれば適用、なければ上書き」というロジックを素の Python で真似してみました。もちろん実際の内部実装はもっと複雑なはずですが、考え方としてはこんな感じです。
import operator
from typing import Annotated, TypedDict, get_type_hints
class State(TypedDict):
messages: Annotated[list, operator.add]
status: str
def apply_update(state: dict, update: dict) -> dict:
hints = get_type_hints(State, include_extras=True)
new_state = dict(state)
for key, value in update.items():
reducer = getattr(hints[key], "__metadata__", [None])[0]
if reducer:
new_state[key] = reducer(state.get(key, []), value)
else:
new_state[key] = value # reducerなし＝上書き
return new_state
state = {"messages": ["hi"], "status": "start"}
state = apply_update(state, {"messages": ["hello"], "status": "running"})
state = apply_update(state, {"messages": ["bye"], "status": "done"})
print(state)実行すると messages は [‘hi’, ‘hello’, ‘bye’] と積み上がって、status は最後の ‘done’ だけが残ります。Annotated のメタデータに入れた関数が「古い値と新しい値をどう合体させるか」を決めている、ということですね。
実際の LLM の会話では、operator.add の代わりに LangGraph が用意している add_messages を使うのが定番です。こちらはメッセージIDを見て、同じIDなら置き換え、新しければ追加、という少し賢い動きをするとされています。さらに、messages キーだけを持つ State がよく使われるので、MessagesState という定義済みのクラスも用意されています。
Claude APIをつないでツールを呼ぶ
ここからが本番です。Claude API は普段から使っているので、langchain-anthropic 経由でつないでみます。冒頭のコスト集計の話に寄せて、「指定日のAWSコストを返すツール」をダミーで用意しました。
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
@tool
def get_daily_cost(date: str) -> str:
"""指定した日付のAWS利用料金を返す"""
return f"{date} のコストは 1.23 USD です" # 本当はDynamoDBとかから引く
tools = [get_daily_cost]
llm = ChatAnthropic(model="claude-sonnet-4-5").bind_tools(tools)
def agent(state: MessagesState):
return {"messages": [llm.invoke(state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()
result = graph.invoke(
{"messages": [{"role": "user", "content": "2026-10-01のAWSコストを教えて"}]}
)
print(result["messages"][-1].content)流れとしてはこうなっています。
- agent ノードで Claude を呼ぶ
tools_conditionが「ツール呼び出しがあるか」を見て、あれば tools ノードへ、なければ END へ- tools ノード（
ToolNode）がツールを実行して、結果を messages に追加 - また agent に戻る
自前で while ループを書いていたときの「tool_use なら実行して tool_result を詰めて再送…」という処理が、エッジ2本で表現できるのが気持ちいいです。@tool デコレータの docstring がそのままツールの説明として LLM に渡るので、docstring はちゃんと書いたほうがいいです。ここをサボると、LLM がツールをうまく使ってくれないことがあります。
ちなみに、この「LLM＋ツールのループ」はあまりに定番なので、LangChain 1.0 以降は langchain.agents の create_agent で一発で作れるようになっているようです。中身は LangGraph で動いているとされています。以前よく使われていた langgraph.prebuilt.create_react_agent は、こちらへの移行が推奨されているようです。
では自分でグラフを組む意味があるのか、という話ですが、最初は手で組んで仕組みを理解して、定番パターンなら create_agent に任せる、くらいが良いのかなと思っています。「コストを取得→異常判定→異常ならSlack通知、正常なら終了」みたいな、エージェントというよりワークフロー寄りの処理は、自分でノードを並べたほうがわかりやすい気がします。
会話を覚えさせる・人間の承認を挟む
LangGraph らしい機能として、もう2つ押さえておきたいものがあります。
checkpointer と thread_id
上のグラフは invoke するたびにまっさらな状態から始まります。会話を続けたいときは、compile 時に checkpointer を渡して、実行時に thread_id を指定します。
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "nyanchu-1"}}
graph.invoke({"messages": [{"role": "user", "content": "10/1のコストは？"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "じゃあその前日は？"}]}, config)同じ thread_id なら、前回の State を引き継いでくれます。checkpointer は各ステップごとに State のスナップショットを保存しているので、途中から再開したり、過去の状態をさかのぼったりもできるとされています。
InMemorySaver はプロセスが落ちたら消えるので、開発・お試し用です。本番では SQLite や Postgres 用の checkpointer パッケージが別途用意されています。Lambda で動かすなら DynamoDB に保存したくなりますが、そのあたりはまだ調べきれていません。
interrupt で一時停止
「ツールを実行する前に人間がOKを出す」みたいな処理は interrupt で書けます。
from langgraph.types import interrupt, Command
def confirm(state):
# ここ注意：再開時はこのノードが最初から実行し直される
answer = interrupt({"question": "通知を送ってもいい？"})
if answer == "yes":
return {"status": "approved"}
return {"status": "rejected"}
# 1回目：interrupt のところで止まる（checkpointer必須）
graph.invoke(inputs, config)
# 2回目：Command(resume=...) で値を渡して再開
graph.invoke(Command(resume="yes"), config)ハマりポイントっぽいのが、コメントに書いた「再開時にノードが頭から実行し直される」という仕様です。interrupt の前に API 呼び出しや DB への書き込みを置いていると、それが2回走ってしまいます。副作用のある処理は interrupt の後ろに置くか、別ノードに分けるのが安全そうです。また、interrupt は状態の保存が前提なので、checkpointer なしでは動きません。
途中経過を見たいときは invoke の代わりに graph.stream(inputs, config, stream_mode="updates") を使うと、ノードごとの更新内容が順番に流れてきます。デバッグにはかなり便利です。
まとめ
- LangGraph は State・Node・Edge で LLM の処理フローを組むライブラリ
- ノードは「変えたいキーだけ」返す。追記したいキーには reducer（add_messages など）を指定する
- ツールループは ToolNode と tools_condition で書ける。定番構成なら create_agent に任せる手もある
- checkpointer と thread_id で記憶、interrupt で人間の承認。再開時にノードが頭から再実行される点に注意
放置しているコスト集計の Lambda に「LLM が異常を判定して、怪しいときだけ通知する」グラフを組み込めないか考えています。ただ、Lambda のコールドスタートで langgraph と langchain 系をまとめて読み込むとどれくらい重くなるのかが気になっていて、まずはそこを計測するところからになりそうです。
