コンテンツにスキップ

LangGraph 基本構文 — 写経レクチャー学習メモ & 用語集

StateGraph を素手で組み、add_edge(静的)→ add_conditional_edges(ルーター分岐)→ Command(goto=)(ノード自身が次を決める)→ Send(動的並列)→ create_react_agent(prebuilt)と、ルーティングの抽象度を 1 段ずつ上げながら学ぶフォルダ。


0. このフォルダのファイル構成(写経用と完成版が同居している)

このフォルダには 2 系統のファイルが混在している。混乱しやすいので先に整理する。

系統 ファイル 位置づけ
完成版(読む用) ex01_minimal_graph.pyex07_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つの対比

  1. ex03 vs ex04: 「順番が固定(静的エッジ)」と「途中で行き先が変わる(条件エッジ)」の違い
  2. ex04 vs ex05: 「ビルダー側のルーター関数が次を決める」と「ノード自身が Command(goto=) で次を決める」の違い(同じ自己改善ループを 2 通りで実装している)
  3. 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_nodeadd_edgecompileinvoke)だけを、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 に飛びうる」という可視化用ヒント(辞書ではなくリストでも可)
  • MessagesStateadd_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. CommandSend の違い(名前が似ていて最も混同する)

どちらも 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 も追記される」 → : 追記されるのはリデューサを付けたキーだけ。リデューサのないキーは常に上書き。resultsoperator.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_nodeadd_edgecompileinvoke の 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. 拡張アイデア

  1. ex03 を 3 段に拡張: 翻訳 → 要約 → さらに「箇条書き化」ノードを足し、name で誰が書いたか追う。add_edge を足すだけで伸びる感覚を掴む。
  2. ex04 と ex05 を同一プロンプトで実行し挙動を比較: 同じ自己改善ループが、conditional edge 版と Command 版で何ステップで PASS に至るか、recursion_limit を 3 まで絞って GraphRecursionError を意図的に出してみる。
  3. ex06 を「fan-out → 集約ノード」型に拡張: explain の後に END ではなく aggregate ノードを足し、N 件の results を 1 つの総括文に LLM でまとめる(map-reduce パターン)。
  4. ex05 を 3 ノードの supervisor 化: supervisor ノードが Command(goto=)researcher / writer のどちらかへ委譲するマルチエージェントに拡張(第18回の前哨戦)。
  5. graph.stream() で途中経過を観察: ex07 を invoke ではなく stream で回し、Thought / Tool Call / Observation が 1 ステップずつ流れる様子を print する。
  6. checkpointer を付けて会話を継続: compile(checkpointer=MemorySaver()) + thread_id で、ex02 を「state が次の invoke に引き継がれるチャット」にする。

6. 現代版に移植するなら(古い書き方への注釈)

  • create_react_agentprompt 引数: 旧 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.pygraph.invoke(initial, recursion_limit=5) とキーワード直渡し、完成版 ex04_conditional_edges.pygraph.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. 記事参照


作成: 2026-06-12 / 最終更新: 2026-06-12