LLM（大規模言語モデル）を活用したアプリケーション開発において、単純なプロンプト応答を超えた「エージェント」の構築が注目を集めています。エージェントとは、与えられた目標に対して自律的に判断し、ツールを使い分けながらタスクを遂行するシステムです。LangGraphは、LangChainチームが開発したグラフベースのフレームワークで、複雑なエージェントの状態管理とワークフロー制御を実現します。
従来のLangChainのAgentExecutorでは、エージェントのフローが暗黙的で制御が難しいという課題がありました。LangGraphはこの問題を、明示的なグラフ構造でワークフローを定義することで解決しています。
LangGraphの核心は、エージェントの動作をグラフ（ノードとエッジ）で表現する点にあります。各ノードは処理ステップ、エッジは遷移条件を表し、全体のフローを視覚的かつ明示的に制御できます。
入会申込
ログインfrom langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 状態の型定義
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
current_step: str
tool_results: dict
# グラフの構築
graph = StateGraph(AgentState)
# ノードの追加
graph.add_node("analyze", analyze_task)
graph.add_node("select_tool", select_tool)
graph.add_node("execute_tool", execute_tool)
graph.add_node("evaluate", evaluate_result)
# エッジの定義
graph.set_entry_point("analyze")
graph.add_edge("analyze", "select_tool")
graph.add_edge("select_tool", "execute_tool")
graph.add_conditional_edges(
"evaluate",
should_continue,
{"continue": "analyze", "finish": END}
)このコードでは、タスクの分析、ツールの選択、ツールの実行、結果の評価という4つのステップをノードとして定義し、条件分岐により処理を繰り返すか終了するかを制御しています。
LangGraphの状態管理は、TypedDictベースの状態クラスで行われます。特にAnnotatedとoperator.addを使うことで、メッセージリストへの追記的な更新が可能になります。これにより、各ノードは状態の一部だけを返せばよく、状態全体を管理する負担が軽減されます。
def analyze_task(state: AgentState) -> dict:
messages = state["messages"]
# LLMにタスク分析を依頼
response = llm.invoke(messages)
return {
"messages": [response],
"current_step": "analysis_complete"
}エージェントの真価は、外部ツールとの連携にあります。LangGraphでは、LangChainのツール定義と組み合わせることで、LLMが状況に応じて適切なツールを選択・実行できます。
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
@tool
def search_database(query: str) -> str:
"""データベースからクエリに一致する情報を検索する"""
# 実際のDB検索ロジック
results = db.execute(query)
return str(results)
@tool
def calculate(expression: str) -> str:
"""数式を計算して結果を返す"""
return str(eval(expression))
@tool
def send_notification(message: str, channel: str) -> str:
"""指定チャネルに通知を送信する"""
slack_client.send(channel, message)
return f"通知を{channel}に送信しました"
# ツールをLLMにバインド
tools = [search_database, calculate, send_notification]
llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools(tools)ツールの呼び出しと結果の処理を行うノードを実装します。LLMがツール呼び出しを決定した場合、そのツールを実際に実行し結果を状態に反映します。
from langgraph.prebuilt import ToolNode
# ToolNodeは定義済みツールの実行を自動的に処理する
tool_node = ToolNode(tools)
# 条件分岐：ツール呼び出しが必要か判定
def should_use_tool(state: AgentState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "use_tool"
return "respond"
# グラフへの組み込み
graph.add_node("agent", call_agent)
graph.add_node("tools", tool_node)
graph.add_conditional_edges(
"agent",
should_use_tool,
{"use_tool": "tools", "respond": END}
)
graph.add_edge("tools", "agent")LangGraphの強力な機能の一つが、チェックポイントによる状態の永続化です。これにより、長時間にわたるタスクの中断・再開や、会話の履歴管理が実現できます。
from langgraph.checkpoint.sqlite import SqliteSaver
# チェックポイントの設定
memory = SqliteSaver.from_conn_string(":memory:")
# グラフのコンパイル時にチェックポイントを指定
app = graph.compile(checkpointer=memory)
# スレッドIDを指定して実行
config = {"configurable": {"thread_id": "user_session_001"}}
result = app.invoke(
{"messages": [{"role": "user", "content": "売上レポートを作成して"}]},
config=config
)チェックポイントはSQLite以外にも、PostgreSQLやRedisなど複数のバックエンドをサポートしています。本番環境ではPostgreSQLを使用し、開発環境ではSQLiteを使うといった使い分けが可能です。
エージェントが重要な操作（データの削除、外部への通知送信など）を行う前に、人間の承認を求める仕組みを組み込むことは、実運用において非常に重要です。
from langgraph.graph import StateGraph, END
def request_approval(state: AgentState) -> dict:
"""人間の承認を求めるノード"""
pending_action = state["tool_results"].get("pending_action")
return {
"messages": [{
"role": "assistant",
"content": f"以下の操作を実行してよろしいですか？\n{pending_action}"
}],
"current_step": "awaiting_approval"
}
# interrupt_before で特定ノードの前に中断
app = graph.compile(
checkpointer=memory,
interrupt_before=["execute_dangerous_tool"]
)interrupt_beforeパラメータを使うことで、指定したノードの実行前にグラフの実行が中断されます。ユーザーの承認後に再開することで、安全なエージェント運用が可能になります。
エージェントがツール実行に失敗した場合のリトライロジックや、LLMの応答が不適切だった場合のフォールバック処理は、実運用で必ず必要になります。グラフの条件分岐を使って、エラー状態を明示的にハンドリングすることが重要です。
複数のツールを同時に実行できる場合、LangGraphのファンアウト・ファンイン機構を使って並列処理が可能です。例えば、複数のデータソースから同時に情報を取得し、結果を統合するパターンは頻繁に使用されます。
複雑なエージェントは、サブグラフに分割してモジュール化することで保守性が向上します。例えば、情報収集フェーズ、分析フェーズ、レポート生成フェーズをそれぞれ独立したサブグラフとして定義し、メインのグラフから呼び出す構成が考えられます。
LangGraphは、AIエージェント開発における状態管理とワークフロー制御の課題を、グラフベースのアプローチで解決する強力なフレームワークです。明示的なグラフ構造によりデバッグが容易になり、チェックポイント機能で長時間タスクの管理も実現できます。ツール連携とHuman-in-the-Loopの仕組みを組み合わせることで、実用的で安全なAIエージェントを構築できるでしょう。まずは小さなワークフローから始めて、徐々に複雑なエージェントへと発展させていくアプローチをお勧めします。
