LangGraphで学ぶエージェント設計の基本：単一エージェント構成をGoogle Colabで実装してみた！

こんにちは。

[株式会社KANGEN Holdings](https://kangen-holdings.co.jp/)の野中康生です！

生成AI・LLMの進化により、複雑なタスクをエージェントに任せる時代が本格化しています。その中でもぜひ注目したい存在が、**LangGraph** です。LangChainの拡張でありながら、「状態遷移」の考え方を取り入れて、より柔軟にエージェントを構成できるのが魅力です。

今回の記事では、このLangGraphを使って、代表的なエージェントアーキテクチャのうち、「**単一エージェント（Single Agent）**」について解説していきます。実際にGoogle Colab上で動かせる事例も紹介します。

# LangGraphとは？
[LangGraph](https://www.langchain.com/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キーの設定

```python
import os
from google.colab import userdata
os.environ["OPENAI_API_KEY"] = userdata.get('OPENAI_API_KEY')
```

# 3. 単一エージェントのLangGraph構築
では実際に、単一エージェントについてLangGraphで構築したコードを紹介します。

```python
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の <code>gpt-4o-mini</code> モデルを使ってLLMを初期化します。
また、計算用ツールとして <code>numexpr</code> を使ったmultiply関数を LangChain の @tool デコレータで定義しツールとして登録しています。

## LLMとツールのバインド

LangGraphでは <code>llm.bind_tools</code>を使って、LLMとツール群を直接バインドできます。
これにより、LLMがツール呼び出しを判断し、自動で適切なツールを選択できるようになります。

https://python.langchain.com/docs/how_to/tool_calling/#defining-tool-schemas

## 状態の定義とノード処理

LangGraphでは、各ノード（エージェント・ツール・終了条件など）が明示的な「**状態（State）**」を受け取り・返すことで、制御の流れを構築します。
今回の例では、状態として messages（やりとりの履歴）を持つ **AgentState** を定義しています。

## ツール実行ノードの実装
LangGraphでは、エージェントがツールを使うべきと判断した場合、<code>tool_calls</code>という形式でその呼び出しが出力されます。
その呼び出しに応じて、実際のツールを実行するノードを用意する必要があります。

このノードでは以下の流れで処理を行います。

1. 直前のLLM出力（messages[-1]）から tool_calls を取得
2. 呼び出されたツール名と引数を読み取り
3. 該当する**ツール関数**を実行
4. 結果を <code>ToolMessage</code> として状態に追加

こうすることで、LLMの「Action」に対応する「Observation（=ツール実行結果）」を次に渡すことができます。

## 継続判定ノードの実装
LangGraphは状態遷移グラフなので、「次にどのノードに進むか」を制御するための分岐条件を設けることができます。

ここではシンプルに、

LLMの出力に tool_calls が含まれていれば → tools ノードへ

それ以外（ツールを呼び出していない） → 終了

という条件分岐を <code>should_continue()</code>  関数で定義しています。これにより、「**ツールを実行してから再度LLMに戻る**」というループが可能になります。

## LangGraphのワークフロー構築
ここまでに定義したノード（agent、tools、should_continue）を使って、実際のグラフ構造を構築していきます。

- <code>add_node()</code> ：各ステップの関数をノードとして登録

- <code>set_entry_point()</code> ：ワークフローの開始地点を指定

- <code>add_conditional_edges()</code> ：LLM出力に応じた分岐の定義

- <code>add_edge()</code> ：通常の遷移ルート（tools → agent）を明示

- <code>compile()</code> ：グラフ全体を実行可能なワークフローに変換

このようにLangGraphでは、制御構造そのものを グラフとして視覚的・論理的に定義できるのが特徴です。

## エージェントの実行
ここでは、入力を HumanMessage として初期状態（AgentState）に渡し、LangGraphのワークフローを実行しています。

この <code>graph.invoke()</code>の中で、以下の一連の処理が行われています。

1. **LLMがツールを呼び出すか判断**
2. **必要に応じてツールを実行**
3. **結果を受け取ってLLMが最終回答を生成**
4. **条件に応じて処理を終了**

この一連の流れを状態（State）を介しながら、滑らかに進行させるのがLangGraphの魅力です。

## 結果出力
最後に「120 * 100」を入力値として結果を出力していますが、ここでは少しテクニカルなことをしています。
通常LLMはツールの結果をもとに「**自然な文章でまとめた最終回答**」を返します。しかし、ユーザーが計算結果だけ（数値）を欲しい場合に、LLMのまとめを経由すると余計な装飾が入ってしまいます。

例として：

自然な回答例: The result of 120 * 100 is 12,000.

必要な値: 12000

このようなケースでは、LangGraphの状態（messages）から **ToolMessage** の出力だけを取り出すことで、必要な値のみ（12000）を取得できます。

# おわりに

ここまで読んでくださってありがとうございました！

今回は、LangGraphの基本である単一エージェント構成について、実際に手を動かしながら理解してみました。LangChainとOpenAIの組み合わせに、LangGraphという「状態管理」が加わることで、より洗練されたエージェントが作れるようになりました！

次の記事では、複数のエージェントが連携して動作する「水平アーキテクチャ」を取り上げる予定です。

次回もよろしくお願いします！