Multi-agent Systemsの翻訳です。

本書は著者が手動で翻訳したものであり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。

エージェントとは、アプリケーションの制御フローを決定するためにLLMを用いるシステムです。これらのシステムを開発していくと、これらは時間と共により複雑になり、管理やスケールが困難になります。例えば、以下のような問題に遭遇するかもしれません:

エージェントが利用できるツールが大量にあり、次にどのツールを呼び出すのかに関する意思決定が貧弱になる。
単一のエージェントが追跡するにはコンテキストが複雑になり過ぎてしまう。
システムにおいて複数の特化領域が必要となる(プランナー、リサーチャー、数学の専門家など)。

これらに取り組むために、あたなのアプリケーションを複数の小規模かつ独立したエージェントに分割し、マルチエージェントシステムとして構成したいと考えるかもしれません。これら独立したエージェントはプロンプトやLLM呼び出しくらいにシンプルにすることができますし、ReActエージェントのように複雑にすることもできます(これら以外にも！)。

マルチエージェントシステムを活用することによる主要なメリットは:

モジュール性: 個別のエージェントによって、エージェントシステムの開発、テスト、維持が簡単に。
特化: 全体的なシステムのパフォーマンスの助けとなる、特定ドメインにフォーカスした専門家エージェントを作成することが可能。
コントロール: (関数呼び出しに依存するのではなく)エージェントがどのようにコミュニケーションするのかを明示的に制御するとこが可能。

マルチエージェントのアーキテクチャ

マルチエージェントシステムでエージェントを接続するには複数の方法があります。

ネットワーク: それぞれのエージェントは他のすべてのエージェントとコミュニケーションできます。すべてのエージェントが次にどのエージェントを呼び出すのかを決定できます。
スーパーバイザー: それぞれのエージェントは単一のスーパーバイザーエージェントとコミュニケーションします。スーパーバイザーエージェントはどのエージェントを次に呼び出すのかを決定します。
スーパーバイザー(ツール呼び出し): これはスーパーバイザーアーキテクチャの特殊なケースです。個々のエージェントはツールとして表現することができます。このケースでは、スーパーバイザーエージェントはどのエージェントツール呼び出すのかを決定するために、ツール呼び出しLLMとそれらのエージェントに引き渡す引数を使用します。
階層型: スーパーバイザーのスーパーバイザーを用いてマルチエージェントシステムを定義することができます。これは、スーパーバイザーアーキテクチャを汎化したものであり、より複雑な制御フローを可能にします。
カスタムのマルチエージェントワークフロー: それぞれのエージェントはエージェントのサブセットのみとコミュニケーションします。フローのパーツは決定論的であり、いくつかのエージェントのみが他のどのエージェントを呼び出すのかを決定することができます。

ハンドオフ

マルチエージェントアーキテクチャにおいては、エージェントはグラフのノードとして表現することができます。それぞれのエージェントノードはそのステップを実行し、実行を終了するか他のエージェントにルーティングするのかを決定します。ルーティング先には自分自身が含まれる場合があります(ループの実行など)。マルチエージェントのインタラクションにおける一般的なパターンは、一つのエージェントが制御を他に引き渡すハンドオフです。ハンドオフによって、以下を指定することができます:

宛先: 移動先のターゲットエージェント(移動先のノード名など)
ペイロード: 当該エージェントに引き渡す情報(状態のアップデートなど)

LangGraphでハンドオフを実装するために、エージェントノードではコントロールフロート状態のアップデートの両方を組み合わせることができる、Commandオブジェクトを返却することができます:

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # the condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc.
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # Specify which agent to call next
        goto=goto,
        # Update the graph state
        update={"my_state_key": "my_state_value"}
    )

それぞれのエージェントノード自身がグラフであるような(サブグラフなど)より複雑なシナリオにおいては、エージェントのサブグラフの一つにあるノードは、別のエージェントにナビゲートしたい場合があるかもしれません。例えば、aliceとbobの2つのエージェント(親のグラフにおけるサブグラフのノード)があり、aliceがbobにナビゲートする必要がある場合、Commandオブジェクトでgraph=Command.PARENTを設定することができます。

def some_node_inside_alice(state)
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        # specify which graph to navigate to (defaults to the current graph)
        graph=Command.PARENT,
    )

注意
Command(graph=Command.PARENT)を用いてコミュニケーションを行うサブグラフの可視化をサポートする必要がある場合には、例えば以下のようにするのではなく、Commandアノテーションを持ちいったノード関数でそれらをラッピングする必要があります:

builder.add_node(alice)

以下のようにする必要があるでしょう:

def call_alice(state) -> Command[Literal["bob"]]:
    return alice.invoke(state)

builder.add_node("alice", call_alice)

ツールとしてのハンドオフ

最も一般的なエージェントタイプの一つは、ReActスタイルのツール呼び出しエージェントです。これらのタイプのエージェントにおける一般的なパターンは、以下のようにツール呼び出しでハンドオフをラッピングするというものです:

def transfer_to_bob(state):
    """Transfer to bob."""
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        graph=Command.PARENT,
    )

これは、状態のアップデートに加えて制御フローも含まれる、ツールからグラフの状態を更新する特殊なケースです。

重要
Commandを返すツールを使いたい場合には、構築済みのcreate_react_agent / ToolNodeを使うか、以下のようにツールによって返却されるCommandオブジェクトを収集し、それらのリストを返却する自身のツール実行ノードを実装することができます:

def call_tools(state):
    ...
    commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
    return commands

それでは、様々なマルチエージェントアーキテクチャを詳しく見ていきましょう。

ネットワーク

このアーキテクチャでは、エージェントはグラフのノードとして定義されます。それぞれのエージェントは他のすべてのエージェントとコミュニケーションを行い、どのエージェントを次に呼び出すのかを決定することができます。このアーキテクチャは、明確なエージェントの階層構造を持たない問題や、エージェントの呼び出しに特定の順序がない問題に適しています。

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END

model = ChatOpenAI()

def agent_1(state: MessagesState) -> Command[Literal["agent_2", "agent_3", END]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which agent to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_agent" field)
    response = model.invoke(...)
    # route to one of the agents or exit based on the LLM's decision
    # if the LLM returns "__end__", the graph will finish execution
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

def agent_2(state: MessagesState) -> Command[Literal["agent_1", "agent_3", END]]:
    response = model.invoke(...)
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

def agent_3(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    ...
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
builder.add_node(agent_3)

builder.add_edge(START, "agent_1")
network = builder.compile()

API Reference: ChatOpenAI | Command | StateGraph | START | END

スーパーバイザー

このアーキテクチャにおいては、ノードとしてエージェントを定義し、どのエージェントノードを次に呼び出すべきかを決定するスーパーバイザーノード(LLM)を追加します。スーパーバイザーの決定に基づいた適切なエージェントノードに実行をルーティングするためにCommandを使用します。また、このアーキテクチャは複数のエージェントの並列実行やmap-reduceパターンの使用に適しています。

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END

model = ChatOpenAI()

def supervisor(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which agent to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_agent" field)
    response = model.invoke(...)
    # route to one of the agents or exit based on the supervisor's decision
    # if the supervisor returns "__end__", the graph will finish execution
    return Command(goto=response["next_agent"])

def agent_1(state: MessagesState) -> Command[Literal["supervisor"]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # and add any additional logic (different models, custom prompts, structured output, etc.)
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

def agent_2(state: MessagesState) -> Command[Literal["supervisor"]]:
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

builder = StateGraph(MessagesState)
builder.add_node(supervisor)
builder.add_node(agent_1)
builder.add_node(agent_2)

builder.add_edge(START, "supervisor")

supervisor = builder.compile()

API Reference: ChatOpenAI | Command | StateGraph | START | END

スーパーバイザーマルチエージェントアーキテクチャの例については、こちらのチュートリアルをチェックしてください。

スーパーバイザー(ツール呼び出し)

スーパーバイザーアーキテクチャのこの派生形では、個々のエージェントをツールとして定義し、スーパーバイザーノードでツール呼び出しLLMを使用します。これは、LLMノード(スーパーバイザー)とツール(この場合はエージェント)を実行するツール呼び出しノードの2つのノードを持つReActスタイルのエージェントとして実装することができます。

from typing import Annotated
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import InjectedState, create_react_agent

model = ChatOpenAI()

# this is the agent function that will be called as tool
# notice that you can pass the state to the tool via InjectedState annotation
def agent_1(state: Annotated[dict, InjectedState]):
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # and add any additional logic (different models, custom prompts, structured output, etc.)
    response = model.invoke(...)
    # return the LLM response as a string (expected tool response format)
    # this will be automatically turned to ToolMessage
    # by the prebuilt create_react_agent (supervisor)
    return response.content

def agent_2(state: Annotated[dict, InjectedState]):
    response = model.invoke(...)
    return response.content

tools = [agent_1, agent_2]
# the simplest way to build a supervisor w/ tool-calling is to use prebuilt ReAct agent graph
# that consists of a tool-calling LLM node (i.e. supervisor) and a tool-executing node
supervisor = create_react_agent(model, tools)

API Reference: ChatOpenAI | InjectedState | create_react_agent

階層型

システムにより多くのエージェントを追加していくと、スーパーバイザーがそれら全てを管理することが非常に困難になる場合があります。スーパーバイザーはどのエージェントを次に呼び出すのかに関する意思決定をうまく行えなくなったり、単一のスーパーバイザーで追跡するにはコンテキストが複雑になり過ぎる場合があります。言い換えると、初めにマルチエージェントアーキテクチャの動機づけとなったのと同じ問題に陥ることになります。

これに対応するために、システムを階層的に設計することができます。例えば、個々のスーパーバイザーによって管理される特化したエージェントのチームと、チームを管理するトップレベルのスーパーバイザーを個別に作成することができます。

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command
model = ChatOpenAI()

# define team 1 (same as the single supervisor example above)

def team_1_supervisor(state: MessagesState) -> Command[Literal["team_1_agent_1", "team_1_agent_2", END]]:
    response = model.invoke(...)
    return Command(goto=response["next_agent"])

def team_1_agent_1(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})

def team_1_agent_2(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})

team_1_builder = StateGraph(Team1State)
team_1_builder.add_node(team_1_supervisor)
team_1_builder.add_node(team_1_agent_1)
team_1_builder.add_node(team_1_agent_2)
team_1_builder.add_edge(START, "team_1_supervisor")
team_1_graph = team_1_builder.compile()

# define team 2 (same as the single supervisor example above)
class Team2State(MessagesState):
    next: Literal["team_2_agent_1", "team_2_agent_2", "__end__"]

def team_2_supervisor(state: Team2State):
    ...

def team_2_agent_1(state: Team2State):
    ...

def team_2_agent_2(state: Team2State):
    ...

team_2_builder = StateGraph(Team2State)
...
team_2_graph = team_2_builder.compile()


# define top-level supervisor

builder = StateGraph(MessagesState)
def top_level_supervisor(state: MessagesState) -> Command[Literal["team_1_graph", "team_2_graph", END]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which team to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_team" field)
    response = model.invoke(...)
    # route to one of the teams or exit based on the supervisor's decision
    # if the supervisor returns "__end__", the graph will finish execution
    return Command(goto=response["next_team"])

builder = StateGraph(MessagesState)
builder.add_node(top_level_supervisor)
builder.add_node("team_1_graph", team_1_graph)
builder.add_node("team_2_graph", team_2_graph)
builder.add_edge(START, "top_level_supervisor")
builder.add_edge("team_1_graph", "top_level_supervisor")
builder.add_edge("team_2_graph", "top_level_supervisor")
graph = builder.compile()

API Reference: ChatOpenAI | StateGraph | START | END | Command

カスタムのマルチエージェントワークフロー

このアーキテクチャではカスタムのワークフローにおいて、グラフノードとして個々のエージェントを追加し、事前にどの順番でエージェントを呼び出すのかを定義します。LangGraphにおいては、2つの方法でワークフローを定義することができます:

明示的な制御フロー(通常のエッジ): LangGraphによって、通常のグラフエッジを通じてアプリケーションの制御フロー(エージェントがコミュニケーションする順番など)を明示的に定義することができます。これは、上述のアーキテクチャにおいて最も決定論的な派生形となります - 事前に我々はどのエージェントが次に呼び出されるのかを常に知っています。
動的な制御フロー(Command): LangGraphにおいては、アプリケーションの制御フローの部品をLLMに決定させることができます。これは、Commandを通じて達成することができます。これの特殊なケースは、スーパーバイザー(ツール呼び出し)です。その場合では、スーパーバイザーエージェントで使用されるツール呼び出しLLMは、どのツールを次に呼び出すのかに関する順序について意思決定を行います。

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START

model = ChatOpenAI()

def agent_1(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}

def agent_2(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
# define the flow explicitly
builder.add_edge(START, "agent_1")
builder.add_edge("agent_1", "agent_2")

API Reference: ChatOpenAI | StateGraph | START

エージェント間のコミュニケーション

マルチエージェントシステムを構築する際に最も重要なことは、エージェントがどのようにコミュニケーションするのかを明確にすることです。いくつかのことなる検討事項があります:

エージェントはグラフの状態、あるいはツール呼び出し経由でコミュニケーションするのか？
2つのエージェントが異なる状態スキーマを持っている場合にはどうするのか？
共有メッセージリストを通じてどのようにコミュニケーションするのか？

グラフの状態 vs ツール呼び出し

エージェント間でやり取りされる「ペイロード」とは何でしょうか？上で議論したアーキテクチャのほとんどにおいて、エージェントはグラフの状態を通じてコミュニケーションします。ツール呼び出しを用いたスーパーバイザーの場合は、ペイロードはツール呼び出しの引数となります。

グラフの状態

グラフの状態を通じてコミュニケーションを行うには、個々のエージェントがグラフのノードとして定義される必要があります。これらは、関数あるいはサブグラフ全体として追加することができます。グラフ実行のそれぞれのステップにおいて、エージェントノードはグラフの現在の状態を受け取り、エージェントコードを実行し、次のノードに更新された状態を引き渡します。

通常、エージェントノードは単一の状態スキーマを共有します。しかし、異なる状態スキーマを持つエージェントを設計したいと考えるかもしれません。

様々な状態スキーマ

あるエージェントでは、他のエージェントとは異なる状態スキーマを持たせる必要が出てくるかもしれません。例えば、検索エージェントはクエリーと収集したドキュメントだけを追跡できれば良いかもしれません。LangGraphでは、2つの方法でこれを達成できます:

個別の状態スキーマを用いてサブグラフエージェントを定義します。サブグラフと親のグラフ間で共有の状態キー(チャネル)がない場合は、親のグラフがサブグラフとどのようにコミュニケーションしたらいいのかがわかるように、入力 / 出力の変換処理を追加することが重要です。
全体的なグラフの状態スキーマとは異なる、プライベート入力状態スキーマを用いてエージェントノード関数を定義します。これによって、特定のエージェントの実行にのみ必要な情報を引き渡すことができます。

共有メッセージリスト

エージェントがコミュニケーションを行う最も一般的な方法は、共有状態チャネル、特にメッセージのリストを通じたものです。これは、エージェントによって共有される状態に常に少なくとも単一のチャネル(キー)が存在していることを前提としています。共有メッセージリストを通じたコミュニケーションを行う際には、追加の検討事項があります: エージェントはそれらの思考プロセスの完全な履歴を共有するのか、最終的な結果のみを共有するべきか？ということです。

完全な履歴の共有

エージェントは、他のすべてのエージェントと自分の思考プロセスの完全な履歴(スクラッチパッド)を共有することができます。このスクラッチパッドは通常、メッセージのリストのように見えるものとなります。完全な思考プロセスを共有することのメリットは、他のエージェントが優れた意思決定を行う助けとなり、全体としてのシステムの推論能力を改善するというものです。欠点は、エージェントの数が増加し、複雑性が増えると、「スクラッチパッド」はすぐに大きくなり、メモリー管理における更なる戦略が必要となるかもしれないということです。

最終的な結果の共有

エージェントは自身のプライベートの「スクラッチパッド」を持ち、他のエージェントとは最終結果のみを共有することができます。このアプローチは、多数のエージェントやより複雑なエージェントを持つシステムではよりよく動作する場合があります。この場合、異なる状態スキーマを持つエージェントを定義する必要があるでしょう。

ツールとして呼び出されるエージェントにおいては、スーパーバイザーはツールのスキーマに基づいて入力を特定します。さらに、LangGraphによって、実行時に個々のツールに状態を引き渡すことができ、配下のエージェントは必要に応じて親の状態にアクセスすることができます。