Google ADK でマルチエージェント・ワークフローを設計する実践ガイド

pip install google-adk

from google.adk.agents import LlmAgent, SequentialAgent, ParallelAgent
from google.adk.tools import google_search
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
 
# ① 各競合社を検索するエージェント（並列実行用）
def create_search_agent(company_name: str) -> LlmAgent:
    return LlmAgent(
        name=f"search_{company_name}",
        model="gemini-2.0-flash",
        instruction=f"""
        {company_name} の最新機能リリースや製品アップデートを検索してください。
        過去30日以内の情報に絞り、URLとともに箇条書きで返してください。
        """,
        tools=[google_search],
        output_key=f"search_result_{company_name}",
    )
 
# ② 検索結果を分析するエージェント
analysis_agent = LlmAgent(
    name="analysis_agent",
    model="gemini-2.0-flash",
    instruction="""
    前のステップで収集された競合他社の情報を分析してください。
    以下の観点で整理してください。
    - 最も注目すべき機能変更は何か
    - 自社に影響する可能性がある変化は何か
    - 競合間での差別化ポイントはどこか
    """,
    output_key="analysis_result",
)
 
# ③ レポートを生成するエージェント
report_agent = LlmAgent(
    name="report_agent",
    model="gemini-2.0-flash-thinking-exp",  # 推論モデルで品質向上
    instruction="""
    分析結果をもとに、経営層向けの競合動向サマリーレポートを作成してください。
    - 全体概要（3文以内）
    - 要注意の動向（優先度順に3点）
    - 推奨アクション（具体的に2点）
    という構成でまとめてください。
    """,
    output_key="final_report",
)

COMPETITORS = ["OpenAI", "Anthropic", "Mistral"]
 
# 競合社ごとの検索エージェントを並列実行
parallel_search = ParallelAgent(
    name="parallel_search",
    sub_agents=[create_search_agent(c) for c in COMPETITORS],
)
 
# 全体のワークフロー：検索（並列）→ 分析 → レポート
competitive_analysis_workflow = SequentialAgent(
    name="competitive_analysis",
    sub_agents=[parallel_search, analysis_agent, report_agent],
)

session_service = InMemorySessionService()
runner = Runner(
    agent=competitive_analysis_workflow,
    app_name="competitive_analysis_app",
    session_service=session_service,
)
 
async def run_analysis():
    session = await session_service.create_session(
        app_name="competitive_analysis_app",
        user_id="user_001",
    )
 
    message = types.Content(
        role="user",
        parts=[types.Part(text="競合調査を実行してください")],
    )
 
    async for event in runner.run_async(
        user_id="user_001",
        session_id=session.id,
        new_message=message,
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)
 
    # セッション状態から中間結果も取得できる
    final_session = await session_service.get_session(
        app_name="competitive_analysis_app",
        user_id="user_001",
        session_id=session.id,
    )
    print("分析結果:", final_session.state.get("analysis_result"))
    print("最終レポート:", final_session.state.get("final_report"))

def create_resilient_search_agent(company_name: str) -> LlmAgent:
    return LlmAgent(
        name=f"search_{company_name}",
        model="gemini-2.0-flash",
        instruction=f"""
        {company_name} の最新情報を検索してください。
        検索が失敗した場合は「{company_name}: 情報取得失敗」とだけ返してください。
        絶対にエラーをそのまま返さないでください。
        """,
        tools=[google_search],
        output_key=f"search_result_{company_name}",
    )

adk web