15. LangGraphで最初のグラフを構築する

第IV部では、ツールを定義し、それをLLMに接続し、決定と実行のサイクルを繰り返すエージェントループを完成させました。ループを駆動すること、LLMが要求したときにツールを実行すること、いつ停止すべきかを判断すること — このフローのすべての部分を手作業でコーディングしました。

本章では、まったく異なる方法で同じエージェントを構築します。フローを直接記述する代わりに、ステップ(ノード)と接続ルール(エッジ)をLangGraphフレームワークに登録し、実行を任せます。動作は第14章と同じですが、構築の仕方が変わります。

本章では、4つのコアコンセプト — StateGraph、ノード(node)、エッジ(edge)、State — を扱い、その後、第14章のエージェントループをLangGraphのグラフにリファクタリングします。続く章では、第16章で条件付きルーティングと事前構築済みコンポーネントを、第17章でエージェントが中断された場所から再開できるようにする状態の永続化を扱います。

15.1) なぜグラフなのか?

15.1.1) 既存のエージェントループの限界

第14章のエージェントループを振り返ってみましょう。エラー処理やその他の詳細を取り除くと、中心となる構造は次のようになっていました。

# 第14章のエージェントループ — 中心となる構造(簡略化)
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content=user_input),
]
 
for step in range(max_steps):
    # LLMに次のアクションを決定させる
    ai_message = llm_with_tools.invoke(messages)
    messages.append(ai_message)
 
    # ツール呼び出しが要求されなければ、最終的な回答を返す
    if not ai_message.tool_calls:
        return ai_message.content
 
    # 要求されたツールを実行する
    for tool_call in ai_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        messages.append(tool_message)

このコードは基本的なことだけをカバーしており、それ以外は何もありません。しかし、実際の本番環境では、はるかに多くのことが必要になります。いくつか例を挙げます。

• クラッシュからの復旧 — エージェントが10ステップのリサーチタスクの7ステップ目でクラッシュした場合、最初からやり直すのではなく、7ステップ目から再開できるべきです。
• 承認リクエスト — エージェントが重要な操作を実行する前に、一時停止して人間に「続行してもよいですか?」と尋ねられるべきです。
• リアルタイムモニタリング — ユーザーは、エージェントが今何をしているのか、どのツールを呼び出しているのかを確認できるべきです。
• 可視化とデバッグ — エージェントがどのように動作するかを示す図が利用でき、問題が発生したときにどのステップで間違ったのかを追跡できるべきです。

これらの機能を自分で実装することは不可能ではありませんが、簡単でもありません。クラッシュからの復旧だけでも、各ステップの状態をシリアライズし、ディスクに保存し、復元し、正確な位置から再開するコードを書く必要があります。ビジネスロジックよりもインフラのコードのほうが多くなってしまうかもしれません。

LangGraphは、これらの機能をフレームワークレベルで提供するために作られました。クラッシュからの復旧、承認リクエスト、モニタリング、可視化 — これらすべてをフレームワークが処理します。ただし、1つの要件があります。それは、フレームワークが理解できる構造でエージェントを構築しなければならないということです。

第14章のエージェントループはすべてのロジックを直接処理しているため、フレームワークが介入できる余地がありません。LangGraphが提供するものを活用するには、LangGraphが理解する構造 — つまりグラフ(graph) — でエージェントを再構築する必要があります。それが本章のテーマです。

15.1.2) LangGraphとは何か?

LangGraphは、エージェントのワークフローをグラフとして定義し実行するオーケストレーションフレームワークです。ここでいうグラフとは、エージェントが実行する各ノード(ステップ)がエッジ(接続ルール)によって接続された構造を意味します。

LangGraphでは、ワークフローを独立したノードに分割し、それらをエッジで接続します。するとLangGraphはグラフをたどり、途中で各ノードを実行します。第14章のエージェントループをグラフとして表現すると、次のようになります。

長方形のボックスがノードで、矢印がエッジです。ひし形は、条件に基づいて異なるパスに分岐する条件付きエッジ(conditional edge)を表します。

第14章では、ワークフロー全体がforループ、ifチェック、その他の手書きのコードの中に存在していました。LangGraphでは、各ノードが何をするかを定義し、ノード同士をエッジで結びつけます。要するに、ワークフローをコーディングする方式から、それを構造として宣言する方式へと移行するのです。

LangGraphは、第12章〜第14章で学んだことを何一つ置き換えるものではありません。ツールの定義、bind_tools()、tool_calls、ToolMessage — これらすべては、以前とまったく同じように、ノードの中で引き続き使用されます。

次のセクションでは、LangGraphのコアコンポーネント — StateGraph、State、ノード、エッジ — を1つずつ扱います。

15.2) LangGraphのコンポーネント: StateGraph、State、ノード、エッジ

このセクションでは、LangGraphの4つのコアコンポーネントを1つずつ見ていきます。まずStateGraph — State、ノード、エッジをまとめてグラフにするクラス — から始め、その後、その中に入る各パーツ(State、ノード、エッジ)を扱います。

15.2.1) StateGraph

StateGraphは、LangGraphでグラフを構築するために使用するクラスです。グラフが管理するStateを指定し、ノードを追加し、それらをエッジで接続し、最後にコンパイルして実行可能なグラフを生成します。

どのように動作するか見てみましょう。

StateGraphインスタンスの作成

StateGraphコンストラクタを呼び出してインスタンスを作成します。パラメータとしてStateのスキーマ(クラスそのもの)を渡す必要があります。ここでは、メッセージリストを管理するためにLangGraphが提供する定義済みのStateであるMessagesStateを使用しています。詳細は15.2.2で扱います。

from langgraph.graph import StateGraph, MessagesState
 
builder = StateGraph(MessagesState)

ノードの追加

add_node()を使ってノードを登録します。ノードは、現在のStateを受け取り、変更したい部分を返すPython関数です。ノード関数については15.2.3で詳しく扱います。

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder.add_node(say_hello)    # ノード名は "say_hello" になる

エッジの接続

add_edge(source, target)を使ってノードを接続します。sourceはエッジの起点で、targetは行き先です。STARTとENDは、グラフの入口と出口を表す特別なマーカーです。エッジについては15.2.4で扱います。

from langgraph.graph import START, END
 
builder.add_edge(START, "say_hello")   # グラフの開始 → say_hello を実行
builder.add_edge("say_hello", END)     # say_hello の完了 → グラフの終了

コンパイルと実行

compile()を呼び出すと、グラフ構造が検証され、実行可能なオブジェクトが生成されます。コンパイルされたグラフはinvoke()で実行し、初期のState値を渡します。

graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)

では、これらすべてをまとめて、シンプルなグラフを構築してみましょう。

from langgraph.graph import StateGraph, MessagesState, START, END
 
def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder = StateGraph(MessagesState)
 
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
 
graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

出力:

hello world

invoke()を呼び出すと、グラフはSTART → say_hello → ENDの順に実行されます。say_helloはmessagesをキーとする辞書を返し、その値がMessagesStateのmessagesリストに追加されました。これがどのように機能するかは15.2.2で詳しく掘り下げます。要点は、最後のメッセージのコンテンツを取り出すと"hello world"が得られるということです。

15.2.2) State: グラフを流れるデータ

Stateは、グラフ内のすべてのノードが共有するデータです。ノードが実行されると、現在のStateを受け取り、作業を行い、変更したい部分だけを返します。LangGraphはそれらの変更をStateに折り込み、更新されたバージョンを次のノードに渡します。

Stateの定義

TypedDictをサブクラス化してStateを定義します。エージェントが追跡する必要があるものに合わせて、フィールドと型を選びます。以下はシンプルな例です。

from typing_extensions import TypedDict
 
class AgentState(TypedDict):
    messages: list       # メッセージリスト
    llm_calls: int       # LLM呼び出し回数

ここから、StateGraphを作成する際にAgentStateを渡し、ノード関数の型ヒントとして使用します。

リデューサー(Reducer)

ノードが値を返すと、対応するStateフィールドが更新されます。デフォルトの動作は上書きです — ノードが{"llm_calls": 3}を返すと、llm_callsは以前の値が何であっても単に3になります。

しかし、一部のフィールドは上書きではなく追加が必要です。messagesが上書きされたらどうなるでしょうか?ノードが新しいメッセージを返すたびに、会話履歴全体が消えてしまいます。messagesにとっては、追加が正しい動作です。

LangGraphでは、リデューサー(reducer)関数を通じて、フィールドごとに異なる更新戦略を設定できます。リデューサーはAnnotatedの第2引数として指定します。

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # リデューサー: 追加
    llm_calls: int                            # リデューサーなし: 上書き

add_messagesは、LangGraphが提供するリデューサーです。リストを置き換える代わりに、新しいメッセージを既存のものに追加します。このリデューサーがmessagesフィールドに設定されているため、ノードがmessagesに対して返すあらゆる値が追加されます。llm_callsにはリデューサーがないため、返された値は以前の値を単に上書きします。

これこそが、15.2.1でsay_helloが返したメッセージがmessagesを置き換えるのではなく追加された理由です — リデューサーがそれを処理したのです。

MessagesState

LangGraphには、MessagesStateという定義済みのStateが同梱されています。内部では次のようになっています。

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

先ほど扱ったのと同じ構造です — add_messagesリデューサーがすでに配線されたmessagesフィールドです。

追加のフィールドが必要な場合は、単にサブクラス化します。

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int  # デフォルトの上書き動作

15.2.3) ノード: Stateを更新する関数

ノードは、グラフ内で単一の特定のタスクを実行するPython関数です。

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

ノード関数を書くときに知っておくべきことが2つあります。

ルール1: 引数として現在のStateを受け取ります。 LangGraphは、ノードを実行する際に現在のStateオブジェクトを渡します。

ルール2: State全体ではなく、変更したい部分だけを返します。 ノードはStateを直接変更しません。更新したいフィールドだけを返せば、LangGraphが各フィールドのリデューサールールに従って、それらを既存のStateにマージします。

add_node()を使ってStateGraphにノードを追加します。

builder.add_node(say_hello)          # 関数名 "say_hello" がノード名になる
builder.add_node("my_node", my_func) # 名前を明示的に指定することもできる

15.2.4) エッジ: ノードを接続するルール

エッジは、「このノードが終わった後、次に何を実行するか?」を決定します。2つのタイプがあります。

通常のエッジ

通常のエッジは、2つのノード間の固定された「移動先」を配線します。add_edge(source, target)を使います — sourceが起点のノード、targetが行き先です。

builder.add_edge(START, "say_hello")       # グラフの開始時に say_hello を実行
builder.add_edge("say_hello", "llm_call")  # say_hello の後に llm_call を実行
builder.add_edge("llm_call", END)          # llm_call の後にグラフを終了

条件付きエッジ

条件付きエッジは、現在のStateに基づいて実行時に次のノードを選びます。add_conditional_edges(source, routing_function)を使います — sourceが起点のノードで、routing_functionは現在のStateを受け取り、次のノードの名前を返す関数です。

from langgraph.graph import END
 
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"   # ツール呼び出しが要求された → tool_node へ
    return END               # ツール呼び出しなし → 終了
 
builder.add_conditional_edges("llm_call", should_continue)

add_conditional_edges("llm_call", should_continue)は、LangGraphに「llm_callが終わったら、should_continueを呼び出して次に何を実行するか決めなさい」と伝えます。should_continueは、最後のメッセージにtool_callsがあれば"tool_node"へ、なければENDへルーティングします。実際には、これはLLMがツール呼び出しを要求したときにグラフがツール実行ノードへ進み、要求しなかったときに終了することを意味します。

これで4つのコンポーネントをすべて扱ったので、次のセクションではそれらを使って第14章のエージェントループをLangGraphのグラフにリファクタリングします。

15.3) エージェントループをグラフにリファクタリングする

第14章のエージェントループをLangGraphを使って再構築してみましょう。動作は第14章と同じです — LLMが決定し、LLMの要求に基づいてツールが実行され、完了するまでサイクルが繰り返されます。変わるのは、このフローをどう構造化するかだけです。

完成したグラフは次のようになります。

グラフは、LLMがツール呼び出しの要求をやめるまでllm_callとtool_nodeの間を循環し、その時点でENDに抜けます。ステップごとに構築していきましょう。

15.3.1) Stateの定義

15.2.2のMessagesStateをサブクラス化して、エージェントのStateを定義します。messagesフィールドはMessagesStateから継承され、LLM呼び出しの回数を追跡するためにllm_callsフィールドを追加します。

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int    # LLM呼び出し回数(上書き)

messagesリストには、ユーザー入力(HumanMessage)、LLMの応答(AIMessage)、ツール実行結果(ToolMessage)が順に蓄積されていきます。

15.3.2) ノードの構築

まず、第14章のツールとモデルをセットアップしましょう。

from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """都市の現在の天気を取得します。"""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """簡単な算術式を計算します。例: '3 * 21'。"""
    return str(eval(expression))  # 警告: eval() はセキュリティリスクです。本番環境では使用しないでください。
 
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)

次に、2つのノード関数を書きましょう。

llm_callノード — LLMを呼び出し、応答を返します。

def llm_call(state: AgentState):
    """LLMを呼び出し、応答を返します。"""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

model_with_tools.invoke()がLLMを呼び出し、応答は戻り値の辞書のmessagesキーの下にパッケージ化されます。リデューサーがそれをAgentStateの既存のmessagesに追加します。llm_callsは現在のカウントに1を加えた値を返し、以前の値を上書きします。

tool_nodeノード — LLMが要求したツールを実行し、結果を返します。

def tool_node(state: AgentState):
    """LLMが要求したツールを実行します。"""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        results.append(tool_message)
    return {"messages": results}

tool_nodeは常にllm_callの直後に実行されるため、messagesの最後のメッセージは、LLMがちょうど生成したAIMessageであることが保証されます。そのメッセージのtool_callsフィールドには、LLMが要求したツール呼び出しが含まれています。ノードは各ツールを実行し、結果をresultsに集め、messagesキーの下でそれらを返します — リデューサーが既存のリストへの追加を処理してくれます。

15.3.3) 条件付きエッジ

llm_callが終わったら、tool_nodeを実行するかグラフを終了するかを決める条件付きエッジが必要です。これは15.2.4と同じパターンに従います。

from typing import Literal
from langgraph.graph import END
 
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """ツールを実行するかグラフを終了するかを決定します。"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

last_message.tool_callsが存在する場合、LLMはツール呼び出しを要求しているので、tool_nodeへルーティングします。それ以外の場合は、ENDへルーティングしてグラフを終了します。

Literal["tool_node", "__end__"]という戻り値の型ヒントは、この関数が返しうる行き先を宣言します。LangGraphは、グラフの可視化において条件付きエッジのパスを正しく描画するためにこのヒントを必要とします。実行時の動作には影響しません。

"__end__"はENDの基礎となる文字列値です。Literalは文字列リテラルしか受け付けないため、ENDの代わりに"__end__"と書きます。

15.3.4) グラフの組み立てと実行

すべてを配線するときが来ました。State、ノード、条件付きエッジをStateGraphに組み立ててコンパイルしましょう。

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
 
builder.add_edge(START, "llm_call")                         # 開始 → llm_call
builder.add_conditional_edges("llm_call", should_continue)  # llm_call → tool_node または END
builder.add_edge("tool_node", "llm_call")                   # tool_node → llm_call(ループ)
 
agent = builder.compile()

tool_nodeからllm_callへ戻るエッジがループを作ります。LLMが別のツール呼び出しを要求する代わりに最終的な回答で応答するまで、実行は循環し続け、その時点でループを抜けます。

実行してみましょう。

from langchain_core.messages import HumanMessage
 
result = agent.invoke({
    "messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nTotal LLM calls: {result['llm_calls']}")

出力:

Current temperature in Cairo: 31°C. Multiplied by 3 = 93.
 
Total LLM calls: 3

エージェントはget_weather("Cairo")を呼び出し、その結果を見て、calculate("31 * 3")を呼び出し、最終的な回答を生成しました — 第14章で得たのと同じ結果です。

完全なメッセージ履歴を調べると、すべてのステップがmessagesに順に記録されているのがわかります。

for message in result["messages"]:
    message.pretty_print()

出力:

================================ Human Message =================================
Get the temperature in Cairo, then multiply the number by 3.
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_DiL9WF)
  Args:
    city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, sunny
================================== Ai Message ==================================
Tool Calls:
  calculate (call_wa6RqWST)
  Args:
    expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
Current temperature in Cairo: 31°C. Multiplied by 3 = 93.

15.3.5) グラフの可視化

Jupyterノートブックでは、agent.get_graph().draw_mermaid_png()がグラフ構造を画像としてセルの出力にそのまま描画します。

from IPython.display import Image, display
 
display(Image(agent.get_graph().draw_mermaid_png()))

ターミナル環境では、代わりにPNGファイルとして保存します。

agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")

生成された画像:

実線が通常のエッジで、点線が条件付きエッジです。この図はコードから自動的に生成されます。

15.3.6) 再帰制限(Recursion Limit)

第14章で無限ループを防ぐためにmax_stepsを使ったのと同様に、LangGraphには組み込みのセーフティネットがあります。グラフ実行中にノードが実行されるたびに、内部カウンターが1つずつ増加します。そのカウンターが設定された制限を超えると、LangGraphはGraphRecursionErrorを発生させます。

カウントがどのように機能するかを見るために、先ほどの実行を見てみましょう。get_weatherとcalculateの両方を呼び出すと、次の順序でノードを訪問しました。

llm_call(1) → tool_node(2) → llm_call(3) → tool_node(4) → llm_call(5) → END

合計で5回のノード訪問です。recursion_limitを3に設定すると、3回目の訪問で制限が発動し、実行が途中で打ち切られます。

from langgraph.errors import GraphRecursionError
 
try:
    result = agent.invoke(
        {"messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
         "llm_calls": 0},
        config={"recursion_limit": 3},
    )
except GraphRecursionError:
    print("Agent hit the recursion limit — stopping execution.")

出力:

Agent hit the recursion limit — stopping execution.

invoke()にconfig={"recursion_limit": 数値}を渡して制限を設定します。適切な値は、ユースケースとグラフの複雑さによって異なります。まずは余裕のある数値から始め、テストを通じて調整してください。