LangGraph 基本構文 — 写経レクチャー学習メモ & 用語集¶
StateGraphを素手で組み、add_edge(静的)→add_conditional_edges(ルーター分岐)→Command(goto=)(ノード自身が次を決める)→Send(動的並列)→create_react_agent(prebuilt)と、ルーティングの抽象度を 1 段ずつ上げながら学ぶフォルダ。
- サンプルの並び・起動手順は
README.md- 連載対応: 第17回 LangGraph Studio(
../../software-design/17/STUDY_NOTES.md)、第18回 マルチエージェント(../../software-design/18/STUDY_NOTES.md)- 上位ガイド: リポジトリルート
../../CLAUDE.mdの「写経用レクチャーフォルダ」節
0. このフォルダのファイル構成(写経用と完成版が同居している)¶
このフォルダには 2 系統のファイルが混在している。混乱しやすいので先に整理する。
| 系統 | ファイル | 位置づけ |
|---|---|---|
| 完成版(読む用) | ex01_minimal_graph.py 〜 ex07_react_agent.py |
# ①②③… の番号コメント付き。このメモが解説する正本 |
| 写経用(自分で書く用) | ex01.py, ex02.py, ex04.py, ex05.py |
完成版を削ぎ落とした手習い用の写し。ex03.py は空ファイル(自分で埋める課題) |
写経用
ex0N.pyは完成版とほぼ同内容(ex04.pyだけrecursion_limit=5、完成版は6という細かい差はある)。学ぶときはex0N_*.py(完成版)を読み、ex0N.py側で手を動かすのが想定された使い方のはず。以降の解説はすべて完成版ex0N_*.pyを指す。
1. 全体像 — ルーティング抽象度の階段¶
LangGraph で「次にどのノードへ進むか(=ルーティング)」を決める方法は複数あり、このフォルダはそれを抽象度の低い順に並べてある。学ぶ順序そのものが設計の選択肢の一覧になっている。
| # | ファイル | 学ぶ概念 | ルーティングの決め方 | LLM | 外部 API |
|---|---|---|---|---|---|
| 01 | ex01_minimal_graph.py |
StateGraph 最小構成(add_node/add_edge/compile/invoke) |
静的(START→node→END 一本道) | ❌ | ❌ |
| 02 | ex02_messages_state.py |
MessagesState + ChatOpenAI で 1 ノード LLM 呼び出し |
静的 | ✅ | OpenAI |
| 03 | ex03_static_edges.py |
複数ノードを add_edge で直列接続 |
静的(固定の順番) | ✅ | OpenAI |
| 04 | ex04_conditional_edges.py |
add_conditional_edges でルーター関数による分岐ループ |
動的・ビルダー側(ルーター関数が次ノード名を返す) | ✅ | OpenAI |
| 05 | ex05_command_routing.py |
Command(goto=) で動的ハンドオフ(add_edge を書かない) |
動的・ノード側(ノード自身が goto を返す) | ✅ | OpenAI |
| 06 | ex06_send_parallel.py |
Send で動的並列 fan-out + operator.add 集約 |
動的・並列(N 個同時起動) | ✅ | OpenAI |
| 07 | ex07_react_agent.py |
create_react_agent prebuilt で ReAct ループ |
prebuilt が内部で自動 | ✅ | OpenAI + Tavily |
観察してほしい3つの対比¶
- ex03 vs ex04: 「順番が固定(静的エッジ)」と「途中で行き先が変わる(条件エッジ)」の違い
- ex04 vs ex05: 「ビルダー側のルーター関数が次を決める」と「ノード自身が
Command(goto=)で次を決める」の違い(同じ自己改善ループを 2 通りで実装している) - ex06: 「リストで
Sendを返す = N 個並列起動」は、普通のエッジとは別物の挙動
処理フロー(ex04 / ex05 の自己改善ループ)¶
ex04 と ex05 はまったく同じ振る舞い(回答→自己評価→PASS なら終了 / FAIL なら再回答)を、実装手段を変えて書いたもの。
flowchart TD
START([START]) --> answer[answer<br/>LLM が回答]
answer --> review[review<br/>LLM が自己評価<br/>JUDGE: PASS / FAIL]
review -->|JUDGE: PASS| END([END])
review -->|JUDGE: FAIL| answer
- ex04 はこの分岐を
add_conditional_edges("review", route_after_review, {...})でビルダー側に書く - ex05 はこの分岐を
reviewノードがreturn Command(goto=...)でノード側に書く(ビルダーにはSTART→answerの 1 本しか edge がない)
処理フロー(ex06 の Send fan-out)¶
flowchart TD
START([START]) -->|fan_out が Send を N 個返す| s1[explain topic=LangGraph]
START --> s2[explain topic=LCEL]
START --> s3[explain topic=ReAct]
s1 -->|results に append| END([END])
s2 --> END
s3 --> END
fan_out が [Send("explain", {...}), Send("explain", {...}), ...] を返すと、explain ノードがトピック数だけ並列に起動する。各 explain の戻り値 {"results": [...]} は operator.add リデューサで 1 本のリストに連結される。
2. サンプル別の要点¶
ex01 — StateGraph の最小構成(LLM なし)¶
LangGraph の骨格 4 ステップ(add_node→add_edge→compile→invoke)だけを、LLM もツールも使わず体感するサンプル。「state を受け取って加工して返す関数」を 1 個ノードにする。
class MyState(TypedDict): # ① state の型。ここでは number:int だけ
number: int
def add_one(state: MyState) -> dict: # ② ノード関数。更新したい差分だけ dict で返す
new_number = state["number"] + 1
return {"number": new_number}
builder = StateGraph(MyState) # ③ ビルダーに state 型を渡す(必須)
builder.add_node("add_one", add_one) # ④ ("名前", 関数) で登録
builder.add_edge(START, "add_one") # ⑤ START → add_one
builder.add_edge("add_one", END) # add_one → END
graph = builder.compile() # ⑥ 実行可能な graph に確定
result = graph.invoke({"number": 10}) # ⑦ 初期 state を渡す。戻り値は最終 state
| 行 | やっていること | なぜそうするか |
|---|---|---|
| ① | TypedDict で state のスキーマを宣言 |
LangGraph は state の型を見て「どのキーをどう更新するか」を管理する。Pydantic でも可だが TypedDict が最軽量 |
| ② | ノードは全 state を返さず差分だけ返す | LangGraph 側がリデューサで既存 state にマージする。返した dict のキーだけが更新される |
| ③ | StateGraph(MyState) に型を渡す |
この型がないとビルダーは state の構造を知れない |
| ⑥ | compile() で確定 |
これ以降ノード追加不可になり、実行可能な CompiledStateGraph が返る |
START / END は特殊定数。ノード名の文字列としても扱えるが、実体は LangGraph が予約した「グラフの入口・出口」マーカー。graph.get_graph().draw_mermaid() で構造を Mermaid 文字列として出力できる(公式機能、デバッグに便利)。
ex02 — MessagesState + ChatOpenAI¶
LangGraph がビルトインで用意する MessagesState を使い、LLM 応答を会話履歴に積み上げる最小サンプル。
def call_llm(state: MessagesState) -> dict:
response = llm.invoke(state["messages"]) # 会話履歴を丸ごと LLM へ
return {"messages": [response]} # 差分(追加メッセージ)だけ返す
MessagesStateは{"messages": Annotated[list[BaseMessage], add_messages]}と等価なビルトイン型。自分で書かなくても「メッセージのリストを持つ state」が手に入る。add_messagesリデューサが肝。ノードが{"messages": [response]}と差分 1 件だけ返しても、リデューサが既存リストに追記(append)してくれる。普通の dict 更新なら上書きされるところを、メッセージ列だけは「連結」される特別扱い。("user", "テキスト")はHumanMessage("テキスト")の糖衣構文(タプルの第 1 要素が role)。
ex03 — add_edge で直列パイプライン¶
「翻訳 → 要約」の 2 段パイプラインを add_edge でつなぐ。各ノードは自分の役割だけ書き、フロー制御はビルダー側にある。
builder.add_edge(START, "translate") # ④ 一直線
builder.add_edge("translate", "summarize") # translate が終わったら必ず summarize
builder.add_edge("summarize", END)
add_edge("A", "B")=「A が終わったら必ず B を呼ぶ」固定エッジ。条件分岐なし。- 各ノードは
state["messages"][-1].content(直前メッセージ)を入力にし、AIMessage(..., name="translator")のようにnameを付けて結果を積む。後で誰が書いたか判別するため。 - LCEL(
prompt | llm | parser)との対比: LCEL は 1 直線パイプライン。StateGraphは state を持つ有限状態機械で、途中 state を後から参照できる点が違う。実行後result["messages"]には user→translator→summarizer の 3 件が残る。
ex04 — add_conditional_edges でルーター分岐ループ¶
「回答 → 自己評価 → PASS なら終了 / FAIL なら再回答」の自己改善ループ。分岐をビルダー側のルーター関数で書く。
def route_after_review(state: MessagesState) -> Literal["answer", "end"]: # ③
last = state["messages"][-1].content
if "JUDGE: PASS" in last:
return "end" # ルーター戻り値(次の行き先のキー)を文字列で返す
return "answer"
builder.add_conditional_edges( # ⑤
"review", # 出発ノード
route_after_review, # ルーター関数(state → 次キー)
{"answer": "answer", "end": END}, # キー → 実ノードの対応辞書
)
result = graph.invoke(initial, {"recursion_limit": 6}) # ⑥ ループ暴走防止
| 引数 | 役割 |
|---|---|
第1引数 "review" |
どのノードの後で分岐するか |
第2引数 route_after_review |
state を見て次のキー(文字列)を返す純関数 |
第3引数 {...} |
キー→実ノード名のマッピング。LangGraph Studio はこの辞書から「ありうるエッジ」を把握するので、可視化のためにも明示が要る |
- ルーター関数は state を更新しない(読むだけ)。あくまで「次どこへ行くか」を返すだけ。
- ループを作ると無限ループの危険があるため、
recursion_limit(既定 25)をinvokeの config で渡す。超えるとGraphRecursionError。
ex05 — Command(goto=) で動的ハンドオフ(第18回 sd_18 と同じ設計)¶
ex04 と同じ振る舞いを、add_edge を 1 本も書かずに実装する。ノード関数自身が Command(goto=...) を返して次を能動的に指定する。
def review(state: MessagesState) -> Command[Literal["answer", END]]: # ②
...
goto = END if "JUDGE: PASS" in content else "answer"
return Command( # ③
update={"messages": [AIMessage(content=content, name="reviewer")]},
goto=goto,
)
builder = StateGraph(MessagesState)
builder.add_node("answer", answer)
builder.add_node("review", review)
builder.add_edge(START, "answer") # ④ edge はこの 1 本だけ。あとは Command が動的に決める
| 要素 | 役割・内部メカニズム |
|---|---|
Command(update=..., goto=...) |
LangGraph 0.2+ の動的ルーティング機構。state 更新(update)と次ノード指定(goto)を 1 個の戻り値で同時に行う |
戻り値の型ヒント Command[Literal["answer", END]] |
実行には影響しないが、Studio や mypy に「このノードからの遷移先候補」を伝えるメタ情報。Studio はこれを読んでグラフを描く |
add_edge が START→answer の 1 本だけ |
残りの遷移はすべてノード側の goto が決めるので、ビルダーには書かない。これがハンドオフ(hand-off)パターンの核 |
- ex04 との本質的な違い: ex04 は「ビルダーが分岐表を持つ」、ex05 は「各ノードが自分の次を知っている」。マルチエージェントの supervisor / hand-off はこの ex05 型でないと表現しづらい(エージェントが状況を見て次のエージェントへ委譲する=ノード側が goto を決める形だから)。
- 第18回
sd_18の ReflectionAgent はこの応用版で、LLM 出力を regex で解析してgotoを決めるはず。
ex06 — Send で動的並列 fan-out(第17回 task_executor と同じ設計)¶
「3 トピックを並列で説明 → 結果集約」。Send をリストで返すと、その個数だけノードが並列起動する。
class State(TypedDict):
topics: list[str]
results: Annotated[Sequence[str], operator.add] # ① 複数ノードが書くので連結リデューサ
def fan_out(state: State) -> list[Send]: # ③
return [Send("explain", {"topic": t}) for t in state["topics"]]
builder.add_conditional_edges(START, fan_out, ["explain"]) # ⑥
builder.add_edge("explain", END)
| 要素 | 役割・内部メカニズム |
|---|---|
Send("explain", {"topic": t}) |
「explain ノードを、この個別 state で起動せよ」という指示オブジェクト。並列ノードは親 state ではなく Send で渡された小さな state({"topic": ...})を受け取る |
fan_out がリストを返す |
リストの要素数だけ explain が並列起動する。これが fan-out の正体。普通のエッジ 1 本では起きない |
Annotated[Sequence[str], operator.add] |
results は複数ノードから同時に書かれるので、リデューサで連結が必須。これがないと最後の 1 個で上書きされ結果が 1 件になる |
add_conditional_edges(START, fan_out, ["explain"]) |
第3引数 ["explain"] は「次は explain に飛びうる」という可視化用ヒント(辞書ではなくリストでも可) |
MessagesStateのadd_messagesと同じ発想: 「複数の書き込みを連結したいキー」には専用リデューサを付ける。違いは append 専用(add_messages)か汎用連結(operator.add)か。
ex07 — create_react_agent prebuilt で ReAct ループを 1 行¶
ReAct パターン(Thought → Action → Observation を回す)を、自分で StateGraph を組まずに 1 行で得る prebuilt 関数。Tavily 検索ツールを渡して自由に検索させる。
agent = create_react_agent(
llm,
tools=[tavily],
prompt="あなたは情報収集の専門家です。…", # 旧 state_modifier。システムプロンプトとして冒頭注入
)
result = agent.invoke({"messages": [("user", query)]})
create_react_agentの中身は「LLM がtool_callを返すかどうかを見て、ツール実行 → 結果を戻して再度 LLM …を繰り返す小さなStateGraph」(内部でtools_condition+ToolNodeを組んだもの)。返り値もCompiledStateGraphなのでinvoke/streamできる。- ex01〜06 で手で組んだ「条件分岐ループ」を既製品として 1 行で手に入れる省略形。第17回 task_executor の中身・第18回 Agent クラスの内部もこれを使っているはず。
result["messages"]を全部見ると、AIMessage(tool_calls=...)(Action)→ToolMessage(Observation)→AIMessage(最終回答)の遷移が観察できる。
3. 用語集(最重要)— 写経で混乱する語を 1 枚に¶
3-1. グラフの 5 部品: node / edge / state / reducer / START・END¶
LangGraph の最頻出語。役割の階層で並べる。
| 用語 | 何を指すか | 具体例(このフォルダの何か) | 作り方 / 使う API |
|---|---|---|---|
| state | グラフ全体で共有されるデータ。ノード間の唯一の受け渡し手段 | ex01 の MyState、ex02 以降の MessagesState |
TypedDict を定義し StateGraph(型) に渡す |
| node(ノード) | state を受け取り差分 dict を返す関数 | add_one / call_llm / answer / explain |
builder.add_node("名前", 関数) |
| edge(エッジ) | ノード間の遷移。「次にどこへ行くか」 | ex03 の translate→summarize |
builder.add_edge("A","B")(静的) |
| reducer(リデューサ) | 同じ state キーへの複数の書き込みをどうマージするかを決める関数 | add_messages(ex02), operator.add(ex06) |
Annotated[型, リデューサ] で state に宣言 |
| START / END | グラフの入口・出口を表す予約定数 | 全 ex で登場 | from langgraph.graph import START, END |
判定基準: ノードが返す dict のキーが「上書きしてほしい」なら reducer 不要、「追記・連結してほしい」なら reducer 必須。メッセージ列や並列結果は後者。
3-2. ルーティングの 3 流派: 静的 edge / conditional edge / Command(goto=)¶
このフォルダの最大の混乱ポイント。「次のノードを誰がどう決めるか」が 3 通りある。
| 流派 | 誰が次を決めるか | 書く場所 | このフォルダの例 | 使う API |
|---|---|---|---|---|
| 静的エッジ | 固定(常に同じ次) | ビルダー | ex01〜03 | add_edge("A","B") |
| 条件エッジ | ルーター関数(state を見て次キーを返す) | ビルダー | ex04 | add_conditional_edges(出発, ルーター, {キー:行き先}) |
| Command ハンドオフ | ノード自身(戻り値 Command(goto=)) |
ノード関数の中 | ex05 | return Command(update=, goto=) |
判定基準(conditional edge と Command の使い分け): 分岐ロジックをグラフ構造として外に見せたいなら conditional edge(Studio が分岐表を描ける)。ノードが自分の処理結果から直接次を決めたい・state 更新と遷移を同時にやりたいなら
Command(goto=)。マルチエージェントのハンドオフは後者が自然。
3-3. Command と Send の違い(名前が似ていて最も混同する)¶
どちらも langgraph.types から import する動的制御プリミティブだが、目的が正反対。
Command(goto=) |
Send(node, state) |
|
|---|---|---|
| 目的 | 1 つの次ノードへ遷移(直列ハンドオフ) | N 個のノードを並列起動(fan-out) |
| 返し方 | ノードが単体で返す | ルーター関数がリストで返す |
| 渡す state | update= で共有 state を更新 |
各 Send に個別の小さな state を渡す |
| 結果の集約 | 共有 state にマージ | reducer(operator.add)で連結 |
| このフォルダ | ex05 | ex06 |
| 一言で | 「次はここ 1 つへ行け」 | 「これら全部を同時に走らせろ」 |
判定基準: 分岐して1 本道に進むなら
Command、同じ処理を複数データに一斉適用するならSend。
3-4. MessagesState まわりの糖衣¶
| 書き方 | 等価な正式形 | 備考 |
|---|---|---|
MessagesState |
class S(TypedDict): messages: Annotated[list[BaseMessage], add_messages] |
ビルトイン。自分で書かなくてよい |
("user", "text") |
HumanMessage("text") |
タプル第1要素が role。("assistant", ...)→AIMessage |
ノードが {"messages": [x]} を返す |
x が履歴に追記される | add_messages リデューサのおかげ。上書きではない |
3-5. よくある誤解の訂正¶
-
誤解: 「ノードは state 全体を返さないといけない」 → 正: 返すのは更新したいキーの差分 dict だけ。残りは LangGraph が保持する。
return {"number": x}で OK。 -
誤解: 「
add_messagesを使えば普通の dict も追記される」 → 正: 追記されるのはリデューサを付けたキーだけ。リデューサのないキーは常に上書き。resultsもoperator.addを付けて初めて連結される(ex06)。 -
誤解: 「ex05 は
add_edgeを書き忘れているから動かないはず」 → 正:Command(goto=)がエッジの代わりに動的遷移を担うので、START→answerの 1 本だけで完全に動く。add_edgeを書かないのは設計であってバグではない。 -
誤解: 「
Sendは条件エッジの一種」 → 正:Sendは並列起動の指示。条件エッジは分岐(1 本選ぶ)。ex06 でadd_conditional_edges(START, fan_out, ["explain"])と書くのは、fan_out が Send リストを返す入口を登録しているだけで、分岐の意味ではない。
3-6. クイック早見表(困ったらここ)¶
| 困りごと | 見る / 使うもの |
|---|---|
| 一番小さいグラフの骨格を知りたい | ex01(add_node→add_edge→compile→invoke) |
| LLM 応答を会話履歴に積みたい | MessagesState + {"messages":[response]}(ex02) |
| ノードを固定順でつなぎたい | add_edge("A","B")(ex03) |
| state を見て次を分岐させたい(構造を見せたい) | add_conditional_edges(ex04) |
| ノードが自分で次を決めたい / 更新と遷移を同時に | return Command(update=, goto=)(ex05) |
| 同じ処理を複数データに並列適用したい | [Send("node", {...}) for ...] + operator.add(ex06) |
| 複数ノードの結果が 1 件に潰れる | state キーに Annotated[..., operator.add] を付け忘れ |
ループが止まらない / GraphRecursionError |
invoke(..., {"recursion_limit": N}) を調整(ex04/05) |
| ReAct エージェントを手早く | create_react_agent(llm, tools=[...], prompt=...)(ex07) |
| グラフ構造を図で確認したい | graph.get_graph().draw_mermaid()(ex01) |
4. 学んだこと(要点)¶
- LangGraph の骨格は
add_node→add_edge→compile→invokeの 4 ステップ。state はTypedDict、ノードは差分 dict を返すだけ。 - 「次のノードを誰が決めるか」に 静的 / 条件エッジ /
Command(goto=)の 3 流派がある。ex04 と ex05 は同じ自己改善ループを別流派で書いた対比サンプル。 - reducer が state の挙動を決める。
add_messages(メッセージ追記)もoperator.add(並列結果連結)も「複数書き込みのマージ規則」。リデューサのないキーは上書き。 Command(1 つへ直列遷移)とSend(N 個へ並列起動)は目的が逆。名前が似ているので混同注意。create_react_agentは ex01〜06 の手組みを1 行に畳んだ既製品。中身はtools_condition+ToolNodeの小さなStateGraph。- 戻り値の型ヒント(
Command[Literal[...]]、ルーター関数の-> Literal[...])は実行に影響しないが Studio / mypy への遷移先ヒントになる。
5. 拡張アイデア¶
- ex03 を 3 段に拡張: 翻訳 → 要約 → さらに「箇条書き化」ノードを足し、
nameで誰が書いたか追う。add_edgeを足すだけで伸びる感覚を掴む。 - ex04 と ex05 を同一プロンプトで実行し挙動を比較: 同じ自己改善ループが、conditional edge 版と Command 版で何ステップで PASS に至るか、
recursion_limitを 3 まで絞ってGraphRecursionErrorを意図的に出してみる。 - ex06 を「fan-out → 集約ノード」型に拡張:
explainの後に END ではなくaggregateノードを足し、N 件のresultsを 1 つの総括文に LLM でまとめる(map-reduce パターン)。 - ex05 を 3 ノードの supervisor 化:
supervisorノードがCommand(goto=)でresearcher/writerのどちらかへ委譲するマルチエージェントに拡張(第18回の前哨戦)。 graph.stream()で途中経過を観察: ex07 をinvokeではなくstreamで回し、Thought / Tool Call / Observation が 1 ステップずつ流れる様子をprintする。- checkpointer を付けて会話を継続:
compile(checkpointer=MemorySaver())+thread_idで、ex02 を「state が次の invoke に引き継がれるチャット」にする。
6. 現代版に移植するなら(古い書き方への注釈)¶
create_react_agentのprompt引数: 旧 API ではstate_modifier、さらに前はmessages_modifierだった。現行(langgraph >= 0.6, このフォルダの依存)はprompt=が正。古い記事のコードを見たら読み替える。Command/Sendの import 元: 現行はfrom langgraph.types import Command, Send。古い資料ではlanggraph.graph直下やlanggraph.constantsから import している例があるが、現行はlanggraph.types。TavilySearchの import: 現行はfrom langchain_tavily import TavilySearch(独立パッケージlangchain-tavily)。旧from langchain_community.tools.tavily_search import TavilySearchResultsは非推奨方向。recursion_limitの渡し方: 写経用ex04.pyはgraph.invoke(initial, recursion_limit=5)とキーワード直渡し、完成版ex04_conditional_edges.pyはgraph.invoke(initial, {"recursion_limit": 6})と config dict 渡し。config dict 渡しが正式(invoke(input, config)の第2引数が RunnableConfig)。キーワード直渡しはバージョンによっては通らない可能性があるので config dict 形式に揃えるのが安全。
7. 既知の注意点¶
ex03.pyは空ファイル(0 バイト)。写経課題として意図的に空にしてあるはず。完成版ex03_static_edges.pyを見ながら自分で埋める想定。- 写経用と完成版で
recursion_limitの値・渡し方が違う(上記 §6 参照)。挙動は同じだが、揃えるなら config dict 形式に。 - ex02 以降は OpenAI API キー、ex07 は加えて Tavily API キーが要る。README は
.env.sampleコピー方式だが、このリポジトリの方針では.env.op+op run --env-file=.env.op -- uv run python ex07_react_agent.pyを使うのが正(~/.claude/docs/secret-handling.md参照)。
8. 記事参照¶
- Software Design 連載「実践LLMアプリケーション開発」第17回(LangGraph Studio)・第18回(マルチエージェント)の構文の素振りフォルダ
- 連載各回メモ:
../../software-design/17/STUDY_NOTES.md,../../software-design/18/STUDY_NOTES.md - LangGraph 公式 Tutorial: https://langchain-ai.github.io/langgraph/tutorials/
作成: 2026-06-12 / 最終更新: 2026-06-12