はじめに
課題
何かに悩んだり困った際にはweb検索をしたり、上司や友人、家族などに相談することは多いと思いますが生成AIの名前を聞くようになってからは生成AIに相談する人も多いのではないでしょうか。
また生成AIに心理カウンセラーなど役割やペルソナを与えることで特定の知識を持ったAIに回答させることでより良い助言をもらえるようにもなりました。
一方で役割ごとに複数のAIを用意した場合には個別のAIに都度質問することが面倒に感じることがありました。
解決策
ちょうどそのようなことを考えていた時に以下の記事に出会いました。
ルートエージェントとサブエージェントという仕組みを使うことで、ルートエージェントが質問を受け付け、質問の内容に応じて特定のサブエージェントが回答することが実現できるようです。
つまり質問する入り口を１つでも複数の役割を持ったAIエージェントに質問することができるので個別のAIに毎回質問するという手間が省けると感じました。
そこで今回は上記の記事を元にマルチエージェントのシステムを開発したいと思います。
実装内容
以下を実装しました。
- サブエージェントの役割に応じて質問を投げるコーディネーター
- サブエージェント
- 営業の進め方や提案内容を回答する営業エージェント
- Google Cloudの技術的な質問に回答するエンジニアエージェント
- その他の一般的な質問に回答する一般エージェント
今回はColabにて実装しています。
実装
上記の記事を参考に実装しました。
準備
まずは必要なパッケージをインストールします。
pip install google-adk
必要なモジュールをインポートします。
import uuid, re
from google.adk.agents.llm_agent import LlmAgent
from google.genai.types import Part, UserContent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
GoogleのAPIキーを設定します。
from google.colab import userdata
import os
os.environ["GOOGLE_API_KEY"] = userdata.get("GOOGLE_API_KEY")
Local Appの実装
エージェントと会話するために簡易的なアプリのクラスLocalAppを用意します。
class LocalApp:
def __init__(self, agent, runner, session, user_id):
self._agent = agent
self._runner = runner
self._session = session
self._user_id = user_id
@classmethod
async def create(cls, agent):
user_id = 'local_app'
runner = Runner(
app_name=agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
session = await runner.session_service.create_session(
app_name=agent.name,
user_id=user_id,
state={},
session_id=uuid.uuid4().hex,
)
return cls(agent, runner, session, user_id)
async def _stream(self, query_text):
content = UserContent(parts=[Part.from_text(text=query_text)])
current_agent_being_processed = self._agent
if DEBUG:
print(f"--- Invoking agent: {current_agent_being_processed.name} with query: '{query_text[:100]}...' ---")
async_events = self._runner.run_async(
user_id=self._user_id,
session_id=self._session.id,
new_message=content,
)
text_parts_result = []
routing_target_name = None
async for event in async_events:
if DEBUG:
print(f"----\nEVENT (author: {event.author}, expected: {current_agent_being_processed.name})\n{event}\n----")
if (event.content and event.content.parts):
response_text_from_event = ''
for p in event.content.parts:
if p.text:
response_text_from_event += f'[{event.author}]\n\n{p.text}\n'
if response_text_from_event:
pattern = 'transfer_to_agent\(agent_name=["\']([^"]+)["\']\)'
matched = re.search(pattern, response_text_from_event)
if not routing_target_name and matched:
routing_target_name = matched.group(1)
if DEBUG:
print(f"--- Agent {event.author} responded with a routing command to: {routing_target_name} ---")
else:
if matched and routing_target_name:
if DEBUG: print(f"--- Agent {event.author} also tried to route (already have main route '{routing_target_name}'). Printing as text. ---")
elif DEBUG:
print(f"--- Agent {event.author} responded with direct answer ---")
print(response_text_from_event)
text_parts_result.append(response_text_from_event)
return text_parts_result, routing_target_name
async def stream(self, query):
coordinator_text_parts, routing_target_name_from_coordinator = await self._stream(query)
if routing_target_name_from_coordinator:
if DEBUG:
print(f'----\nCoordinator decided to route to: {routing_target_name_from_coordinator}\n----')
target_agent_instance = None
if hasattr(self._agent, 'sub_agents'):
for sa in self._agent.sub_agents:
if sa.name == routing_target_name_from_coordinator:
target_agent_instance = sa
break
if target_agent_instance:
if DEBUG:
print(f'----\nCreating new LocalApp context for {target_agent_instance.name} and running with ORIGINAL query.\n----')
target_app = await LocalApp.create(target_agent_instance)
final_result_parts, sub_agent_routing_decision = await target_app._stream(query)
if sub_agent_routing_decision:
if DEBUG:
print(f"--- WARNING: Sub-agent {target_agent_instance.name} also tried to route to {sub_agent_routing_decision}. This secondary routing is not handled further. ---")
return final_result_parts
else:
if DEBUG:
print(f"--- ERROR: Target agent '{routing_target_name_from_coordinator}' not found in coordinator's sub_agents. ---")
return []
else:
if DEBUG:
print("--- Coordinator did not provide a routing command. Returning its direct response (if any). ---")
return coordinator_text_parts
create(cls, agent)
今回は利用していませんがセッション情報を非同期で準備しています。
_stream(self, query_text)
ユーザーからの質問を処理してその応答を受け取っています。
DEBUGでどのサブエージェントが呼び出されたかを表示します。
async for event in async_events:ではエージェントから送られてくる個々の「イベント」を処理し、それがユーザーへの直接の回答なのか、他のエージェントへの処理の引き継ぎなのかを判断しています。
stream(self, query)
複数のエージェントが連携するシステムの動作を実現します。Coordinatorに問い合わせ、その指示に従って適切なサブエージェントを呼び出すという一連の流れを実現します、
AIエージェントの実装
共通部分
全てのAIエージェントで共通な設定を先に記述します。
LLMのモデルについてはAIエージェントを実装する際に個別に設定もできます。
global_instruction="Answer questions in Japanese"
llm_model = "gemini-2.5-flash-preview-05-20"
AIエージェントの指示や記述は英語のため、日本語で答えるよう指示とLLMのモデルをgemini 2.5 flashとしました。
営業エージェント
営業の進め方や提案内容を助言します。
sales_agent
sales_agent = LlmAgent(name="Sales_agent",
model=llm_model,
description="Consultations and questions about how to proceed with sales activities and proposals to customers.",
global_instruction=global_instruction,
instruction="""You are an excellent sales representative.
Please provide specific advice on how to proceed with sales and make proposals in response to the following customer requests and situations.
""")
エンジニアエージェント
Google Cloudの技術的な質問に答えます。
engineer_agent
engineer_agent = LlmAgent(name="Engineer_agent",
model=llm_model,
description="Answer technical questions about Google Cloud in an easy-to-understand manner based on your expertise.",
global_instruction=global_instruction,
instruction="""You are an experienced Google Cloud engineer.
Please answer the following technical IT questions in an easy-to-understand manner based on your expertise.
""")
一般エージェント
他のサブエージェントが答えられない質問に答えます。
general_agent
general_agent = LlmAgent(
name="General_agent",
model=llm_model,
description="Answers general questions and provides information when other specialized agents cannot handle the request.",
global_instruction=global_instruction,
instruction="""You are a helpful AI assistant designed to answer general questions.
When a user asks a question that is not about specific topics, provide a clear, comprehensive, and friendly answer.
If you are unsure or the question is too complex, acknowledge that you may not have all the information but try to provide the best possible general guidance.
""")
プロパティを説明します。
name：エージェントの名前
model：LLMのモデル。個別に設定可能
description：他のエージェントが対象のエージェントをどう使うかを決定する
instruction：LLMエージェントの行動を決定するためのもの
またglobal_instructionとして全てのAIエージェントが共通の指示も含めています。
descriptionとinstructionの違いは、前者は他のエージェントからどのように見られるか、後者はそのエージェント自体がどのように行動するかという点になります。
コーディネーター
最後にサブエージェントに質問を投げるコーディネーターを実装します。
coordinator = LlmAgent(
name="Coordinator",
model=llm_model,
description="Analyze the user's request and determine which agent is most appropriate to process it.",
global_instruction=global_instruction,
instruction="""You are an AI routing coordinator. Your SOLE and EXCLUSIVE responsibility is to analyze the user's query and delegate it to the MOST appropriate sub-agent from the list of available agents you have access to. You MUST NOT answer the query yourself under any circumstances.
You have access to a list of sub-agents. Each sub-agent has a 'description' that outlines its specific function and area of expertise. One of these sub-agents is named 'General_agent', which is designated to handle queries that do not clearly match the expertise of any other specialized sub-agent.
Follow this process meticulously:
1. Carefully and thoroughly analyze the user's query to understand its core intent.
2. Review the 'description' of EACH available sub-agent (excluding 'General_agent' in this initial review) to understand their specific capabilities and designated purpose.
3. Determine if the user's query directly and unequivocally aligns with the described expertise of ONE of these specialized sub-agents.
4. If you identify ONE specialized sub-agent whose 'description' clearly and uniquely matches the user's query:
Your entire response MUST be ONLY the following string, replacing '[Name_of_the_matched_specialized_agent]' with the actual name of that agent:
transfer_to_agent(agent_name="[Name_of_the_matched_specialized_agent]")
5. In ALL OTHER SITUATIONS (e.g., if no specialized sub-agent's description is a clear match, if the query is ambiguous, if it seems to be a very general question, or if multiple specialized agents seem partially relevant but none are a perfect fit):
You MUST delegate the query to the 'General_agent'. Your entire response MUST be ONLY the following string:
transfer_to_agent(agent_name="General_agent")
Your response MUST strictly be the 'transfer_to_agent' command and nothing else. Do not add any introductory phrases, explanations, or any text beyond the specified command format.
""",
sub_agents=[sales_agent, engineer_agent, general_agent]
)
sub_agents：質問を投げるサブエージェントをリストで追加
一般エージェント以外が答えられる質問は対象のサブエージェントに振り、それ以外を一般エージェントが答えるという指示にすることで今後特定のサブエージェントが増えたとしてもinstructionを変更する必要がないようにしました。
実行結果
それでは実施にいくつかクエリを試してみましょう。
出力がエラーとなる場合にはDEBUG=Trueで実行して原因を調べてみてください。
実行１：営業の進め方の助言
仮想の顧客と相談内容をqueryに入れて実行してみましょう
DEBUG = False
client = await LocalApp.create(coordinator)
query = f'''
ゲーム業界のお客様より弊社のホームページに問い合わせがありました。
現在データの分析をスプレッドシートを使って手動で行っているためデータ分析基盤とBIツールを検討しているとのことです。
お客様との初回の打ち合わせはどのように進めれば良いでしょうか。
'''
_ = await client.stream(query)
結果
Sales agentから打ち合わせの進め方の助言をもらうことができました。
[Sales_agent]
ゲーム業界のお客様との初回打ち合わせですね。データ分析基盤とBIツールをご検討とのこと、非常に良い機会です。
初回の打ち合わせで最も重要なのは、お客様の現状の課題と、将来的に実現したいことを深く理解することです。具体的な進め方として、以下のポイントを押さえて準備・実施してください。
### 1. 打ち合わせ前の準備
* **お客様の事前調査:**
* お客様の会社ウェブサイトを確認し、提供しているゲームの種類、ターゲット層、企業規模などを把握します。
* 直近のニュースリリースや業界トレンドも調べ、お客様が置かれている状況や課題を推測します。
* 御社の過去のゲーム業界での導入事例があれば、目を通しておきましょう。
* **ヒアリング項目の準備:**
* スプレッドシートでの手動分析の「具体的な」課題（例：時間がかかる、ミスが多い、データ量が増えて限界、リアルタイム性がない、属人化しているなど）。
* 現在分析しているデータの内容（例：ユーザー行動ログ、課金データ、マーケティングデータ、ゲーム内イベントデータなど）。
* 誰が、何のために分析しているのか（例：ゲームプランナー、マーケター、経営層など）。
* データ分析基盤とBIツール導入で「何を解決したいか」「何を達成したいか」という具体的な目標。
* 現状利用しているツールやシステム。
* 今回のプロジェクトにおける関係者、意思決定プロセス、予算感、導入時期の目安。
* **御社ソリューションの概要資料準備:**
* 詳細な説明は不要ですが、お客様の課題解決に繋がりそうなソリューションの全体像がわかる資料（会社紹介、製品概要、ゲーム業界向け活用イメージなど）を準備しておくと良いでしょう。
### 2. 打ち合わせの進め方
1. **丁寧な導入と自己紹介:**
* お客様へ感謝を述べ、本日の打ち合わせの目的（お客様の現状と課題を理解し、最適なご提案の方向性を探ること）を明確に伝えます。
* 自己紹介と、御社がお客様の課題解決に貢献できることの簡単な説明を行います。
2. **お客様の現状と課題のヒアリング（最も重要！）:**
* 「現在スプレッドシートで手動分析されているとのことですが、具体的にどのような点が課題となっていますか？」といった質問から掘り下げていきます。
* 準備したヒアリング項目をベースに、オープンな質問を多用し、お客様にたくさん話していただくことに注力してください。
* **「なぜその課題が発生しているのか？」「その課題を解決することで、お客様のビジネスにどのような良い影響があるのか？」**といった、課題の背景とインパクトを深掘りすることが重要です。
* 現在分析されているデータの種類や、分析の目的、それを誰がどのように活用しているのかを具体的に聞きます。
* 「理想的には、どのようなデータ分析環境が実現できれば、ビジネス目標達成に貢献できるとお考えですか？」と、目指す姿を聞き出します。
3. **目指す姿と期待値の確認:**
* 「データ分析基盤とBIツールを導入することで、具体的にどのような目標を達成したいとお考えですか？」
* 「例えば、ユーザー離反率の改善、LTV向上、新規施策の効果測定の迅速化など、具体的にどのような効果を期待されていますか？」
* このフェーズで、お客様が考えている具体的な効果や、導入後の成功イメージを引き出します。
4. **御社ソリューションの紹介（お客様の課題に合わせて）:**
* ヒアリングで把握したお客様の課題に対して、御社のデータ分析基盤やBIツールがどのように貢献できるかを、具体的なメリットに紐付けて説明します。
* **この段階では、詳細な機能説明よりも、お客様の課題解決にフォーカスした「価値」を伝えることを意識してください。**
* もしゲーム業界や類似業界での成功事例があれば、簡潔に紹介すると、お客様の共感を得やすくなります。
5. **次回のステップの提案と合意形成:**
* 本日のヒアリング内容を簡単にまとめ、お客様の課題と目指す姿への理解を共有します。
* 「本日お伺いした内容を基に、より具体的なご提案を作成させていただくために、次回は〇〇（例：技術的な要件を深掘り、デモンストレーション、より詳細な事例紹介など）のお打ち合わせをさせて頂くのはいかがでしょうか？」と、次回の打ち合わせ内容と目的を提案し、合意を得ます。
* 関係者の追加、予算・導入時期の再確認など、今後のプロジェクト推進に必要な情報も確認しておくと良いでしょう。
### 3. 打ち合わせ後のフォローアップ
* **お礼メールの送付:**
* 打ち合わせ後、迅速にお礼のメールを送りましょう。
* メールには、本日のヒアリング内容の要点と、次回の打ち合わせ内容、日時、役割などを簡潔に記載し、認識の相違がないことを確認します。
* 必要であれば、関連資料も添付します。
* **社内連携:**
* ヒアリングで得た情報を社内のエンジニアやコンサルタントと共有し、次回の打ち合わせに向けた準備を進めます。
この初回打ち合わせを通じて、お客様との信頼関係を築き、お客様の「真の課題」と「成功の定義」を深く理解することが、今後の提案活動の成功に繋がります。頑張ってください！
実行２：Google Cloudの技術的な質問
続いてGoogle Cloudの技術的な質問もしてみましょう。
DEBUG = False
client = await LocalApp.create(coordinator)
query = f'''
Google Cloudのプロジェクトを新規で作成しようと考えています。
現在Google Workspaceを利用しているのですがプロジェクトをGoogle Workspaceの組織に紐づけるメリットを教えてください。
'''
_ = await client.stream(query)
結果
Engineer agentが答えてくれました。
[Engineer_agent]
Google Cloud プロジェクトを Google Workspace の組織に紐づけることには、以下のような多くのメリットがあります。
1. **一元的なユーザー管理と認証:**
* Google Workspace で管理しているユーザーやグループを、そのまま Google Cloud の IAM (Identity and Access Management) で利用できます。これにより、新しいユーザーごとに認証情報を設定する手間が省け、既存の組織構造を活かして権限管理が行えます。
* シングルサインオン (SSO) の恩恵も受けられ、ユーザーは同じ認証情報で Google Workspace と Google Cloud の両方にアクセスできます。
2. **強固なセキュリティとコンプライアンス:**
* **組織ポリシーサービス:** 組織レベルでポリシーを設定し、配下のすべてのプロジェクトに強制的に適用できます。例えば、特定の API の無効化、リソースロケーションの制限、公開 IP アドレスの利用制限など、組織全体のセキュリティとコンプライアンスを強化できます。
* **監査ログ:** Google Workspace の監査ログと連携し、組織全体でのアクティビティをより詳細に追跡・監査することが容易になります。
3. **効率的なリソース管理と階層構造:**
* **リソース階層:** 組織は、Google Cloud のリソース階層における最上位のノードとなります。その下にフォルダを作成し、さらにその下にプロジェクトを配置することで、部門や用途に応じた論理的な階層構造を構築できます。これにより、リソースの整理、管理、課金を効率的に行えます。
* **ポリシーの継承:** 組織レベルで設定した IAM ポリシーや組織ポリシーは、その下のフォルダやプロジェクトに自動的に継承されます。これにより、個々のプロジェクトごとに同じ設定を繰り返す手間が省け、設定漏れや不整合を防ぐことができます。
4. **一元的な課金管理:**
* 組織レベルで請求アカウントを管理し、複数のプロジェクトの課金情報を一元的に把握・管理できます。これにより、コストの可視性が向上し、予算管理がしやすくなります。
5. **開発・運用チームの連携強化:**
* 既存の Google Workspace グループを活用することで、開発チームや運用チームに対する Google Cloud リソースへのアクセス権限付与が容易になります。これにより、チーム間の連携がスムーズになり、ガバナンスが向上します。
まとめると、Google Workspace の組織に Google Cloud プロジェクトを紐づけることで、**セキュリティ、管理の効率性、ガバナンス、そして運用コストの最適化**といった面で大きなメリットを享受できます。特に大規模な組織や複数のプロジェクトを運用する場合には、この連携が不可欠となります。
実行３：一般的な質問
最後に関係のない一般的な質問もしてみましょう
DEBUG = False
client = await LocalApp.create(coordinator)
query = f'''
2014年の日本の内閣総理大臣は誰ですか？
'''
_ = await client.stream(query)
実行結果
General agentが回答してくれました。
[General_agent]
2014年の日本の内閣総理大臣は安倍晋三氏です。
感想
Agent Development Kit（ADK）を利用してマルチエージェント構成を実装してみました。
今回はシンプルな実装になりましたが追加でセッション情報として会話の情報やユーザー情報も保持することができます。
またエージェントを順番に実行（SqquentialAgent）したり、複数のエージェントに同時に実行（ParallelAgent）することもできます。
これらを組み合わせることで複雑なワークフローにも対応することができるのではないでしょうか。
今後は他の機能を使ってみたいと思います。
またGoogle CloudのAgent Engineにもデプロイしたいと思います。
最後に
ADKはリリースされたばかりのサービスでAIに頼ったコーディングができませんでしたが、
ドキュメントもわかりやすく手軽に実装することができました。
最近はLLMに頼ったコーディングばかりしていたのでドキュメントを読んだ上で実装することはやはり知識が深まるなと感じました。
最後までお読みいただきありがとうございました。