LINE botにマルチエージェントシステムを組み込もう【multi-Agent LINE bot ②】

## 概要
LINEからマルチエージェントシステムと対話できるbotを作成しました。
LINE Messaging APIにどうやってマルチエージェントシステムを組み込んだかを解説します。

親記事：[マルチエージェントと対話できるLINE botを開発しよう](https://qiita.com/MokonaSato/items/62d4dc2d28bafc48eabd)
サンプルコード：https://github.com/MokonaSato/line-multi-agent-sample

## 環境・ライブラリなど
- Python: 3.12
- google-adk: 0.5.0
- line-bot-sdk: 3.17.1
- fastapi: 0.115.12
- devcontainer上で開発（image: python:3.12-slim）

## 参考文献
[Google Agent Development Kit](https://google.github.io/adk-docs/)

## 実装ポイント
### シングルトンパターンでの実装
シングルトンパターンにより、
- アプリケーション全体でセッションサービスを共有します（ただし、ユーザーIDごとにセッションは分けています。`セッション管理`の項を参照。）
- エージェントの初期化はアプリケーション起動時1回のみにできます
```python:src/services/agent_service_impl.py
from google.adk.artifacts.in_memory_artifact_service import (
    InMemoryArtifactService,
)
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from src.services.agent_service.message_handler import MessageHandler
from src.services.agent_service.session_manager import SessionManager


class AgentService:
    _instance = None

    def __new__(cls):
        """シングルトンインスタンスを作成または返却"""
        if cls._instance is None:
            cls._instance = super(AgentService, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        """初期化（一度だけ実行）"""
        if not self._initialized:
            # サービス
            self.session_service = InMemorySessionService()
            self.artifacts_service = InMemoryArtifactService()

            # コンポーネントの初期化
            self.session_manager = SessionManager(self.session_service)
            self.message_handler = MessageHandler()

            # エージェント関連
            self.root_agent = None
            self.exit_stack = None
            self.runner = None
            self.executor = None

            # 初期化完了
            self._initialized = True
```

### EventHandlerからのAgent呼び出し
`LineEventHandler`クラスの
- `handle_text_message()`は上記`AgentService`クラスの`call_agent_text()`を
- `handle_image_message()`は上記`AgentService`クラスの`call_agent_with_image()`を
呼び出しています。

そして、両関数ともに`_call_agent_internal`を呼び出してエージェントを実行して応答を取得しています。
```python:src/services/agent_service_impl.py
class AgentService:
    async def call_agent_text(
        self, message: str, user_id: str, session_id: Optional[str] = None
    ) -> str:
        """テキストメッセージを送信して応答を取得

        Args:
            message: ユーザーからのメッセージ
            user_id: ユーザーID
            session_id: セッションID（未指定時は新規作成）

        Returns:
            エージェントからの応答文字列
        """
        return await self._call_agent_internal(
            message=message,
            user_id=user_id,
            session_id=session_id,
            image_data=None,
            image_mime_type=None,
        )

    async def call_agent_with_image(
        self,
        message: str,
        image_data: bytes,
        image_mime_type: str,
        user_id: str,
        session_id: Optional[str] = None,
    ) -> str:
        """画像付きメッセージを送信して応答を取得

        Args:
            message: ユーザーからのメッセージ
            image_data: 画像のバイナリデータ
            image_mime_type: 画像のMIMEタイプ（例: "image/jpeg"）
            user_id: ユーザーID
            session_id: セッションID（未指定時は新規作成）

        Returns:
            エージェントからの応答文字列
        """
        return await self._call_agent_internal(
            message=message,
            user_id=user_id,
            session_id=session_id,
            image_data=image_data,
            image_mime_type=image_mime_type,
        )

    async def _call_agent_internal(
        self,
        message: str,
        user_id: str,
        session_id: Optional[str] = None,
        image_data: Optional[bytes] = None,
        image_mime_type: Optional[str] = None,
    ) -> str:
        """エージェントへの内部呼び出し（テキスト・画像両対応）

        Args:
            message: ユーザーからのメッセージ
            user_id: ユーザーID
            session_id: セッションID（未指定時は新規作成）
            image_data: 画像のバイナリデータ（オプション）
            image_mime_type: 画像のMIMEタイプ（オプション）

        Returns:
            エージェントからの応答文字列
        """
        # エージェントを初期化
        await self.init_agent()

        # セッションを管理
        session_id = await self.session_manager.get_or_create_session(
            user_id, session_id
        )

        # メッセージをContent型に変換
        content = self.message_handler.create_message_content(
            message, image_data, image_mime_type
        )

        # エージェントを実行して応答を取得
        return await self.executor.execute_and_get_response(
            message, user_id, session_id, content, image_data
        )
```

### セッション管理
セッション管理はSessionManagerクラスで、ユーザーIDに基づいて実施されています。
```python:src/services/agent_service/session_manager.py
from typing import Optional

from google.adk.sessions import InMemorySessionService, Session

from src.services.agent_service.constants import APP_NAME
from src.utils.logger import setup_logger

logger = setup_logger("session_manager")


class SessionManager:
    """セッション管理クラス

    ユーザーセッションの作成と管理を担当します。
    """

    def __init__(self, session_service: InMemorySessionService):
        """初期化

        Args:
            session_service: セッションサービスのインスタンス
        """
        self.session_service = session_service

    async def get_or_create_session(
        self, user_id: str, session_id: Optional[str] = None
    ) -> str:
        """セッションを取得または作成

        既存のセッションがある場合はそれを返し、なければ新規作成します。

        Args:
            user_id: ユーザーID
            session_id: セッションID（未指定時は生成）

        Returns:
            有効なセッションID
        """
        # セッションIDが指定されていない場合はユーザーIDから生成
        if session_id is None:
            session_id = f"session_{user_id}"

        # 既存セッションを取得
        session = self._get_session(user_id, session_id)

        # セッションがない場合は新規作成
        if not session:
            session = self._create_session(user_id, session_id)
            logger.info(f"Created new session: {session.id}")
        else:
            logger.info(f"Using existing session: {session.id}")

        return session.id
```

### Root Agentの実装
本アプリケーションでは単一のRoot Agentの配下にSubAgentやToolを集約し、ユーザーからの入力を受け付けるようにしています。
```python:src/agents/root_agent.py
from contextlib import AsyncExitStack
from typing import Tuple

from google.adk.agents.llm_agent import LlmAgent

from src.agents.agent_factory import AgentFactory
from src.agents.config import AGENT_CONFIG
from src.agents.prompt_manager import PromptManager
from src.utils.logger import setup_logger

# ロガーを設定
logger = setup_logger("root_agent")

# グローバル変数
_root_agent = None
_exit_stack = AsyncExitStack()


async def create_agent() -> Tuple[LlmAgent, AsyncExitStack]:
    """ルートエージェントとすべてのサブエージェントを作成する

    Returns:
        Tuple[LlmAgent, AsyncExitStack]: ルートエージェントとリソース管理用のexitスタック
    """
    global _root_agent, _exit_stack

    # すでに作成済みの場合はそれを返す
    if _root_agent is not None:
        logger.info("既存のルートエージェントを返します")
        return _root_agent, _exit_stack

    try:
        logger.info("新しいルートエージェントを作成します")

        # プロンプトマネージャーからすべてのプロンプトを読み込む
        prompt_manager = PromptManager()
        prompts = prompt_manager.get_all_prompts()

        # ファクトリークラスを初期化
        factory = AgentFactory(prompts, AGENT_CONFIG)

        # すべての標準エージェントを作成
        agents = factory.create_all_standard_agents()

        # ルートエージェントを作成
        _root_agent = factory.create_root_agent(agents)

        logger.info("ルートエージェントの作成に成功しました")

    except Exception as e:
        logger.error(f"ルートエージェント作成中にエラーが発生: {e}")
        raise

    return _root_agent, _exit_stack
```
Root Agentの作成は`AgentFactory`クラスで実施します。
プロンプト管理については、下記記事をご覧ください。
[]()
```python:src/agents/agent_factory.py
from typing import Dict

from google.adk.agents import Agent, SequentialAgent
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools import agent_tool, google_search

from src.tools.calculator_tools import calculator_tools_list
from src.tools.notion import notion_tools_list
from src.tools.web_tools import fetch_web_content
from src.utils.logger import setup_logger

# ロガー
logger = setup_logger("agent_factory")


class AgentFactory:
    """エージェント生成のファクトリークラス

    各種エージェントの生成ロジックをカプセル化して提供します。
    """

    def __init__(self, prompts: Dict[str, str], config: Dict):
        self.prompts = prompts
        self.config = config

    def create_calculator_agent(self) -> Agent:
        """計算エージェントを作成"""
        cfg = self.config["calculator"]
        return Agent(
            name=cfg["name"],
            model=cfg["model"],
            description=cfg["description"],
            instruction=self.prompts[cfg["prompt_key"]],
            tools=calculator_tools_list,
        )

    def create_google_search_agent(self) -> Agent:
        """Google検索エージェントを作成"""
        cfg = self.config["google_search"]
        return Agent(
            name=cfg["name"],
            model=cfg["model"],
            description=cfg["description"],
            instruction=cfg.get(
                "instruction", "インターネット検索であなたの質問に答えます。"
            ),
            tools=[google_search],
        )
        
    ## 中略 ##

    def create_root_agent(self, sub_agents: Dict[str, Agent]) -> LlmAgent:
        """ルートエージェントを作成"""
        cfg = self.config["root"]

        return LlmAgent(
            model=cfg["model"],
            name=cfg["name"],
            instruction=self.prompts[cfg["prompt_key"]],
            description=cfg["description"],
            tools=[
                agent_tool.AgentTool(agent=sub_agents["google_search_agent"]),
            ],
            sub_agents=[
                sub_agents["calc_agent"],
                sub_agents["url_recipe_workflow_agent"],
                sub_agents["image_recipe_workflow_agent"],
                sub_agents["notion_agent"],
                sub_agents["vision_agent"],
            ],
        )
```