Semantic Kernel Agent Frameworkでマルチエージェント実装手法を解説

当記事では、Microsoft がオープンソースとして提供している Semantic Kernel Agent Framework（Python SDK） を活用し、シングルエージェントおよびマルチエージェントのサンプルコードをもとに、その実装手法を解説します。

Semantic Kernel Agent Framework

概要

Semantic Kernel Agent Framework （以降、Semantic Kernel Agents）は、Microsoft によって開発されたオープンソースのエージェント SDK であり、エージェントの構築・統合・オーケストレーションを容易にするための仕組みを提供します。

2025年4月時点では、C# (.NET)、Python、Java 向けに SDK が提供されており、 C# (.NET) および Python の SDK はすでに GA（一般提供） となっているためエンタープライズでの本番環境の実装にも適しています。

参考：Semantic Kernel Agents are now Generally Available

プラグイン

プラグイン とは、エージェントが外部の機能やサービスを活用するためのツール群のまとまりです。これにより、エージェントは外部 API や既存のコードを通じて、より高度な処理やタスクを実行できるようになります。

プラグインは、1つ以上のカーネル関数（Kernel Function）で構成されており、各関数はエージェントから呼び出すことができます。

現在サポートされているプラグインの種類（2025年4月時点）。

• ネイティブコードプラグイン
  既存の Python コードやクラスの関数をカーネル関数として登録・活用するプラグイン。アプリケーション内部のロジックをそのままエージェントに提供できます。
• OpenAPI 仕様プラグイン
  OpenAPI ドキュメントに基づき、自動的に関数を生成するプラグイン。外部サービスとの連携や、異なる言語・プラットフォーム間でのプラグイン共有に適しています。

Agent Class

Agent クラスは、Semantic Kernel Agents におけるすべてのエージェントの基底クラスです。
このクラスを継承することで、異なるエージェント間でも、抽象的なインターフェース（get_response, invoke など）でエージェントを操作できるため開発効率が高まります。

主な派生クラス

• ChatCompletionAgent
  Azure、OpenAI、Anthropic、Vertex AI など、各社が提供する会話型 API（Chat Completion API）に対応した汎用エージェント。ChatCompletionClientBase を介することでプロバイダーの違いを吸収して柔軟に統合できます。

• AzureAIAgent / BedrockAgent / AutoGenConversableAgent 等
  これらは特定のクラウドサービスや外部エージェントフレームワークと連携するためのエージェントクラスです。既存のエージェント基盤を活かしつつ、Semantic Kernel Agents として機能させることができます。

つまり、Semantic Kernel Agents を用いることで、モデルプロバイダーやエージェントフレームワークの違いにかかわらず、共通の設計方針でエージェントを構築し、拡張できるメリットがあります。

AgentThread Class

Agent クラスとセットで使うのが、AgentThread クラスです。
こちらは、エージェントの会話状態（スレッド）を抽象化・管理するための基底クラスです。
AgentThread を用いることで、エージェントごとに異なる履歴やコンテキストの保持方式に対応し、共通のインターフェースで統一的に扱うことができます。

参考：Agent Architecture - AgentThread

シングルエージェント

まずは、Semantic Kernel Agents の基本的な動作を確認するため、シングルエージェントを動かしてみましょう。

事前準備

まずは、必要なライブラリのインストールと、環境変数の読み込みを行います。以下のコードをノートブック上で順に実行してください。

ライブラリのインストール等は折りたたみます。

ソースコード

ライブラリのインストール

!pip install semantic-kernel==1.27.2 \
    python-dotenv==1.0.1 \
    azure-ai-projects==1.0.0b7 --quiet

環境変数の読み込み

import os
from dotenv import load_dotenv

load_dotenv(override=True)

AZURE_OPENAI_API_KEY=os.getenv("AZURE_OPENAI_API_KEY")
AZURE_DEPLOYMENT_NAME=os.getenv("AZURE_DEPLOYMENT_NAME")
AZURE_OPENAI_ENDPOINT=os.getenv("AZURE_OPENAI_ENDPOINT")
AZURE_OPENAI_API_VERSION=os.getenv("AZURE_OPENAI_API_VERSION")

ユーティリティ関数を定義

import datetime
import json
from semantic_kernel.contents.function_call_content import FunctionCallContent
from semantic_kernel.contents.function_result_content import FunctionResultContent
from semantic_kernel.contents.text_content import TextContent


async def print_thread_message_details(thread: str):
    """
    スレッドのメッセージ詳細を表示します。

    Args:
        thread (str): スレッドのインスタンス
    """
    async for message in thread.get_messages():
        print("-----")

        for item in message.items:
            if isinstance(item, FunctionCallContent):
                print(f"[Function Calling] by {message.ai_model_id}")
                print(f" - Function Name : {item.name}")
                print(f" - Arguments     : {item.arguments}")

            elif isinstance(item, FunctionResultContent):
                print(f"[Function Result]")
                # 文字列のデコード変換
                if isinstance(item.result, str):
                    try:
                        decoded = json.loads(item.result)
                        print(f" - Result        : {decoded}") # デコード成功時は変換後の値を表示
                    except json.JSONDecodeError:
                        print(f" - Result        : {item.result}")  # デコード失敗時はそのまま
                else:
                    print(f" - Result        : {item.result}")

            elif isinstance(item, TextContent):
                if message.name:
                    print(f"[Agent Response] from {message.ai_model_id}")
                else:
                    print("[User Message]")
                print(f" - Content       : {item.text}")

            else:
                print(f"[Unknown Item Type] ({type(item).__name__})")
                print(f" - Raw Item      : {item}")


def log_with_timestamp(message: str) -> None:
    """
    現在時刻付きでメッセージを標準出力にログとして表示します。

    Args:
        message (str): 出力するログメッセージ。
    """
    timestamp = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
    print(f"[{timestamp}] {message}")

プラグイン定義

エージェントがユーザー情報にアクセスできるよう、UserPlugin というプラグインを定義します。このプラグインには、ダミーデータから契約プラン情報を取得するカーネル関数を含めます。

import json
from typing import Annotated
from semantic_kernel.functions import kernel_function


class UserPlugin:
    @kernel_function(
        name="get_plan",
        description="指定されたユーザーIDから契約情報を取得します",
    )
    def get_plan(
        self, 
        user_id: Annotated[int, "ユーザーID（4桁の数字）"]
    ) -> Annotated[str, "ユーザーの契約プラン名（例: '安心保障プラン'）"]:
        
        sample_user = [
            {"id": 1234, "name": "佐藤太郎", "email": "[e-mail]", "plan": "安心保障プラン"},
            {"id": 2345, "name": "鈴木花子", "email": "[e-mail]", "plan": "総合生活サポートプラン"},
            {"id": 3456, "name": "田中一郎", "email": "[e-mail]", "plan": "シンプルプラン"},
        ]

        for user in sample_user:
            if user["id"] == user_id:
                return json.dumps(user["plan"])
        return "該当するユーザーが見つかりませんでした。"

シングルエージェントの作成

ここでは、ChatCompletionAgent を用いて、Azure の Chat Completion API でエージェントを構築します。

from semantic_kernel.agents import (
    ChatCompletionAgent,
    ChatHistoryAgentThread,
)
from semantic_kernel.connectors.ai.open_ai import (
    AzureChatCompletion,
)

# Azure ChatCompletionのインスタンスを作成
azure_chat_completion = AzureChatCompletion(
    api_key=AZURE_OPENAI_API_KEY,
    deployment_name=AZURE_DEPLOYMENT_NAME,
    endpoint=AZURE_OPENAI_ENDPOINT,
)

# エージェントを作成
agent = ChatCompletionAgent(
    service=azure_chat_completion,
    name="EmployeeAssistant",
    instructions=(
        "あなたはコールセンターで働く従業員向けサポートAIアシスタントです。"
        "従業員の質問に答えたり、サポートを提供したりします。"
    ),
    plugins=[UserPlugin()]
)

動作確認

作成したエージェントに対して get_response メソッドでレスポンスを取得します。なお、スレッドには AgentThread を継承した ChatHistoryAgentThread を使用します。

# スレッドを作成
thread = ChatHistoryAgentThread()

# レスポンスを取得
response = await agent.get_response(
    messages=["ユーザー ID 3456 のお客様の保険の加入状況を教えて。"],
    thread=thread,
)

print(response)

出力例

ユーザー ID 3456 のお客様は「シンプルプラン」に加入されています。

スレッドの詳細を確認
ユーティリティ関数 print_thread_message_details を使用することで、エージェントがどのように応答を構築したかを詳細に確認できます。

await print_thread_message_details(thread)

出力例

-----
[User Message]
 - Content       : ユーザー ID 3456 のお客様の保険の加入状況を教えて。
-----
[Function Calling] by gpt-4o-mini
 - Function Name : UserPlugin-get_user
 - Arguments     : {"user_id":3456}
-----
[Function Result]
 - Result        : シンプルプラン
-----
[Agent Response] from gpt-4o-mini
 - Content       : ユーザー ID 3456 のお客様は「シンプルプラン」に加入しています。何か他にお手伝いできることはありますか？

このように、エージェントが自律的に関数呼び出しから関数実行まで行い、その結果を踏まえて応答を生成していることが確認できます。
Semantic Kernel Agents を用いることで、少ないコード量で柔軟かつ自律的なエージェントが実装できることを実感できたでしょう。

Tips ) 関数の情報をリッチにする方法

@kernel_function を使って関数を登録する際、関数名・説明・引数・戻り値のメタデータは「自動取得」または「明示的に指定」することが可能です。

関数名と説明の指定

@kernel_function は以下のいずれかの形式で使用できます。

• 自動取得
  「関数名」はそのまま、「関数の説明」は docstring から自動的に反映されます。

  @kernel_function()
  def get_user(user_id: int) -> str:
        """指定されたユーザーIDから契約情報を取得します"""
      ...
  

• 明示的に指定
  @kernel_function の引数で「関数名」と「関数の説明」を明示的に指定できます。
  @kernel_function の「関数の説明」と dockstring が両方指定されている場合、@kernel_function の設定が優先されます。

  @kernel_function(name="get_plan", description="指定されたユーザー ID から契約情報を取得します")
  def get_user(user_id: int) -> str:
          """
        指定されたユーザーIDから契約情報を取得します
  
        Args:
            user_id (int): ユーザーID（4桁の数字）
  
        returns:
            str: ユーザーの契約プラン名（例: '安心保障プラン'）
        """
      ...
  

引数と戻り値の説明

「引数の説明」と「戻り値の説明」を追加するには、typing.Annotated を使用します。Annotated により、「型」と一緒に「説明」を提供できます。

from typing import Annotated

@kernel_function(name="get_plan", description="指定されたユーザー ID から契約情報を取得します")
def get_plan(user_id: Annotated[int, "ユーザー ID"]) -> Annotated[str, "契約中のプラン名（例: '安心保障プラン'）"]:
    ...

Agent オブジェクトの中身を確認

上記で定義された関数は、内部的には以下のようなメタデータを Agent オブジェクトに渡します。

"get_user": {
  "metadata": {
    "name": "get_user",
    "description": "指定されたユーザー ID から契約情報を取得します",
    "parameters": [
      {
        "name": "user_id",
        "description": "ユーザーID（4桁の数字）",
        "type_": "int",
        "is_required": true,
        ...

このように、@kernel_function による明示的な定義と Annotated の併用により、関数の情報を正確かつ詳細に提供でき、エージェントが呼び出す関数呼び出しの精度が向上します。

ただし、情報をリッチにすれば、それに伴ってトークンの消費量も増加する点には注意が必要です。

参考：kernel_function_decorator Module

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

マルチエージェント・アーキテクチャには、ユースケースやビジネスロジックに応じて柔軟に構成を選択できるよう、いくつかの代表的なパターンが存在します。

なかでも、実務で頻出する以下の 3 パターンを押さえておくことで、複雑かつ拡張性のあるマルチエージェント設計を効果的に組み立てることができます。

代表的なアーキテクチャパターン

• 協力型（Cooperative）
  各エージェントが役割分担しながら、共通の目的に向かって協力する構成。問題解決力や合意形成の精度が求められるシナリオに適しています。
• 競争型（Competitive）
  対立する立場や評価軸を持つエージェント同士が出力を競い合い、最適なアイデアや回答を導き出す構成。クリエイティブ生成や意思決定支援に有効です。
• ワークフロー型（Workflow）
  あらかじめ定義されたプロセスに沿って、エージェントが順番に処理を引き継いでいく構成。業務プロセスの自動化やタスク分担に適しています。

当記事では、これらのパターンを Semantic Kernel Agents で実装するサンプルコードとともに解説していきます。

また、マルチエージェントの設計の Tips については、以下の YouTube 動画でも紹介しています。興味のある方はぜひご覧ください。

マルチエージェント実装&解説

協力型

はじめに、複数のエージェントが協調してタスクを遂行する「協力型マルチエージェント」の実装例をご紹介します。構成は以下の通りです。

事前準備

プラグイン定義
以下のようなプラグインを定義します。

• TimeWeatherPlugin
  • fetch_current_datetime: 現在の時刻を取得
  • fetch_weather: 指定された都市の天気を取得（※ダミー関数）
• ConvertTemperaturePlugin
  • convert_temperature: 摂氏から華氏への温度変換
• SendEmailPlugin
  • send_email: 指定された件名と本文のメールを送信（※ダミー関数）

ソースコード

import json
from typing import Annotated
import datetime
from semantic_kernel.functions import kernel_function


class TimeWeatherPlugin:

    @kernel_function(
        name="fetch_current_datetime",
        description="現在の時刻を JSON 文字列として取得します。オプションでフォーマットを指定できます。",
    )
    def fetch_current_datetime(
        self,
        format: Annotated[str, "現在の時刻を返す形式（例: '%Y/%m/%d %H:%M'）。未指定時はデフォルト形式。"] = "",
    ) -> Annotated[str, "現在の時刻を含む JSON 文字列（例: {'current_time': '2023-10-01 12:00:00'}）"]:

        time_format = format or "%Y-%m-%d %H:%M:%S"
        current_time = datetime.datetime.now().strftime(time_format)
        return json.dumps({"current_time": current_time})


    @kernel_function(
        name="fetch_weather",
        description="指定された場所の天気情報を取得します。",
    )
    def fetch_weather(
        self, 
        location: Annotated[str, "天気情報を取得する都市名（例: Tokyo, New York, London）"]
    ) -> Annotated[str, "天気情報を含む JSON 文字列（例: {'weather': 'Sunny, 25°C'}）"]:

        # ダミーの天気データを使用
        dummy_weather_data = {
            "New York": "Sunny, 25°C",
            "London": "Cloudy, 18°C",
            "Tokyo": "Rainy, 22°C"
        }
        weather = dummy_weather_data.get(
            location, "Weather data not available for this location."
        )

        return json.dumps({"weather": weather})


class ConvertTemperaturePlugin:

    @kernel_function(
        name="convert_temperature",
        description="温度を摂氏から華氏に変換します。",
    )
    def convert_temperature(
        self,
        celsius: Annotated[float, "摂氏温度（例: 25.0）"]
    ) -> Annotated[str, "華氏温度を含む JSON 文字列（例: {'fahrenheit': 77.0}）"]:

        fahrenheit = (celsius * 9 / 5) + 32

        return json.dumps({"fahrenheit": fahrenheit})


class SendEmailPlugin:

    @kernel_function(
        name="send_email",
        description="指定の件名と本文を含むメールを宛先に送信します。",
    )
    def send_email(
        self,
        recipient: Annotated[str, "メールの宛先アドレス（例: [e-mail]）"],
        subject: Annotated[str, "メールの件名"],
        body: Annotated[str, "メールの本文"]
    ) -> Annotated[str, "完了通知を含む文字列（例: {'message': Email successfully sent to [e-mail].}）"]:
        print(f"Sending email to {recipient}...")
        print(f"Subject: {subject}")
        print(f"Body:\n{body}")
        return json.dumps({"message": f"Email successfully sent to {recipient}."})

マルチエージェント作成

それぞれのプラグインを専門のエージェントに紐づけ、最後にそれらのエージェントを制御・振り分けるエージェント（triage_agent）にプラグインとして統合します。

from semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.connectors.ai.open_ai import OpenAIChatPromptExecutionSettings
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.functions.kernel_arguments import KernelArguments

# Azure Chat Completion の初期化
azure_chat_completion = AzureChatCompletion(
    api_key=AZURE_OPENAI_API_KEY,
    deployment_name=AZURE_DEPLOYMENT_NAME,
    endpoint=AZURE_OPENAI_ENDPOINT,
)

# 専門エージェントの作成
time_weather_agent = ChatCompletionAgent(
    service=azure_chat_completion, 
    name="TimeWeatherAgent", 
    instructions="あなたは時間と天気のクエリの専門的なエージェントです。",
    plugins=[TimeWeatherPlugin()]
)

temperature_agent = ChatCompletionAgent(
    service=azure_chat_completion, 
    name="TemperatureAgent", 
    instructions="あなたは温度変換の専門的なエージェントです。",
    plugins=[ConvertTemperaturePlugin()]
)

send_email_agent = ChatCompletionAgent(
    service=azure_chat_completion, 
    name="SendEmailAgent", 
    instructions="あなたは電子メールを送信するための専門のエージェントです。",
    plugins=[SendEmailPlugin()]
)

# 振り分けを行うトリアージエージェントを作成
triage_agent = ChatCompletionAgent(
    service=azure_chat_completion, 
    name="TriageAgent",
    instructions=(
        "ユーザーのリクエストを評価し、適切なエージェント（TimeWeatherAgent, TemperatureAgent, SendEmailAgent ）"
        "に転送して、適切なサポートを提供します。エージェントからの情報を含め、ユーザーに完全な回答を提供します。"
        "元のユーザーリクエストが完全に処理されたどうか確認してください。"
    ),
    plugins=[time_weather_agent, temperature_agent, send_email_agent],
)

triage_agent の詳細を確認
triage_agent がどのように他のエージェントを制御・振り分けるのかを確認するため、以下のコードを実行します。

formatted_json = json.dumps(triage_agent.model_dump(), indent=2, ensure_ascii=False, default=str)
print(formatted_json)

出力例（抜粋）

"plugins": {
    "TimeWeatherAgent": {
        "functions": {
            "TimeWeatherAgent": {
                "description": "あなたは時間と天気のクエリの専門的なエージェントです。",
                ...

このように、各エージェントは「関数」のように振る舞い、triage_agent が入力内容に応じて適切なエージェント（関数）を選択・実行する構成となっています。

動作確認

以下のような複数のタスクを一度に含めたユーザー入力を与えて、マルチエージェントの動作を検証します。

thread = ChatHistoryAgentThread()

user_input = (
    "はじめに、現在の時刻を '%Y-%m-%d %H:%M:%S' 形式で、また Tokyo の天気を教えてください。"
    "次に、Tokyo の気温を華氏に変換してください。"
    "最後に、結果の概要を記載したメールをサンプル受信者に送信してください。"
)

response = await triage_agent.get_response(
    messages=user_input,
    thread=thread
)

出力例

Sending email to [e-mail]...
Subject: 現在の時刻とTokyoの天気概要
Body:
現在の時刻は 2025年4月14日 20:45 です。

Tokyoの天気は雨で、気温は22°Cです。これを華氏に変換すると、約71.6°Fです。

スレッドの詳細を確認

await print_thread_message_details(thread)

出力例（一部）

-----
[User Message]
 - Content       : はじめに、現在の時刻を '%Y-%m-%d %H:%M:%S' 形式で、また Tokyo の天気を教えてください。次に、Tokyo の気温を華氏に変換してください。最後に、結果の概要を記載したメールをサンプル受信者に送信してください。
-----
[Function Calling] by gpt-4o-mini
 - Function Name : TimeWeatherAgent-TimeWeatherAgent
 - Arguments     : {"messages": "現在の時刻とTokyoの天気を教えてください。"}
[Function Calling] by gpt-4o-mini
 - Function Name : TemperatureAgent-TemperatureAgent
 - Arguments     : {"messages": "Tokyoの気温を華氏に変換してください。"}
-----
[Function Result]
 - Result        : 現在の時刻は 2025年4月14日 20:45 です。  
東京の天気は雨で、気温は22℃です。
-----
[Function Result]
 - Result        : 東京の現在の気温を知る必要があります。もし具体的な気温を教えていただければ、その温度を華氏に変換します。または、一般的な気温（例えば25度など）を提供していただいても結構です。
-----
...

この例では、TemperatureAgent による気温変換が想定どおりに行われていないことが確認できます。

これは、TimeWeatherAgent による天気情報の取得が完了する前に、TemperatureAgent を実行してしまったためです。

Tips ) 並列関数呼び出しの注意点

並列関数呼び出しはデフォルトで有効化されいます。
もちろん、一度に複数の関数を呼び出せるためパフォーマンスの向上に適していますが、今回のケースのように、あるエージェントの出力を別のエージェントが前提とするタスク間の依存関係がある場合、並列関数呼び出し（Parallel Function Calling）が原因で処理順序が崩れ、期待どおりに動作しないことがあります。

並列関数呼び出しの無効化方法

OpenAI / Azure OpenAI のコネクタを使用している場合、以下のように指定（ parallel_tool_calls=False）することで、並列呼び出しを無効にできます。

# 並列関数呼び出しを無効にする
arguments = KernelArguments(
    settings = OpenAIChatPromptExecutionSettings(
        function_choice_behavior=FunctionChoiceBehavior.Auto(),
        parallel_tool_calls=False
    )
)

response = await triage_agent.get_response(
    messages=user_input,
    thread=thread,
    arguments=arguments,
)

参考：Auto Function Invocation

無効化後の動作例

並列呼び出しを無効にして再実行すると、以下のように各エージェントが順を追って正しく呼び出され、期待通りの処理が行われていることが確認できます。

-----
[Function Calling] by gpt-4o-mini
 - Function Name : TimeWeatherAgent-TimeWeatherAgent
 - Arguments     : {"messages":"現在の時刻と東京の天気を教えてください。"}
-----
[Function Result]
 - Result        : 現在の時刻は **2025年4月14日 17:58** です。東京の天気は **雨**で、気温は **22°C** です。
-----
[Function Calling] by gpt-4o-mini
 - Function Name : TemperatureAgent-TemperatureAgent
 - Arguments     : {"messages":"22°C を華氏に変換してください。"}
-----
[Function Result]
 - Result        : 22°C は 71.6°F に相当します。
-----

このように、順序依存のあるプロンプトにおいては、並列実行を無効にすることで処理の正確性が確保されます。

競争型

続いては、互いに異なる視点や立場から意見を出し合い、最終的により良いアウトプットを生み出す 競争型マルチエージェント の実装例をご紹介します。

競争型マルチエージェントの構築には、AgentGroupChat クラスを活用します。このクラスは、複数のエージェントが順番に発言しながらディスカッションする構造を提供します。

2025年4月現在、AgentGroupChat クラスは実験段階（Experimental）なステータスです。将来的に破壊的変更が加えられる可能性があるため、本番環境での利用は推奨されません。

参考：Exploring Agent Collaboration in AgentChat

事前準備

各エージェントの作成
今回の実装では、以下の2つの役割を持つエージェントを作成します。

• Reviewer Agent：
  ユーザーが与えたキーワードや下書きに対して、追加すべき要素を提案する。
• Writer Agent：
  Reviewer の提案に従い、マークダウン形式で記事を執筆・更新する。

from semantic_kernel.agents import  ChatCompletionAgent


REVIEWER_NAME = "Reviewer"
WRITER_NAME = "Writer"


# Azure ChatCompletionのインスタンスを作成
azure_chat_completion = AzureChatCompletion(
    api_key=AZURE_OPENAI_API_KEY,
    deployment_name=AZURE_DEPLOYMENT_NAME,
    endpoint=AZURE_OPENAI_ENDPOINT,
)


# レビュアーエージェントを作成
agent_reviewer = ChatCompletionAgent(
    service=azure_chat_completion,
    name=REVIEWER_NAME,
    instructions="""
あなたの役割は、ユーザーが提示するキーワード、要点、草稿（マークダウン形式）に対して、
追加すべき情報や観点を具体的に提案するレビュワーです。

初期入力がキーワードのみであっても、読者にとって有益な情報を盛り込むために必要なトピック、
構成要素、視点などを具体的に提示してください。

制約：
- 記事の直接修正や文章生成は行わないでください。
- 提案はリスト形式で簡潔に提示してください。
"""
)


# ライターエージェントを作成
agent_writer = ChatCompletionAgent(
    service=azure_chat_completion,
    name=WRITER_NAME,
    instructions="""
あなたの役割は、レビュワーが提案した内容に基づき、
マークダウン形式で記事を段階的に肉付けしていくライターです。

制約：
- 提案された内容をすべて反映し、マークダウン形式で執筆してください。
- 記事本文以外の出力は含めないでください。
"""
)

選択戦略・終了戦略で使うプロンプトの定義

AgentGroupChat では、どのエージェントが次に発言するかを決める 選択戦略、およびディスカッションを終了するかどうかを判断する 終了戦略 を定義します。

ここでは、選択戦略および終了戦略で用いるプロンプトを定義しておきます。

from semantic_kernel.functions import KernelFunctionFromPrompt


# 選択戦略で使うプロンプトを定義
selection_function = KernelFunctionFromPrompt(
    function_name="selection", 
    prompt=f"""
提供された直前の発言内容を確認し、次に発言すべき参加者を判断してください。  
出力には、選んだ参加者の名前のみを記述し、理由や補足説明は一切記載しないでください。  
また、同じ参加者が続けて発言することはできません。

選択可能な参加者：  
- {REVIEWER_NAME}  
- {WRITER_NAME}  

ルール：  
- 入力がユーザーによる初回キーワード（例：「Azure AI Agent Service」など）の場合 → {REVIEWER_NAME} を選んでください。  
- 発言者が {REVIEWER_NAME} の場合 → 次は {WRITER_NAME} を選んでください。  
- 発言者が {WRITER_NAME} の場合 → 次は {REVIEWER_NAME} を選んでください。  

直前の発言：  
{{{{$lastmessage}}}}
"""
)


# 終了戦略で使うプロンプトを定義
termination_keyword = "yes"

termination_function = KernelFunctionFromPrompt(
    function_name="termination", 
    prompt=f"""
以下の回答を確認し、その内容が適切かどうかを判断してください。  
適切である場合は、説明や補足なしで「{termination_keyword}」という単語のみを出力してください。

判定基準：  
- 回答に具体的な修正提案が含まれている場合 → 不適切です。  
- 回答に修正提案が含まれていない場合 → 適切です。  

確認対象の回答：  
{{{{$lastmessage}}}}
"""
)

マルチエージェントの作成
競争型マルチエージェントを構築する AgentGroupChat の引数には、ディスカッションに参加する 各エージェント と、選択戦略 および 終了戦略 を指定するひつようがありｍ。

ユーザー入力を受け取った後のフローは以下の通りです。

1. エージェント選択（選択戦略）
2. 選ばれたエージェントを実行
3. 結果を会話履歴に記録
4. 終了判定（終了戦略）
5. End or 終了条件を満たすまで繰り返し

from semantic_kernel import Kernel
from semantic_kernel.agents import AgentGroupChat
from semantic_kernel.agents.strategies import (
    KernelFunctionSelectionStrategy,
    KernelFunctionTerminationStrategy,
)
from semantic_kernel.contents import ChatHistoryTruncationReducer


# Kernel を作成
kernel = Kernel(services=azure_chat_completion)


# エージェントの会話履歴管理（直近 4 つのメッセージのみを保持するよう指定）
history_reducer = ChatHistoryTruncationReducer(target_count=4)


# AgentGroupChat を初期化
chat = AgentGroupChat(

    # ディスカッションに参加するエージェントのリスト
    agents=[agent_reviewer, agent_writer],

    # 選択戦略
    selection_strategy=KernelFunctionSelectionStrategy(
        initial_agent=agent_reviewer,
        function=selection_function,
        kernel=kernel,
        result_parser=lambda result: str(result.value[0]).strip() if result.value[0] is not None else WRITER_NAME,
        history_variable_name="lastmessage",
        history_reducer=history_reducer,
    ),

    # 終了戦略
    termination_strategy=KernelFunctionTerminationStrategy(
        agents=[agent_reviewer],
        function=termination_function,
        kernel=kernel,
        result_parser=lambda result: termination_keyword in str(result.value[0]).lower(),
        history_variable_name="lastmessage",
        maximum_iterations=8, # 最大 8 回のやり取りを許可
        history_reducer=history_reducer,
    ),
)

動作確認

ユーザー入力から記事を執筆するマルチエージェントを動かしてみます。

user_input = ("生成 AI によるマルチエージェントのユースケースについて")

print(f"=== User ===")
print(user_input)

await chat.add_chat_message(message=user_input)

try:
    turn = 1
    async for response in chat.invoke():
        if response is None or not response.name:
            continue
        print(f"\n=== TURN {turn} ({response.name.upper()}) ===")
        print(response.content)
        turn += 1
except Exception as e:
    print(f"Error during chat invocation: {e}")

出力例（抜粋）

=== User ===
生成 AI によるマルチエージェントのユースケースについて

=== TURN 1 (REVIEWER) ===
生成AIによるマルチエージェントのユースケースに関する

...（中略）...

=== TURN 2 (WRITER) ===
# 生成AIによるマルチエージェントのユースケース

...（中略）...

=== TURN 8 (WRITER) ===
# 生成AIによるマルチエージェントのユースケース

## 概要
生成AIは、マルチエージェントシステムにおいて重要な役割を果たしています。このシステムでは、多数のエージェントが協力し、情報を共有することで複雑な問題を解決します。この記事では、生成AIがもたらすマルチエージェントのユースケースについて詳述します。

## 1. 基本概念の説明
### マルチエージェントシステムとは
マルチエージェントシステム（MAS）は、複数のエージェント（自律的なプログラム）が互いに作用して特定の目標を達成するシステムです。このシステムにより、分散処理や協力的問題解決が可能となります。

### 生成AIの効果
生成AIは、マルチエージェントシステム内でエージェントの行動を強化し、適応的な反応を可能にし、よりリアルなインタラクションを提供します。これにより、エージェントは環境の変化に迅速に対応することができます。

## 2. ユースケースの具体例
- **ゲーム開発**
  - プレイヤーの行動に応じてNPC（ノンプレイヤーキャラクター）が適応し、リアルな動きや反応を示します。この要素はゲーム体験を大いに向上させ、プレイヤーの没入感を高めます。

- **交通管理**
  - 複数のエージェントがリアルタイムで交通データを処理し、渋滞を回避するための最適な交通流を維持します。ここでは、信号機の制御や交差点のダイナミックな調整が含まれます。

- **市場取引**
  - 自律型エージェントがストックマーケットのデータをリアルタイムで処理し、効率的な取引を実現します。これにより、投資家のリスクを軽減し、利益を最大化することが可能になります。

## 3. 生成AIの役割
生成AIは、以下の方法でエージェントの行動を強化します：
- **強化学習**: エージェントが環境からのフィードバックを基に行動を調整します。
- **深層学習アルゴリズム**: 複雑なパターンや相関関係を理解し、エージェントが新たな情報に基づいて柔軟に意思決定を行う能力を向上させます。

## 4. 具体的な事例研究
実際に生成AIを用いた成功事例：
- **医療分野**: 生成AIによるマルチエージェントシステムが患者データを効率的に分析し、リスクの高い患者を早期に特定することで、治療の精度を向上させた事例があります。具体的なデータとして、診断時間が30%短縮されたケースや、治療成功率が20%向上した例が挙げられます。

## 5. メリットとデメリット
### メリット
- **迅速な問題解決**: 複数のエージェントの協力により、迅速に問題を解決できます。
- **効率性向上**: 資源の最適配分が可能となり、コスト削減にも寄与します。

### デメリット
- **プライバシー問題**: データの取り扱いに関する倫理的な懸念が生じる可能性があります。
- **技術的課題**: エージェント間の衝突解決やデータの整合性保持は依然として難しい問題です。

## 6. 未来の展望
生成AIとマルチエージェントシステムの発展に伴い、個別化されたユーザーサービスの提供が可能になることが期待されています。特に、教育分野やヘルスケア分野での応用が進展するでしょう。

### 教育やトレーニングの応用
生成AIとマルチエージェントシステムが教育の場でどのように活用され、学習効果が向上するかを示す具体例や指標を提供します。

## 7. 法的規制と倫理的観点
生成AIやマルチエージェントシステムの導入に伴う法律面での課題や、データ保護規制（GDPRなど）への対応について考察します。特に、倫理的視点を重視することがますます重要です。

## 8. 将来の技術動向
マルチエージェントシステムと生成AIは、今後の技術的なトレンドと新しいアプローチによって発展し続けるでしょう。特に、AIの透明性や説明責任が求められる時代において、これらの技術の倫理的な利用が肝要となります。

## 結論
生成AIによるマルチエージェントのユースケースは多岐にわたり、今後の技術革新によってさらなる発展が期待されます。読者には、自身の業界においてどのようなユースケースが興味深いかを考え、実践に結びつける機会を持つことを促します。
Sending email to [e-mail]...

このように、AgentGroupChat を用いることで、役割上は相互に刺激しあいながら、質の高いアウトプットに仕上げる競争型マルチエージェント構成が実現できます。

ワークフロー型

Process Framework で実装する方法

ワークフロー型マルチエージェントの構築は、 Semantic Kernel Process Framework （Process Framework）というフレームワークと併用することで実装可能です。

この Process Framework を活用すれば、各ステップを Python 関数で自由に記述できるため、これまで作成した Semantic Kernel Agents を各ステップで動かすことが可能です。

もちろん、ルールベースの決定論的処理を組み込んでエージェントの処理と併用することもできます。

2025年4月現在、Process Framework は実験段階（Experimental）なステータスです。将来的に破壊的変更が加えられる可能性があるため、本番環境での利用は推奨されません。

当記事の GitHub リポジトリ では、Process Framework と Semantic Kernel Agents を活用したワークフロー型マルチエージェントのサンプルコードも公開しています。具体的な実装に興味がある方はぜひご覧ください。

さらに、以下の Qiita 記事では Process Framework の入門として、代表的な処理パターンのサンプルコードを解説しています。はじめて Process Framework に触れる方は、こちらもあわせてチェックしてみてください。

今後の展望

先日公開された Semantic Kernel の 2025 年上半期ロードマップ において、今後の注目ポイントのひとつとして Semantic Kernel Agents を用いたワークフロー構築機能の強化 が示されました。

こちらは、VS Code 拡張機能を通じて、視覚的かつ直感的にエージェント連携を設計できる機能のようです。

以下の YouTube 動画では、yaml 形式でエージェント間のステップを定義し、VS Code 上でワークフローをプレビューするデモが紹介されています。

こうした機能により、ワークフロー型マルチエージェント構成の開発体験はさらに効率的になることが期待されます。

引き続き、この新機能の進展をウォッチしていきます。

参考：New capabilities in Azure AI Foundry to build advanced agentic applications

GitHub

当記事で紹介しているコード一式は、以下の GitHub リポジトリにまとめています。