マルチーエージェントの基礎を学ぶ（LangGraphのNode、SupervisorAgent）

Last updated at 2025-07-28Posted at 2025-07-26

概要

マルチエージェントシステムについて調査した際の備忘録
LangGraph を使用

LangGraph について

LangGraph は以下のようなフレームワーク
- LangChain をベースにしていて LLM を活用した複雑な処理フローやマルチエージェントシステムを構築出来る
- グラフ構造を用いて柔軟で視覚的なワークフロー設計が可能
  - グラフと言われると GraphDB とかを思い描いてなんかややこしそう、となるかもしれないですが、そこまでややこしくは無さそうです

LangGraph の要素	説明
State	会話履歴や処理結果などを保持し、ノード間で共有・更新出来る。これにより、長い対話や複数ステップの処理が安定して行える。
Node	実際の処理を行う関数やエージェント。Stateを参照・更新出来る
Edge	ノード間の接続。「あらかじめ定義したノードに遷移」「条件に応じて次のノードを選択」「一定条件で繰り返す」など柔軟なノード遷移の設定が可能。Edge を定義せず LangGraph の goto の関数でも遷移可能
Graph	Node と Edge で構成される全体の処理フロー

例として以下のようなフローを考える

機能

ユーザの年齢を聞くと年齢を答える
ユーザ２人の年齢差を聞くと年齢差を答える

ツール

ユーザ情報（生年月日）取得ツール
年齢計算ツール
年齢比較ツール

シングルエージェントにツールを持たせた時のイメージ

エージェントはユーザの会話を理解してどのツールをどの順序で使うか判断する
そのためのプロンプトも細かく書く必要がある

プロンプトの例

# あなたは以下に書かれた「# タスク」を行うエージェント。
- タスクを行う際は「# 使用可能なツール」から選んで使う。
- 「# 条件」を厳守すること。

# タスク
## ユーザの年齢を計算
- 「ユーザ情報取得ツール」でユーザ情報を取得する。
- 「年齢計算ツール」でユーザの年齢を計算して、回答する。
## 2人のユーザの年齢差を計算
- 「ユーザ情報取得ツール」で2人分のユーザ情報を取得する。
- 「年齢計算ツール」で2人分の年齢を計算する。
- 「年齢比較ツール」で年齢差を計算して、回答する。

# 使用可能なツール
- ユーザ情報取得ツール：ユーザ名を入力して、ユーザ情報（生年月日）を取得する。
- 年齢計算ツール：生年月日を入力して、現在の年齢を計算する。
- 年齢比較ツール：年齢を2つ入力して、年齢差を計算する。

# 条件
- ツールを使わずに「年齢」「年齢差」を計算する事を禁止する。
- 「# タスク」に書かれている事以外は、「分かりません」と回答する。

LangGraph の Node でフローを定義した時のイメージ

エージェントはユーザとの会話を通してどの「ノード（処理）」に遷移するか判断して、ノード内である程度絞られた処理を実行する
- プロンプトを書く際に1つの処理に注力出来るのでプロンプトがコンパクトになる
- ただしリクエストを複数回実行する必要がある
ノードを振り分けた後の処理も固定化され（その時のフローにもよるが）、LLM による判断が減る（ツールの実行順序などの判断を無くせたり）

プロンプトの例

以下のようにノードごと（実際はノードの中でいくつかプロンプトを使う事もある）にプロンプトを分けて考える事が出来る
年齢計算、年齢比較ノード内でユーザ名を抽出した後の処理は固定化されるので LLM による思考は不要
実際の実装では with_structured_output を使うので、プロンプトはもっとシンプルに書けると思います

処理振り分けノード

# あなたは同僚にタスクを依頼するエージェントです。
# 依頼するタスクは以下の2つから選択して下さい。
- タスク名：年齢計算タスク
- タスク名：年齢比較タスク
- タスク名以外の回答は不要です。

年齢計算ノード

ユーザ名を抽出するためのプロンプトを記載

# あなたはユーザとの会話から年齢を計算するユーザ名を抽出するエージェントです。
- 「さん」などの敬称は不要です。例：Aさん → A
- ユーザ名以外の回答は不要です。

年齢比較ノード

ユーザ名を2人分抽出するためのプロンプトを記載

# あなたはユーザとの会話から年齢を比較するユーザ名を2人分抽出するエージェントです。
- 「さん」などの敬称は不要です。例：AさんとBさん → A, B
- ユーザ名以外の回答は不要です。

LangGraph の SupervisorAgent のイメージ

SupervisorAgent はタスク内容を理解して、専門のエージェントにタスクを依頼する
振り分けられた専門のエージェントはタスクを実施して、SupervisorAgent に結果を返す

SupervisorAgent の種類

以下のような種類がある
- 専門のエージェントに会話履歴を全て渡すタイプ
- SupervisorAgent が会話履歴からタスク（文章）を考えて、専門エージェントに渡すタイプ
会話履歴を渡すタイプはトークン数がどんどん増えてしまうが、タスク実行時の会話漏れが減りそう
SupervisorAgent が考えたタスクを渡すタイプは専門エージェントに渡るプロンプトはコンパクトになるが、SupervisorAgent が会話を理解して正確で漏れが無いタスクを依頼させる事が出来るかどうかの大変さがある

プロンプトの例

エージェントごとにプロンプトを分けて考える
ツールを使う順序などはエージェントが考える事になるので、タスク数が少ない場合はシングルエージェントと大きな差が無さそう
また今回の例では被った機能を持つエージェントが出来てしまったので、ちょっと例が微妙？

SupervisorAgent

# あなたは以下の「# 同僚エージェント」に仕事を依頼するSupervisorAgentです。

# 同僚エージェント
- 年齢計算エージェント：年齢計算のタスクを実施します
- 年齢比較計算エージェント：2人の年齢を比較するタスクを実施します

# タスク
## ユーザの年齢を計算
## 2人のユーザの年齢差を計算

# 条件
- 一度に複数のエージェントを呼び出す事は禁止です。
- あなた自身の思考で年齢や年齢比較をしないで下さい。
- 「# タスク」に書かれている事以外は、「分かりません」と回答する。

年齢計算エージェント

# あなたはユーザの年齢を計算するエージェント。
- 「# 手順」「# 条件」を厳守すること。

# 手順
1. 「ユーザ情報取得ツール」でユーザ情報を取得する。
2. 「年齢計算ツール」でユーザの年齢を計算して、回答する。

# ツール
- ユーザ情報取得ツール：ユーザ名（敬称不要）を入力して、ユーザ情報（生年月日）を取得する。
- 年齢計算ツール：生年月日（YYYY/MM/DD）を入力して、現在の年齢を計算する。

# 条件
- ツールを使わずに「年齢」を計算する事を禁止する。

年齢比較エージェント

# あなたはユーザ2人の年齢を比較するエージェント。
- 「# 手順」「# 条件」を厳守すること。

# 手順
1. 「ユーザ情報取得ツール」で2人分のユーザ情報を取得する。（必要があれば2回実行）
2. 「年齢計算ツール」で2人分の年齢を計算する。（必要があれば2回実行）
3. 「年齢比較ツール」で年齢差を計算して、回答する。

# 使用可能なツール
- ユーザ情報取得ツール：ユーザ名（敬称不要）を入力して、ユーザ情報（生年月日）を取得する。
- 年齢計算ツール：生年月日（YYYY/MM/DD）を入力して、現在の年齢を計算する。
- 年齢比較ツール：年齢を2つ入力して、年齢差を計算する。

# 条件
- ツールを使わずに「年齢」「年齢差」を計算する事を禁止する。

通常の LLM リクエスト

LangGraph でやっている事は LLM の判断結果で分岐・繰り返しをしたり、結果をオブジェクトに入れて更新したり基本的な事をしている感じなので、通常の LLM リクエストだけでやろうと思えば出来そう
- Node 間の遷移（Edge）は if、Node のループも for、State もユーザ自身で管理出来なくはない
ただ LangGraph （フレームワーク）を使う事で上記のようなややこしい処理を一貫性を保って実装する事が出来たり、処理フローを整理しやすくなるので積極的に利用した方が良い

実装例

共通

requirements.txt

gradio==5.37.0
langgraph==0.5.3
langchain==0.3.26
langgraph-supervisor==0.0.28
langchain-google-vertexai==2.0.27
langchain-openai==0.3.28 # エラーの検証で使った。なくても良い
python-dotenv==1.1.1

LangGraph Node

コード

ファイル（.env とか requirements.txt とかは省略）
- app.py
- actions/chat.py
- _langgraph/graph/langgraph_flow.py
- _langgraph/graph/node/routing_node.py
- _langgraph/graph/node/age_calculator_node.py
- _langgraph/graph/node/age_difference_node.py
- _langgraph/graph/func/age_func.py
- _langgraph/graph/func/user_info.py

app.py

import gradio as gr
from actions.chat import send

with gr.Blocks() as demo:
    chatbot = gr.Chatbot(
        show_label=False,
        type="messages"
    )
    
    chat_input = gr.Textbox(
        show_label=False,
        submit_btn=True
    )
    
    chat_msg = chat_input.submit(
        send, [chatbot, chat_input], [chatbot, chat_input]
    )

demo.launch()

actions/chat.py

from langchain_google_vertexai import ChatVertexAI

from dotenv import load_dotenv

from _langgraph.graph.langgraph_flow import build_graph

load_dotenv()

def send(
        messages,
        user_input,
    ):
    # 環境変数で GOOGLE_APPLICATION_CREDENTIALS="gcp_cred.json" を指定済み
    # llm = ChatVertexAI(model="gemini-2.5-flash") # 性能が低くユーザ名抽出に難あり
    llm = ChatVertexAI(model="gemini-2.5-pro")

    # 会話に今回のユーザ入力を追加
    messages.append({"role": "user", "content": user_input})
    # print(messages)
    # グラフに渡す State を作成
    state={
        "messages": messages,
        "user_input": user_input,
        "llm": llm,
    }
    # グラフの作成
    graph = build_graph()
    # グラフの呼び出し
    state_response = graph.invoke(state)

    # graph から返却された state > messages の末尾を会話履歴に追加
    messages.append({"role": "assistant", "content": state_response["messages"][-1].content})

    return messages, ""

_langgraph/graph/langgraph_flow.py

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_google_vertexai import ChatVertexAI

from _langgraph.graph.node.routing_node import routing_node
from _langgraph.graph.node.age_calculator_node import age_calculator_node
from _langgraph.graph.node.age_difference_node import age_difference_node

class State(TypedDict):
    messages: Annotated[list, add_messages]
    user_input: str
    llm: ChatVertexAI

def build_graph():
    # グラフビルダーを定義
    graph_builder = StateGraph(State)

    # グラフにノードを追加
    graph_builder.add_node("routing_node", routing_node) # 処理分岐ノード
    graph_builder.add_node("age_calculator_node", age_calculator_node) # 年齢計算ノード
    graph_builder.add_node("age_difference_node", age_difference_node) # 年齢差計算ノード

    # エッジを追加
    graph_builder.add_edge(START, "routing_node")

    # グラフをコンパイル
    graph = graph_builder.compile()
    return graph

_langgraph/graph/node/routing_node.py

from typing import Literal
from pydantic import BaseModel, Field

from langgraph.graph import END
from langgraph.types import Command

class Routing(BaseModel):
    """あなたの回答と次に行う処理のルーティング"""
    # gemini では Literal を使うとエラーになる？
    llm_response: str = Field(..., description="あなた自身の会話に対する回答。'age_calculator' or 'age_difference' の時は '' とだけ回答する。")
    route: str = Field(..., description="次に行う処理. 'normal_chat' or 'age_calculator' or 'age_difference'")

# 会話から次のノードを決定するノード
def routing_node(state: dict):
    llm = state["llm"]
    messages = state["messages"]
    # 分岐判断
    routing_prompt=[]
    routing_system_prompt="""あなたは以下の処理のルーティングを確実に行って下さい。
なお、名前の情報があれば遷移先で生年月日は取得出来るものとする。
年齢計算：age_calculator
年齢の差計算：age_difference
それ以外：normal_chat"""
    routing_prompt.extend([{"role":"system", "content": routing_system_prompt}])
    routing_prompt.extend(messages)
    routing_response = llm.with_structured_output(Routing).invoke(routing_prompt)
    print(f"routing_response: {routing_response.route}")

    # LLM の判断によって分岐
    match routing_response.route:
        case "age_calculator":
            return Command(goto="age_calculator_node")
        case "age_difference":
            return Command(goto="age_difference_node")
        case _:
            # 分岐無しの場合は "llm_response" を messages の末尾に追加して返却して終了
            state["messages"].append({"role": "assistant", "content": routing_response.llm_response})
            return Command(goto=END, update=state)

_langgraph/graph/node/age_calculator_node.py

# from typing import Literal, List
from pydantic import BaseModel, Field

from langgraph.graph import END
from langgraph.types import Command

from _langgraph.graph.func.user_info import user_info
from _langgraph.graph.func.age_func import get_age

class User(BaseModel):
    # gemini では Literal を使うとエラーになる？
    name: str = Field(..., description="抽出したユーザ名。敬称不要。例：X君 → X")

# ユーザの年齢を調べます
def age_calculator_node(state: dict):
    print("年齢計算ノード")
    llm = state["llm"]
    messages = state["messages"]
    
    # 名前の抽出
    user_info_prompt=[]
    user_info_prompt.extend([{"role":"system", "content":"会話から年齢を計算するユーザ名を抽出して下さい。"}])
    user_info_prompt.extend(messages)
    user_info_response = llm.with_structured_output(User).invoke(messages)
    print(f"user_info_response: {user_info_response.name}")

    # ユーザ情報（生年月日）の取得
    ymd = user_info(user_info_response.name)
    print(f"ymd: {ymd}")
    # 年齢の計算
    age = get_age(ymd)
    print(f"age: {age}")

    state["messages"].append({"role": "assistant", "content": f"{user_info_response.name} さんの年齢：{age}"})
    return Command(goto=END, update=state)

_langgraph/graph/node/age_difference_node.py

from typing import Literal, List
from pydantic import BaseModel, Field

from langgraph.graph import END
from langgraph.types import Command

from _langgraph.graph.func.user_info import user_info
from _langgraph.graph.func.age_func import get_age, age_diff

class User(BaseModel):
    # gemini では Literal を使うとエラーになる？
    name: List[str] = Field(..., description="抽出したユーザ名（2人）。敬称不要。例：XさんYさん → [X, Y]")

# 年齢差を調べます
def age_difference_node(state: dict):
    print("年齢比較ノード")
    llm = state["llm"]
    messages = state["messages"]
    
    # 年齢比較する二人の名前を抽出
    user_info_prompt=[]
    user_info_prompt.extend([{"role":"system", "content":"会話から年齢を比較するユーザを2名抽出して下さい。"}])
    user_info_prompt.extend(messages)
    user_info_response = llm.with_structured_output(User).invoke(messages)
    print(f"user_info_response: {user_info_response.name}")

    # ユーザ情報（生年月日）の取得
    ymd1 = user_info(user_info_response.name[0])
    ymd2 = user_info(user_info_response.name[1])
    print(f"ymd1: {ymd1}")
    print(f"ymd2: {ymd2}")
    # 年齢の計算
    age1 = get_age(ymd1)
    age2 = get_age(ymd2)
    # 年齢差の計算
    diff = age_diff(age1, age2)
    print(f"age1: {age1}")
    print(f"age2: {age2}")

    state["messages"].append({"role": "assistant", "content": f"年齢差：{diff}"})
    return Command(goto=END, update=state)

_langgraph/graph/func/age_func.py

from dateutil import parser
from dateutil.relativedelta import relativedelta
from datetime import date

# 生年月日を受け取って年齢を返す
def get_age(ymd: str):
    birth = parser.parse(ymd).date()
    # 年齢を計算
    age = relativedelta(date.today(), birth)
    return age.years

# 年齢を2つ受け取って差を返す
def age_diff(age1: int, age2: int):
    # 年齢差を計算
    diff = abs(age1-age2)
    return diff

_langgraph/graph/func/user_info.py

# ユーザの名前を受け取って、生年月日を返却する
user_data={
    "A": {
        "ymd": "1990/01/01"
    },
    "B": {
        "ymd": "1985/05/05"
    }
}
def user_info(user_name: str):
    return user_data[user_name]["ymd"]

実行結果

画面・ログ
1回目のチャット
- Aさんの年齢を聞いたので「処理分岐ノード」→「年齢計算ノード」に遷移する
- ユーザ名「A」を抽出
  → 生年月日を取得
  → 年齢を計算
2回目のチャット
- 続けてBさんとの年齢差を聞いたので「処理分岐ノード」→「年齢比較ノード」に遷移する
- 会話履歴から A さんとの比較という事を読み取ってユーザ名「A」と「B」を抽出
  → 2人の生年月日を取得
  → 2人の年齢を計算
  → 年齢差を計算

SupervisorAgent

Gemini の留意点

SupervisorAgent で Gemini を使っていて messages のラストが "assistant" の状態で LLM にリクエストが送られた時の挙動で気になる点があった（2025/07/25 時点）
- SupervisorAgent の実行結果になぜ起きたかは記載
例としては以下のように AzureOpenAI は最後の "assistant" を無視するようなTemplateが用意されている？などで直前の会話履歴に関わらず常にフルのテキストを生成するが、
Gemini は最後の "assistant" に続ける形でメッセージを生成しようとするため、開発時に留意する必要がある
- チャット欄に表示するメッセージリストを更新する際に messages.append({"role": "assistant", "content": state_response["messages"][-1].content}) とすると、「30歳、Aさん」ではなく「ん、 30歳」となってしまう。
イメージ画像
コード

test.py

from langchain_google_vertexai import ChatVertexAI
from langchain_openai import AzureChatOpenAI
from dotenv import load_dotenv

load_dotenv()

messages=[
    {"role":"system", "content": "あなたはユーザの名前と年齢を返します。出力例（順序は逆でも良いです）：XXXさん、v歳"},
    {"role":"user", "content": "私は A です。30歳です。"},
    {"role":"assistant", "content": "Aさ"},
]
llm = AzureChatOpenAI(deployment_name="gpt-4.1")
r = llm.invoke(messages)
print(f"AzureOpenAI: {r.content}")

llm = ChatVertexAI(model="gemini-2.5-pro")
r = llm.invoke(messages)
print(f"gemini: {r.content}")

出力例

AzureOpenAI: 30歳、Aさん
gemini:  ん、 30歳

system prompt を変えて試したが、Gemini が続きから生成する部分には影響しなさそう

コード

ファイル（.env とか requirements.txt とかは省略）
- app.py
- actions/chat_supervisor.py
- _langgraph/supervisor/supervisor_agent.py
- _langgraph/supervisor/tools/age_func.py
- _langgraph/supervisor/tools/handoff.py
- _langgraph/supervisor/tools/user_info.py

app.py

import gradio as gr
from actions.chat import send
from actions.chat_supervisor import send as send_supervisor

with gr.Blocks() as demo:
    chatbot = gr.Chatbot(
        show_label=False,
        type="messages"
    )
    
    chat_input = gr.Textbox(
        show_label=False,
        submit_btn=True
    )
    
    # chat_msg = chat_input.submit(
    #     send, [chatbot, chat_input], [chatbot, chat_input]
    # )
    chat_msg = chat_input.submit(
        send_supervisor, [chatbot, chat_input], [chatbot, chat_input]
    )

demo.launch()

actions/chat_supervisor.py

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_google_vertexai import ChatVertexAI

from dotenv import load_dotenv

from _langgraph.supervisor.supervisor_agent import (
    build_graph_create_supervisor, 
    build_graph_supervisor_agent,
    build_graph_supervisor_agent_with_description,
)

load_dotenv()

def send(
        messages: list[dict],
        user_input: str,
    ):
    # 環境変数で GOOGLE_APPLICATION_CREDENTIALS="gcp_cred.json" を指定済み
    # llm = ChatVertexAI(model="gemini-2.5-flash")
    llm = ChatVertexAI(model="gemini-2.5-pro")

    # 会話に今回のユーザ入力を追加
    messages.append({"role": "user", "content": user_input})
    # print(messages)
    # グラフに渡す State を作成
    state={
        "messages": messages,
    }
    # グラフの作成（3種試す）
    supervisor = build_graph_create_supervisor(llm)
    # supervisor = build_graph_supervisor_agent(llm)
    # supervisor = build_graph_supervisor_agent_with_description(llm)
    # グラフの呼び出し
    state_response = supervisor.invoke(state)

    for message in state_response["messages"]:
        if isinstance(message, HumanMessage):
            print(f"Input Message: {message.content}")
        elif isinstance(message, AIMessage): # and message.content != "":
            print(f"{message.name} Message: {message.content}")
        elif isinstance(message, ToolMessage) and message.content != "":
            print(f"Tool Message: {message.content}")

    # graph から返却された state > messages の末尾を会話履歴に追加
    messages.append({"role": "assistant", "content": state_response["messages"][-1].content})

    return messages, ""

_langgraph/supervisor/supervisor_agent.py

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.managed import  RemainingSteps
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import AnyMessage, add_messages
from langchain_google_vertexai import ChatVertexAI

from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

from _langgraph.supervisor.tools.age_func import get_age, age_diff
from _langgraph.supervisor.tools.user_info import user_info
from _langgraph.supervisor.tools.handoff import create_handoff_tool, create_task_description_handoff_tool

class CustomState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    remaining_steps: RemainingSteps = 5

def create_agents(llm: ChatVertexAI):
    age_calc_agent = create_react_agent(
        model=llm,
        tools=[user_info, get_age],
        prompt=(
            "# あなたはユーザの「年齢計算」タスクを行うエージェントです。# タスクの手順 を参考にして下さい。\n"
            "- # 条件 を厳守して下さい。"
            "\n"
            "# タスクの手順:\n"
            "1. ユーザ名を入力として「user_info_tool」でユーザの生年月日を取得する。\n"
            "2. ユーザの生年月日を入力として「age_calc_tool」でユーザの年齢を取得する。\n"
            "\n"
            "# 条件:\n"
            "- タスクが完了したら Supervisor に年齢を回答して下さい。\n"
            "- ツールで取得した年齢のみ回答して下さい。例：30"
            "- ツールを同時に複数呼び出す事は禁止です。"
            "- 「年齢計算」タスクのみ行って下さい。"
        ),
        name="age_calc_agent",
    )
    age_diff_agent = create_react_agent(
        model=llm,
        tools=[user_info, get_age, age_diff],
        prompt=(
            "# あなたは2人のユーザの「年齢差の計算」タスクを行うエージェントです。# タスクの手順 を参考にして下さい。\n"
            "- # 条件 を厳守して下さい。"
            "\n"
            "# タスクの手順:\n"
            "1. ユーザ名を入力として「user_info_tool」でユーザの生年月日を取得する。\n"
            "2. ユーザの生年月日を入力として「age_calc_tool」でユーザの年齢を取得する。\n"
            "3. 比較する全ユーザの年齢が取得出来るまで、手順 1. と 2. を繰り返す\n"
            "4. 比較対象ユーザの年齢を入力として「age_diff_tool」で年齢差を計算する。\n"
            "\n"
            "# 条件:\n"
            "- タスクが完了したら Supervisor に年齢差を回答して下さい。\n"
            "- ツールで取得した年齢差のみ回答して下さい。例：3"
            "- ツールを同時に複数呼び出す事は禁止です。"
            "- 「年齢差計算」タスクのみ行って下さい。"

        ),
        name="age_diff_agent",
    )
    return age_calc_agent, age_diff_agent

# LangGraph の create_supervisor()を使用した SupervisorAgent（一番シンプル）
def build_graph_create_supervisor(
        llm: ChatVertexAI
    ):
    # 同僚エージェントの作成
    age_calc_agent, age_diff_agent = create_agents(llm)
    # SupervisorAgent の作成
    supervisor = create_supervisor(
        model=llm,
        agents=[age_calc_agent, age_diff_agent],
        prompt=(
            "# あなたは2人の同僚エージェントに指示を出す SupervisorAgent です。\n"
            "- # 条件 を厳守して下さい。"
            "\n"
            "# INSTRUCTIONS:\n"
            "- age_calc_agent には「年齢計算」のタスクを依頼して下さい。\n"
            "- age_diff_agent には「年齢差計算」のタスクを依頼して下さい。\n"
            "\n"
            "# 条件:\n"
            "- エージェントを並行して呼び出さないでください。\n"
            "- 一度にタスクを依頼出来るのは1人のエージェントのみです。\n"
            "- 自分でタスクを行わないでください。\n"
        ),
        add_handoff_back_messages=True,
        output_mode="full_history",
        state_schema=CustomState
    ).compile()
    return supervisor

# handoff ツール を使用する SupervisorAgent
def build_graph_supervisor_agent(
        llm: ChatVertexAI
    ):
    # 同僚エージェントの作成
    age_calc_agent, age_diff_agent = create_agents(llm)
    # Handoffs
    assign_to_age_calc_agent = create_handoff_tool(
        agent_name="age_calc_agent",
        description="年齢計算タスクを依頼します。",
    )
    assign_to_age_diff_agent = create_handoff_tool(
        agent_name="age_diff_agent",
        description="年齢差計算タスクを依頼します。",
    )
    supervisor_agent = create_react_agent(
        model=llm,
        tools=[assign_to_age_calc_agent, assign_to_age_diff_agent],
        prompt=(
            "# あなたは2人の同僚エージェントに指示を出す SupervisorAgent です。\n"
            "- # 条件 を厳守して下さい。"
            "\n"
            "# INSTRUCTIONS:\n"
            "- age_calc_agent には「年齢計算」のタスクを依頼して下さい。\n"
            "- age_diff_agent には「年齢差計算」のタスクを依頼して下さい。\n"
            "\n"
            "# 条件:\n"
            "- エージェントを並行して呼び出さないでください。\n"
            "- 一度にタスクを依頼出来るのは1人のエージェントのみです。\n"
            "- 自分でタスクを行わないでください。\n"
        ),
        name="supervisor",
    )
    supervisor = (
        StateGraph(CustomState)
        # NOTE: `destinations` is only needed for visualization and doesn't affect runtime behavior
        .add_node(supervisor_agent, destinations=("age_calc_agent", "age_diff_agent", END))
        .add_node(age_calc_agent)
        .add_node(age_diff_agent)
        .add_edge(START, "supervisor")
        # always return back to the supervisor
        .add_edge("age_calc_agent", "supervisor")
        .add_edge("age_diff_agent", "supervisor")
        .compile()
    )
    return supervisor

# SupervisorAgent が考えたタスクの依頼文のみ受け渡すタイプの SupervisorAgent
def build_graph_supervisor_agent_with_description(
        llm: ChatVertexAI
    ):
    # 同僚エージェントの作成
    age_calc_agent, age_diff_agent = create_agents(llm)
    # Handoffs
    assign_to_age_calc_agent = create_task_description_handoff_tool(
        agent_name="age_calc_agent",
        description="年齢計算タスクを依頼します。ユーザ名が必要です。",
    )
    assign_to_age_diff_agent = create_task_description_handoff_tool(
        agent_name="age_diff_agent",
        description="年齢差計算タスクを依頼します。ユーザ名が必要です。",
    )
    supervisor_agent = create_react_agent(
        model=llm,
        tools=[assign_to_age_calc_agent, assign_to_age_diff_agent],
        prompt=(
            "# あなたは2人の同僚エージェントに指示を出す SupervisorAgent です。\n"
            "- # 条件 を厳守して下さい。"
            "\n"
            "# INSTRUCTIONS:\n"
            "- age_calc_agent には「年齢計算」のタスクを依頼して下さい。\n"
            "- age_diff_agent には「年齢差計算」のタスクを依頼して下さい。\n"
            "\n"
            "# 条件\n"
            "- エージェントを並行して呼び出さないでください。\n"
            "- 一度にタスクを依頼出来るのは1人のエージェントのみです。\n"
            "- 自分でタスクを行わないでください。\n"
        ),
        name="supervisor",
    )
    supervisor = (
        StateGraph(CustomState)
        # NOTE: `destinations` is only needed for visualization and doesn't affect runtime behavior
        .add_node(supervisor_agent, destinations=("age_calc_agent", "age_diff_agent", END))
        .add_node(age_calc_agent)
        .add_node(age_diff_agent)
        .add_edge(START, "supervisor")
        # always return back to the supervisor
        .add_edge("age_calc_agent", "supervisor")
        .add_edge("age_diff_agent", "supervisor")
        .compile()
    )
    return supervisor

_langgraph/supervisor/tools/age_func.py

from dateutil import parser
from dateutil.relativedelta import relativedelta
from datetime import date
from langchain_core.tools import tool

# 生年月日を受け取って年齢を返す
@tool("age_calc_tool", description="年齢を計算する。")
def get_age(ymd: str):
    """生年月日を受け取って、年齢を計算する。
    Args:
        ymd: YYYY/MM/DD の生年月日

    """
    print(f"年齢計算ツール。入力: {ymd}")
    birth = parser.parse(ymd).date()
    # 年齢を計算
    age = relativedelta(date.today(), birth)
    return age.years

# 年齢を2つ受け取って差を返す
@tool("age_diff_tool", description="年齢差を計算する。")
def age_diff(age1: int, age2: int):
    """2人の年齢を入力して、年齢差を計算する。
    Args:
        age1: 1人目の年齢
        age2: 2人目の年齢
    """
    print(f"年齢差計算ツール。入力1: {age1}、入力2: {age2}")
    # 年齢差を計算
    diff = abs(age1-age2)
    return diff

_langgraph/supervisor/tools/handoff.py

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.types import Command, Send

# 会話履歴をそのまま同僚エージェントに渡す
def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto=agent_name,
            update={**state, "messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )

    return handoff_tool

# SupervisorAgent が考えたタスクの依頼文のみ同僚エージェントに渡す
def create_task_description_handoff_tool(
    *, agent_name: str, description: str | None = None
):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        # this is populated by the supervisor LLM
        task_description: Annotated[
            str,
            "Description of what the next agent should do, including all of the relevant context.",
        ],
        # these parameters are ignored by the LLM
        state: Annotated[dict, InjectedState],
    ) -> Command:
        task_description_message = {"role": "user", "content": task_description}
        agent_input = {**state, "messages": [task_description_message]}
        return Command(
            goto=[Send(agent_name, agent_input)],
            graph=Command.PARENT,
        )

    return handoff_tool

_langgraph/supervisor/tools/user_info.py

from pydantic import BaseModel, Field
from langchain_core.tools import tool

user_data={
    "A": {
        "ymd": "1990/01/01"
    },
    "B": {
        "ymd": "1985/05/05"
    }
}

class user(BaseModel):
    name: str = Field(..., description="ユーザ名。「さん」や「君」などの敬称不要。例：「太郎さん」の場合、「太郎」")

# ユーザの名前を受け取って、生年月日を返却する
@tool("user_info_tool", args_schema=user, description="ユーザの生年月日を取得する。")
def user_info(name: str):
    """ユーザ名を入力して、生年月日を返却する。
    Args:
        name: ユーザ名
    """
    print(f"ユーザ情報取得ツール。入力: {name}")
    return user_data[name]["ymd"]

実行結果

actions/chat_supervisor.py の以下の関数のコメントを付け替えて3種類試す
- build_graph_create_supervisor()
  - langgraph_supervisor ライブラリの create_supervisor 関数を使った一番シンプルなやり方
  - AI Message が連続しないため Gemini の留意点の問題も発生しなさそう
- build_graph_supervisor_agent()
  - SupervisorAgent が同僚エージェントに会話履歴ごとタスク移譲する handoff ツールを作成するやり方
  - 回答生成時に AI Message が連続するため、Gemini の留意点の問題が有る
とりあえずの対処として以下に変更してチャットした
actions/chat_supervisor.py
```
    messages.append({"role": "assistant", "content": state_response["messages"][-2].content + state_response["messages"][-1].content})
```
- build_graph_supervisor_agent_with_description()
  - SupervisorAgent が同僚エージェントにタスク移譲する handoff ツールを作成するやり方（会話履歴ではなく、タスクの依頼テキストのみを渡すやり方）

エージェントの失敗例

ツールの使用可否判断を誤ってユーザに聞き返す例
- ユーザ名「Bさん」があるにも関わらず、エージェントが「ユーザ情報取得ツール」を使用せずユーザに聞き返している。

まとめ

LangGraph ノードの方が処理を制御しやすい
（処理の実行順や、実行可否を自律的に動くエージェントより制御しやすい）
SupervisorAgent は自律的に動くので、年齢差を聞いた時に既に過去の会話から A さんの年齢が分かっている場合は無駄な処理（Aさんの年齢計算）を省略するなど、効率的な動作をする場合がある
（ただし LangGraph のノード処理でもユーザの年齢を State 管理すれば同じように省略する事が出来る。なので設計次第で同等の性能になるのでそこまでのアドバンテージではないと思う）
SupervisorAgent を使った方が自律的に動いている感は有るので、マルチエージェントを使っている感は強い（が外側から見ている分には分からないので精度高い LangGraph ノードで良い）
- ちなみに SupervisorAgent の実行例は思い通りに動かないパターンがそこそこあったので何度か実行し直しをしている。プロンプトを頑張ればパターンをもっと絞れると思うが、本来はもっとややこしい処理のはずなのでそうなると大変そう
現時点ではプロジェクトで使う場合は LangGraph のノードで進める方が安定はしそう

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up