LangGraph×Bedrockによる複数のAgentic Workflowを利用したSupervisor型マルチエージェントの実装：広告素材作成アプリケーション

本記事は、以下 2 名による共著です。

• NTT データ – デジタルサクセスコンサルティング事業部 原 弥麻人
• NTT データ – デジタルサクセスコンサルティング事業部 鯨田 連也

本記事は、Bedrock Night オンライン 〜AWSで生成AIアプリ開発！ 最新ナレッジ共有〜で登壇した際の内容です。以下の登壇資料も合わせてご覧ください！

はじめに

株式会社NTTデータ デジタルサクセスコンサルティング事業部の@yamato0811, @ren8kです。

昨今、AI Agentの開発が活発化しており、幅広い分野のタスクに対応するために、Multi-Agentの実装が求められています。Multi-Agentの実装には、LangGraphやAmazon Bedrock Agentなどのフレームワークが利用されることが多いですが、その実装方法についてはまだ十分な情報が共有されていません。また、Agentic WorkflowをMulti-Agentに組み込む方法などは、執筆時点（2025/03/12）ではほぼ情報がありません。我々もMulti-Agentを実装するにあたり、ネット上の情報や実装例が少なく、数多くの試行錯誤を重ねました。

本記事では、LangGraphを利用した、複数のAgentic Workflowを利用したSupervisor型のMulti-Agentの実現方法について解説します。特に、Agentic Workflowの強みである「決定的なタスクの実行」と、Agent（ReAct）の強みである「柔軟な対応力」を両立させることで、より幅広い用途に使用可能なAI Agentを実現できることを示します。

また、広告素材（コピー文、画像）作成アプリケーションを題材とし、LangGraphでMulti-Agentを実装する際のポイントや工夫点を、Pythonのコード例とともに解説します。特に、Sub Agentの定義方法や、Agent間の制御の委譲方法（handoff）について、実装上の工夫点を解説します。なお、実装にあたり、利用しているLLMや画像生成AIにはAmazon Bedrockを利用しています。

解説用に実装した簡易アプリケーションのリポジトリは、以下のリンクからご確認いただけます。

※ 以前、StreamlitとLangGraphで実装したHuman-in-the-loop Agentic Workflowについて解説した記事を執筆しました。まだご覧になっていない方は、そちらを合わせてお読みいただくとより理解が深まるかと思います！

想定読者

• Streamlitを用いたアプリケーション開発経験がある方
• LangGraphの概要を理解している方
• LangGraphでのMulti-Agent実装に興味がある方

Agentic Workflow vs Agent

AI Agentの開発においては、「Agentic Workflow」と「Agent（ReAct）」という2つの主要なアプローチがあります。それぞれの特徴と違いを理解し、適切な場面で使い分けることが重要です。

Agentic Workflow

Agentic Workflowは、事前に明確に定義された実行順序で、複数のタスク（LLMやツール）を順番に実行する仕組みです。このアプローチの利点は、最終的な出力が一貫性を持ちやすく、正確であることです。

各タスクの実行順序は予め決められているため、プロセスが制御しやすく、結果も安定しやすいのが特長です。

Agent

Agent（ReAct）は、状況に応じてLLMがどのツールを使うべきかを自律的に決定し、ツールの実行結果をもとに次のアクションを柔軟に選択する仕組みです。この仕組みでは、LLMが考え、行動し、その結果を観察して再び考える（Reasoning, Acting, Observation）というサイクルを繰り返します。

実行タスクやその実行順序を動的に決定できるため、ユーザーのニーズに対し柔軟に対応可能です。

Agentic WorkflowとAgentの比較

Agentic Workflowはプロセスが明確で制御が容易な反面、状況変化への対応力や柔軟性に欠けることがあります。一方で、Agent（ReAct）は柔軟な判断が可能なため、予測不能な状況や、複雑で柔軟な対応が求められるシナリオで効果を発揮します。ただし、Agentはその柔軟性ゆえに動作が予測しづらく、調整やチューニングが必要な場合があります。

項目	Agentic Workflow	Agent（ReAct）
回答の一貫性	◯ タスク実行順序は決定的で制御しやすい	△ 想定通りのタスクの実行順序になるとは限らない
全体的なレイテンシー	◯ タスクを直列に実行するのみ	△ タスク（Tool）実行の度にLLMの思考がある
柔軟性	△ 一部のタスクのみを実行することはできない	◯ LLMが実行すべきタスクや実行順序を動的に決定する
機能追加の容易性	△ フローが複雑になるほど実装コストが高い（タスク間の依存関係の管理など）	◯ Toolを利用するのみで、改修範囲は狭い

Agentic WorkflowやAgentについては、Anthropicの技術記事Building effective agentsやLangGraphのドキュメントAgent architecturesの記事が大変参考になります。

Multi-Agent

AI Agentが管理するツールの増加や、AI Agentへの指示の複雑化により、単一のエージェントでの対応が困難になる場合があります。そのような状況で有効な解決策となるのがMulti-Agentのアプローチです。
Multi-Agentシステムでは、専門分野に特化した複数のSub Agentが協調して動作するので、複雑なタスクを処理することが可能です。

Multi-Agentの利点

• 各Agentのモジュール化により、Agentごとに開発・テスト・管理が容易になる
• 特定の領域に特化した専門Agentを構築でき、システム全体のパフォーマンスが向上する
• エージェント間の通信方法を明確に制御できる

Multi-Agentの構成パターン

Multi-Agentの構成にはいくつかのパターンが存在します。

1. ネットワーク型：全てのエージェントが互いに自由に通信でき、どのエージェントも次にどのエージェントを呼ぶか決定できる
2. Supervisor型：各Agentが単一のSupervisor Agentと通信し、Supervisorが次の呼び出しAgentを決定する
3. Supervisor（ツール型）：Agentをツールとして扱い、Supervisor Agentがツール呼び出し可能なLLMを使って、どのエージェントを呼ぶか、および引数を決定する
4. 階層型：Supervisorの上位にさらに別のSupervisorを置くことで、階層構造を作り管理を効率化する
5. カスタム型：特定のニーズに応じてエージェント同士の接続を自由に定義する

上記図や整理内容はLangGraphドキュメントのMulti-agent Systemsから引用しています。

作成したアプリケーション

広告の「キャッチコピー文」と「画像」を生成するMulti-Agentアプリケーションを実装しました。Multi-Agentの構成として、複数の専門エージェントを定義することができ、拡張性の高いSupervisor型を採用しています。

本アプリケーションは、ユーザーが広告に使用する素材の作成を要望すると、Supervisorが各Sub Agent（キャッチコピー生成Agent, 画像生成Agent）を適切に呼び出し、目的に応じた広告素材を作成する仕組みとなっています。

Python実装は以下に公開しています。LLMとしてAmazon BedrockのClaude 3.7 Sonnetを、画像生成AIとしてAmazon BedrockのAmazon Nova Canvasを利用しています。

WorkflowをMulti-Agentで利用するアイデア

WorkflowとMulti-Agentの両方のメリットを享受するため、Multi-Agentの各Agent（Sub Agent）としてAgentic Workflowを利用しました。これにより、ユーザーの要望を基に実行すべきWorkflowを柔軟に選択しつつ、決定的な実行順序で正確なタスク実行が可能となります。

上記のアイデアを実現するために、LangGraphの以下の機能を利用しています。

• SubGraph: グラフ（Agentic Workflow, Agent）をノードとして利用する機能
• tool use(@tool) + handoff(Command): 実行すべきAgentを選択し、そのAgentへ実行制御と必要情報を引き継ぐ機能

Supervisor型のMulti-Agentを実装する際、ユーザーが入力したプロンプトからSub Agentへ渡すべき情報をどのように抽出し、伝達するかが課題でした。例えば、広告のキャッチコピー文を生成するAgentic Workflow（Sub Agent）の実行において、「どのようなテーマでコピー文を生成するか」という情報が必要です。しかし、ユーザーのプロンプトには必ずしもテーマのみが含まれるわけではなく、以下のようなケースが考えられます。

• (1) 広告のテーマ以外にも様々な要望などの情報が含まれるケース
• (2) ユーザーの入力が不完全で、テーマが含まれないケース

(1)の場合には、テーマのみを抽出する必要があり、(2)の場合には、ユーザーに（Sub Agentの実行に必要な）テーマを入力するよう依頼する必要があります。

この課題に対し、LangGraphの機能を調査・検証した結果、@toolデコレータとhandoff(Command)を組み合わせた手法が有効であると考えました。具体的には、@toolデコレータを利用してAgentic Workflow（Sub Agent）を実行するためのツールを定義します。その際、ツールの引数として、Agentic Workflowの最初のノードの実行に必要な情報を設定し、ツールの説明には、Agentic Workflowの説明を記述します。これにより、ユーザーのプロンプトから必要な情報だけを正確に抽出することが可能となり、ユーザーの入力が不完全な場合には、ユーザーに再入力を促すことができます。

# handoff用のツールの簡易実装例
from langgraph.types import Command

@tool
def handoff_to_copy_generator(
    theme_copy: Annotated[str, "コピーのテーマ"] # ツールの引数として、Workflowの実行に必要な情報を設定
    ) -> Command:
    """Agentic Workflow（Sub Agent）の説明"""
    return Command(
        goto="copy_generator_subgraph", # Supervisorが次に実行すべきノード（Sub Agent）
        update={"theme_copy": theme_copy}, # Workflowの最初のノードでは、state["theme_copy"]を利用してコピー文を生成する
    )

また、ツールの返り値としてCommandオブジェクトを返すように設定します。Commandとは、LangGraphのノード内で次に実行するノードの指定や、AgentのStateの更新を同時に行う機能です。Commandオブジェクトには、次に実行すべきSub Agentと、Sub Agentの実行に必要な情報（引数）を設定します。これにより、Supervisorがユーザーのプロンプトから抽出した情報（生成された引数）をSub Agentに引き渡し、Sub Agentを実行（handoff）することが可能となります。なお、SupervisorとSub Agentの情報の連携はStateを介して行います。

以上の仕組みにより、ユーザーのプロンプトから、次に実行すべきSub Agentの決定と、Sub Agentの実行に必要な情報の抽出が可能になります。

tool useを利用したエージェント間でのハンドオフの実装は、LangGraphのドキュメントHow to implement handoffs between agentsを参考にしています。

LangGraphのグラフ構造

以下に、今回作成したアプリケーションのグラフ構造を示します。

アプリの機能

Supervisorは以下2つの機能を持つSub Agentを管理しており、ユーザーからの要望に応じて各Sub Agentに指示を出します。

• コピー文生成エージェント: 与えられたキーワードをもとに効果的なコピー文を作成する
  • generate_copyノード：Supervisorが指示したキーワードに基づいて初期コピー文を生成
  • refine_copyノード：生成した初期コピー文をさらに改善
• 画像生成エージェント: 画像の主題に基づいて、画像を生成する
  • generate_promptノード：Supervisorが指示した画像の主題に基づいて画像生成用のプロンプトを作成
  • generate_imageノード：作成したプロンプトを元に画像を生成

LangGraphでMulti-Agentの実装

本章では、LangGraphを使用したAgentグラフの実装方法について、具体的なコード例を交えて説明します。特に、Sub AgentとしてWorkflowを定義するために利用した、SubGraphと、handoff(Command) という機能を中心に解説します。

※ プログラムの一部のみ抜粋していますので、適宜GitHubのコードを参照ください。

Stateの定義

グラフのノード間を遷移するState情報はagent/state.pyに以下のように定義しています。

display_message_dictは、Streamlitに表示するためのメッセージをまとめたものです。各ノードでこの内容を更新することで、各ノードの実行結果をStreamlit上に表示します。

agent/state.py

from langgraph.graph.message import add_messages
from typing_extensions import Annotated, Literal, NotRequired, TypedDict


class DisplayMessageDict(TypedDict):
    role: Literal["user", "assistant"]
    title: str
    icon: str
    content: str
    images: NotRequired[list[str]]


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

    # Copy Generator
    theme_copy: str
    draft_copy: str

    # Image Generator
    visual_concept: str
    img_prompt: str

    # for Streamlit Display
    display_message_dict: DisplayMessageDict

messages: Annotated[list, add_messages]の​add_messagesはリデューサーと呼ばれ、状態（State）の更新時の処理を指定する際に利用します。​add_messagesはLangGraphにおける組み込みの関数で、状態（State）内のメッセージリストを更新する際、上書きせずにリストに追加（append）することが可能です。
https://langchain-ai.github.io/langgraph/tutorials/introduction/#part-1-build-a-basic-chatbot

SubGraph

Workflowで構成した各Sub Agentを階層的に管理するため、SubGraphの機能を使用しました。SubGraphを使用することで、グラフ全体の管理性や拡張性を高めることが可能となります。

Supervisorの定義

agent/supervisor.pyでSupervisorのグラフを構築しています。まず、__init__関数で、Supervisorが利用するツールの設定を行います。グラフの定義およびコンパイル処理は、build_graph関数で行っています。

すでにコンパイル済みのグラフ（copy_generatorやimage_generator）を直接ノードとしてadd_nodeすることで、Agentic Workflowをサブグラフとして利用することができます。

ツールとして、handoffによりSub Agentに制御を委譲する（Sub Agentを実行する）ための関数（handoff_to_copy_generatorやhandoff_to_image_generator）を定義しています。これら解説については、後述のセクションで詳しく行います。

agent/supervisor.py

import json

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import Command
from typing_extensions import Literal

from agent.copy_generator import CopyGenerator
from agent.image_generator import ImageGenerator
from agent.state import AgentState
from agent.tools import handoff_to_copy_generator, handoff_to_image_generator
from models.llm import LLM


class Supervisor:
    def __init__(
        self, llm: LLM, copy_generator: CopyGenerator, image_generator: ImageGenerator
    ) -> None:
        self.tools = [handoff_to_copy_generator, handoff_to_image_generator]
        self.tools_by_name = {tool.name: tool for tool in self.tools}
        self.llm_with_tools = llm.model.bind_tools(self.tools)
        self.checkpointer = MemorySaver()
        self.graph = self.build_graph(copy_generator, image_generator)

    def build_graph(
        self, copy_generator: CopyGenerator, image_generator: ImageGenerator
    ) -> CompiledStateGraph:
        graph_builder = StateGraph(AgentState)
        graph_builder.add_node(self.supervisor)
        graph_builder.add_node("copy_generator_subgraph", copy_generator.graph)
        graph_builder.add_node("image_generator_subgraph", image_generator.graph)
        graph_builder.add_node(self.end_node)
        graph_builder.add_edge("copy_generator_subgraph", "supervisor")
        graph_builder.add_edge("image_generator_subgraph", "supervisor")
        graph_builder.set_entry_point("supervisor")
        return graph_builder.compile(checkpointer=self.checkpointer)

サブグラフを追加する際は、サブグラフの状態（state）のスキーマにおけるプロパティを1つ以上、共有キーとして親グラフのstateに含める必要があります。これは、共有キーを介して親グラフとサブグラフの状態を連携させるためです。
もし親グラフと全く異なるスキーマ（共有キーなし）を定義したい場合は、サブグラフを呼び出すノード関数を定義する必要があります。

https://langchain-ai.github.io/langgraph/how-tos/subgraph/

Sub Agentの定義

Sub Agentのグラフの代表例として、CopyGeneratorのSubGraphについて説明します。
CopyGeneratorは、キャッチコピーの生成を行うノード（generate_copy）とコピーの改善を行うノード（refine_copy）を定義したAgentic Workflowです。

このSubGraphは単純な直列的な構造で、generate_copyからrefine_copyへと順番に処理を進めます。

最終的にSupervisorに制御が移るように、以下のようにSupervisorのグラフ内でエッジを定義しています。

graph_builder.add_edge("copy_generator_subgraph", "supervisor")

agent/copy_generator.py

from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph

from agent.state import AgentState
from models.llm import LLM


class CopyGenerator:
    def __init__(self, llm: LLM) -> None:
        self.llm = llm
        self.graph = self.build_graph()

    def build_graph(self) -> CompiledStateGraph:
        graph_builder = StateGraph(AgentState)
        graph_builder.add_node(self.generate_copy)
        graph_builder.add_node(self.refine_copy)

        graph_builder.set_entry_point("generate_copy")
        graph_builder.add_edge("generate_copy", "refine_copy")
        graph_builder.set_finish_point("refine_copy")
        return graph_builder.compile()

CopyGeneratorの各ノードの処理関数は以下のように記載しています。

generate_copy()では、Supervisorから受け取ったコピーのテーマstate['theme_copy']を基にLLMを実行し、キャッチコピーを生成します。生成結果はdraft_copyとして返します。refine_copy()では、generate_copy()で生成したdraft_copyを受け取り、LLMの実行により改善したコピーを返します。

また、display_message_dictには、Streamlitに表示するためのメッセージを格納しています。

agent/copy_generator.py

    def generate_copy(self, state: AgentState) -> dict:
        response = self.llm(
            [
                (
                    "system",
                    "あなたはプロのコピーライターです。",
                )
            ]
            + [
                (
                    "human",
                    f"""以下のテーマで、キャッチコピーを1つ生成してください。必ず結果のみ出力してください。
                    <theme_copy>
                    {state['theme_copy']}
                    </theme_copy>
                    """,
                )
            ]
        )

        display_message_dict = {
            "role": "assistant",
            "title": "Copy Generatorの生成結果",
            "icon": "📝",
            "content": response.content,
        }

        return {
            "messages": response,
            "draft_copy": response.content,
            "display_message_dict": display_message_dict,
        }

    def refine_copy(self, state: AgentState) -> dict:
        response = self.llm(
            [
                (
                    "system",
                    "あなたはプロのコピーライターです。",
                )
            ]
            + [
                (
                    "human",
                    f"""多角的な観点で、以下のキャッチコピーを改善してください。必ず結果のみ、改善した1つのコピーだけ出力してください。
                    <draft_copy>
                    {state['draft_copy']}
                    </draft_copy>
                    """,
                )
            ]
        )

        display_message_dict = {
            "role": "assistant",
            "title": "Copy Generatorの改善結果",
            "icon": "📝",
            "content": response.content,
        }

        return {
            "messages": response,
            "display_message_dict": display_message_dict,
        }

handoff(Command)

handoffは、あるAgentが別のAgentに制御を渡す考え方のことで、LangGraphのCommandという機能を利用して実現する事ができます。

Supervisorの処理関数の実装

Supervisorのsupervisor()関数では、tool useとhandoff（Command）を使用して、Supervisor Agentが他のSub Agentに処理を委譲する（Sub Agentをツールとして呼び出す）仕組みを実装しています。

まず、Supervisorは、過去の会話履歴とユーザーからの指示に基づき、次にどのSub Agent（ツール）を利用すべきかを判断します。その際、応答（response）は会話のコンテキストとしてstate["messages"]に追加しています。

agent/supervisor.py

def supervisor(self, state: AgentState) -> Command[
    Literal["copy_generator_subgraph", "image_generator_subgraph", "end_node"]
]:
    response = self.llm_with_tools.invoke(
        [
            (
                "system",
                """
                あなたは、Sub Agentの会話を管理する役割を持つ監督者です。
                ユーザーのリクエストに基づき、どのSub Agentを指示するか（どのツールを呼び出すか）を決定します。
                ツール呼び出しの必要がない場合は、ユーザーのサポートを行います。
                <rules>
                - Sub Agent呼び出しが必要あれば、Sub Agentを呼び出してください。その際、なぜそのSub Agentを呼び出すのかの理由も説明してください。
                - Sub Agent呼び出しが不要な場合は、Sub Agentを呼び出す必要はありません。
                - 直前にSub Agentを呼び出した場合、Sub Agentの結果を整理して報告してください。
                </rules>
                """
            )
        ]
        + state["messages"]
        + [
            (
                "human",
                """
                <instructions>
                会話を基にSub Agentを呼び出してください。呼び出しの必要がなければ、Sub Agentの結果を整理して報告してください。
                </instructions>
                """
            )
        ]
    )

    # LLM の応答をメッセージ履歴に追加
    state["messages"].append(response)

その後、Sub Agentの処理委譲が必要かどうかをtool_callsを使用して判定します。（toolの定義内容については後述します。）

agent/supervisor.py

    if len(response.tool_calls) > 0:
        for tool_call in response.tool_calls:
            tool = self.tools_by_name[tool_call["name"]]
            tool_response = tool.invoke(
                {**tool_call, "args": {**tool_call["args"], "state": state}}
            )
            invoke_result = json.loads(tool_response.content)

ツール呼び出しの結果から得られたgotoとupdateを使い、Commandオブジェクトを返します。これにより、処理は次のSub Agent（gotoに格納されているcopy_generator_subgraphやimage_generator_subgraph）へ移譲されます。

agent/supervisor.py

        # for bedrock
        content = response.content[0]["text"]

        display_message_dict = {
            "role": "assistant",
            "title": "Supervisorの思考が完了しました。",
            "icon": "👨‍🏫",
            "content": content,
        }

        return Command(
            goto=invoke_result["goto"],
            update={
                **invoke_result["update"],
                "display_message_dict": display_message_dict,
            },
        )

Commandは、ノード内でAgentのStateの更新と、次に実行するノードの指定を同時に行う、2024年12月に発表された機能です。gotoには次に実行するノード名を、updateには更新するStateを記載します。

https://langchain-ai.github.io/langgraph/how-tos/command/

ツール呼び出しが含まれていない場合は、単に結果を報告し、gotoを"end_node"として会話を終了する Commandオブジェクトを返します。

agent/supervisor.py

    else:
        display_message_dict = {
            "role": "assistant",
            "title": "Supervisorの回答",
            "icon": "👨‍🏫",
            "content": response.content,
        }
        return Command(
            goto="end_node",
            update={
                "messages": response,
                "display_message_dict": display_message_dict,
            },
        )

toolの定義

Supervisorで利用するtoolは以下のように定義しています。@toolでデコレートしたhandoff用の関数（ツール）をSupervisorが呼び出すことで、以下の情報を取得可能です。

• 次に実行すべきSub Agent（Workflow）のノード名（copy_generator_subgraph）
• ツールメッセージ（[tool_msg]）
• Workflowの実行に必要なStateの情報（tool useによって生成されたツールの引数 theme_copy）

その後、SupervisorのCommandオブジェクト内で、上記の情報をそれぞれ引数gotoとupdateに指定してreturnで返しています。この結果、Supervisorはtool useで（間接的に）Sub Agentを呼び出し、同時にtool useでWorkflowの実行に必要なStateの情報も生成しています。

なお、LLMでツール呼び出しを行った際には会話履歴にtool messageを含める必要があるため、toolの返り値の要素のmessagesにtool_msgを追加している点には注意してください。

また、toolに記載する説明はAnthoropicのBest practices for tool definitionsに従い、できるだけ詳細に記載することが重要です。（本来は今よりも更に詳細に記載することが望ましいです）

agent/tools.py

@tool
def handoff_to_copy_generator(
    theme_copy: Annotated[
        str,
        "コピーのテーマ。どのようなコピー文を生成すべきかという内容が記載されている。",
    ],
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> dict:
    """
    コピー生成を行うために、copy_generatorに引き継ぎます。
    ユーザーがコピー生成を要望している場合、このツールを呼び出します。

    copy_generatorの役割は以下の通り
    - テーマに基づいてコピー文を生成します
    """
    print("## Called Copy Generator")

    tool_msg = {
        "role": "tool",
        "content": "Successfully transferred to Copy Generator.",
        "tool_call_id": tool_call_id,
    }

    return {
        "goto": "copy_generator_subgraph",
        "update": {
            "messages": [tool_msg],
            "theme_copy": theme_copy,
        },
    }


@tool
def handoff_to_image_generator(
    visual_concept: Annotated[
        str,
        "画像の主題（ビジュアルコンセプト）。どのような画像を生成すべきかという内容が記載されている。",
    ],
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> dict:
    """
    画像生成を行うために、image_generatorに引き継ぎます。
    ユーザーが画像生成を要望している場合、このツールを呼び出します。

    image_generatorの役割は以下の通り
    - 画像の主題（ビジュアルコンセプト）に基づいて画像を生成します
    """
    print("## Called Image Generator")

    tool_msg = {
        "role": "tool",
        "content": "Successfully transferred to Image Generator.",
        "tool_call_id": tool_call_id,
    }

    return {
        "goto": "image_generator_subgraph",
        "update": {
            "messages": [tool_msg],
            "visual_concept": visual_concept,
        },
    }

ツールの引数以外の値を関数に渡したい場合は、InjectedToolCallIdのようなInjectedArgの注釈を付与します。InjectedArgの注釈が付与されたパラメータは、tool use時にLLMに認識されなくなり、ツールの引数として生成されなくなります。

https://langchain-ai.github.io/langgraph/how-tos/pass-run-time-values-to-tools/

StreamlitでのUI実装

Streamlitの実装はapp.pyに記載しています。

ユーザーがメッセージを送信すると、グラフの実行が始まります。生成された結果はストリーム形式で受け取り、その都度display_message()を使って画面に反映させています。

agent/app.py

import streamlit as st
from langchain_core.messages import HumanMessage

from agent.copy_generator import CopyGenerator
from agent.image_generator import ImageGenerator
from agent.supervisor import Supervisor
from models.bedrock_img_gen_model import BedrockImageModel
from models.llm import LLM
from utils.app_util import display_message, display_messages

MODEL = "claude-3-7-sonnet"  # you can use "claude-3-5-haiku"
IMG_GEN_MODEL = "nova-canvas"
THREAD_ID = "1"
TEMPERATURE = 0.2


def main() -> None:
    # Page Config
    st.set_page_config(
        page_title="Streamlit×LangGraph MultiAgent | 広告素材生成アプリケーション",
        page_icon="🤖",
    )
    st.title("Streamlit×LangGraph MultiAgent | 広告素材生成アプリケーション")

    # Init Actors
    llm = LLM(MODEL, TEMPERATURE)
    bedrock_image_model = BedrockImageModel(IMG_GEN_MODEL)
    copy_generator = CopyGenerator(llm)
    image_generator = ImageGenerator(llm, bedrock_image_model)

    # Set session state
    if "supervisor" not in st.session_state:
        st.session_state.supervisor = Supervisor(llm, copy_generator, image_generator)
        st.session_state.supervisor.write_mermaid_graph()
    if "display_messages" not in st.session_state:
        init_display_message_dict = {
            "role": "assistant",
            "title": "Supervisorの回答",
            "icon": "👨‍🏫",
            "content": """
            こんにちは！何かお手伝いできることはありますか？\u0020\u0020
            以下の機能を利用することができます。

            - コピー生成
            - 画像生成
            """,
        }
        st.session_state.display_messages = [init_display_message_dict]

    # Display All Messages
    display_messages(st.session_state.display_messages)

    # User Input
    user_input = st.chat_input("メッセージを入力してください:")
    if user_input:
        display_message_dict = {
            "role": "user",
            "title": "ユーザーの入力",
            "icon": "👤",
            "content": user_input,
        }
        display_message(display_message_dict)
        st.session_state.display_messages.append(display_message_dict)
    else:
        st.stop()

    # Core Algorithm
    inputs = {"messages": [HumanMessage(user_input)]}
    config = {"configurable": {"thread_id": THREAD_ID}}

    event_prev = {}
    for event in st.session_state.supervisor.graph.stream(
        inputs, config, stream_mode="values", subgraphs=True
    ):
        # Skip when transition between parent and child
        if event_prev == event[1]:
            continue
        event_prev = event[1]
        # Display Message
        if display_message_dict := event[1].get("display_message_dict"):
            display_message(display_message_dict)
            st.session_state.display_messages.append(display_message_dict)


if __name__ == "__main__":
    main()

display_message()はutil関数として以下のように定義しています。

utils/app_util.py

def display_message(message: dict) -> None:
    """
    messageをstreamlit上に表示する関数
    """
    with st.chat_message(message["role"]):
        with st.expander(
            message["title"],
            expanded=True,
            icon=message["icon"],
        ):
            if "images" in message:
                st.write(message["content"])
                # images is saved as List[Image.Image]
                images = [convert_base64_2_img(img) for img in message["images"]]
                _show_images(images)
            else:
                st.write(message["content"], unsafe_allow_html=True)

苦労した点

Sub Agent（サブグラフ）内のエッジの定義にCommandを利用すると、期待した挙動とならない不具合（バグ）が発生しました。具体的には、同一のWorkflowを定義しているにもかかわらず、Commandを利用した場合とadd_edgeを利用した場合のグラフで、以下に示すように挙動が異なっております。

基本的に、Commandはadd_edgeで代替することが可能です。しかし、ツール（@toolでデコレートしたhandoff用の関数）の実行結果に応じて、次に実行するサブグラフを動的に決定したり、異なるStateの要素を更新する場合、Commandを利用しなければ実装が非常に複雑になってしまいます。

Commandを利用する場合

出力

((), {'node_name': ['__start__']})
((), {'node_name': 'node_1'})
(('node_2:d41305f4-c2ba-78d9-5d9b-ed1c6b8549da',), {'node_name': 'node_1'})
(('node_2:d41305f4-c2ba-78d9-5d9b-ed1c6b8549da',), {'node_name': 'subgraph_node_1', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'subgraph_node_2'})
((), {'node_name': 'node_3'})

コード

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import Literal


class State(TypedDict):
    node_name: str
    foo: str


def subgraph_node_1(state: State) -> Command[Literal["subgraph_node_2"]]:
    return Command(
        goto="subgraph_node_2",
        update={
            "node_name": "subgraph_node_1",
            "foo": "Update at subgraph_node_1!",
        },
    )


def subgraph_node_2(state: State) -> Command:
    return Command(
        goto="node_3",
        update={"node_name": "subgraph_node_2"},
        graph=Command.PARENT,
    )


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_2", END)
subgraph = subgraph_builder.compile()


# Define main graph
def node_1(state: State) -> Command[Literal["node_2"]]:
    return Command(
        goto="node_2",
        update={"node_name": "node_1"},
    )


def node_3(state: State) -> Command[Literal["__end__"]]:
    return Command(
        goto=END,
        update={"node_name": "node_3"},
    )


main_builder = StateGraph(State)
main_builder.add_node("node_1", node_1)
main_builder.add_node("node_2", subgraph)
main_builder.add_node("node_3", node_3)
main_builder.add_edge(START, "node_1")
main_builder.add_edge("node_2", "node_3")
main_graph = main_builder.compile()


# Build subgraph
with open("graph.md", "w") as file:
    file.write(f"\n{main_graph.get_graph(xray=1).draw_mermaid()}")

initial = {"node_name": ["__start__"]}
for chunk in main_graph.stream(initial, stream_mode="values", subgraphs=True):
    print(chunk)

Commandを利用せずadd_edgeを利用する場合（期待する動作）

出力

((), {'node_name': ['__start__']})
((), {'node_name': 'node_1'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'node_1'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'subgraph_node_1', 'foo': 'Update at subgraph_node_1!'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'subgraph_node_2', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'subgraph_node_2', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'node_3', 'foo': 'Update at subgraph_node_1!'})

コード

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    node_name: str
    foo: str


def subgraph_node_1(state: State):
    return {
        "node_name": "subgraph_node_1",
        "foo": "Update at subgraph_node_1!",
    }


def subgraph_node_2(state: State):
    return {"node_name": "subgraph_node_2"}


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph_builder.add_edge("subgraph_node_2", END)
subgraph = subgraph_builder.compile()


# Define main graph
def node_1(state: State):
    return {"node_name": "node_1"}


def node_3(state: State):
    return {"node_name": "node_3"}


main_builder = StateGraph(State)
main_builder.add_node("node_1", node_1)
main_builder.add_node("node_2", subgraph)
main_builder.add_node("node_3", node_3)
main_builder.add_edge(START, "node_1")
main_builder.add_edge("node_1", "node_2")
main_builder.add_edge("node_2", "node_3")
main_builder.add_edge("node_3", END)
main_graph = main_builder.compile()


# Build subgraph
with open("graph.md", "w") as file:
    file.write(f"```mermaid\n{main_graph.get_graph(xray=1).draw_mermaid()}```")

initial = {"node_name": ["__start__"]}
for chunk in main_graph.stream(initial, stream_mode="values", subgraphs=True):
    print(chunk)

Commandは当時リリースされた直後で、Issueなどにも本事象は報告されておらず、何が原因で想定通りの挙動にならないのかが不明でした。そこで、Commandについて、ドキュメントで仕様を調査し、add_edgeを利用した場合の実行結果と比較することで、以下の2点が不具合の原因であると確認しました。また、以下の2点は、サブグラフ特有のバグである点も確認しました。

• SupervisorとSub Agent間のStateの連携がなされない
• streamメソッドの出力に、Sub Agentの最終ノードの情報が含まれない

上記の回避策として、サブグラフにはCommandを利用せず、add_edgeを利用することで、期待通りの挙動となることを確認しております。その際、サブグラフと親グラフ間のエッジをadd_edgeで定義することが必要です。

本件については、LangGraphのGitHubのIssue（#3115, #3362）にて報告しており、開発チームと議論を行っています。執筆時点（2025/03/12）ではまだ未解決ですが、Issueの進展があり次第、本記事でも情報を更新する予定です。

まとめ

本記事では、LangGraphとAmazon Bedrockを利用し、Supervisor型のMulti-Agentシステムで、複数のAgentic WorkflowをSub Agentとして利用する方法を解説しました。また、LangGraphで、Agentic WorkflowをSub Agentとして利用するための機能であるSubGraphと、Agent間で制御を委譲する機能であるhandoff（Command）について解説しました。執筆時点（2025/03/12）において、Multi-Agentの実装事例が少ない中で、本記事が参考になれば幸いです。

本記事の内容は、弊社でのAWS Japan生成AI実用化推進プログラムでの取り組みの一環です。プログラムにおける取り組み内容は、以下のnoteにて外部発信していますのでぜひご覧ください！

謝辞

記事執筆にあたり、同チームの藤田さんにレビューとアドバイスをいただきました。
この場を借りて、心より感謝申し上げます。

仲間募集

NTTデータ テクノロジーコンサルティング事業本部 では、以下の職種を募集しています。

1. クラウド技術を活用したデータ分析プラットフォームの開発・構築(ITアーキテクト/クラウドエンジニア)

クラウド／プラットフォーム技術の知見に基づき、DWH、BI、ETL領域におけるソリューション開発を推進します。
https://enterprise-aiiot.nttdata.com/recruitment/career_sp/cloud_engineer

2. データサイエンス領域（データサイエンティスト／データアナリスト）

データ活用／情報処理／AI／BI／統計学などの情報科学を活用し、よりデータサイエンスの観点から、データ分析プロジェクトのリーダーとしてお客様のDX／デジタルサクセスを推進します。
https://enterprise-aiiot.nttdata.com/recruitment/career_sp/datascientist

3.お客様のAI活用の成功を推進するAIサクセスマネージャー

DataRobotをはじめとしたAIソリューションやサービスを使って、
お客様のAIプロジェクトを成功させ、ビジネス価値を創出するための活動を実施し、
お客様内でのAI活用を拡大、NTTデータが提供するAIソリューションの利用継続を推進していただく人材を募集しています。
https://nttdata.jposting.net/u/job.phtml?job_code=804

4.DX／デジタルサクセスを推進するデータサイエンティスト《管理職/管理職候補》

データ分析プロジェクトのリーダとして、正確な課題の把握、適切な評価指標の設定、分析計画策定や適切な分析手法や技術の評価・選定といったデータ活用の具現化、高度化を行い分析結果の見える化・お客様の納得感醸成を行うことで、ビジネス成果・価値を出すアクションへとつなげることができるデータサイエンティスト人材を募集しています。

https://nttdata.jposting.net/u/job.phtml?job_code=898

ソリューション紹介

Trusted Data Foundationについて

～データ資産を分析活用するための環境をオールインワンで提供するソリューション～
https://www.nttdata.com/jp/ja/lineup/tdf/
最新のクラウド技術を採用して弊社が独自に設計したリファレンスアーキテクチャ（Datalake+DWH+AI/BI）を顧客要件に合わせてカスタマイズして提供します。
可視化、機械学習、DeepLearningなどデータ資産を分析活用するための環境がオールインワンで用意されており、これまでとは別次元の量と質のデータを用いてアジリティ高くDX推進を実現できます。

TDFⓇ-AM（Trusted Data Foundation - Analytics Managed Service）について

～データ活用基盤の段階的な拡張支援（Quick Start）と保守運用のマネジメント（Analytics Managed）をご提供することでお客様のDXを成功に導く、データ活用プラットフォームサービス～
https://www.nttdata.com/jp/ja/lineup/tdf_am/
TDFⓇ-AMは、データ活用をQuickに始めることができ、データ活用の成熟度に応じて段階的に環境を拡張します。プラットフォームの保守運用はNTTデータが一括で実施し、お客様は成果創出に専念することが可能です。また、日々最新のテクノロジーをキャッチアップし、常に活用しやすい環境を提供します。なお、ご要望に応じて上流のコンサルティングフェーズからAI/BIなどのデータ活用支援に至るまで、End to Endで課題解決に向けて伴走することも可能です。

NTTデータとDatabricksについて

NTTデータは、お客様企業のデジタル変革・DXの成功に向けて、「databricks」のソリューションの提供に加え、情報活用戦略の立案から、AI技術の活用も含めたアナリティクス、分析基盤構築・運用、分析業務のアウトソースまで、ワンストップの支援を提供いたします。

https://www.nttdata.com/jp/ja/lineup/databricks/

NTTデータとTableauについて

ビジュアル分析プラットフォームのTableauと2014年にパートナー契約を締結し、自社の経営ダッシュボード基盤への採用や独自のコンピテンシーセンターの設置などの取り組みを進めてきました。さらに2019年度にはSalesforceとワンストップでのサービスを提供開始するなど、積極的にビジネスを展開しています。

これまでPartner of the Year, Japanを4年連続で受賞しており、2021年にはアジア太平洋地域で最もビジネスに貢献したパートナーとして表彰されました。
また、2020年度からは、Tableauを活用したデータ活用促進のコンサルティングや導入サービスの他、AI活用やデータマネジメント整備など、お客さまの企業全体のデータ活用民主化を成功させるためのノウハウ・方法論を体系化した「デジタルサクセス」プログラムを提供開始しています。
https://www.nttdata.com/jp/ja/lineup/tableau/

NTTデータとAlteryxについて

Alteryxは、業務ユーザーからIT部門まで誰でも使えるセルフサービス分析プラットフォームです。

Alteryx導入の豊富な実績を持つNTTデータは、最高位にあたるAlteryx Premiumパートナーとしてお客さまをご支援します。

導入時のプロフェッショナル支援など独自メニューを整備し、特定の業種によらない多くのお客さまに、Alteryxを活用したサービスの強化・拡充を提供します。

https://www.nttdata.com/jp/ja/lineup/alteryx/

NTTデータとDataRobotについて

DataRobotは、包括的なAIライフサイクルプラットフォームです。

NTTデータはDataRobot社と戦略的資本業務提携を行い、経験豊富なデータサイエンティストがAI・データ活用を起点にお客様のビジネスにおける価値創出をご支援します。

https://www.nttdata.com/jp/ja/lineup/datarobot/

NTTデータとInformaticaについて

データ連携や処理方式を専門領域として10年以上取り組んできたプロ集団であるNTTデータは、データマネジメント領域でグローバルでの高い評価を得ているInformatica社とパートナーシップを結び、サービス強化を推進しています。
https://www.nttdata.com/jp/ja/lineup/informatica/

NTTデータとSnowflakeについて

NTTデータでは、Snowflake Inc.とソリューションパートナー契約を締結し、クラウド・データプラットフォーム「Snowflake」の導入・構築、および活用支援を開始しています。

NTTデータではこれまでも、独自ノウハウに基づき、ビッグデータ・AIなど領域に係る市場競争力のあるさまざまなソリューションパートナーとともにエコシステムを形成し、お客さまのビジネス変革を導いてきました。
Snowflakeは、これら先端テクノロジーとのエコシステムの形成に強みがあり、NTTデータはこれらを組み合わせることでお客さまに最適なインテグレーションをご提供いたします。

https://www.nttdata.com/jp/ja/lineup/snowflake/