LangGraphって何？とりあえず動かしてみよう

今回は、LangGraphとは何か、を動かしながら私の言葉で説明していきます。

LangGraphとは

LangGraphは、LangChain社が開発したステートフルなエージェントを構築・管理・デプロイのフレームワークです。MITライセンスで公開されており、Python版とJavaScript版があります。私はPython版を利用します。

https://github.com/langchain-ai/langgraph

ステートフルなエージェントとは？

次のような機能を持つエージェントがステートフルなエージェントです。

会話履歴
タスクの途中経過
人間の承認待ち状態

つまり、開始から完了まで長い時間がかかり、人間とやりとりしながら動作するエージェントになります。

エージェントの構築だけでなく管理・デプロイ？

LangChain社は、LangGraphに関連して、ステートフルなエージェントを構築するだけでなく、管理やデプロイに関する機能やサービスを提供しています。

前提として、ステートフルなエージェントのデプロイは、通常のWebアプリのデプロイでは問題にならない要因が問題になることがあるようです。で、LangChain社はそれを解決するために有料の「LangGraph Platform（LangSmith Deployment）」を提供している様子です。私は使っていませんが、この背景についても一応、理解・説明しておきたいと思います。

この理由は、公式ブログの以下で確認できます。

https://www.langchain.com/blog/langgraph-platform-ga

簡単に言うと、トランザクションが長期間に及びやすいし、途中で中断されて非同期に再開もするし、定期実行されやすいから一気に負荷も高まりやすい、と言う話だと理解してます。

動かしてみる

公式のハンズオンをやっていきます。LLMは、AnthropicのClaudeを利用します。

https://docs.langchain.com/oss/python/langgraph/quickstart

今回は、公式のハンズオンをやりつつ、わからないことをClaude opus 4.7に調べさせつつ、疑問を解消しながら進んでいきます。

ハンズオンでは電卓エージェントを作って、LangGraphの基本概念（State / Node / Edge / 条件分岐 / ループ）を一通り触ります。「3と4を足して」と聞くと、LLMがaddツールを呼び、結果を返してくれるエージェントを作り、途中で何が起きているかを観察します。

「なぜ、こんな面白くないハンズオンを？」と言う疑問をClaudeに聞いてみたところ、簡単な仕組みだからこそ、仕組みが理解できるとのことです。なるほど。

ではやっていきます。まず、環境を作ってスタートです。

uv init
uv add langgraph langchain "langchain[anthropic]" python-decople
Code language: JavaScript (javascript)

最初に、ANTHROPIC_API_KEYを .envに設定して、一旦疎通確認します。

> uv run python
Python 3.12.1 (v3.12.1:2305ca5144, Dec  7 2023, 17:23:38) [Clang 13.0.0 (clang-1300.0.29.30)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from decouple import config
>>> from langchain.chat_models import init_chat_model
>>>
>>> # .env から ANTHROPIC_API_KEY を読み込む
>>> ANTHROPIC_API_KEY = config("ANTHROPIC_API_KEY")
>>>
>>> # chat model を初期化
>>> chat_model = init_chat_model(
...     "claude-sonnet-4-6",
...     model_provider="anthropic",
...     api_key=ANTHROPIC_API_KEY,
... )
>>>
>>> # 動作確認
>>> response = chat_model.invoke("こんにちは")
>>> print(response.content)
こんにちは！👋

お元気ですか？何かお手伝いできることはありますか？😊
Code language: PHP (php)

電卓エージェントの準備

次に、電卓エージェントを作ります。これは、足し算、掛け算、割り算をツールとして持つシンプルなエージェントです。この時点では、LangGraphは使っておらず、LangChainで構築しています。ちなみに @tool のdocstringがLLM向けの説明になります。

from decouple import config
from langchain.tools import tool
from langchain.chat_models import init_chat_model

ANTHROPIC_API_KEY = config("ANTHROPIC_API_KEY")

model = init_chat_model(
	"claude-sonnet-4-6",
	model_provider="anthropic",
	api_key=ANTHROPIC_API_KEY,
	temperature=0
)

@tool
def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """Divide a and b."""
    return a / b

tools = [add, multiply, divide]
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools)
Code language: JavaScript (javascript)

LangGraphの前提

ここからLangGraphの知識が必要になります。LangGraphでのグラフの構築とエージェント実行の流れを簡単に説明します。

グラフ間で持ち歩く状態（State）を定義します。
その状態（State）を更新するノード（Node）を定義します。
状態（State）をみて、次に進むノードを決定する条件分岐（エッジ）を定義します。
ノードとエッジを繋げてグラフを作ります。そして、実行します。

上記の流れになります。と言うわけで、これをやっていきます。

状態の定義

以下が状態定義の例です。今回は、追記され続ける会話履歴と、LLM呼び出し回数が状態になってます。

from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int
Code language: CSS (css)

「状態定義は、TypedDictがMUSTなの？」と言う疑問が湧いたので、その回答を調べさせました。状態の定義方法は以下の3つがあるようです。

TypedDict: デフォルト推奨。軽量、Pythonの型ヒントだけ
dataclass: デフォルト値が欲しい時
Pydantic BaseModel: ランタイムでのバリデーションが欲しい時。ただし重い

出典： https://docs.langchain.com/oss/python/langgraph/graph-api#schema

また、 Annotated[list[AnyMessage], operator.add] と定義されてますが、LangGraphでは、 Annotatedの2つ目の引数をreducerとして受けとります。そこで「reducerって何？」と言う疑問が出たのでこれも調べておいたところ、以下の情報も得られました。

もともと、reducerの起源は関数型プログラミングの「畳み込み」
フロントエンド界隈のReduxで状態管理用語として広く定着した
LangGraphも同じ概念を採用していて、(current, new) -> merged と言う状態管理を行う

つまり、Reduxが状態遷移関数のためにReducerという用語を定義していたのを拝借して、LangGraphでも同じように状態遷移関数をReducerと呼んでると言うことです。個人的にはAnnotatedの使い方として気持ち悪いですが、一旦スルーします。

ノードの定義

次に、今回は、二つのノードを定義します。

LLMがツールを使うかどうかを判断するノード
直前のメッセージで呼び出すべきと判断されたツールを呼び出すノード

それぞれのノードは、直前のstateを受け取って、stateの更新分を辞書で返す関数になっています。その中で、LLM呼び出しを行います。LLM呼び出しは各ノードの中で明示的に定義することになります。LangGraphでは、暗黙的にLLM呼び出しは行われません。

from langchain.messages import SystemMessage

def llm_call(state: dict):
    """LLMがツールを呼ぶか否かを判断するノード"""
    return {
        "messages": [
            model_with_tools.invoke(
                [SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")]
                + state["messages"]
            )
        ],
        "llm_calls": state.get("llm_calls", 0) + 1
    }
Code language: JavaScript (javascript)

from langchain.messages import ToolMessage

def tool_node(state: dict):
    """直前のLLM出力に含まれるツール呼び出しを全部実行するノード"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]] # 指定されたtoolを取り出す
        observation = tool.invoke(tool_call["args"]) # 実際にtoolを呼び出す
        result.append(
	        ToolMessage(content=observation,
	        tool_call_id=tool_call["id"])
		) # 会話履歴に結果を入れる
    return {"messages": result}
Code language: PHP (php)

エッジを定義する

エッジを定義します。エッジは、LangGraph上でルーティングの条件になり、次のノードを文字列で返します。

前提として

LangGraphではノードとエッジは次のように役割が分かれてます。
- ノード：LLM呼び出しなど、Stateを更新する処理を記述する
- エッジ：決定論的に分岐する処理を記述する
なので、エッジでLLMを呼び出してはならず、それをやるくらいなら、直前にLLMを呼び出すノードを作り、それをStateに込めてエッジで処理することになります。

今回は、呼び出すツールをLLMに選ばせるノードの次に配置するエッジです。そして、ツールがなければ終わりに進むエッジが以下です。

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

def should_continue(state: MessagesState) -> Literal["tool_node", "__end__"]:
    """LLMがツールを呼んでいればtool_nodeへ、呼んでなければ終了"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END
Code language: JavaScript (javascript)

グラフを組み立てる

今回は以下のようにグラフを組み立てます。

これを実際のコードにすると以下のようになります。

agent_builder = StateGraph(MessagesState)

# ノードを登録
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# エッジを張る
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")  # ツール実行後はLLMに戻ってループ

agent = agent_builder.compile()
Code language: PHP (php)

実行してみる

実行は以下になります。

from langchain.messages import HumanMessage

messages = [HumanMessage(content="3と4を足すと？")]
result = agent.invoke({"messages": messages})

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

print(f"\nLLM呼び出し回数: {result['llm_calls']}")
Code language: JavaScript (javascript)

実際に実行してみると以下になります。

uv run python main.py
Code language: CSS (css)

> uv run python main.py
================================ Human Message =================================

3と4を足すと？
================================== Ai Message ==================================

[{'text': '3と4を足し算します！', 'type': 'text'}, {'id': 'toolu_01F8LAdbTQJSeMHXHbhDSbLQ', 'caller': {'type': 'direct'}, 'input': {'a': 3, 'b': 4}, 'name': 'add', 'type': 'tool_use'}]
Tool Calls:
  add (toolu_01F8LAdbTQJSeMHXHbhDSbLQ)
 Call ID: toolu_01F8LAdbTQJSeMHXHbhDSbLQ
  Args:
    a: 3
    b: 4
================================= Tool Message =================================

7
================================== Ai Message ==================================

3と4を足すと、**7**になります！ 😊

LLM呼び出し回数: 2
Code language: JavaScript (javascript)

ストリーミングで挙動を観察する

各ノードの実行結果を途中で見たい場合は、 agent.invoke の代わりに agent.stream を使います。

for chunk in agent.stream({"messages": messages}, stream_mode="updates"):
    for node_name, update in chunk.items():
        print(f"\n=== Node: {node_name} ===")
        if "messages" in update:
            for m in update["messages"]:
                m.pretty_print()
Code language: PHP (php)

グラフを可視化する

以下でmermaidのテキストを出力できます。JupyterやColabだとそれ向けに出力できるようです。（試してません）

print(agent.get_graph().draw_mermaid())
Code language: CSS (css)

実際に出力されるのは以下です。

graph TD;
	__start__([<p>__start__</p>]):::first
	llm_call(llm_call)
	tool_node(tool_node)
	__end__([<p>__end__</p>]):::last
	__start__ --> llm_call;
	llm_call -.-> __end__;
	llm_call -.-> tool_node;
	tool_node --> llm_call;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc
Code language: CSS (css)

描画すると以下になります。

終わり

一旦、簡単なハンズオンは終了です。今回のハンズオンは、古典的なエージェントのアーキテクチャであるReActになっています。

ハンズオンの終わりに、Claudeから次に学ぶべき情報のリンクを教えてもらいました。以下になります。今回は、一旦ここまで。

永続化（チェックポイント）: https://docs.langchain.com/oss/python/langgraph/persistence
中断と再開（Human-in-the-loop）: https://docs.langchain.com/oss/python/langgraph/interrupts
メモリ（短期・長期）: https://docs.langchain.com/oss/python/langgraph/add-memory
サブグラフ: https://docs.langchain.com/oss/python/langgraph/use-subgraphs
ローカルサーバーで動かす: https://docs.langchain.com/oss/python/langgraph/local-server

Post Views: 25

LangGraphって何？とりあえず動かしてみよう

LangGraphとは

ステートフルなエージェントとは？

エージェントの構築だけでなく管理・デプロイ？

動かしてみる

電卓エージェントの準備

LangGraphの前提

状態の定義

ノードの定義

エッジを定義する

グラフを組み立てる

実行してみる

ストリーミングで挙動を観察する

グラフを可視化する

終わり

Bunとは何か。何が起きているの？

Alpine.jsって何？

pytest-watchの代替を探す

Alpine.jsのx-cloakって何？

hx-confirm って何？

Alpine.jsのx-bindって何？

LangGraphとは

ステートフルなエージェントとは？

エージェントの構築だけでなく管理・デプロイ？

動かしてみる

電卓エージェントの準備

LangGraphの前提

状態の定義

ノードの定義

エッジを定義する

グラフを組み立てる

実行してみる

ストリーミングで挙動を観察する

グラフを可視化する

終わり

類似投稿