第7章 LangGraph入門

1. LangGraphの概要

LangGraphは、LLMアプリケーションを有向グラフとして構築するためのフレームワークである。ノード（処理）とエッジ（遷移）を定義することで、複雑なワークフローやエージェントを表現できる。2024年初頭にリリースされ、LangChainエコシステムの中核コンポーネントとなっている。

1.1 グラフベースアーキテクチャ

LangGraphでは、アプリケーションを「状態」「ノード」「エッジ」の3要素で表現する。

Table 1. LangGraphの基本要素

Fig 1. グラフベースのエージェント構造

        ┌──────────┐
        │  START   │
        └────┬─────┘
             │
        ┌────▼─────┐
        │  agent   │ ←── LLMが判断
        └────┬─────┘
             │
       ┌─────┴─────┐
       │           │
  ┌────▼────┐ ┌────▼────┐
  │  tools  │ │   END   │
  └────┬────┘ └─────────┘
       │
       └──────────→ agent へ戻る

1.2 vs LangChain Agent

従来のLangChain Agent（AgentExecutor）と比較して、LangGraphには以下の利点がある。

Table 2. LangChain Agent vs LangGraph

LangGraphは、エージェントの動作をより細かく制御したい場合、複雑なワークフローを構築したい場合、本番環境での運用を見据えている場合に特に有効である。

2. 状態管理

LangGraphの中核機能は、グラフ全体で共有される状態（State）の管理である。各ノードは状態を読み取り、更新することで、処理間でデータを受け渡す。

2.1 StateGraph

StateGraphは、状態付きグラフを構築するためのクラスである。TypedDictまたはPydanticモデルで状態の型を定義する。

Fig 2. StateGraphの基本構造

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

# 状態の定義
class AgentState(TypedDict):
    messages: Annotated[list, add]  # メッセージリスト（追加で更新）
    next_action: str                # 次のアクション

# グラフの作成
graph = StateGraph(AgentState)

# ノードの追加
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)

# エッジの追加
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route_function)
graph.add_edge("tools", "agent")

# グラフのコンパイル
app = graph.compile()

2.2 State定義

状態の各フィールドには、更新方法を指定するReducerを設定できる。Annotatedと関数を組み合わせて定義する。

Table 3. 主要なReducerパターン

Fig 3. Reducerの動作例

from operator import add
from typing import Annotated

class State(TypedDict):
    # 上書き更新
    current_step: str
    
    # リストへの追加
    messages: Annotated[list, add]
    
    # カスタムReducer（最大値を保持）
    max_score: Annotated[float, lambda old, new: max(old, new)]

# ノードからの更新
def my_node(state: State) -> dict:
    return {
        "current_step": "step2",           # 上書き
        "messages": [new_message],         # 既存リストに追加
        "max_score": 0.95                  # 現在値と比較して大きい方を保持
    }

3. ノードとエッジ

ノードは状態を受け取り、更新を返す関数である。エッジはノード間の遷移を定義し、条件分岐も可能である。

Fig 4. ノードの実装例

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

model = ChatOpenAI(model="gpt-4o")

def agent_node(state: AgentState) -> dict:
    """エージェントノード: LLMを呼び出して次のアクションを決定"""
    messages = state["messages"]
    response = model.invoke(messages)
    
    return {
        "messages": [response],
        "next_action": determine_action(response)
    }

def tools_node(state: AgentState) -> dict:
    """ツールノード: ツールを実行して結果を返す"""
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    
    results = []
    for call in tool_calls:
        result = execute_tool(call)
        results.append(result)
    
    return {"messages": results}

Fig 5. 条件付きエッジの実装

def route_function(state: AgentState) -> str:
    """次に遷移するノードを決定"""
    last_message = state["messages"][-1]
    
    # ツール呼び出しがある場合
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 終了条件
    return END

# 条件付きエッジの追加
graph.add_conditional_edges(
    "agent",           # 元のノード
    route_function,    # ルーティング関数
    {
        "tools": "tools",  # "tools" → toolsノードへ
        END: END           # END → 終了
    }
)

4. 実践：エージェント構築

実際にLangGraphを使用してReActエージェントを構築する例を示す。

Fig 6. 完全なReActエージェントの実装

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
from operator import add

# 状態の定義
class AgentState(TypedDict):
    messages: Annotated[list, add]

# ツールの定義
@tool
def search(query: str) -> str:
    """Web検索を実行"""
    return f"'{query}'の検索結果: ..."

@tool
def calculator(expression: str) -> str:
    """数式を計算"""
    return str(eval(expression))

tools = [search, calculator]

# モデルの設定（ツールをバインド）
model = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# ノードの定義
def agent(state: AgentState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

tool_node = ToolNode(tools)

# ルーティング関数
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# グラフの構築
graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

# コンパイル
app = graph.compile()

# 実行
result = app.invoke({
    "messages": [HumanMessage(content="2024年の東京オリンピックについて調べて")]
})

print(result["messages"][-1].content)

Fig 7. ストリーミング実行

# ストリーミングで各ステップを確認
for event in app.stream({"messages": [HumanMessage(content="質問")]}):
    for node_name, output in event.items():
        print(f"--- {node_name} ---")
        print(output)

Fig 8. チェックポイントによる永続化

from langgraph.checkpoint.sqlite import SqliteSaver

# チェックポイントの設定
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    
    # スレッドIDを指定して実行
    config = {"configurable": {"thread_id": "user-123"}}
    
    # 最初の実行
    result1 = app.invoke({"messages": [HumanMessage(content="こんにちは")]}, config)
    
    # 同じスレッドで継続（履歴が保持される）
    result2 = app.invoke({"messages": [HumanMessage(content="続きを教えて")]}, config)

チェックポイント機能を使用することで、会話の状態を永続化し、中断・再開が可能になる。本番環境では、PostgreSQLやRedisなどのバックエンドを使用する。