【LangGraph】マルチエージェント　ハンドオフの使い分け

公式ドキュメントの、このチュートリアル部分です
https://langchain-ai.github.io/langgraph/how-tos/agent-handoffs/

公式ドキュメントではlangchain-anthropicを使っていますが、料金やツール呼び出しのスピード改善のためlangchain_openaiでgpt-4.1-nanoを使うように書き直しました。（ツールの呼び出し方が一部異なるため、modelを書き直しただけではエラーとなります）

コマンドによるハンドオフ

• エージェントノードが直接Commandオブジェクトを返して次のノードを指定します。

サンプルコード

from typing_extensions import Literal
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.types import Command

model = ChatOpenAI(model="gpt-4.1-nano",temperature=0.0)


@tool
def transfer_to_multiplication_expert():
    """乗算エージェントに助けを求める"""
    # このツールは何も返しません：LLMが別のエージェントに引き継ぎを求めるためのシグナルとして使用しています
    return

@tool
def transfer_to_addition_expert():
    """加算エージェントに助けを求める"""
    return


def addition_expert(
    state: MessagesState,
) -> Command[Literal["multiplication_expert", "__end__"]]:
    system_prompt = (
        "あなたは加算エージェントです。乗算エージェントに助けを求めることができます。"
        "引き継ぎの前に計算を行ってください。"
    )
    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    ai_msg = model.bind_tools([transfer_to_multiplication_expert]).invoke(messages)
    
    # APIの応答からツール呼び出しを確認
    # dict形式のai_msgからtool_calls属性を取得
    if hasattr(ai_msg, "additional_kwargs") and "tool_calls" in ai_msg.additional_kwargs:
        tool_calls = ai_msg.additional_kwargs["tool_calls"]
        if tool_calls:
            # 全てのtool_call_idに対するツールメッセージを作成
            tool_messages = []
            for tool_call in tool_calls:
                if tool_call.get("function", {}).get("name") == "transfer_to_multiplication_expert":
                    tool_call_id = tool_call["id"]
                    tool_msg = ToolMessage(
                        content="Successfully transferred",
                        tool_call_id=tool_call_id,
                    )
                    tool_messages.append(tool_msg)
            
            # ツール呼び出しがあり、それが乗算エージェントへの移動の場合
            if tool_messages:
                return Command(
                    goto="multiplication_expert", 
                    update={"messages": [ai_msg] + tool_messages}
                )

    # 専門家が回答を持っている場合は、ユーザーに直接返す
    return Command(goto="__end__", update={"messages": [ai_msg]})


def multiplication_expert(
    state: MessagesState,
) -> Command[Literal["addition_expert", "__end__"]]:
    system_prompt = (
        "あなたは乗算エージェントです。加算エージェントに助けを求めることができます。"
        "引き継ぎの前に計算を行ってください。"
    )
    messages = [{"role": "system", "content": system_prompt}] + state["messages"]
    ai_msg = model.bind_tools([transfer_to_addition_expert]).invoke(messages)
    
    # APIの応答からツール呼び出しを確認
    if hasattr(ai_msg, "additional_kwargs") and "tool_calls" in ai_msg.additional_kwargs:
        tool_calls = ai_msg.additional_kwargs["tool_calls"]
        if tool_calls:
            # 全てのtool_call_idに対するツールメッセージを作成
            tool_messages = []
            for tool_call in tool_calls:
                if tool_call.get("function", {}).get("name") == "transfer_to_addition_expert":
                    tool_call_id = tool_call["id"]
                    tool_msg = ToolMessage(
                        content="正常に転送されました",
                        tool_call_id=tool_call_id,
                    )
                    tool_messages.append(tool_msg)
            
            # ツール呼び出しがあり、それが加算エージェントへの移動の場合
            if tool_messages:
                return Command(
                    goto="addition_expert", 
                    update={"messages": [ai_msg] + tool_messages}
                )

    return Command(goto="__end__", update={"messages": [ai_msg]})


builder = StateGraph(MessagesState)
builder.add_node("addition_expert", addition_expert)
builder.add_node("multiplication_expert", multiplication_expert)
# we'll always start with the addition expert
builder.add_edge(START, "addition_expert")

graph = builder.compile()

# グラフの可視化
import os
file_name = os.path.basename(__file__)
graph.get_graph().draw_mermaid_png(output_file_path=f"{file_name.replace('.py', '.png')}")

# メッセージの表示
from langchain_core.messages import convert_to_messages
def pretty_print_messages(update):
    if isinstance(update, tuple):
        ns, update = update
        # skip parent graph updates in the printouts
        if len(ns) == 0:
            return

        graph_id = ns[-1].split(":")[0]
        print(f"Update from subgraph {graph_id}:")
        print("\n")

    for node_name, node_update in update.items():
        print(f"Update from node {node_name}:")
        print("\n")

        for m in convert_to_messages(node_update["messages"]):
            m.pretty_print()
        print("\n")


# 実行
for chunk in graph.stream(
    {"messages": [("user", "(3 + 5) * 12を計算してください")]},
):
    pretty_print_messages(chunk)

実行結果

AIがツールを呼び出し、その呼び出しを受け取って別のエージェントを呼び出していることがわかります。

Update from node addition_expert:


================================== Ai Message ==================================
Tool Calls:
  transfer_to_multiplication_expert (call_SPPJW3S6RslGq4dm0EiCWMQ4)
 Call ID: call_SPPJW3S6RslGq4dm0EiCWMQ4
  Args:
  transfer_to_multiplication_expert (call_zmp1GcIVcyK2bQpQmC2QHKHS)
 Call ID: call_zmp1GcIVcyK2bQpQmC2QHKHS
  Args:
================================= Tool Message =================================

Successfully transferred
================================= Tool Message =================================

Successfully transferred


Update from node multiplication_expert:


================================== Ai Message ==================================

まず、加算部分の計算を行います。3 + 5の結果は8です。その後、その結果に12を掛けます。
Tool Calls:
  transfer_to_multiplication_expert (call_zMwax6HaufMzMwM7RUdFjkXu)
 Call ID: call_zMwax6HaufMzMwM7RUdFjkXu
  Args:
    value1: 8
    value2: 12
  transfer_to_multiplication_expert (call_VgsJ1yed4CDZo94MPshP6brQ)
 Call ID: call_VgsJ1yed4CDZo94MPshP6brQ
  Args:
    value1: 8
    value2: 12

ツールによるハンドオフ

• LLMがmake_handoff_toolというツールを呼び出し、こいつがCommandを返すという間接的なプロセスを踏みます。
  （今のところmake_handoff_toolは自作関数ですが、将来的にはlanggraphの組み込み関数になるかもしれません）
• 各エージェントはサブグラフとして実装します。組み込み関数のcreate_react_agentを使うと簡単に実装できます。create_react_agentは、ツールを実行しコマンドを返す関数です。

サンプルコード

"""

https://langchain-ai.github.io/langgraph/how-tos/agent-handoffs/#implement-handoffs-using-tools
"""

from typing import Annotated

from langchain_core.tools import tool
from langchain_core.tools.base import InjectedToolCallId
from langchain_core.messages import ToolMessage, HumanMessage
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.types import Command

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4.1-nano", temperature=0.0)

def make_handoff_tool(*, agent_name: str):
    """Commandを介してハンドオフを返すツールを作成する"""
    tool_name = f"transfer_to_{agent_name}"

    @tool(tool_name)
    def handoff_to_agent(
        # オプションで現在のグラフの状態をツールに渡す（LLMには無視される）
        state: Annotated[dict, InjectedState],
        # オプションで現在のツール呼び出しIDを渡す（LLMには無視される）
        tool_call_id: Annotated[str, InjectedToolCallId],
    ):
        """他のエージェントに助けを求める"""
        # ToolMessageオブジェクトを使用して正しいツールメッセージを作成
        tool_msg = ToolMessage(
            content=f"{agent_name}に引き継ぎます",
            tool_call_id=tool_call_id,
            name=tool_name,
        )
        return Command(
            # 親グラフの別のエージェントノードに移動
            goto=agent_name,
            graph=Command.PARENT,
            # これは、エージェント`agent_name`が呼び出されたときに見ることになる状態更新です。
            # エージェントの完全な内部メッセージ履歴を渡し、ツールメッセージを追加して、
            # 結果のチャット履歴が有効であることを確認します。詳細は上記の段落を参照してください。
            update={"messages": state["messages"] + [tool_msg]},
        )

    return handoff_to_agent

# 加算エージェントを設定
@tool
def add(a: int, b: int) -> int:
    """2つの数を足します"""
    return a + b

# 乗算エージェントを設定
@tool
def multiply(a: int, b: int) -> int:
    """2つの数を掛けます"""
    return a * b

# 加算エージェントの設定
addition_expert = create_react_agent(
    model=model,
    tools=[add, make_handoff_tool(agent_name="multiplication_expert")],
    name="addition_expert",
    prompt=(
        "あなたは加算エージェントです。足し算の計算を行う専門家です。"
        "乗算が必要な場合は、multiplication_expertに引き継いでください。"
        "常に一度に1つのツールのみを使用してください。"
    ),
)

# 乗算エージェントの設定
multiplication_expert = create_react_agent(
    model=model,
    tools=[multiply, make_handoff_tool(agent_name="addition_expert")],
    name="multiplication_expert",
    prompt=(
        "あなたは乗算エージェントです。掛け算の計算を行う専門家です。"
        "加算が必要な場合は、addition_expertに引き継いでください。"
        "常に一度に1つのツールのみを使用してください。"
    ),
)

# グラフの構築
builder = StateGraph(MessagesState)
builder.add_node("addition_expert", addition_expert)
builder.add_node("multiplication_expert", multiplication_expert)
builder.add_edge(START, "addition_expert")
graph = builder.compile()

# pretty_print_messages関数の定義
def pretty_print_messages(update):
    if isinstance(update, tuple):
        ns, update = update
        # skip parent graph updates in the printouts
        if len(ns) == 0:
            return

        graph_id = ns[-1].split(":")[0]
        print(f"Update from subgraph {graph_id}:")
        print("\n")

    for node_name, node_update in update.items():
        print(f"Update from node {node_name}:")
        print("\n")

        # MessagesStateからメッセージを取得
        if "messages" in node_update:
            from langchain_core.messages import convert_to_messages
            for m in convert_to_messages(node_update["messages"]):
                m.pretty_print()
        print("\n")

if __name__ == "__main__":
    # ユーザーメッセージをHumanMessageとして作成
    user_message = HumanMessage(content="(3 + 5) * 12を計算してください")
    
    for chunk in graph.stream(
        {"messages": [user_message]}, subgraphs=True
    ):
        pretty_print_messages(chunk)

グラフの構築において、エッジの接続はSTARTと"addition_expert"の間にしかないことに注目してください

builder.add_edge(START, "addition_expert")

LLMがmake_handoff_toolを介して乗算エージェントを呼び出しますので、エッジの接続は不要です。
図示しても、以下のように乗算エージェントは単独でいます。

実行結果

同様にAIがツールを呼び出し、その呼び出しを受け取って別のエージェントを呼び出していることがわかります。

Update from subgraph addition_expert:


Update from node agent:


================================== Ai Message ==================================
Name: addition_expert
Tool Calls:
  transfer_to_multiplication_expert (call_y9hXpAxT2vbgAHvWzAGTHR4m)
 Call ID: call_y9hXpAxT2vbgAHvWzAGTHR4m
  Args:


Update from subgraph multiplication_expert:


Update from node agent:


================================== Ai Message ==================================
Name: multiplication_expert
Tool Calls:
  multiply (call_wbaM8ANQfrSeXhdTnVgpt6Wg)
 Call ID: call_wbaM8ANQfrSeXhdTnVgpt6Wg
  Args:
    a: 8
    b: 12


Update from subgraph multiplication_expert:


Update from node tools:


================================= Tool Message =================================
Name: multiply

96


Update from subgraph multiplication_expert:


Update from node agent:


================================== Ai Message ==================================
Name: multiplication_expert

(3 + 5) * 12 の計算結果は 96 です。

両者の違いと使い分け

• コマンドベースのハンドオフは、直接エージェントやハンドオフを記述するのでわかりやすいですが、エージェントが増えてくると、エージェント同士のつながりを修正したい時に書き直しが多く必要となってきます
• 一方でツールベースのハンドオフは、エージェントをサブグラフとして定義しますので、エージェントの切り分けを容易にできます。
• 一般的にツールベースのハンドオフのほうが推奨されます。そもそもLangGraphがマルチエージェントを容易に実装できるように、create_react_agentを提供してくれているので、素直にそのやり方に乗っかれば良いです