💡
Semantic KernelのAgent Frameworkによるマルチエージェント実装入門
Semantic Kernelについて、Agent Frameworkを用いたマルチエージェントを実装について確認していきます。
（今回は、Agent FrameworkのみでProcess Frameworkについては、別途確認予定）
Agent Frameworkでのマルチエージェント
Agent Frameworkでマルチエージェントについて、以下の3種類を確認してみます。
| 種類 | 内容 |
|---|---|
| GroupChat方式 | 複数エージェントを同じ会話に参加させ、議論するようなマルチエージェント |
| Handoff方式 | OpenAIのswarmで提唱されていた方式。「トリアージ」エージェントを置き、入力内容に応じて、回答を生成するエージェントを切りかえる |
| Orchestrator方式 | 明示的に「司令塔」エージェントを用意し、複数のサブエージェントを管理する構成。人間のPMのようにタスク管理を行うようなイメージ |
それぞれについて、簡単な実装例を確認していきます。
GroupChat方式
今回は、4つのエージェントを用意し、4人が同じ会話（GroupChat）に参加し、リサーチ → アイデア出し → 批評 → まとめまでをラウンドロビンで回す形で作成しています。
モジュールのインポート
import os
import asyncio
from dotenv import load_dotenv
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.agents.group_chat.agent_group_chat import AgentGroupChat
from semantic_kernel.agents.strategies.selection.sequential_selection_strategy import (
SequentialSelectionStrategy,
)
from semantic_kernel.agents.strategies.termination.termination_strategy import (
TerminationStrategy,
)
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.contents.utils.author_role import AuthorRole
まずは、必要なモジュールをインストールします。
AgentGroupChatは、複数エージェントを1つの会話に束ねるハブとして利用します。
SequentialSelectionStrategyで、順番に発言させる（ラウンドロビン）ようにします。
API/カーネル設定
# =========================
# 設定/初期化
# =========================
def build_kernel_and_settings():
load_dotenv()
kernel = Kernel()
service = AzureChatCompletion(
api_key=os.environ["AZURE_OPENAI_API_KEY"],
endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"],
)
kernel.add_service(service)
return kernel
.env から Azure の3点（APIキー / エンドポイント / デプロイ名）を読みこみ、KernelにAzureChatCompletionサービスを追加しています。
（以下、.envファイルに環境設定を行っている前提のコードになっています）
エージェント定義
def make_agents(kernel):
researcher = ChatCompletionAgent(
kernel=kernel,
name="Researcher",
description="市場/現場リサーチ担当",
instructions=(
"あなたはリサーチ担当。与えられたテーマについて、"
"想定ユーザー・ニーズ・先行事例・リスクを3〜5点で簡潔に列挙してください。"
"URLは不要。具体的かつ短く。"
),
)
ideator = ChatCompletionAgent(
kernel=kernel,
name="Ideator",
description="アイデア担当",
instructions=(
"あなたはアイデア担当。Researcherの示唆を受け、"
"実装可能性を踏まえたアイデアを3案提案。各案は"
"『狙い→主要機能→期待効果→KPI(1つ)』の順で1〜2行ずつ。"
),
)
critic = ChatCompletionAgent(
kernel=kernel,
name="Critic",
description="批評/改善担当",
instructions=(
"あなたは批評担当。Ideatorの案それぞれに対して、"
"リスク・前提・代替案を短く指摘し、各案に1つ改善提案を添えてください。"
),
)
synthesizer = ChatCompletionAgent(
kernel=kernel,
name="Synthesizer",
description="統括まとめ担当",
instructions=(
"あなたは統括担当。これまでの出力を踏まえ、"
"最終案を1つに統合して提示してください。"
"最後に <<FINAL>> と1行で明記して終了してください。"
),
)
return [researcher, ideator, critic, synthesizer]
各エージェントの定義を行います。
今回は、以下、4エージェント（名前・説明・指示文）を用意しました。
- Researcher：事実ベースの示唆を短く列挙する
- Ideator：その示唆を受けた実装可能な3案を提示する
- Critic：各案のリスクや代替案＋改善提案提示する
- Synthesizer：最終案に統合し、最後に <<FINAL>> と書く
会話終了の設定
class FinalTokenTerminationStrategy(TerminationStrategy):
"""
Synthesizer（まとめ役）が <<FINAL>> を発言したら終了。
ループ暴走防止に maximum_iterations も併用。
"""
def __init__(self, *, maximum_iterations: int = 12, agents=None):
super().__init__(maximum_iterations=maximum_iterations, agents=agents)
async def should_agent_terminate(self, agent, history):
# 直近履歴から <<FINAL>> を検知したら終了
for message in reversed(history):
content = getattr(message, "content", None) or ""
if "<<FINAL>>" in content:
return True
return False
会話の終了条件を定義しています。
今回は、Synthesizerエージェントが「FINAL」を宣言し、その文言が出現したら会話を終了するように設定しています。
また、会話が長引いた場合を想定して、MAX12までで終了するよう設定しています。
GroupChat構築
async def run_brainstorm_non_stream(user_goal: str):
kernel = build_kernel_and_settings()
agents = make_agents(kernel)
selection = SequentialSelectionStrategy() # ラウンドロビン
termination = FinalTokenTerminationStrategy(maximum_iterations=12, agents=[agents[-1]]) # Synthesizer を指定
chat = AgentGroupChat(
agents=agents,
selection_strategy=selection,
termination_strategy=termination,
)
chat.history.add_message(ChatMessageContent(role=AuthorRole.USER, content=user_goal))
print("\n=== Multi-Agent GroupChat (non-stream) ===\n")
while not chat.is_complete:
async for message in chat.invoke():
if message.content:
speaker = getattr(message, "name", None) or str(message.role)
print(f"[{speaker}] {message.content}")
if message.content and "<<FINAL>>" in message.content:
chat.is_complete = True
break
if __name__ == "__main__":
topic = (
"テーマ: 製造現場の不良削減と段取り時間短縮に効く、"
"『現場向けAIアシスタント（モバイル/タブレット）』の新機能をブレストし、"
"最終的に1案へ統合してください。制約: 3ヶ月でPoC可能な範囲。"
)
asyncio.run(run_brainstorm_non_stream(topic))
AgentGroupChatに以下を設定。
- agents：4エージェント
- selection_strategy：SequentialSelectionStrategy（登録順に発言）
- termination_strategy：先ほどの FinalTokenTerminationStrategy
ユーザーのテーマ(user_goal)を会話履歴に追加して会話を実施します。
実行結果は以下の通り。
[Researcher]
【想定ユーザー】
1. 製造ラインの現場作業者
2. ラインリーダーや品質管理担当者
3. 設備保全スタッフ
【ニーズ】
1. 不良品発生の早期検知と原因特定支援
2. 段取り替え手順の迅速かつ正確な案内
3. 作業状況のリアルタイムフィードバック
4. 誰でも使いやすい直感的なUI
【先行事例】
1. AIによる画像検査で不良検出率向上のスマートカメラ導入
2. モバイル端末で段取りマニュアル動画を表示する現場支援アプリ
3. IoTセンサー連携による稼働状況の可視化ツール
【リスク】
1. AI判定の誤検知による現場混乱
2. 作業者のICTリテラシー不足による導入障壁
3. 現行システムや工程との連携不整合
4. 短期間でのPoCにおけるデータ不足
【提案新機能案（統合）】
「AI画像認識による不良品リアルタイム検知機能＋段取り替え時の作業員向けARガイド表示機能」
→ モバイル/タブレットのカメラを活用し、不良部品や段取りミスをAIがリアルタイム判定。加えて、AR技術で段取り替え手順を直感的に表示し、短期間のPoCで現場の不良削減と段取り時間短縮を同時に支援。
[Ideator]
案1
狙い：不良品の早期発見で歩留まり向上を目指す
主要機能：モバイル端末のカメラで製品を撮影し、AIが不良品をリアルタイム検知・通知する
期待効果：不良の早期対応で再作業減、品質改善に寄与
KPI：不良検知率
案2
狙い：段取り替え時間の短縮とミス削減を実現する
主要機能：段取り替え手順をARでタブレット上に表示し、作業者の確認と進捗管理をサポート
期待効果：段取り時間短縮と手順ミス減少に効果的
KPI：段取り時間の短縮率
案3
狙い：不良検知と段取り支援を一体化し工数削減を図る
主要機能：不良品検知AIと段取り替えARガイドを同一アプリで実装し、作業効率を上げる
期待効果：不良削減と段取り時間短縮の同時達成で総合的な生産性向上
KPI：不良品発生率の低減
【統合案】
狙い：3ヶ月でPoC可能なAI＋ARの現場支援アプリ開発で、不良削減と段取り時間短縮を同時に実現する
主要機能：モバイル端末のカメラで製品の不良検知をリアルタイムに行い、段取り替え時にはARで作業手順を直感的に表示し進捗をサポート
期待効果：現場の即時不良検知と段取り支援により品質改善と工数削減を促進
KPI：不良品発生率の低減
[Critic]
案1：不良品のリアルタイムAI検知
リスク：画像データの質や多様性が不足し誤検知増加の可能性。
前提：作業環境で安定したカメラ撮影が可能であること。
代替案：撮影自動化や手動報告併用。
改善提案：誤検知時の作業員による簡単なフィードバック機能を加え、AIを学習させる。
案2：段取り替え支援のARガイド表示
リスク：AR操作の習熟に時間がかかり逆に工数増加の恐れ。
前提：現場でタブレットのAR利用が許容されること。
代替案：動画や画像での手順案内。
改善提案：シンプル操作とステップ別の小さな案内画面で直感的使いやすさを強化。
案3：不良検知と段取りガイドの統合アプリ
リスク：機能統合により開発工数が増えPoC期間内完了が困難。
前提：AI画像認識とAR技術双方で最低限の実装レベルが確保できること。
代替案：優先順位をつけ段取りガイドの簡略版から先に実装。
改善提案：PoCではまず不良検知機能に集中し、段取りガイドは簡易版で組み込み段階的に機能拡張。
【最終統合案】
「モバイル端末カメラ利用の不良品リアルタイムAI検知機能と、段取り替え時に利用可能なシンプルAR手順ガイドを組み合わせ、作業効率と品質向上を同時に目指すアプリ」
3ヶ月PoCでの実装を意識し、不良検知に重点を置きつつ段取り支援は直感的で簡易的なAR案内を併設。現場フィードバックで機能精度向上を目標にする。
[Synthesizer]
【最終統合案】
■狙い
3ヶ月のPoC期間で、製造現場の不良削減と段取り時間短縮を同時に実現し、作業効率と品質向上を促進する。
■主要機能
1. モバイル／タブレットのカメラを用いたAI画像認識による不良品のリアルタイム検知・通知
・撮影環境に適応した画像収集と簡易フィードバック機能でAI精度向上を図る
2. 段取り替え作業時に使えるシンプルなAR手順ガイド表示
・直感的かつ段階的に作業手順を提示し、作業ミスや段取り時間を削減
・複雑な操作は避け、現場作業者のICTリテラシーに配慮
■PoC体制
・まず不良検知機能を中心に開発・評価し、段取り替えARガイドは簡易版として同時並行で開発
・現場での操作性・効果検証を経てフィードバックで改善し、段階的に機能をブラッシュアップ
・品質管理担当者やリーダーも巻き込み現場受容性を確認
■期待効果
・不良品を早期に検知し再作業・廃棄削減を促進（不良品発生率の低減）
・段取り替え作業の時間短縮と手順ミス減少により稼働効率が向上（段取り時間短縮率の向上）
・現場の業務負荷軽減と品質安定化
この統合案は現場の使いやすさと精度向上を両立し、限られた3ヶ月でPoC実施可能な範囲に留めた実現性高いアプローチです。
Handoff方式
今回は、3つのエージェント（要約、翻訳、天気回答）を用意し、トリアージエージェントがユーザの質問に応じて、エージェントを選択して回答を生成する形にします。
モジュールのインポート/サービスの設定
import os
import asyncio
from dotenv import load_dotenv
from datetime import datetime
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.agents.chat_completion.chat_completion_agent import ChatHistoryAgentThread
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.functions import kernel_function
# =========================
# .env 読み込み
# =========================
load_dotenv()
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")
service = AzureChatCompletion(
api_key=AZURE_OPENAI_API_KEY,
deployment_name=AZURE_DEPLOYMENT_NAME,
endpoint=AZURE_OPENAI_ENDPOINT,
)
Weather用関数（function calling）
# =========================
# Weather 用のダミー関数（東京/大阪のみ）
# =========================
class WeatherPlugin:
@kernel_function(name="get_weather", description="指定都市（東京/大阪/Tokyo/Osaka）の本日の天気を返す")
def get_weather(self, city: str) -> str:
c = (city or "").strip().lower()
if c in ("東京", "tokyo"):
return "Tokyo: 晴れ, 最高32℃ / 最低26℃, 降水確率10%"
if c in ("大阪", "osaka"):
return "Osaka: くもり一時晴れ, 最高33℃ / 最低27℃, 降水確率20%"
return "対応都市は Tokyo/Osaka（東京/大阪）のみです。"
ここでは、事前にWeatherエージェントを呼んだ際に使用する関数をダミーで定義しています。
（ダミーの天気情報ではなく実際の天気情報を取得するのであればAPIなどを利用する）
@kernel_function を付けたメソッドは LLM から呼べる関数になります。
Weatherエージェントにこのプラグインを登録し、LLM が function calling で呼べるようにしています。
エージェントの定義
# =========================
# 専門エージェント
# - Translator: 翻訳
# - Summarizer: 要約
# - Weather: 天気（function calling で WeatherPlugin を呼ぶ）
# =========================
translator_agent = ChatCompletionAgent(
service=service,
name="Translator",
instructions=(
"あなたはプロの翻訳者です。ユーザーの文を自然で読みやすい日本語/英語に翻訳します。"
"不要な前置きは避け、翻訳文のみを返してください。"
),
)
summarizer_agent = ChatCompletionAgent(
service=service,
name="Summarizer",
instructions=(
"あなたは要約の専門家です。入力テキストを3〜5行で要約します。"
"重要点を簡潔に、箇条書き可。"
),
)
weather_agent = ChatCompletionAgent(
service=service,
name="Weather",
instructions=(
"ユーザーが天気を尋ねたら、都市名（東京/大阪/Tokyo/Osaka）を抽出し、"
"Weather.get_weather を function calling で呼び、返り値を1〜3行で要約して回答してください。"
"都市名が不明なら1回だけ質問して特定してください。"
),
plugins=[WeatherPlugin()],
function_choice_behavior=FunctionChoiceBehavior.Auto(), # プラグイン関数を自動選択
)
ここでは、各エージェントの定義を行っています。
Translator と Summarizerについては、instructionsのみで動くシンプルなエージェントにしていますが、Weatherのみplugins=[WeatherPlugin()]で、ツールとして、WeatherPluginを了できるようにしています。
こうすることで、LLMが「天気の質問だ」と判断したら、get_weather(city) を自ら呼び、戻り値を使って回答します。
Triage が叩く「呼び出し関数」の定義
# =========================
# Triage が使う「呼び出し用プラグイン」
# - Triage はこの関数群を function calling で選び、実行します
# - 中で各専門エージェントを起動し、回答結果を返します。
# =========================
class AgentInvokerPlugin:
def __init__(self, *, thread: ChatHistoryAgentThread | None = None):
self.thread = thread
def _ensure_thread(self):
if self.thread is None:
raise RuntimeError("thread が未設定です。実行前に invoker.thread = thread をセットしてください。")
@kernel_function(
name="use_translator",
description="翻訳担当を起動し、入力テキストを適切な言語へ翻訳して返す。引数 text にユーザー文を渡す。")
async def use_translator(self, text: str) -> str:
self._ensure_thread()
resp = await translator_agent.get_response(messages=text, thread=self.thread)
return str(resp)
@kernel_function(
name="use_summarizer",
description="要約担当を起動し、入力テキストを3〜5行で要約して返す。引数 text に元テキストを渡す。")
async def use_summarizer(self, text: str) -> str:
self._ensure_thread()
resp = await summarizer_agent.get_response(messages=text, thread=self.thread)
return str(resp)
@kernel_function(
name="use_weather",
description="天気担当を起動し、ユーザー発話をそのまま渡す。必要に応じて都市名の質問→Weather.get_weather を呼ぶ。")
async def use_weather(self, text: str) -> str:
self._ensure_thread()
resp = await weather_agent.get_response(messages=text, thread=self.thread)
return str(resp)
TriageAgentエージェントが、各専門エージェントを呼ぶための窓口関数を用意します。
use_translator / use_summarizer / use_weather の3つを kernel_function として公開しており、ユーザの質問に応じて選択します。
TriageAgentの定義
# =========================
# TriageAgent（AIに任せるハンドオフ）
# - 目的：ユーザー意図を判断し、上記 use_* 関数のいずれか（複数可）を function calling で選択・実行
# =========================
invoker = AgentInvokerPlugin()
triage_agent = ChatCompletionAgent(
service=service,
name="TriageAgent",
instructions=(
"あなたはオーケストレーションエージェントです。ユーザーのリクエストを読み取り、"
"必要に応じて次の関数から**最も適切なもの**（複数可）を選んで実行してください："
"use_translator, use_summarizer, use_weather。"
"関数の出力を確認・要約し、必要なら不足情報を1回だけ質問してから、"
"ユーザーにとって最も役立つ**統合済みの最終回答**を返してください。"
"直接の独断回答は避け、原則として少なくとも1回は関数を呼んでください。"
),
plugins=[invoker],
function_choice_behavior=FunctionChoiceBehavior.Auto(),
)
TriageAgent は「オーケストレータ」としての役割をしています。
ユーザの質問に応じて、どのAIエージェントに回答を生成させるかを決め、そのエージェントの回答を返します。
「Triage → Invoker → 専門エージェント」 の流れで、プログラムは動作します。
実行
# =========================
# デモ実行
# =========================
async def main():
user_inputs = [
"東京の天気を教えて。傘は要る？",
"次の文を英語に翻訳して: 今日は雨ですが出かけます。",
"以下を3行で要約してください。生成AIの導入による最大のメリットは、生産性と付加価値の大幅な向上にあります。まず、文章作成やデータ分析、プログラム生成といった高度かつ時間を要するタスクを自動化することで、従業員は単純作業から解放され、より戦略的で創造性を発揮できる業務に集中できます。これにより作業効率が向上し、同じ時間でより多くの成果を生み出せるようになります。また、生成AIは膨大な情報を高速かつ正確に処理できるため、企画立案や意思決定の場面でも迅速なサポートが可能です。さらに、顧客対応ではチャットボットやFAQの自動応答を通じて24時間のサポート体制を実現し、顧客満足度の向上にもつながります。加えて、生成AIは既存のナレッジを学習し新しい知見を創出する力を持つため、研究開発や製品設計においてもイノベーションを加速させます。このように、生成AIの導入はコスト削減と業務効率化だけでなく、競争力強化や新たなビジネス価値創出の基盤となる点で極めて大きな意義があります。",
]
thread = ChatHistoryAgentThread()
invoker.thread = thread
for user_message in user_inputs:
response = await triage_agent.get_response(messages=user_message, thread=thread)
print(f"User: {user_message}\n[Agent] {response}\n{'-'*40}")
if __name__ == "__main__":
asyncio.run(main())
実行結果は以下の通り。
Orchestrator方式
今回は、Orchestrator（PM役）が自分で考えて、タスク追加/更新/一覧と担当エージェント呼び出しを選択・実行するプログラムの作成を行います。
目的 → タスク → 担当へ依頼 → 進捗記録 → サマリを単一エージェントで実施し、Orchestrator が全体を束ねるような作りにします。
モジュールのインポート/サービスの設定
import os
import asyncio
from dataclasses import dataclass, field
from typing import Dict, List
from dotenv import load_dotenv
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.agents.chat_completion.chat_completion_agent import ChatHistoryAgentThread
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior
from semantic_kernel.functions import kernel_function
# =========================
# .env & Azure OpenAI
# =========================
load_dotenv()
service = AzureChatCompletion(
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
deployment_name=os.getenv("AZURE_DEPLOYMENT_NAME"),
endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
)
専門エージェントの設定
# =========================
# 専門エージェント（実務担当）
# - Researcher: 事前調査・競合・KPI候補
# - Designer: ユーザーフロー/画面ラフ
# - Engineer: 実装方針/PoC計画
# =========================
researcher = ChatCompletionAgent(
service=service,
name="Researcher",
instructions=(
"あなたは市場/ユーザー調査の担当です。依頼に対し、ターゲット/ニーズ/競合/リスク/計測KPI候補を"
"3〜6点で箇条書きにしてください。出力は簡潔に。"
),
)
designer = ChatCompletionAgent(
service=service,
name="Designer",
instructions=(
"あなたはプロダクトデザイン担当です。依頼に対して、主要ユーザーフロー（3〜5段階）と"
"画面ラフ説明（2〜4枚想定）を箇条書きで提示してください。"
),
)
engineer = ChatCompletionAgent(
service=service,
name="Engineer",
instructions=(
"あなたはエンジニア担当です。PoCの技術方針/アーキテクチャ/必要なAPI/スプリント計画（2週間想定）を"
"箇条書きで示し、最小スコープを明記してください。"
),
)
それぞれ役割専用の指示（instructions）を与えたエージェントを用意します。
Orchestratorはここで定義したエージェントにタスクを割り振るようにします。
タスクボード（PMのToDo管理）
# =========================
# タスクボード（PMのToDo管理）
# =========================
@dataclass
class Task:
id: str
title: str
assignee: str # Researcher/Designer/Engineer
status: str = "todo" # todo / doing / done
notes: List[str] = field(default_factory=list)
due: str = ""
class TaskBoardPlugin:
"""Orchestrator用：タスク作成・更新・一覧を扱うダミーTaskBoard。"""
def __init__(self):
self._seq = 0
self.tasks: Dict[str, Task] = {}
def _next_id(self) -> str:
self._seq += 1
return f"T{self._seq:03d}"
@kernel_function(
name="add_task",
description="新しいタスクを追加。引数: title(必須), assignee=[Researcher|Designer|Engineer], due(任意)"
)
def add_task(self, title: str, assignee: str, due: str = "") -> str:
tid = self._next_id()
self.tasks[tid] = Task(id=tid, title=title, assignee=assignee, due=due)
return f"ADDED {tid} [{assignee}] {title} (due:{due or '-'})"
@kernel_function(
name="update_task",
description="既存タスクの状態やメモを更新。引数: id, status=[todo|doing|done], note(任意)"
)
def update_task(self, id: str, status: str = "", note: str = "") -> str:
t = self.tasks.get(id)
if not t:
return f"ERROR: task {id} not found"
if status:
t.status = status
if note:
t.notes.append(note)
return f"UPDATED {id} status={t.status} notes={len(t.notes)}"
@kernel_function(name="list_tasks", description="全タスクを表形式のテキストで返す")
def list_tasks(self) -> str:
if not self.tasks:
return "(no tasks)"
lines = ["ID | Assignee | Status | Title | Due | Notes",
"---|----------|--------|-------|-----|------"]
for t in self.tasks.values():
lines.append(f"{t.id} | {t.assignee} | {t.status} | {t.title} | {t.due or '-'} | {len(t.notes)}")
return "\n".join(lines)
@kernel_function(name="get_task", description="タスク詳細を返す。引数: id")
def get_task(self, id: str) -> str:
t = self.tasks.get(id)
if not t:
return f"ERROR: task {id} not found"
notes = "\n - ".join(t.notes) if t.notes else "-"
return (f"{t.id} [{t.assignee}] {t.title}\n"
f"status: {t.status}\n"
f"due: {t.due or '-'}\n"
f"notes:\n - {notes}")
Orchestrator が自分でこの関数を選び、タスク追加/更新/一覧を操作します。
今回は、コード内で関数を定義し、タスクを管理していますが、タスク自体は、DB等管理することもできます。
タスクの追加/更新/一覧操作について、@kernel_function が付いているので、LLMから“関数（ツール）として呼べる”ようになります。
各エージェントの呼び出し
# =========================
# 専門家呼び出し
# =========================
class AgentInvokerPlugin:
def __init__(self, thread: ChatHistoryAgentThread | None = None):
self.thread = thread
def _need_thread(self):
if self.thread is None:
raise RuntimeError("thread 未設定。実行前に invoker.thread = thread をセットしてください。")
@kernel_function(name="use_researcher",
description="Researcherを起動。引数 text に要件を渡す。")
async def use_researcher(self, text: str) -> str:
self._need_thread()
resp = await researcher.get_response(messages=text, thread=self.thread)
return str(resp)
@kernel_function(name="use_designer",
description="Designerを起動。引数 text に要件を渡す。")
async def use_designer(self, text: str) -> str:
self._need_thread()
resp = await designer.get_response(messages=text, thread=self.thread)
return str(resp)
@kernel_function(name="use_engineer",
description="Engineerを起動。引数 text に要件を渡す。")
async def use_engineer(self, text: str) -> str:
self._need_thread()
resp = await engineer.get_response(messages=text, thread=self.thread)
return str(resp)
Orchestratorは、ここで定義した関数を function calling で呼び、内部で専門エージェントに処理させます。
新しいエージェントを追加する場合は、エージェントの定義を行い、ここに関数を1つ足すだけで専門エージェントを増やすことが可能です。
Orchestratorの定義
# =========================
# Orchestrator（PM役）
# - 目的分解→タスク化→担当へ依頼→進捗記録→最終レポート
# - すべて function calling で自律的に進める
# =========================
taskboard = TaskBoardPlugin()
invoker = AgentInvokerPlugin() # thread は実行時に注入
orchestrator = ChatCompletionAgent(
service=service,
name="Orchestrator",
instructions=(
"あなたはPM（プロジェクトマネージャー）です。ユーザーの高レベル目標を受け、"
"1) タスク分解（2〜6個）→ 2) 担当割当（Researcher/Designer/Engineer）→ "
"3) Invoker関数で担当に依頼 → 4) TaskBoardに進捗/メモを記録 → 5) 最終レポートを返す、を実施します。\n"
"利用できる関数:\n"
"- add_task(title, assignee, due?) / update_task(id, status?, note?) / list_tasks() / get_task(id)\n"
"- use_researcher(text) / use_designer(text) / use_engineer(text)\n"
"推奨フロー:\n"
"A. 目標を読み取り、add_task を複数回で作成。各taskのidを控える。\n"
"B. それぞれに対応する use_* を呼んで成果を得る。update_task で status=doing/done と notes を追記。\n"
"C. list_tasks でボードの最新を取得。『リスク/依存/次の一手』も簡潔にまとめてください。\n"
"不明点は1回だけ質問して不足を補ってください。"
),
plugins=[taskboard, invoker],
function_choice_behavior=FunctionChoiceBehavior.Auto(),
)
orchestratorの内容を instructions に定義します。
orchestratorが使用できる関数としては、
- TaskBoardPlugin の関数（add/update/list/get）
- AgentInvokerPlugin の関数（use_researcher/designer/engineer）
上記2つの関数を駆使して、orchestratorは、「目的 → タスク化 → 担当に依頼 → 記録 → まとめ」 を自分で考えて進めるようになります。
実行
# =========================
# 実行
# =========================
async def main():
thread = ChatHistoryAgentThread()
invoker.thread = thread
user_goal = (
"製品不良を検知する新機能を2週間でPoCにまとめたい。まず市場/ユーザー調査、"
"主要フローの画面ラフ、最小スコープの実装方針まで進めて。"
"KPI候補と主要リスクも洗い出して。"
)
print("=== Orchestrator demo ===\n")
resp = await orchestrator.get_response(messages=user_goal, thread=thread)
print(str(resp))
print("\n--- Current TaskBoard ---")
print(taskboard.list_tasks())
if __name__ == "__main__":
asyncio.run(main())
以下、実行イメージになります。
User Goal
↓
Orchestrator (PM)
├─ add_task() × N … タスク作成
├─ use_researcher() … 調査依頼 → 結果を update_task(note) で記録
├─ use_designer() … 画面ラフ → 記録
├─ use_engineer() … PoC方針 → 記録
├─ list_tasks() … 最新ボードの取得
└─ 最終レポート作成 … KPI/リスク/次の一手 などをまとめる
上記を実行した結果、以下のような出力となりました。
=== Orchestrator demo ===
2週間での製品不良検知新機能PoCに向け以下の進捗です。
1. 市場/ユーザー調査（完了）
IoTやAIを活用した不良検知市場が拡大中。主なターゲットは製造業の品管理担当者やライン管理者。ユーザー課題は高精度の不良検知、導入コスト低減、既存システムとの連携の難しさ。
2. 主要フローの画面ラフ（完了）
主要ユーザーフローは以下4段階。画面は選択、検知開始、結果表示、履歴管理の4画面構成で、シンプルかつ直感的なUIを意識しました。
- 不良検知対象製品の選択
- 検知開始およびデータ取得
- 検知結果表示（良品/不良品の判定）
- 結果履歴と追加対応画面
3. 最小スコープの実装方針（完了）
単一種類の不良検知に絞り、既存のAI画像解析モデルを活用し検知判定を行う。画面はReactで構築し、バックエンドはPython。AWS EC2でのPoC運用を想定。主な技術リスクはAI精度不足とリアルタイム性。
4. KPI候補（完了）
- 検知精度（正確率）
- 検知時間（アラート発信までの時間）
- 検知率（総不良品に対する検出件数割合）
- 誤検知件数（誤報の頻度）
- ユーザーフィードバック数
- システム稼働率
- ユーザー満足度評価
5. 主要リスク（完了）
- AIモデルの検出精度不足による誤検知・見逃しリスク
- リアルタイム処理遅延の可能性
- ユーザーの誤操作や誤認識による誤対応
- システム障害発生時の対応不足
- 既存業務フローとの非整合性
- 競合サービスとの差異化困難
【リスク/依存/次の一手】
- AI精度向上やモデル改良はPoC後の課題。
- 既存システム連携要件が明確になると設計変更リスク。
- 次はPoC用プロトタイプ画面の詳細設計とAIモデルのテスト実施が望ましいです。
以上、ご要望のPoC準備フェーズを進めました。ご質問や追加希望があればお知らせください。
--- Current TaskBoard ---
ID | Assignee | Status | Title | Due | Notes
---|----------|--------|-------|-----|------
T001 | Researcher | done | 市場/ユーザー調査を行う | - | 1
T002 | Designer | done | 主要フローの画面ラフを作成する | - | 1
T003 | Engineer | done | 最小スコープの実装方針を検討する | - | 1
T004 | Researcher | done | KPI候補の洗い出し | - | 1
T005 | Researcher | done | 主要リスクの洗い出し | - | 1
各専門エージェントにタスクの割り振りを実施できており、必要な情報を出力できているかと思います。
まとめ
Agent Frameworkでのマルチエージェントの実装例をいくつか確認しました。
ここで確認した内容だけでもプロンプトさえ精緻なものに変えれば、より高度なマルチエージェントシステムを作成できることがわかりました。
そのうえで、Process Frameworkを加えたマルチーエージェントシステムにすることで、より高度な業務フローにも対応したエージェントが作成できそうです。
次は、Process Frameworkについても確認しようと思います。
Discussion