🤖

LINE と Agent Development Kit (ADK) でボットを作ってみた vol.1

に公開
LINE
Cloud Run
Google Cloud
LINE Bot
Agent Development Kit
tech

はじめに

こんにちは、クラウドエースの岸本です。
今回は、2025年 4 月に Google Cloud Next で発表された Agent Development Kit（以下、ADK）と LINE の接続方法を解説しようと思います。
最後には、LINE から、Cloud Run 上で動作する ADK の Google Search エージェントを呼び出してみます。

使用する主要技術

本記事で活用する主要な技術スタックは以下の通りです。

  • LINE Developers:
    LINE Developers は、LINE Messaging API チャネルの設定と管理を行うためのポータルサイトです。

  • ADK:
    ADK は、エージェント開発のためのオープンソースのフレームワークです。
    詳細はこちらから確認できます。

  • Cloud Run:
    Cloud Run は、Google Cloud のサービスで、コンテナ化されたアプリケーションをフルマネージドで実行できるため、サーバー管理の手間を大幅に削減し、開発に集中できる環境を提供します。

  • Python:
    ADK は、Python または Java で開発することを前提としています。
    今回は、Python を使用します。

なぜ LINE と ADK と Cloud Run なのか？

LINE 公式アカウントを活用している企業にとって、顧客との接点をどう深めるかは重要なテーマです。
さらに、変化の早い市場環境では、開発スピードや将来的な拡張性も欠かせない要素となります。
こうした観点から考えたときに、LINE、ADK、Cloud Run の組み合わせが、ひとつの良い選択肢だと感じました。
その理由は、以下の通りです。

  • 理由 1
    LINE は日本をはじめとするアジア圏で絶大なユーザー数を誇るコミュニケーションプラットフォームです。
    顧客との接点を増やす強力なツールとなります。

  • 理由 2
    ADK は、エージェント開発を効率化するためのフレームワークです。
    市場のスピードに合わせて、エージェントの開発とサービス提供を迅速に行えます。

  • 理由 3
    Cloud Run は、エージェントをスケーラブルに実行できるサービスです。
    Google Cloud のサービスとの連携が容易で、拡張性が高いです。
    例えば、BigQuery との連携を行い、エージェントの会話履歴を保存・分析することで、パーソナライズされたエージェントを開発できます。

上記の理由から、LINE と ADK と Cloud Run を組み合わせることで、エージェントを迅速に開発でき、市場のスピードに合わせてサービスを提供できます。

接続の全体像

LINE と Cloud Run 上の ADK を接続する流れは以下の通りです。

  1. ユーザーが LINE でメッセージを送信
  2. LINE プラットフォームがメッセージを Webhook として Cloud Run に送信
  3. Cloud Run が Webhook を受信し、ADK がメッセージを受信
  4. ADK で作成したエージェントがメッセージを確認
  5. 確認後、Google Search を使用して返信するか、簡単に返信するかを判断
  6. ADK が LINE Messaging API を通じてユーザーに応答を送信

上記を実現するために、以下の手順が必要です。

  1. LINE Developers でチャネルを作成
  2. ソースコードを作成
  3. Cloud Run にデプロイ
  4. 接続の確認

1. ソースコードの作成

以下のコードでは、LINE からのメッセージを受信し、ADK を使ってメッセージを処理するコードを例に挙げます。
Cloud Shell または、ローカル環境でコードを作成してください。
コードは以下の通りです。

main.py
import logging
import os
import uuid
from fastapi import FastAPI, Request
import uvicorn
from linebot.v3 import WebhookParser
from linebot.v3.messaging import (
    ApiClient,
    Configuration,
    MessagingApi,
    ReplyMessageRequest,
    TextMessage,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.memory import InMemoryMemoryService
from google.adk.artifacts import InMemoryArtifactService
from google.adk.runners import Runner
from google.cloud import logging as cloud_logging
from google.genai.types import Content, Part
from google.adk.tools import google_search

# Cloud Logging の設定
def setup_cloud_logging():
    try:
        # Cloud Logging クライアントの初期化
        client = cloud_logging.Client() 
        client.setup_logging()
        logging.info("Cloud Logging の設定が完了しました")
    except Exception as e:
        logging.error(f"Cloud Logging の設定に失敗しました: {e}")

setup_cloud_logging()

# FastAPI の設定
app = FastAPI()

# LINE API 設定（LINE Developers で取得した値を設定してください。）
CHANNEL_ACCESS_TOKEN = "LINE_CHANNEL_ACCESS_TOKEN"
CHANNEL_SECRET = "LINE_CHANNEL_SECRET"

# アプリケーション設定
APP_NAME = "line_adk_app"

# LINE API クライアントの初期化
configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(CHANNEL_SECRET)

# ADK エージェントの初期化
agent = LlmAgent(
    name="line_adk_app",
    model="gemini-2.0-flash-lite-001",
    description="Google 検索を活用して、ユーザーの質問に回答する LINE ボット",
    instruction="あなたは日本語で応答する LINE ボットです。LINE からのメッセージを受信し、必ず日本語で自然で親しみやすい会話スタイルで応答してください。簡潔で分かりやすい日本語の回答を心がけ、必要に応じて Google 検索を活用して情報を提供してください。",
    tools=[google_search]
)

class LineApp:
    def __init__(self, agent):
        self._agent = agent
        self._runner = Runner(
            app_name=self._agent.name,
            agent=self._agent,
            artifact_service=InMemoryArtifactService(),
            session_service=InMemorySessionService(),
            memory_service=InMemoryMemoryService(),
        )
        self._sessions = {}  # ユーザー ID ごとのセッションを管理
        
    async def get_or_create_session(self, user_id):
        """ユーザー ID に対応するセッションを取得または作成"""
        if user_id not in self._sessions:
            session = await self._runner.session_service.create_session(
                app_name=self._agent.name,
                user_id=user_id,
                session_id=uuid.uuid4().hex,
            )
            self._sessions[user_id] = session
        return self._sessions[user_id]
        
    async def process_message(self, user_id, message_text):
        """メッセージを処理して応答を取得"""
        try:
            # セッションの取得または作成
            session = await self.get_or_create_session(user_id)
            
            # メッセージの作成
            content = Content(
                role='user',
                parts=[Part.from_text(text=message_text)]
            )

            # エージェントの実行と応答の取得
            final_response_text = "(No response)"
            async for event in self._runner.run_async(
                user_id=user_id,
                session_id=session.id,
                new_message=content,
            ):
                if event.is_final_response() and event.content and event.content.parts:
                    final_response_text = event.content.parts[0].text
                    break

            return final_response_text
            
        except Exception as e:
            logging.error(f"Error in process_message: {e}")
            return f"エラーが発生しました: {str(e)}"

# LineApp インスタンスの作成
line_app = LineApp(agent)

async def process_message_and_reply(body: str, signature: str):
    try:
        # イベントのパース
        events = parser.parse(body, signature)
        
        # 各イベントを処理
        for line_event in events:
            if isinstance(line_event, MessageEvent) and isinstance(line_event.message, TextMessageContent):
                user_id = line_event.source.user_id
                message_text = line_event.message.text

                logging.info(f"user_id: {user_id}")
                logging.info(f"message: {message_text}")

                # LineApp でメッセージを処理
                response_text = await line_app.process_message(user_id, message_text)

                # LINE への返信
                with ApiClient(configuration) as api_client:
                    line_api = MessagingApi(api_client)
                    line_api.reply_message(
                        ReplyMessageRequest(
                            reply_token=line_event.reply_token,
                            messages=[TextMessage(text=response_text)],
                        )
                    )
    except Exception as e:
        logging.error(f"Error in process_events: {e}")

@app.post("/callback")
async def callback(request: Request):
    # リクエストボディとシグネチャの取得
    body = await request.body()
    body_text = body.decode("utf-8")
    signature = request.headers.get("X-Line-Signature", "")
    
    # 非同期でメッセージ処理
    await process_message_and_reply(body_text, signature)
    return "OK"

if __name__ == "__main__":
    port = int(os.getenv("PORT", 8080))
    uvicorn.run(app, host="0.0.0.0", port=port)
Dockerfile
FROM python:3.13-slim
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port $PORT"]
requirements.txt
google_adk
line-bot-sdk
fastapi
uvicorn

2. Cloud Run へのデプロイ

エージェントコードを Cloud Run にデプロイします。
はじめに、環境変数とプロジェクトの設定をします。

# 環境変数の設定
export GOOGLE_CLOUD_PROJECT=YOUR_PROJECT_ID
export GOOGLE_CLOUD_LOCATION=us-central1
export GOOGLE_GENAI_USE_VERTEXAI=true

# gcloud コマンドにプロジェクト ID を設定
gcloud config set project $GOOGLE_CLOUD_PROJECT

  • GOOGLE_CLOUD_PROJECT
    ご自身の Google Cloud プロジェクト ID に置き換えてください。

  • GOOGLE_CLOUD_LOCATION
    us-central1 を使用します。

  • GOOGLE_GENAI_USE_VERTEXAI
    Gemini API を Vertex AI で使用するので、true にします。
    詳細はこちらをご覧ください。

次に、ソースコードがあるディレクトリで以下のコマンドを実行して、Cloud Run にデプロイします。

# Cloud Run にデプロイ
gcloud run deploy line-adk-app \
  --source . \
  --region $GOOGLE_CLOUD_LOCATION \
  --project $GOOGLE_CLOUD_PROJECT \
  --allow-unauthenticated \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT,GOOGLE_CLOUD_LOCATION=$GOOGLE_CLOUD_LOCATION,GOOGLE_GENAI_USE_VERTEXAI=$GOOGLE_GENAI_USE_VERTEXAI"

デプロイが完了すると、以下のように Cloud Run のサービス URL が発行されます。

Building using Dockerfile and deploying container to Cloud Run service [line-adk-app] in project [YOUR_PROJECT_ID] region [us-central1]
✓ Building and deploying new service... Done.                                                                    
  ✓ Uploading sources...          
  ✓ Building Container... Logs are available at [https://console.cloud.google.com/cloud-build/builds/BUILD_ID?project=PROJECT_NUMBER].                                                      
  ✓ Creating Revision...          
  ✓ Routing traffic...          
  ✓ Setting IAM Policy...              
Done.                           
Service [line-adk-app] revision [line-adk-app-00001-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://line-adk-app-xxxxxxxxx-uc.a.run.app

デプロイされた Service URL を LINE Developers の Webhook URL に設定します。

3. 接続の確認

最後に、ユーザーが LINE でメッセージを送信し、ADK がメッセージを受信して処理する様子を確認します。

1

上記の例では、健康診断に関する情報を返すための個別の実装は行っていません。
それでも、エージェントが自律的に判断し、Google Search を活用して、一般的な健康診断に関する情報を返すことができます。

まとめ

今回は、Google Search から情報を取得し、回答するエージェントを作成しました。
このシンプルな構成から、さらに高度なエージェントを開発していくことが可能です。
例えば、以下のような機能を追加していくことで、より魅力的なエージェントを作成できます。

  • 外部 API 連携: 天気予報やニュース、E コマース情報など、様々な外部サービスと連携させる。（例: 天気予報 API）
  • データベース連携: ユーザー情報や会話履歴を保存し、パーソナライズされた体験を提供する。

LINE と ADK、Cloud Run を活用して、皆さんのアイデアを形にしてみてください！

参考文献

Discussion