【AutoGen0.7.1】 SelectorGroupChat(4)　Sample CodeⅠ: 2025/8/2

エージェント

　概要

  • AutoGenの SelectorGroupChat を活用したマルチエージェント協調実行のデモコードを解説
  • 本プログラムでは、Planning / Search / Analysis / User 承認という役割を担うエージェントが協力しながら、1つの質問に対して分担して応答を構築

背景と目的

目的

  • AutoGen の SelectorGroupChat を用いて、AIエージェント同士の分業プロセスを実演
  • 計画・検索・分析をそれぞれ別の AI に担当させ、会話の流れに応じて自律的にエージェント選出

処理の流れ

  1. PlanningAgent が検索と分析の流れを指示
  2. SearchAgent がモックのWeb検索を実行
  3. DataAgent が数値分析を担当
  4. 最終的に UserProxyAgent が "APPROVE" で会話を終了

必要なライブラリ

bashCopyEditpip install autogen-agentchat autogen-ext openai

動作環境

  • Python 3.10+
  • AutoGen 0.6.4 以降
  • OpenAI APIキーが OPENAI_API_KEY 環境変数に登録済みであること

コード構成と機能解説

1. モックツール（ツール呼び出しのサンプル）

pythonCopyEditdef search_web_tool(query: str) -> str:
  • 単純な if 文で、事前に決められた文字列（2006-2007など）に対するダミー検索結果を返す
  • AutoGenの Tool 呼び出し機能（Function Calling）を使うためのモック実装
pythonCopyEditdef percentage_change_tool(start: float, end: float) -> str:
  • 2つの数値の変化率を計算して % で返すツール
  • 分析用エージェント DataAgent によって呼び出される

2. 各エージェントの役割と生成

pythonCopyEditplanning = AssistantAgent(name="PlanningAgent", ...)
searcher = AssistantAgent(name="SearchAgent", tools=[search_web_tool])
analyst = AssistantAgent(name="DataAgent", tools=[percentage_change_tool])
user_proxy = UserProxyAgent(name="UserProxy")
  • AssistantAgent は GPT-4o などを使ってタスクを会話形式で実行
  • 各エージェントには description や system_message が設定されており、役割が明確化されている
  • tools=[...] により外部関数（ツール）を呼び出せる

3. チャット終了条件の定義

pythonCopyEdittermination = TextMentionTermination("APPROVE") | MaxMessageTermination(25)
  • "APPROVE" というキーワードが出た時点、または25ターンでチャット終了
  • 複数の終了条件は | 演算子で OR 結合可能

4. SelectorGroupChat チームの構築

pythonCopyEditteam = SelectorGroupChat([...], termination_condition=..., model_client=...)
  • チームに参加するエージェントを配列で定義
  • SelectorGroupChat は会話の文脈に応じて、どのエージェントを次に話させるかを自動で選出する点が特徴
  • model_client はすべてのエージェントに共通で渡す（GPT-4o 使用）

5. メイン実行と非同期処理

pythonCopyEditasync def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o", temperature=0)
    ...
    stream = team.run_stream(task=task)
    await Console(stream)
    await model_client.close()
  • 実行時は非同期（async/await）によりストリーム形式でやり取りが順次表示される
  • 実行後は明示的にクライアントを close() することでAPI接続を解放

6. 実行例

bashCopyEdit$ python selector_group_chat_demo.py

実行すると、以下のような会話の流れが見られます：

luaCopyEdit---------- PlanningAgent ----------
"Let's start by checking Wade's total points for 2006-2007..."

---------- SearchAgent ----------
"Calling search_web_tool('2006-2007')..."
...
---------- DataAgent ----------
"Calling percentage_change_tool(1397, 1175)..."
...
---------- UserProxy ----------
"APPROVE"

補足・カスタマイズ例

  • エージェントごとに別モデル（例：GPT-3.5, Claude）を割り当てたい場合は model_client= を個別に変更可能
  • Tool部分を実際のAPI呼び出し（Web検索、分析システムなど）に置き換えることで本番運用に近づけられる
  • selector_func や candidate_func を導入することで、より高度なエージェント選出ロジックも実装可能

全コード

# =============================================================================
# プログラム名 ：SelectorGroupChat デモ実装（AutoGen × OpenAI GPT-4o）
# -----------------------------------------------------------------------------
# 概要         ：AutoGenのSelectorGroupChat機能を活用し、実行時にユーザーが希望する
#               モード（default / selector_func / candidate_func）を番号で選択し、
#               それに応じてエージェントがタスクを分担・自律選択して処理
# 目的         ：複数のAIエージェント（計画・検索・分析）が協調し、人間の承認を得ながら
#               課題を解決する
# 主要機能     ：
#   - AssistantAgent（PlanningAgent: 指示出し、SearchAgent: 情報検索、DataAgent: 数値分析）
#   - UserProxyAgent（人間ユーザーの割込・承認・最終判断を担当）
#   - SelectorGroupChat（エージェントを文脈に基づいて選出し、順番に発話させる）
#   - モックツール：search_web_tool（疑似検索）、percentage_change_tool（疑似変化率計算）
#   - カスタムロジック：
#       ・selector_func：次の発話者を明示的に選定
#       ・candidate_func：発話候補者の動的制御
#   - 終了条件：
#       ・TextMentionTermination（発話に「APPROVE」が含まれると終了）
#       ・MaxMessageTermination（メッセージ数が上限に達すると終了）
# 動作要件     ：
#   - Python 3.10 以上
#   - AutoGen 0.6.4 以降（autogen-agentchat, autogen-core, autogen-ext）
#   - OpenAI API キーが環境変数 `OPENAI_API_KEY` に設定されていること
# 実行例       ：
#   $ python selector_group_chat_demo.py
#   → 実行時にモード番号を入力 → PlanningAgent が指示 → SearchAgent が検索 → DataAgent が分析 → User がAPPROVE → 終了
# 備考         ：
#   - モック関数は本番APIで置換可能（e.g., 本物のWeb検索API、実データ処理）
#   - 各エージェントの model_client は共通（GPT-4o）を使用
#   - 出力はコンソールにストリーム形式で表示される
#   - 複雑な会話制御を試すためのベース構造として活用可能
# その他       ：
#   - モード選択は番号入力形式（1-default、2-selector_func、3-candidate_func）で実施
#   - 各ターン開始時に担当エージェント名を標準出力
# =============================================================================

# 非同期処理（async/await）用
import asyncio

# 型ヒント用（リスト、シーケンス）
from typing import List, Sequence

# AIとユーザー代理エージェント
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent

# 終了条件クラス
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination

# 自動エージェント選択グループチャット
from autogen_agentchat.teams import SelectorGroupChat

# コンソール出力UI
from autogen_agentchat.ui import Console

# OpenAIモデル利用クライアント
from autogen_ext.models.openai import OpenAIChatCompletionClient

# メッセージとイベント基底クラス
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage

# 模擬Web検索ツール（モック関数）
def search_web_tool(query: str) -> str:
    """簡易的な検索結果を模倣するモック関数"""
    # 2006-2007を含むクエリの場合
    if "2006-2007" in query:
        return "2006-07 Miami Heat: Dwayne Wade 1397 pts, Udonis Haslem 844 pts …"
    # 2007-2008を含むクエリの場合
    elif "2007-2008" in query:
        return "2007-08 Wade rebounds: 214"
    # 条件に合致しない場合
    return "No data found."

# 変化率計算ツール（モック関数）
def percentage_change_tool(start: float, end: float) -> str:
    """始値と終値からパーセント変化率を計算するモック関数"""
    # 変化率を計算（ゼロ除算防止）
    pct = ((end - start) / start) * 100 if start else float('nan')
    # 小数点以下2桁でパーセント表示
    return f"{pct:.2f}%"

# エージェント選択関数（次にPlanningAgentを選ぶ例）
def selector_func(messages):
    """直前の発話者がPlanningAgent以外なら次はPlanningAgentを選択"""
    # 直前がPlanningAgentでない場合
    if messages[-1].source != "PlanningAgent":
        selected_agent = "PlanningAgent"
        # 選択したエージェント名を表示
        print(f"Selector chose: {selected_agent}")
        return selected_agent
    # 条件外ならNoneを返す
    return None

# 候補エージェントを動的に決定する関数
def candidate_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> List[str]:
    """発話履歴に応じて候補となるエージェントリストを動的に決定"""
    # ユーザー入力直後はPlanningAgentを優先
    if messages[-1].source == "user":
        return [planning_agent.name]

    last_message = messages[-1]
    # PlanningAgentの発話内容から次を判断
    if last_message.source == planning_agent.name:
        participants: List[str] = []
        if web_search_agent.name in last_message.to_text():
            participants.append(web_search_agent.name)
        if data_analyst_agent.name in last_message.to_text():
            participants.append(data_analyst_agent.name)
        # 検索や分析エージェントを候補にする
        if participants:
            return participants

    # 発話履歴のエージェント名を取得
    previous_set_of_agents = {msg.source for msg in messages}
    # SearchAgentとDataAgentが話し終えていればPlanningAgentに戻す
    if (
        web_search_agent.name in previous_set_of_agents and
        data_analyst_agent.name in previous_set_of_agents
    ):
        return [planning_agent.name]

    # 条件に該当しない場合は全員候補
    return [planning_agent.name, web_search_agent.name, data_analyst_agent.name]

# チーム構築関数
def create_team(
    model_client: OpenAIChatCompletionClient,
    selector_mode: str = "selector_func",
) -> SelectorGroupChat:
    """エージェントを作成してSelectorGroupChatに登録"""
    global planning_agent, web_search_agent, data_analyst_agent

    # PlanningAgentを生成
    planning_agent = AssistantAgent(
        name="PlanningAgent",
        description="Plan tasks and delegate to search or data agents.",
        system_message="You are a planning agent.",
        model_client=model_client,
    )
    # SearchAgentを生成
    web_search_agent = AssistantAgent(
        name="SearchAgent",
        description="Use search_web_tool to fetch data.",
        system_message="You are a search agent. Call search_web_tool(query).",
        model_client=model_client,
        tools=[search_web_tool],
    )
    # DataAgentを生成
    data_analyst_agent = AssistantAgent(
        name="DataAgent",
        description="Analyze numbers via percentage_change_tool.",
        system_message="You are a data analyst. Call percentage_change_tool(start, end).",
        model_client=model_client,
        tools=[percentage_change_tool],
    )
    # UserProxyAgentを生成
    user_proxy = UserProxyAgent(
        name="UserProxy",
        description="Human user for approval or feedback.",
    )

    # チャット終了条件を設定
    termination = (
        TextMentionTermination("APPROVE") | MaxMessageTermination(max_messages=25)
    )

    # モードに応じて追加設定
    selector_args: dict[str, object] = {
        "termination_condition": termination,
        "model_client": model_client,
    }
    mode = selector_mode.lower() if selector_mode else "default"
    if mode == "selector_func":
        selector_args["selector_func"] = selector_func
    elif mode == "candidate_func":
        selector_args["candidate_func"] = candidate_func

    # SelectorGroupChatのインスタンス生成
    team = SelectorGroupChat(
        [planning_agent, web_search_agent, data_analyst_agent, user_proxy],
        **selector_args,
    )
    return team

# メイン実行関数
async def main() -> None:
    """チャットを開始する非同期メイン関数"""
    # モード選択をユーザーに要求
    selection = input(
        "Select selector mode by number: 1-default, 2-selector_func, 3-candidate_func [1]: "
    ).strip()
    mode_map = {"1": "default", "2": "selector_func", "3": "candidate_func"}
    # 選択されたモードを取得
    mode = mode_map.get(selection, "default")
    # GPT-4oを使用するモデルクライアントを生成
    model_client = OpenAIChatCompletionClient(model="gpt-4o", temperature=0)
    # エージェントチームを構築
    team = create_team(model_client, selector_mode=mode)
    # 実行タスクを定義
    task = (
        "Compare Dwyane Wade total points vs rebounds between the 2006‑2007 "
        "and 2007‑2008 seasons. Provide percentage change."
    )
    # チャットを非同期で実行
    stream = team.run_stream(task=task)
    async for message in stream:
        # 発話者の名前と内容を出力
        print(f"---------- {message.source} ----------")
        print(f"Message object: {message}")
        # コンソールにメッセージを出力
        await Console(stream)
    # クライアントを閉じる
    await model_client.close()

# スクリプト直接実行時のエントリポイント
if __name__ == "__main__":
    # 非同期main関数を実行
    asyncio.run(main())

コメント

タイトルとURLをコピーしました