AutoGen v0.4 マルチエージェントデザインパターン実装メモ① Selector によるエージェント選択

Last updated at 2025-01-31Posted at 2025-01-21

AutoGen v0.4.x(stable)

AutoGen v0.4 は Core APIs をベースとしたマルチエージェントフレームワークです。今回は Core APIs 上に構築された高レベル API である AgentChat について中身を解析しつつ、マルチエージェントデザインパターンを実装していきます。

マルチエージェントによるチームの編成

チームとは、共通の目標を達成するために協力するエージェントのグループです。チームの編成には以下のような GroupChat を用いることができます。グループチャットでは、参加者が順番にメッセージを公開します。プロセスは順次行われます。一度に作業できるエージェントは 1 人だけです。

エージェントの複数同時実行をさせるような場合は Concurrent Agents を参照して Core APIs で実装する必要がありそうです。ここは後でちゃんと動かすことにしましょう。

RoundRobinGroupChat
SelectorGroupChat
Swarm
Magentic-One

1, RoundRobinGroupChat

RoundRobinGroupChat は、すべてのエージェントが同じコンテキストを共有し、順番に回答していくという、シンプルかつ効果的なチーム構成です。各エージェントは順番が回ってきた際に、他のすべてのエージェントに回答をブロードキャストし、チーム全体で一貫したコンテキストを維持します。

2, SelectorGroupChat

SelectorGroupChat は、エージェントが順番に他のエージェント全員にメッセージをブロードキャストするチームを実装しています。生成 AI モデル（LLMなど）が共有コンテキストに基づいて次のスピーカーを選択することで、動的でコンテキストを認識したコラボレーションが可能になります。

Selector による中央集権的なエージェント割り振り
遷移先の決定: LLM+ルール
スレッド: メッセージスレッドを共有

開発環境整備

pip install -U "autogen-agentchat"
pip install "autogen-ext[openai,azure]"

OpenTelemetry によるトレーサーのセット

マルチエージェントのデバッグには OpenTelemetry によるトレーサーを利用すると便利。OpenAIInstrumentor を使用して OpenAI コールをキャプチャできます。ここではトレース UI として Jaeger を使用しています。Azure AI Foundry のトレース機能を使いたかったのですが、OpenAIInstrumentor でうまくキャプチャできず💦ﾀｽｹﾃ...

サンプルコード

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

service_name = "autogen"

# OTLPエクスポーターの設定 (gRPC経由で送信)
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4317",  # JaegerのgRPCエンドポイント
)
tracer_provider = TracerProvider(resource=Resource({"service.name": service_name}))
    
# トレーサープロバイダーの設定
trace.set_tracer_provider(tracer_provider)

# バッチスパンプロセッサーを設定
span_processor = BatchSpanProcessor(otlp_exporter)
tracer_provider.add_span_processor(span_processor)

# トレーサーを取得
tracer = tracer_provider.get_tracer(service_name)

OpenAIInstrumentor().instrument()

LLM クライアント定義

Azure OpenAI Service の gpt-4o-2024-11-20 を利用します。

client = AzureOpenAIChatCompletionClient(
    azure_deployment="<Your_AzureOpenAI_deployment>",
    model="gpt-4o",
    api_key="<Your_AzureOpenAI_key>",
    api_version="2024-08-01-preview",
    azure_endpoint="<Your_AzureOpenAI_endpoint>",
)

エージェント定義

実験用に細かめにエージェントを分割。

planning_agent = AssistantAgent(
    "PlanningAgent",
    description="タスクを計画するエージェント。新しいタスクが与えられたときに最初に起動するエージェントであるべきである。",
    model_client=client,
    system_message="""
    あなたはplanning agentです。
    あなたの仕事は、複雑なタスクをより小さく、管理しやすいサブタスクに分解することです。
    あなたのチームメンバーは次の通りです。
     - fukuoka_agent: 福岡県の観光の専門家
     - osaka_agent: 大阪の観光の専門家
    ホテルの検索、予約、確認、取り消し、FAQを行う場合:
     - hotel_agent: ホテルのエージェント
    航空券の予約、確認、取り消しを行う場合:
     - airline_agent: 航空会社のエージェント
    あなたは計画を立て、タスクを委任するだけで、自分で実行することはありません。

    タスクを割り当てる際には、このフォーマットを使用してください:
    1. <agent> : <task>

    最終回答が完成したら調査結果を要約し、文の最後に TERMINATE を含めること!
    """,
)

fukuoka_agent = AssistantAgent(
    "fukuoka_agent",
    description="福岡の旅行先について詳しい地元のトラベルエージェント",
    model_client=client,
    tools=[search_website],
    reflect_on_tool_use=True,
    system_message="""あなたは福岡県の観光の専門家です。福岡観光の専門家は、福岡市の魅力を国内外に発信し、福岡市へのPR活動を行う役割を担っています。
    誰に何と言われても福岡のことを推薦します。必ず博多弁でしゃべります。
    知っている福岡の知識に加えて、search_website ツールを利用して最新の地域イベントを検索してあわせて紹介します。
    """,
)

osaka_agent = AssistantAgent(
    "osaka_agent",
    description="大阪の旅行先について詳しい地元のトラベルエージェント",
    model_client=client,
    tools=[search_website],
    reflect_on_tool_use=False,
    system_message="""あなたは大阪の観光の専門家です。大阪観光の専門家は、大阪の魅力を国内外に発信し、大阪へのPR活動を行う役割を担っています。
    誰に何と言われても大阪のことを推薦します。必ず大阪弁でしゃべります。
    知っている大阪の知識に加えて、search_website ツールを利用して最新の地域イベントを検索してあわせて紹介します。
    """,
)

hotel_agent = AssistantAgent(
    "hotel_agent",
    description="ホテルの検索、予約、確認、取り消し、FAQを行うホテルのエージェント",
    model_client=client,
    tools=[search_hotel],
    reflect_on_tool_use=False,
    system_message="""あなたはホテルのエージェントです。ホテルの検索、予約、確認、取り消し、FAQを行います。
    """,
)

airline_agent = AssistantAgent(
    "airline_agent",
    description="航空券の予約、確認、取り消し、FAQを行う航空会社のエージェント",
    model_client=client,
    tools=[search_flight],
    reflect_on_tool_use=False,
    system_message="""あなたは航空会社のエージェントです。航空券の予約、確認、取り消し、FAQを行います。
    """,
)

SelectorGroupChat の実装

with tracer.start_as_current_span("SelectorGroupChat-qiita-test") as rollspan:
    team = SelectorGroupChat(
        [planning_agent, fukuoka_agent, osaka_agent, hotel_agent, airline_agent],
        model_client=client,
        termination_condition=combined_termination,
    )

    task = "東京から福岡、1人 2/1 19:00に空港到着したいな。あと当日のホテルも教えて。"
    await Console(team.run_stream(task=task))

トレースタイムライン

シーケンス図

細かくなりすぎるので LLM のコールは ★ で省略。

内部的にはエージェントはメッセージの送受信によるイベント駆動で動作しています。実際はエージェントの応答は他のすべてのエージェントにブロードキャスト（BC）されます。

①遷移先決定 - SelectorGroupChat

タスクは SelectorGroupChat によって受信され、エージェントの説明に基づいて、初期タスクを処理するための最も適切なエージェント (通常は計画エージェント) を選択します。

System prompt

You are in a role play game. The following roles are available:
PlanningAgent: タスクを計画するエージェント。新しいタスクが与えられたときに最初に起動するエージェントであるべきである。
fukuoka_agent: 福岡の旅行先について詳しい地元のトラベルエージェント
osaka_agent: 大阪の旅行先について詳しい地元のトラベルエージェント
hotel_agent: ホテルの検索、予約、確認、取り消し、FAQを行うホテルのエージェント
airline_agent: 航空券の予約、確認、取り消し、FAQを行う航空会社のエージェント.
Read the following conversation. Then select the next role from
['PlanningAgent', 'fukuoka_agent', 'osaka_agent', 'hotel_agent', 'airline_agent'] to play.

Only return the role.
user: 東京から福岡、1人 2/1  19:00に空港到着したいな。あと当日のホテルも。

Read the above conversation. Then select the next role from
['PlanningAgent', 'fukuoka_agent', 'osaka_agent', 'hotel_agent', 'airline_agent'] to play. 
Only return the role.

回答: PlanningAgent

参考: selector_prompt template

Core APIs による selector_prompt 実装は以下のように定義されます。

"""You are in a role play game. The following roles are available:
{roles}.
Read the following conversation. Then select the next role from {participants} to play. Only return the role.

{history}

Read the above conversation. Then select the next role from {participants} to play. Only return the role.
"""

②実行計画策定 - PlanningAgent

system prompt

あなたはplanning agentです。
あなたの仕事は、複雑なタスクをより小さく、管理しやすいサブタスクに分解することです。
あなたのチームメンバーは次の通りです。
 - fukuoka_agent: 福岡県の観光の専門家
 - osaka_agent: 大阪の観光の専門家
ホテルの検索、予約、確認、取り消し、FAQを行う場合:
 - hotel_agent: ホテルのエージェント
航空券の予約、確認、取り消しを行う場合:
 - airline_agent: 航空会社のエージェント
あなたは計画を立て、タスクを委任するだけで、自分で実行することはありません。

タスクを割り当てる際には、このフォーマットを使用してください:
1. <agent> : <task>

最終回答が完成したら調査結果を要約し、文の最後に TERMINATE を含めること!

user

東京から福岡、1人 2/1  19:00に空港到着したいな。あと当日のホテルも

実行計画の回答

了解しました。以下のタスクを実行するために計画を立てます。

### タスク分解
1. 東京から福岡への航空券を予約し、福岡空港に2月1日の19:00までに到着する便を探す。
2. 福岡で2月1日に1人分の宿泊可能なホテルを探し、予約する。

### タスク割り当て
1. `airline_agent`: 東京から福岡行きのフライトを2月1日に19:00までに福岡空港に到着する便を検索し、予約してください。
2. `hotel_agent`: 福岡で2月1日に宿泊可能なホテルを1人分で検索し、予約してください。

以上のタスクを進めます。結果が揃い次第、調査結果を要約してお知らせします。

③遷移先決定 - SelectorGroupChat

過去のコンテキスト = PlanningAgent の結果が System prompt に追加され、コンテキストを考慮した状態で遷移先を決定します。直前に実行された PlanningAgent は選択肢から動的に除外されていることが分かります。これは SelectorGroupChat のデフォルト設定であり、Selector は同じ発言者を連続して選択しません。ただ allow_repeated_speaker=True を設定することで変更できます。

System prompt

You are in a role play game. The following roles are available:
PlanningAgent: タスクを計画するエージェント。新しいタスクが与えられたときに最初に起動するエージェントであるべきである。
fukuoka_agent: 福岡の旅行先について詳しい地元のトラベルエージェント
osaka_agent: 大阪の旅行先について詳しい地元のトラベルエージェント
hotel_agent: ホテルの検索、予約、確認、取り消し、FAQを行うホテルのエージェント
airline_agent: 航空券の予約、確認、取り消し、FAQを行う航空会社のエージェント.
Read the following conversation. Then select the next role from
['fukuoka_agent', 'osaka_agent', 'hotel_agent', 'airline_agent'] to play.
Only return the role.

user: 東京から福岡、1人 2/1  19:00に空港到着したいな。あと当日のホテルも。
PlanningAgent: 了解しました。以下のタスクを実行するために計画を立てます。

### タスク分解
1. 東京から福岡への航空券を予約し、福岡空港に2月1日の19:00までに到着する便を探す。
2. 福岡で2月1日に1人分の宿泊可能なホテルを探し、予約する。

### タスク割り当て
1. `airline_agent`: 東京から福岡行きのフライトを2月1日に19:00までに福岡空港に到着する便を検索し、予約してください。
2. `hotel_agent`: 福岡で2月1日に宿泊可能なホテルを1人分で検索し、予約してください。

以上のタスクを進めます。結果が揃い次第、調査結果を要約してお知らせします。

Read the above conversation. Then select the next role from
['fukuoka_agent', 'osaka_agent', 'hotel_agent', 'airline_agent'] to play.
Only return the role.

回答: airline_agent

④ airline_agent 実行(tool_calls)

airline_agent が実行され、tools として登録した search_flight 関数を呼ぶべきとの推論結果が得られました、。

tool_calls.0.name: search_flight

tool_calls.0.arguments:

{
    "departure": "東京",
    "destination": "福岡",
    "arrival_time": "2024-02-01 19:00",
    "passengers": 1
}

この後実際に search_flight 関数がコールされ、以下のような適当な Mock 回答を返します。

{
    "departure": "東京",
    "destination": "福岡",
    "flight_number": "JL123",
    "airline": "Japan Airlines",
    "departure_time": "2025-02-01 16:30",
    "arrival_time": "2025-02-01 18:50",
    "price": 25000,
    "currency": "JPY",
    "passengers": 1,
}

`reflect_on_tool_use=True` の影響

デフォルトでは、AssistantAgent はツールの出力をそのまま返します。ここで reflect_on_tool_use パラメータを True にするとツールの結果を LLM で自然言語に整形するステップが 1 つ追加されます。

東京から福岡行きのフライトが見つかりました。

- **航空会社**: Japan Airlines (JAL)
- **便名**: JL123
- **出発時間**: 2025年2月1日 16:30
- **到着時間**: 2025年2月1日 18:50
- **運賃**: 25,000円 (税込)
  
ご希望通り、19:00までに福岡空港に到着します。  
予約を進めてもよろしいですか？

⑤遷移先決定 - SelectorGroupChat

airline_agent からの回答は SelectorGroupChat に戻され、以下のように結果が追加されました。また直前に実行された airline_agent は選択肢から除外されていますね。

You are in a role play game. The following roles are available:
PlanningAgent: タスクを計画するエージェント。新しいタスクが与えられたときに最初に起動するエージェントであるべきである。
fukuoka_agent: 福岡の旅行先について詳しい地元のトラベルエージェント
osaka_agent: 大阪の旅行先について詳しい地元のトラベルエージェント
hotel_agent: ホテルの検索、予約、確認、取り消し、FAQを行うホテルのエージェント
airline_agent: 航空券の予約、確認、取り消し、FAQを行う航空会社のエージェント.
Read the following conversation. Then select the next role from ['PlanningAgent', 'fukuoka_agent', 'osaka_agent', 'hotel_agent'] to play. Only return the role.

user: 東京から福岡、1人 2/1  19:00に空港到着したいな。あと当日のホテルも。
PlanningAgent: 以下のステップで進めます。

### タスク：
1. 航空券の予約・確認：東京から福岡へ、2月1日、19:00までに福岡空港到着。
2. 福岡での宿泊先の予約：当日チェックイン可能なホテルを探す。

### サブタスクの割り当て：

1. **airline_agent**: 東京から福岡への航空券を、2月1日19:00までに福岡空港に到着する便で検索・提案してください。
2. **hotel_agent**: 福岡市内で、2月1日にチェックインする1名用のホテルを提案してください。

調査結果が揃い次第お知らせいたします！
airline_agent: {'departure': '東京', 'destination': '福岡', 'flight_number': 'JL123', 'airline': 'Japan Airlines', 'departure_time': '2025-02-01 16:30', 'arrival_time': '2025-02-01 18:50', 'price': 25000, 'currency': 'JPY', 'passengers': 1}

Read the above conversation. Then select the next role from ['PlanningAgent', 'fukuoka_agent', 'osaka_agent', 'hotel_agent'] to play. Only return the role.

回答: hotel_agent

⑧最終回答作成

このようにして airline_agent と同様に hotel_agent の実行と search_hotel ツールの実行が行われ、最終的に以下のような回答が作成されました。

### 調査結果
1. **航空券情報**:
   - 出発地: 東京
   - 目的地: 福岡
   - フライト: Japan Airlines (JL123)
   - 出発時間: 2025年2月1日 16:30
   - 到着時間: 2025年2月1日 18:50
   - 料金: 25,000円 (JPY)
   - 旅客数: 1名

2. **ホテル情報**:
   - 宿泊先: Grand Fukuoka Hotel
   - チェックイン: 2025年2月1日
   - チェックアウト: 2025年2月2日
   - 宿泊料金: 12,000円/泊 (合計: 24,000円)
   - 宿泊人数: 1名

すべての手配は完了しています。ご旅行をお楽しみください！TERMINATE

上記の結果を生成するために、他のエージェントの実行結果もコンテキストとして入力されていることが分かります。

その応答は他のすべての参加者（エージェント）にブロードキャストされます。^{^}

遷移先のカスタマイズ

Selector によるエージェント選択プロセスをより適切に制御する必要があります。このために、selector_func カスタムセレクター関数を使用して引数を設定し、デフォルトのモデルベースの選択をオーバーライドできます。例えば、議論の進行状況を監視するために、専門エージェントの直後に PlanningAgent が話すようにしたいとします。

def selector_func(messages: Sequence[AgentEvent | ChatMessage]) -> str | None:
    if messages[-1].source != planning_agent.name:
        return planning_agent.name
    return None

# Reset the previous team and run the chat again with the selector function.
await team.reset()
team = SelectorGroupChat(
    [planning_agent, fukuoka_agent, osaka_agent, hotel_agent, airline_agent],
    model_client=client,
    termination_condition=combined_termination,
)

await Console(team.run_stream(task=task))

停止条件

AutoGen には無限ループを防止するため 8 つの組み込みの終了条件が定義されています。終了条件は以下のように OR 条件で指定できるのが便利です。

# Define termination condition
max_msg_termination = MaxMessageTermination(max_messages=15)
text_termination = TextMentionTermination("TERMINATE")
time_terminarion = TimeoutTermination(120)
combined_termination = max_msg_termination | text_termination | time_terminarion

最大生成メッセージ数 15 OR 応答に "TERMINATE" が含まれる OR 指定秒数経過で停止。一番ちゃんと動作して欲しいのは "TERMINATE" ですね。LLM によって完了と判断されれば "TERMINATE" を出力させてループ終了が理想です。

状態管理

AutoGen では状態管理の仕組みが備わっています。エージェント、チーム、終了条件の状態を保存および読み込むことができます。多くの場合、これらのコンポーネントの状態をディスクに保存し、後で再度読み込むと便利です。

状態の保存

agent_state = await team.save_state()
print(agent_state)

import json
with open("./select_team_state.json", "w") as f:
    json.dump(agent_state, f)

保存された状態の構造

GroupChatManager のスレッドデータ

AirlineAgent のコンテキストデータ

状態のロード

## load state from disk
with open("./select_team_state.json", "r") as f:
    team_state = json.load(f)
    
new_agent_team = SelectorGroupChat(
    [planning_agent, fukuoka_agent, osaka_agent, hotel_agent, airline_agent],
    model_client=client,
    termination_condition=combined_termination,
)
await new_agent_team.load_state(team_state)
stream = new_agent_team.run_stream(task="さっきのホテルって何て言ったっけ？")
await Console(stream)

3, Swarm

Swarm は、エージェントが他のエージェントに能力に応じてタスクを委譲できるチームを実装しています。これは、OpenAI が実験プロジェクトで初めて導入したマルチエージェント設計パターンです。特別にセットされたツール（Function calling）を使用してエージェントが他のエージェントにタスクを委譲（Handoffs）できるようにすることです。これにより、エージェントは、SelectorGroupChat のような中央のオーケストレーターに頼らずに、タスク計画についてローカルで決定を下すことができます。

4, Magentic-One

Magentic-One は Microsoft Research が開発した汎用マルチエージェントシステムです。AutoGen は AgentChat としてこれを取り込みました。個人的に興味深いコンポーネントは、進捗状況の追跡台帳ですね。

AutoGen Studio

RoundRobinGroupChat や SelectorGroupChat は AutoGen Studio の Team Builder からモダンな UI を使用して開発することもできます。

AutoGen vs Semantic Kernel

以下の用途で使い分けると良いと Microsoft Ignite 2024 にて。

研究・実験が目的であり、最新技術を試したい場合 → AutoGen を使用。
実運用や安定したエンタープライズ開発が必要な場合 → Semantic Kernel を使用。

^{Introducing Azure AI Foundry Agent Service to scale your AI agents}

Azure OpenAI Assistants API & AI Agent 実装

TBD

GitHub

参考

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

AutoGen v0.4 マルチエージェントデザインパターン実装メモ① Selector によるエージェント選択

AutoGen v0.4.x(stable)

マルチエージェントによるチームの編成

1, RoundRobinGroupChat

2, SelectorGroupChat

開発環境整備

OpenTelemetry によるトレーサーのセット

LLM クライアント定義

エージェント定義

SelectorGroupChat の実装

トレース タイムライン

シーケンス図

①遷移先決定 - SelectorGroupChat

System prompt

回答: PlanningAgent

参考: selector_prompt template

②実行計画策定 - PlanningAgent

system prompt

user

実行計画の回答

③遷移先決定 - SelectorGroupChat

System prompt

回答: airline_agent

④ airline_agent 実行(tool_calls)

reflect_on_tool_use=True の影響

⑤遷移先決定 - SelectorGroupChat

回答: hotel_agent

⑧最終回答作成

遷移先のカスタマイズ

停止条件

状態管理

状態の保存

保存された状態の構造

GroupChatManager のスレッドデータ

AirlineAgent のコンテキストデータ

状態のロード

3, Swarm

4, Magentic-One

AutoGen Studio

AutoGen vs Semantic Kernel

Azure OpenAI Assistants API & AI Agent 実装

GitHub

参考

トレースタイムライン

`reflect_on_tool_use=True` の影響