🐈

【LangGraph】マルチエージェントアプリケーションにマルチターン会話を追加する方法

2025/04/29に公開

multiagent

LangGraph

tech

LangGraph公式ドキュメントの解説です。

読みやすいようコメント等を日本語に直しました。
また、langchain_openaiで動作するようにしています。

サンプルコード

以下のようにtravel_advisorエージェントが、必要に応じてhotel_advisorエージェントを呼び出します。
また各エージェントとユーザーがインターラクティブに会話するようにできています。

サンプルコード

クリックして展開

import random
from typing import Annotated, Literal

from langchain_core.tools import tool
from langchain_core.tools.base import InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langchain_core.messages import ToolMessage, HumanMessage
from langchain_core.runnables import RunnableConfig

@tool
def get_travel_recommendations():
    """旅行先のおすすめを取得する"""
    return random.choice(["aruba", "turks and caicos"])

@tool
def get_hotel_recommendations(location: Literal["aruba", "turks and caicos"]):
    """指定された目的地のホテルおすすめを取得する"""
    return {
        "aruba": [
            "The Ritz-Carlton, Aruba (Palm Beach)"
            "Bucuti & Tara Beach Resort (Eagle Beach)"
        ],
        "turks and caicos": ["Grace Bay Club", "COMO Parrot Cay"],
    }[location]


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

    @tool(tool_name)
    def handoff_to_agent(
        state: Annotated[dict, InjectedState],
        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,
            update={"messages": state["messages"] + [tool_msg]},
        )

    return handoff_to_agent


from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4.1-mini")

# 旅行アドバイザーツールとReActエージェントを定義
travel_advisor_tools = [
    get_travel_recommendations,
    make_handoff_tool(agent_name="hotel_advisor"),
]
travel_advisor = create_react_agent(
    model,
    travel_advisor_tools,
    prompt=(
        "あなたは旅行先（国、都市など）をおすすめできる一般的な旅行の専門家です。"
        "ホテルのおすすめが必要な場合は、'hotel_advisor'に助けを求めてください。"
        "別のエージェントに転送する前に、人間が読める応答を含める必要があります。"
    ),
)


def call_travel_advisor(state: MessagesState,) -> Command[Literal["hotel_advisor", "human"]]:
    response = travel_advisor.invoke(state)
    return Command(update=response, goto="human")


# ホテルアドバイザーツールとReActエージェントを定義
hotel_advisor_tools = [
    get_hotel_recommendations,
    make_handoff_tool(agent_name="travel_advisor"),
]
hotel_advisor = create_react_agent(
    model,
    hotel_advisor_tools,
    prompt=(
        "あなたは指定された目的地のホテルの推奨事項を提供できるホテルの専門家です。"
        "旅行先を選ぶのに助けが必要な場合は、'travel_advisor'に助けを求めてください。"
        "別のエージェントに転送する前に、人間が読める応答を含める必要があります。"
    ),
)


def call_hotel_advisor(state: MessagesState) -> Command[Literal["travel_advisor", "human"]]:
    response = hotel_advisor.invoke(state)
    return Command(update=response, goto="human")


def human_node(state: MessagesState, config) -> Command[Literal["hotel_advisor", "travel_advisor", "human"]]:
    """ユーザー入力を収集するためのノード"""

    user_input = interrupt(value="") # Command(resume="")で入力された文が入る

    # 人間に投げかけたエージェント（最後にアクティブだったエージェント）を特定する
    langgraph_triggers = config["metadata"]["langgraph_triggers"]
    if len(langgraph_triggers) != 1:
        raise AssertionError("humanノードでは正確に1つのトリガーが予想されます")

    active_agent = langgraph_triggers[0].split(":")[1]

    # エージェントにユーザーの入力を投げる
    return Command(
        update={"messages": [HumanMessage(content=user_input)]},
        goto=active_agent,
    )


# ノードとエッジの作成
builder = StateGraph(MessagesState)

builder.add_node("travel_advisor", call_travel_advisor)
builder.add_node("hotel_advisor", call_hotel_advisor)
builder.add_node("human", human_node)

builder.add_edge(START, "travel_advisor")

# グラフのビルド
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

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

実行方法

自動で会話をさせるには、以下コードを追加してください。

if __name__ == "__main__":
    import uuid

    thread_config = RunnableConfig({"configurable": {"thread_id": uuid.uuid4()}})

    inputs = [
        # 最初のinputは、stateのmessagesに状態を渡す
        { "messages": [HumanMessage(content="カリブ海で暖かい場所に行きたいです")]},
        # 2回目以降のinputは、Commandプリミティブを使用して再開する必要があります。
        Command(
            resume="エリアの一つでいいホテルを推薦してもらえますか？そしてそれがどのエリアなのか教えてください。"
        ),
        Command(
            resume="最初のが気に入りました。ホテル近くでおすすめのアクティビティはありますか？"
        ),
    ]

    for idx, user_input in enumerate(inputs):
        print()
        print(f"--- 会話ターン {idx + 1} ---", end="\n\n")
        print(f"ユーザー: {user_input}", end="\n\n")
        for update in graph.stream(
            user_input,
            config=thread_config,
            stream_mode="updates",
        ):
            for node_id, value in update.items():
                if isinstance(value, dict) and value.get("messages", []):
                    last_message = value["messages"][-1]
                    if isinstance(last_message, dict) or last_message.type != "ai":
                        continue
                    print(f"{node_id}: {last_message.content}")

または手動で入力するには以下のコードを追加してください。

if __name__ == "__main__":
    import uuid

    thread_config = RunnableConfig({"configurable": {"thread_id": uuid.uuid4()}})

    count = 0

    while True:
        user_input = input("メッセージを入力ください: ")

        if count == 0:
            user_input = { "messages": [HumanMessage(content=user_input)]}
        else:
            user_input = Command(resume=user_input)

        print()
        print(f"--- 会話ターン {count + 1} ---", end="\n\n")
        print(f"ユーザー: {user_input}", end="\n\n")
        for update in graph.stream(
            user_input,
            config=thread_config,
            stream_mode="updates",
        ):
            for node_id, value in update.items():
                if isinstance(value, dict) and value.get("messages", []):
                    last_message = value["messages"][-1]
                    if isinstance(last_message, dict) or last_message.type != "ai":
                        continue
                    print(f"{node_id}: {last_message.content}")
        
        count += 1
        if count == 2:
            break

解説

ハンドオフツールの作成

エージェントの切り替え（ハンドオフ）は、goto="エージェント名"を含んだCommandを返すことによって行うのがLangGraphの基本です。
ツールによるハンドオフでは、Commandを返すツールをエージェントにバインドしておき、エージェントがこのツールを実行するという方法を取ります。

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

    @tool(tool_name)
    def handoff_to_agent(
        state: Annotated[dict, InjectedState],
        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,
            update={"messages": state["messages"] + [tool_msg]},
        )

    return handoff_to_agent

エージェントの生成

LangGraph組み込みのcreate_react_agentメソッドを使ってエージェントを作成します。
create_react_agentにはLLM、ツール、プロンプトの3点を渡します。

from langgraph.prebuilt import create_react_agent
travel_advisor_tools = [
    get_travel_recommendations,
    make_handoff_tool(agent_name="hotel_advisor"),
]
travel_advisor = create_react_agent(
    ChatOpenAI(),
    travel_advisor_tools,
    prompt=(
        "あなたは旅行先（国、都市など）をおすすめできる一般的な旅行の専門家です。"
        "ホテルのおすすめが必要な場合は、'hotel_advisor'に助けを求めてください。"
        "別のエージェントに転送する前に、人間が読める応答を含める必要があります。"
    ),
)

エージェント呼び出しノードの作成

ノードとはつまるところ関数です。一番シンプルな実装は、エージェントにinvokeで状態を渡してやることです。
（もちろん何かしらのロジックを追加することもできます）
Commandでgoto="human"が設定されていることに注目してください。これにより、travel_advisorノードが実行された後にはhumanノードに移行します。
　（なので、add_edgeでhumanノードとtravel_advisorノードを繋ぐ必要はありません）

def call_travel_advisor(state: MessagesState) -> Command[Literal["hotel_advisor", "human"]]:
    response = travel_advisor.invoke(state)
    return Command(update=response, goto="human")

humanノードの作成

外部から受け付けたCommandのresumeがinterruptの返り値となります。
Commandでgoto=active_agentが設定されていることに注目してください。これによりhumanノードが実行された後には該当のエージェントノードに移行します。
configは現在のGraphの状態が保存されています。このconfigに、最後に働いていたエージェントが記録されています。

def human_node(state: MessagesState, config) -> Command[Literal["hotel_advisor", "travel_advisor", "human"]]:
    """ユーザー入力を収集するためのノード"""

    user_input = interrupt(value="") # Command(resume="")で入力された文が入る

    # 人間に投げかけたエージェント（最後にアクティブだったエージェント）を特定する
    langgraph_triggers = config["metadata"]["langgraph_triggers"]
    if len(langgraph_triggers) != 1:
        raise AssertionError("humanノードでは正確に1つのトリガーが予想されます")

    active_agent = langgraph_triggers[0].split(":")[1]

    # エージェントにユーザーの入力を投げる
    return Command(
        update={"messages": [HumanMessage(content=user_input)]},
        goto=active_agent,
    )

ノードとエッジの作成、およびグラフのビルド

エントリーポイントのエッジだけが登録されていることに注目。hotel_advisorやhumanノードへの移行は、travel_advisorがCommandのgotoで指定しますので、エッジを繋ぐ必要がありません。

# ノードとエッジの作成
builder = StateGraph(MessagesState)

builder.add_node("travel_advisor", call_travel_advisor)
builder.add_node("hotel_advisor", call_hotel_advisor)
builder.add_node("human", human_node)

builder.add_edge(START, "travel_advisor")

# グラフのビルド
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

サンプルコード

サンプルコード

実行方法

解説

ハンドオフツールの作成

エージェントの生成

エージェント呼び出しノードの作成

humanノードの作成

ノードとエッジの作成、およびグラフのビルド

Discussion