こんにちは。
株式会社KANGEN Holdingsの野中康生です!
生成AI・LLMの進化により、複雑なタスクをエージェントに任せる時代が本格化しています。その中でもぜひ注目したい存在が、LangGraph です。LangChainの拡張でありながら、「状態遷移」の考え方を取り入れて、より柔軟にエージェントを構成できるのが魅力です。
今回の記事では、このLangGraphを使って、代表的なエージェントアーキテクチャのうち、「単一エージェント(Single Agent)」について解説していきます。実際にGoogle Colab上で動かせる事例も紹介します。
LangGraphとは?
LangGraphは、LangChainエコシステム上に構築されたエージェントフレームワークです。
従来のLangChainエージェントは、線形的な思考パターンが主流でしたが、LangGraphでは、状態遷移グラフの構造を活用することで、複雑な制御構造を持つワークフローを柔軟に表現することができます。
LangChainではチェーン構造によって直線的なワークフローの表現に適したグラフ構造を表現することができました。一方LangGraphでは、チェーンに限らず 一般的なグラフ構造(分岐やループを含む) を用いることで、より柔軟で複雑な制御フローを表現できます。
ちなみにここでいうグラフとは、ノード(頂点) と エッジ(辺) からなる要素間の関係性を表現するもののことです。
LangGraphが扱うグラフ構造には、ノードとエッジに制限がないことから、さまざまな構造を柔軟に表現することを可能としています。
LangGraphを活用した代表的なエージェントアーキテクチャ
LangGraphでは、設計の柔軟性を活かして、さまざまなエージェントの構成を実現できます。代表的なアーキテクチャは以下の3つです。
単一エージェント(Single Agent)
もっともシンプルな構成のエージェントです。1つのエージェントが入力に対して推論・ツール呼び出し・最終出力までをすべて担うアーキテクチャになります。
水平アーキテクチャ(Horizontal Agent Composition)
複数のエージェントが対等な条件で、並列または条件分岐で動作します。水平アーキテクチャは、協調、フィードバック、グループディスカッションなどのタスクで有効です。
垂直アーキテクチャ(Vertical Agent Composition)
上位のエージェント(リーダー)が下位のエージェントを順に制御する構成です。トップレベルの司令塔エージェントが「どのエージェントに何を任せるか」を管理することで、フロー全体を戦略的に制御できます。
そこで今回は、この中でも一番シンプルで理解しやすい「単一エージェント」構成を詳しく見ていきます。
単一エージェントとは?
単一エージェントアーキテクチャは、LangGraphでのエージェント設計において最も基本的な構成です。1つのエージェントが一連の思考とツール呼び出しを繰り返しながら、最終的な回答を導き出します。
このパターンは以下のような特徴を持っています。
-
シンプルな構造:1つのエージェントが全責任を担う
-
導入コストが低い:LangGraphの入門として最適
-
ツール使用の判断も自動化:LLMの出力によって制御される
それでは、ここまでの説明を踏まえて、実際にLangGraphで単一エージェントを構築して動かしてみましょう!
Google Colab上で動かせるコードを紹介するので、ぜひ一緒に試してみてください。
1. 必要なライブラリのインストール
!pip install -q langgraph langchain langchain_openai langchain_community openai
2. OpenAI APIキーの設定
import os
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get('OPENAI_API_KEY')
3. 単一エージェントのLangGraph構築
では実際に、単一エージェントについてLangGraphで構築したコードを紹介します。
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, BaseMessage
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import json
import numexpr
# 計算ツールの定義
@tool
def multiply(expression: str) -> str:
"""与えられた数式(例: 5 * 7)を計算して結果を返す"""
try:
result = numexpr.evaluate(expression).item()
return str(result)
except Exception as e:
return f"計算中にエラーが発生しました: {e}"
# 使用可能なツールを辞書形式で管理
tools = {multiply.name: multiply}
# 状態の定義(LangGraphでは状態を明示的に持つ)
class AgentState(TypedDict):
messages: List[BaseMessage] # ユーザー/エージェント/ツール間の対話履歴
# LLM の初期化とツールのバインド
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tool = llm.bind_tools(list(tools.values()))
# エージェントノード:LLMが次に取るべきアクション(ツール呼び出し or 最終回答)を考える
def agent_node(state: AgentState) -> AgentState:
messages = state["messages"]
response = llm_with_tool.invoke(messages)
messages.append(response)
return {"messages": messages}
# ツール実行ノード:tool_calls を検出して、該当ツールを実行し、その結果を ToolMessage として追加
def tool_node(state: AgentState) -> AgentState:
last_message = state["messages"][-1]
tool_calls = getattr(last_message, "tool_calls", [])
for call in tool_calls:
tool_name = call["name"]
arguments = call.get("args") or call.get("arguments", {})
if isinstance(arguments, str):
arguments = json.loads(arguments)
tool_func = tools.get(tool_name)
if tool_func:
result = tool_func.invoke(arguments)
state["messages"].append(
ToolMessage(tool_call_id=call["id"], content=result)
)
return state
# 継続判定ノード:tool_calls が存在すればツール実行へ、なければ終了
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
# LangGraph のワークフロー定義
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
graph = workflow.compile()
# 実行
query = input("query: ")
state = {"messages": [HumanMessage(content=query)]}
final_state = graph.invoke(state)
# 結果出力:ツールの出力があればそれを優先的に表示
for msg in reversed(final_state["messages"]):
if isinstance(msg, ToolMessage):
print("計算結果:", msg.content)
break
else:
print("単一エージェントの回答:", final_state["messages"][-1].content)
結果:
query: 120 * 100
計算結果: 12000
LLMと計算ツールの定義
上記のコード例では、OpenAIの gpt-4o-mini
モデルを使ってLLMを初期化します。
また、計算用ツールとして numexpr
を使ったmultiply関数を LangChain の @tool デコレータで定義しツールとして登録しています。
LLMとツールのバインド
LangGraphでは llm.bind_tools
を使って、LLMとツール群を直接バインドできます。
これにより、LLMがツール呼び出しを判断し、自動で適切なツールを選択できるようになります。
状態の定義とノード処理
LangGraphでは、各ノード(エージェント・ツール・終了条件など)が明示的な「状態(State)」を受け取り・返すことで、制御の流れを構築します。
今回の例では、状態として messages(やりとりの履歴)を持つ AgentState を定義しています。
ツール実行ノードの実装
LangGraphでは、エージェントがツールを使うべきと判断した場合、tool_calls
という形式でその呼び出しが出力されます。
その呼び出しに応じて、実際のツールを実行するノードを用意する必要があります。
このノードでは以下の流れで処理を行います。
- 直前のLLM出力(messages[-1])から tool_calls を取得
- 呼び出されたツール名と引数を読み取り
- 該当するツール関数を実行
- 結果を
ToolMessage
として状態に追加
こうすることで、LLMの「Action」に対応する「Observation(=ツール実行結果)」を次に渡すことができます。
継続判定ノードの実装
LangGraphは状態遷移グラフなので、「次にどのノードに進むか」を制御するための分岐条件を設けることができます。
ここではシンプルに、
LLMの出力に tool_calls が含まれていれば → tools ノードへ
それ以外(ツールを呼び出していない) → 終了
という条件分岐を should_continue()
関数で定義しています。これにより、「ツールを実行してから再度LLMに戻る」というループが可能になります。
LangGraphのワークフロー構築
ここまでに定義したノード(agent、tools、should_continue)を使って、実際のグラフ構造を構築していきます。
-
add_node()
:各ステップの関数をノードとして登録 -
set_entry_point()
:ワークフローの開始地点を指定 -
add_conditional_edges()
:LLM出力に応じた分岐の定義 -
add_edge()
:通常の遷移ルート(tools → agent)を明示 -
compile()
:グラフ全体を実行可能なワークフローに変換
このようにLangGraphでは、制御構造そのものを グラフとして視覚的・論理的に定義できるのが特徴です。
エージェントの実行
ここでは、入力を HumanMessage として初期状態(AgentState)に渡し、LangGraphのワークフローを実行しています。
この graph.invoke()
の中で、以下の一連の処理が行われています。
- LLMがツールを呼び出すか判断
- 必要に応じてツールを実行
- 結果を受け取ってLLMが最終回答を生成
- 条件に応じて処理を終了
この一連の流れを状態(State)を介しながら、滑らかに進行させるのがLangGraphの魅力です。
結果出力
最後に「120 * 100」を入力値として結果を出力していますが、ここでは少しテクニカルなことをしています。
通常LLMはツールの結果をもとに「自然な文章でまとめた最終回答」を返します。しかし、ユーザーが計算結果だけ(数値)を欲しい場合に、LLMのまとめを経由すると余計な装飾が入ってしまいます。
例として:
自然な回答例: The result of 120 * 100 is 12,000.
必要な値: 12000
このようなケースでは、LangGraphの状態(messages)から ToolMessage の出力だけを取り出すことで、必要な値のみ(12000)を取得できます。
おわりに
ここまで読んでくださってありがとうございました!
今回は、LangGraphの基本である単一エージェント構成について、実際に手を動かしながら理解してみました。LangChainとOpenAIの組み合わせに、LangGraphという「状態管理」が加わることで、より洗練されたエージェントが作れるようになりました!
次の記事では、複数のエージェントが連携して動作する「水平アーキテクチャ」を取り上げる予定です。
次回もよろしくお願いします!