Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

สร้าง Agentic AI App ที่ลูกค้าโต้ตอบด้วยผ่าน Telegram ได้

1. บทนำ

สร้างประสบการณ์การใช้งาน Agentic AI ที่ราบรื่นและโต้ตอบได้ ซึ่งลูกค้าสามารถมีส่วนร่วมได้โดยตรงจากแอปพลิเคชันการรับส่งข้อความที่เคยใช้ ค้นพบวิธีพัฒนาและติดตั้งใช้งานแอปพลิเคชันอัจฉริยะที่ทำงานได้อย่างราบรื่นในอินเทอร์เฟซเว็บและช่องทางการรับส่งข้อความที่ทันสมัย

สิ่งที่คุณจะได้สร้าง

การผสานรวมระหว่าง "เจ้าหน้าที่บริการด้านร้านอาหาร" แบบครบวงจร ซึ่งเป็นแอปพลิเคชันที่ใช้ ADK และขับเคลื่อนโดย Gemini ที่ช่วยให้ผู้รับประทานอาหารเรียกดูเมนูของร้านอาหารและจองโต๊ะได้ กับแอปแชท Telegram คุณสามารถโต้ตอบกับบ็อต Telegram และขอคำอธิบายด้วยภาษาที่เป็นธรรมชาติ เช่น "ฉันอยากได้อาหารรสเผ็ดและเป็นมังสวิรัติ" จากนั้นบ็อตจะเชื่อมต่อกับ Agent ADK ซึ่งอ่านและเขียนไปยังฐานข้อมูล Cloud SQL PostgreSQL ผ่าน MCP Toolbox สำหรับฐานข้อมูลทั้งหมด ซึ่งจัดการการเข้าถึงฐานข้อมูลทั้งหมด รวมถึงการสร้างการฝังอัตโนมัติสำหรับการค้นหาเวกเตอร์ ในขณะเดียวกันผู้ใช้จะเห็นว่าบ็อตรับทราบข้อความและพิมพ์ ... typing เพื่อตอบกลับขณะรอการตอบกลับจาก Agent ADK

สิ่งที่คุณจะได้เรียนรู้

ติดตั้งใช้งาน "เจ้าหน้าที่บริการด้านร้านอาหาร" ที่ใช้งานได้ ซึ่งเป็นแอปพลิเคชันที่ใช้ ADK และขับเคลื่อนโดย Gemini
ตั้งค่าแชทบ็อต Telegram โดยใช้ BotFather
เขียนแอปพลิเคชัน Python เพื่อรับฟังเว็บฮุคของบ็อต
ส่งการดำเนินการแชทเพื่อแสดงการแจ้งเตือน ... typing ใน Telegram เมื่อผู้ใช้ส่งข้อความ และทำการโพลเพื่อส่ง ... typing เป็นระยะๆ ขณะรอการตอบกลับจริง
เรียกปลายทาง Cloud Run ของ Restaurant Concierge เพื่อประมวลผลคำถามของผู้ใช้
จัดการการตอบกลับจาก Agent ADK และส่งข้อความไปยัง Telegram แล้วปิดบัฟเฟอร์
ติดตั้งใช้งานแอปพลิเคชัน Python ใน Cloud Run
โต้ตอบกับบ็อต Telegram

ข้อกำหนดเบื้องต้น

บัญชี Google Cloud ที่มีบัญชีสำหรับการเรียกเก็บเงินช่วงทดลองใช้
มีความคุ้นเคยกับ Python ในระดับพื้นฐาน
ประสบการณ์การใช้งาน ADK และการติดตั้งใช้งาน Cloud Run มาก่อนจะเป็นประโยชน์
บัญชี Telegram
(แนะนำ) ทำ Codelab ต่อไปนี้ให้เสร็จสมบูรณ์
Agentic RAG พร้อม ADK, MCP Toolbox และ Cloud SQL -> คุณสามารถสร้าง Agent ต่อจาก Codelab นี้ได้ เนื่องจากโค้ดเริ่มต้นที่ให้มาเหมือนกัน
(หรือ) Agents at Scale: สถาปัตยกรรมแบบหลาย Agent พร้อมโปรโตคอล A2A ใน Agent Runtime และการผสานรวม ADK -> หากคุณต้องการเพิ่มประสิทธิภาพให้ Agent มากขึ้นโดยใช้สถาปัตยกรรมแบบหลาย Agent

2. การตั้งค่าสภาพแวดล้อม - ดำเนินการต่อจาก Codelab ก่อนหน้า

เรื่องราวที่เรานำเสนอใน Codelab นี้เป็นการดำเนินการต่อจาก Codelab ที่เป็นข้อกำหนดเบื้องต้นนี้: Agentic RAG พร้อม ADK, MCP Toolbox และ Cloud SQL หรือ Agents at Scale: สถาปัตยกรรมแบบหลาย Agent พร้อมโปรโตคอล A2A ใน Agent Runtime และการผสานรวม ADK คุณสามารถทำงานต่อจาก Codelab ก่อนหน้าได้

เราสามารถเริ่มสร้างในไดเรกทอรีที่ใช้งานอยู่ของ Codelab ก่อนหน้าได้ ( ไดเรกทอรีที่ใช้งานอยู่ควรเป็น build-agent-adk-toolbox-cloudsql หรือ adk-a2a-agent-runtime-starter) เพื่อไม่ให้เกิดความสับสน ให้เปลี่ยนชื่อไดเรกทอรีเป็นชื่อไดเรกทอรีเดียวกันกับที่เราใช้เมื่อเริ่มต้นใหม่

หากคุณดำเนินการต่อจาก Lab Agentic RAG พร้อม ADK, MCP Toolbox และ Cloud SQL ให้ทำดังนี้

mv ~/build-agent-adk-toolbox-cloudsql ~/build-agent-adk-telegram

หรือหากคุณดำเนินการต่อจาก Lab Agents at Scale: สถาปัตยกรรมแบบหลาย Agent พร้อมโปรโตคอล A2A ใน Agent Runtime และการผสานรวม ADK ให้ทำดังนี้

mv ~/adk-a2a-agent-runtime-starter ~/build-agent-adk-telegram

จากนั้นเปลี่ยนไดเรกทอรีที่ใช้งานอยู่เป็นไดเรกทอรีนี้

cloudshell workspace ~/build-agent-adk-telegram && cd ~/build-agent-adk-telegram
source .env

หลังจากนั้น ให้ตรวจสอบว่าได้ติดตั้งใช้งาน restaurant-agent แล้วและมี URL สาธารณะที่เข้าถึงได้

AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

หากเข้าถึง URL ได้ แสดงว่าคุณพร้อมที่จะไปยังส่วนถัดไป: Create Telegram Bot

3. การตั้งค่าสภาพแวดล้อม - เริ่มต้นใหม่ด้วยที่เก็บเริ่มต้น

ขั้นตอนนี้จะเตรียมสภาพแวดล้อม Cloud Shell กำหนดค่าโปรเจ็กต์ Google Cloud และโคลนที่เก็บเริ่มต้น

เปิด Cloud Shell

เปิด Cloud Shell ในเบราว์เซอร์ Cloud Shell มีสภาพแวดล้อมที่กำหนดค่าไว้ล่วงหน้าพร้อมเครื่องมือทั้งหมดที่คุณต้องการสำหรับ Codelab นี้ คลิกให้สิทธิ์ เมื่อมีข้อความแจ้ง

จากนั้นคลิก "ดู" -> "เทอร์มินัล" เพื่อเปิดเทอร์มินัล อินเทอร์เฟซของคุณควรมีลักษณะคล้ายกับภาพต่อไปนี้

นี่จะเป็นอินเทอร์เฟซหลักของเรา โดยมี IDE อยู่ด้านบนและเทอร์มินัลอยู่ด้านล่าง

ตั้งค่าไดเรกทอรีที่ใช้งานอยู่

โคลนที่เก็บเริ่มต้น โค้ดทั้งหมดที่คุณเขียนใน Codelab นี้จะอยู่ในที่เก็บนี้

rm -rf ~/build-agent-adk-telegram
git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git build-agent-adk-telegram
cloudshell workspace ~/build-agent-adk-telegram && cd ~/build-agent-adk-telegram

สร้างไฟล์ .env จากเทมเพลตที่ให้มาโดยทำดังนี้

cp .env.example .env

ข้อมูลสำคัญ! เรามีขั้นตอนด้านล่างนี้ไว้เพื่อช่วยตั้งค่าโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google อย่างรวดเร็ว ซึ่งลิงก์กับบัญชีสำหรับการเรียกเก็บเงินแบบทดลองใช้(strict) หากต้องการใช้โปรเจ็กต์ก่อนหน้านี้ของคุณเอง ให้ข้ามขั้นตอนต่อไปนี้และทำขั้นตอนต่อไปนี้แทน

เพิ่มชื่อโปรเจ็กต์ของคุณเองเป็นตัวแปร GOOGLE_CLOUD_PROJECT ในไฟล์ .env
เปิดใช้งานโปรเจ็กต์ในเทอร์มินัลโดยใช้ gcloud config set project your-project-id

หลังจากนั้น คุณสามารถข้ามไปยังส่วนการตั้งค่าโครงสร้างพื้นฐานเริ่มต้นได้เลย

หากไม่แน่ใจ ให้ทำตามส่วนด้านล่าง

หากต้องการตั้งค่าโปรเจ็กต์ในเทอร์มินัลได้ง่ายขึ้น ให้ดาวน์โหลดสคริปต์การตั้งค่าโปรเจ็กต์นี้ลงในไดเรกทอรีที่ใช้งานอยู่

curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh

เรียกใช้สคริปต์ สคริปต์จะยืนยันบัญชีสำหรับการเรียกเก็บเงินแบบทดลองใช้ สร้างโปรเจ็กต์ใหม่ (หรือตรวจสอบโปรเจ็กต์ที่มีอยู่) บันทึกรหัสโปรเจ็กต์ลงในไฟล์ .env ในไดเรกทอรีปัจจุบัน และตั้งค่าโปรเจ็กต์ที่ใช้งานอยู่ใน gcloud

bash setup_verify_trial_project.sh && source .env

สคริปต์จะทำสิ่งต่อไปนี้

ยืนยันว่าคุณมีบัญชีสำหรับการเรียกเก็บเงินแบบทดลองใช้ที่ใช้งานอยู่
ตรวจสอบโปรเจ็กต์ที่มีอยู่ใน .env (หากมี)
สร้างโปรเจ็กต์ใหม่หรือใช้โปรเจ็กต์ที่มีอยู่
ลิงก์บัญชีสำหรับการเรียกเก็บเงินแบบทดลองใช้กับโปรเจ็กต์
บันทึกรหัสโปรเจ็กต์ลงใน .env
ตั้งค่าโปรเจ็กต์เป็นโปรเจ็กต์ gcloud ที่ใช้งานอยู่

ตรวจสอบว่าได้ตั้งค่าโปรเจ็กต์อย่างถูกต้องโดยดูข้อความสีเหลือง ข้างไดเรกทอรีที่ใช้งานอยู่ในการแจ้งของเทอร์มินัล Cloud Shell ข้อความควรแสดงรหัสโปรเจ็กต์

การตั้งค่าโครงสร้างพื้นฐานเริ่มต้น

ก่อนอื่น เราจะต้องติดตั้งการอ้างอิง Python โดยใช้ uv ซึ่งเป็นเครื่องมือจัดการแพ็กเกจและโปรเจ็กต์ Python ที่รวดเร็วซึ่งเขียนด้วย Rust ( เอกสารประกอบของ uv) Codelab นี้ใช้เครื่องมือนี้เพื่อความเร็วและความง่ายในการดูแลโปรเจ็กต์ Python

uv sync

จากนั้นเรียกใช้สคริปต์การตั้งค่าแบบเต็ม ซึ่งจะสร้างอินสแตนซ์ Cloud SQL ป้อนข้อมูลเริ่มต้น และติดตั้งใช้งานบริการ Toolbox ซึ่งจะทำหน้าที่เป็นสถานะเริ่มต้นของ Agent ร้านอาหาร

bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &

หมายเหตุ: สคริปต์นี้ใช้เวลาประมาณ 10-15 นาที โดยจะสร้างอินสแตนซ์ Cloud SQL สร้างฐานข้อมูล ป้อนข้อมูลเมนูเริ่มต้นพร้อมการฝังเวกเตอร์ ดาวน์โหลดและเรียกใช้ไบนารี Toolbox สำหรับการพัฒนาในเครื่อง และติดตั้งใช้งาน Agent ร้านอาหารเริ่มต้นและบริการ Toolbox ใน Cloud Run เพื่อการติดตั้งใช้งานแบบผสานรวมเต็มรูปแบบ สคริปต์จะทำงานอยู่เบื้องหลังและคุณสามารถตรวจสอบบันทึกในไฟล์ logs/full_setup.log

สคริปต์จะทำสิ่งต่อไปนี้

สร้างอินสแตนซ์ Cloud SQL และป้อนข้อมูลเริ่มต้นในฐานข้อมูล (ระยะที่ 1)
สร้างการกำหนดค่าสภาพแวดล้อมของ Agent และเริ่มบริการ Toolbox ในเครื่อง (ระยะที่ 2)
ติดตั้งใช้งานบริการ Toolbox และ Agent ใน Cloud Run (ระยะที่ 3)

หลังจากติดตั้งใช้งานเสร็จแล้ว คุณจะเข้าถึง UI สำหรับนักพัฒนาซอฟต์แวร์ของ ADK ได้ใน URL ของ Cloud Run

source .env
AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

เปิด UI สำหรับนักพัฒนาซอฟต์แวร์ของ ADK เลือก restaurant_agent และทดสอบด้วยข้อความค้นหา เช่น ตัวอย่างต่อไปนี้

What Italian dishes do you have?

หรือ

I want something spicy and creamy

ตอนนี้ การดำเนินการถัดไปคือเราจะย้ายจากอินเทอร์เฟซการพัฒนาเว็บเท่านั้นไปยังช่องทางการรับส่งข้อความ Telegram ได้อย่างไร

4. สร้างบ็อต Telegram

Telegram เป็นแพลตฟอร์มการรับส่งข้อความฟรีที่รู้จักกันดีและมีการใช้งานอย่างกว้างขวางเพื่อการมีส่วนร่วมของชุมชน เหตุผลหนึ่งคือแพลตฟอร์มนี้มีวิธีผสานรวมมากมายที่ทำได้ง่าย ผู้คนจึงสร้างบ็อตของตนเองได้อย่างง่ายดายด้วยฟังก์ชันที่หลากหลาย

ในกรณีนี้ เราจะใช้ BotFather เพื่อสร้างบ็อตของเราเองเป็นครั้งแรก โปรดทราบว่าแม้ว่าเราจะใช้ Telegram สำหรับเซสชันนี้ แต่คุณสามารถใช้วิธีเดียวกันกับ WhatsApp หรือแพลตฟอร์มการรับส่งข้อความอื่นๆ ที่ต้องการได้

ใช้ BotFather เพื่อสร้างบ็อตของคุณเอง

เปิดเว็บเบราว์เซอร์และไปที่ https://telegram.me/BotFather เพื่อเริ่มสร้างบ็อต Telegram ของคุณเอง

เริ่มโต้ตอบกับ BotFather

ส่งคำสั่ง /start

หากต้องการเริ่มต้นใช้งาน BotFather และเริ่มสร้างบ็อตตัวแรก คุณต้องเรียกใช้ข้อความ /start ไปยัง BotFather จากนั้น BotFather จะแชร์คำสั่งทั้งหมดที่คุณใช้โต้ตอบกับ BotFather ได้

/start

เริ่มสร้างบ็อตด้วยคำสั่ง /newbot

มาสร้างบ็อตใหม่โดยส่งคำสั่ง /newbot ไปยัง BotFather ระบบจะขอให้คุณตั้งชื่อบ็อต จากนั้นจะขอให้คุณตั้ง username ของบ็อต ซึ่งต้องลงท้ายด้วย bot เสมอ เช่น TetrisBot หรือ tetris_bot ชื่อผู้ใช้นี้ต้องไม่ซ้ำกัน

เมื่อสร้างบ็อตสำเร็จ คุณจะได้รับข้อความต่อไปนี้จาก BotFather

Done! Congratulations on your new bot. You will find it at t.me/AdkTelegramTest_bot. You can now add a description, about section and profile picture for your bot, see /help for a list of commands. By the way, when you've finished creating your cool bot, ping our Bot Support if you want a better username for it. Just make sure the bot is fully operational before you do this.

Use this token to access the HTTP API:

<YOUR_TELEGRAM_API_KEY>

Keep your token secure and store it safely, it can be used by anyone to control your bot.

For a description of the Bot API, see this page: https://core.telegram.org/bots/api

จด YOUR_TELEGRAM_API_KEY ไว้ เราจะใช้คีย์นี้ในส่วนถัดไป

5. พัฒนาแอปพลิเคชันเว็บฮุค Telegram

มาเตรียมไดเรกทอรีที่ใช้งานอยู่เพื่อเริ่มพัฒนาแอปพลิเคชันเว็บฮุค Telegram กัน

mkdir ~/build-agent-adk-telegram/telegram-integration
cd ~/build-agent-adk-telegram

เพิ่มการอ้างอิงที่จำเป็น

สร้างสคริปต์ requirements.txt ที่มีเนื้อหาต่อไปนี้เพื่อระบุการอ้างอิงที่เพียงพอสำหรับสคริปต์ Listener เว็บฮุค Telegram

cloudshell edit ./telegram-integration/requirements.txt

จากนั้นเพิ่มการอ้างอิงต่อไปนี้

python-telegram-bot[webhooks]
httpx

สร้างสคริปต์สำหรับ Listener เว็บฮุค Telegram

เมื่อติดตั้งทรัพยากร Dependency แล้ว ตอนนี้เราสามารถสร้างสคริปต์ Python main.py สำหรับแอปพลิเคชันการผสานรวมได้แล้ว

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/main.py

จากนั้นคัดลอกโค้ดต่อไปนี้ลงในสคริปต์

# ./telegram-integration/main.py

import asyncio
import os
import sys
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext
from telegram.constants import ChatAction
import httpx

# Read token from environment variable
TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
ADK_SERVER_URL = os.environ.get("ADK_SERVER_URL", "http://localhost:8000")
ADK_APP_NAME = os.environ.get("ADK_APP_NAME", "restaurant_agent")

# Parse base URL out of ADK_SERVER_URL
BASE_URL = ADK_SERVER_URL.rstrip('/')
if BASE_URL.endswith('/run'):
    BASE_URL = BASE_URL[:-4]
elif BASE_URL.endswith('/query'):
    BASE_URL = BASE_URL[:-6]

if not TOKEN:
    print("Error: TELEGRAM_BOT_TOKEN environment variable not set.")
    print("Please set it before running the application.")
    sys.exit(1)

async def start(update: Update, context: CallbackContext) -> None:
    """Send a message when the command /start is issued."""
    await update.message.reply_text('Hi! I am your ADK Integration Bot. Send me a message and I will forward it to the ADK server.')

async def send_typing_loop(chat_id: int, bot, stop_event: asyncio.Event):
    """Send typing action periodically until the stop event is set."""
    while not stop_event.is_set():
        try:
            await bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
            # The research suggested repeating every 4 seconds
            await asyncio.sleep(4)
        except Exception as e:
            print(f"Error sending chat action: {e}")
            await asyncio.sleep(1) # Wait a bit before retrying if error

async def handle_message(update: Update, context: CallbackContext) -> None:
    """Handle incoming user messages."""
    user_message = update.message.text
    chat_id = update.message.chat_id
    raw_user_id = str(update.message.from_user.id)
    
    # Derive unique user_id and session_id for this user
    user_id = f"tg_{raw_user_id}"
    session_id = f"tg_sess_{raw_user_id}"

    print(f"Received message from {user_id}: {user_message}")

    # Create a stop event for the typing loop
    stop_event = asyncio.Event()
    
    # Start the typing loop as a background task
    typing_task = asyncio.create_task(send_typing_loop(chat_id, context.bot, stop_event))

    try:
        async with httpx.AsyncClient() as client:
            # 1. Check if the session exists
            session_url = f"{BASE_URL}/apps/{ADK_APP_NAME}/users/{user_id}/sessions/{session_id}"
            session_check = await client.get(session_url, timeout=10.0)
            
            if session_check.status_code == 404:
                # 2. If session doesn't exist, create it
                print(f"Session {session_id} not found. Creating session...")
                session_create = await client.post(session_url, json={}, timeout=10.0)
                if session_create.status_code != 200:
                    raise Exception(f"Failed to create session: {session_create.status_code} {session_create.text}")
            elif session_check.status_code != 200:
                raise Exception(f"Error checking session: {session_check.status_code} {session_check.text}")
            
            # 3. Run the ADK agent
            run_url = f"{BASE_URL}/run"
            payload = {
                "appName": ADK_APP_NAME,
                "userId": user_id,
                "sessionId": session_id,
                "newMessage": {
                    "role": "user",
                    "parts": [{"text": user_message}]
                }
            }
            response = await client.post(run_url, json=payload, timeout=60.0)
            
        if response.status_code == 200:
            events = response.json()
            if isinstance(events, list) and len(events) > 0:
                # The last event contains the final text response
                last_event = events[-1]
                content = last_event.get("content", {})
                parts = content.get("parts", [])
                if parts and "text" in parts[0]:
                    reply_text = parts[0]["text"]
                else:
                    reply_text = "ADK agent returned an empty or non-text response."
            else:
                reply_text = "No events returned from ADK agent."
        else:
            reply_text = f"Error communicating with ADK server (Status: {response.status_code})."
            
    except Exception as e:
        reply_text = f"Failed to connect to ADK server: {e}"
    finally:
        # Stop the typing loop
        stop_event.set()
        await typing_task

    # Send the final response back to the user
    await update.message.reply_text(reply_text)

def main() -> None:
    """Start the bot."""
    # Create the Application and pass it your bot's token.
    application = Application.builder().token(TOKEN).build()

    # on different commands - answer in Telegram
    application.add_handler(CommandHandler("start", start))

    # on non command i.e message - echo the message on Telegram
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    # Check if running in webhook mode (e.g., on Cloud Run)
    port = os.environ.get("PORT")
    service_url = os.environ.get("SERVICE_URL")

    if port and service_url:
        if not service_url.startswith("http"):
            service_url = f"https://{service_url}"
        
        print(f"Starting bot in WEBHOOK mode on port {port} with url {service_url}")
        
        application.run_webhook(
            listen="0.0.0.0",
            port=int(port),
            url_path=TOKEN,
            webhook_url=f"{service_url}/{TOKEN}",
            allowed_updates=Update.ALL_TYPES
        )
    else:
        print("Starting bot in POLLING mode")
        # Run the bot until the user presses Ctrl-C
        application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

ทำความเข้าใจโค้ดการผสานรวมบ็อต Telegram

เมื่อผู้ใช้ส่งข้อความ ไปป์ไลน์ต่อไปนี้จะทำงานภายใต้ handle_message()

ขั้นตอนที่ 1: การระบุตัวตนและการอนุมานเซสชัน

บ็อตจะจับคู่รหัสผู้ใช้ Telegram กับตัวระบุ ADK ที่ไม่ซ้ำกันเพื่อให้เซสชันของผู้ใช้แยกกัน

user_id = f"tg_{raw_user_id}"
session_id = f"tg_sess_{raw_user_id}"

ขั้นตอนที่ 2: สถานะ "กำลังพิมพ์" แบบไม่พร้อมกัน (บรรทัดที่ 53–58)

เพื่อให้ผู้ใช้ได้รับประสบการณ์การใช้งานที่มีการตอบสนองสูงขณะที่ Agent ADK ประมวลผลข้อความแจ้ง (ซึ่งอาจใช้เวลาหลายวินาที) บ็อตจะเริ่มลูปเบื้องหลังแบบไม่พร้อมกันโดยทำดังนี้

สร้างอินสแตนซ์ asyncio.Event เป็น stop_event
asyncio.create_task จะสร้าง send_typing_loop(...) ในเบื้องหลัง
ลูปจะส่งการดำเนินการ ChatAction.TYPING ไปยัง Telegram ทุกๆ 4 วินาทีจนกว่าจะมีการตั้งค่า stop_event

ขั้นตอนที่ 3: การยืนยันและการสร้างเซสชัน ADK (บรรทัดที่ 61–72)

ก่อนที่จะเรียกใช้ Agent บ็อตจะตรวจสอบว่ามีเซสชันอยู่แล้วหรือไม่โดยทำดังนี้

ส่งคำขอ GET ไปยัง /apps/{appName}/users/{userId}/sessions/{sessionId}
หากการตอบกลับเป็น 404 Not Found บ็อตจะสร้างเซสชันผ่านคำขอ POST ไปยัง URL เดียวกันโดยมีเนื้อหา JSON ว่าง
หากระบบแสดงผลสถานะอื่นที่ไม่ใช่ 200 หรือ 404 ระบบจะแสดงข้อยกเว้น

ขั้นตอนที่ 4: การส่งคำขอไปยัง Agent (บรรทัดที่ 74–85)

ระบบจะส่งต่อเพย์โหลดข้อความไปยังปลายทาง /run ของ ADK

ปลายทาง: POST /run
การหมดเวลาของคำขอจะตั้งค่าเป็น 60.0 วินาทีเพื่อให้มีการอนุมานที่ซับซ้อนหรือเวลาในการตอบสนองของระบบต้นทาง
โครงสร้างเพย์โหลด:

{
  "appName": "restaurant_agent",
  "userId": "tg_<user_id>",
  "sessionId": "tg_sess_<user_id>",
  "newMessage": {
    "role": "user",
    "parts": [{"text": "<user_message>"}]
  }
}

ขั้นตอนที่ 5: การแยกวิเคราะห์การตอบกลับ (บรรทัดที่ 87–101)

เซิร์ฟเวอร์ ADK จะแสดงผลรายการเหตุการณ์ข้อความ บ็อตจะตรวจสอบอาร์เรย์ที่แสดงผลโดยทำดังนี้

ดึงข้อมูลเหตุการณ์สุดท้ายในรายการ (events[-1])
ไปยังเนื้อหาข้อความผ่าน event["content"]["parts"][0]["text"]
หากไม่มีการแสดงผลเหตุการณ์หรือไม่มีโครงสร้างข้อความ ระบบจะตั้งค่าข้อความตัวยึดตำแหน่งที่อธิบาย

ขั้นตอนที่ 6: การล้างข้อมูลและการส่งการตอบกลับ (บรรทัดที่ 103–111)

ในบล็อก finally ระบบจะตั้งค่า stop_event ซึ่งจะหยุดลูปการดำเนินการพิมพ์
บ็อตจะรอให้ typing_task เสร็จสมบูรณ์เพื่อให้มีทรัพยากรที่สะอาด
สุดท้าย บ็อตจะตอบกลับแชท Telegram ด้วยข้อความตอบกลับที่แยกวิเคราะห์แล้ว

6. ติดตั้งใช้งานแอปพลิเคชันเว็บฮุค Telegram ใน Cloud Run

จากนั้นเราจะติดตั้งใช้งาน Listener เว็บฮุค Telegram ใน Cloud Run เพื่อให้บ็อตสื่อสารกับ Listener ได้

สร้าง Dockerfile

ก่อนอื่น เราต้องสร้าง Dockerfile

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/Dockerfile

จากนั้นคัดลอกโค้ดต่อไปนี้ลงใน Dockerfile

# Use an official Python runtime as a parent image
FROM python:3.11-slim

# Prevent Python from writing pyc files to disc and buffering stdout/stderr
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Set the working directory in the container
WORKDIR /app

# Install system dependencies if needed
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Copy the dependencies file to the working directory
COPY requirements.txt .

# Install any needed packages specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of the application code
COPY main.py .

# Expose the port that Cloud Run will provide via environment variable
EXPOSE 8080

# Run main.py when the container launches
CMD ["python", "main.py"]

ระบบจะสร้างคอนเทนเนอร์ของบริการโดยใช้ python:3.11-slim เพื่อให้ขนาดของอิมเมจมีขนาดเล็ก

ติดตั้งการอ้างอิงจาก requirements.txt (python-telegram-bot[webhooks] และ httpx)
เปิดพอร์ตมาตรฐาน 8080
เปิด python main.py

เตรียมตัวแปรสภาพแวดล้อม

หลังจากนั้น ให้ตรวจสอบอีกครั้งว่าได้ติดตั้งใช้งาน Agent สำเร็จหรือไม่

AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

จากนั้นใส่ TELEGRAM_BOT_TOKEN ที่เราได้รับก่อนหน้านี้ลงใน .env

echo "TELEGRAM_BOT_TOKEN=YOUR_TELEGRAM_API_KEY" >> .env

จากนั้นป้อนข้อมูล .env ด้วยค่าอื่นๆ ที่เราต้องการ

echo "ADK_SERVER_URL=$AGENT_URL" >> .env
echo "ADK_APP_NAME=restaurant_agent" >> .env
echo "SERVICE_NAME=telegram-integration" >> .env
source .env

สร้างสคริปต์การติดตั้งใช้งาน

มาสร้างสคริปต์การติดตั้งใช้งานที่จะทำการตรวจสอบแบบสมบูรณ์และติดตั้งใช้งานแอปใน Cloud Run กัน

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/deploy.sh

และคัดลอกโค้ดต่อไปนี้ลงในไฟล์

#!/usr/bin/env bash
# ./telegram-integration/deploy.sh

# Exit immediately if a command exits with a non-zero status
set -euo pipefail

# Color codes for neat terminal output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0;37m' # No Color
# Load environment variables from .env if it exists
if [ -f .env ]; then
    echo -e "${GREEN}✔ Loading environment variables from .env...${NC}"
    export $(grep -v '^#' .env | xargs)
fi

echo -e "${BLUE}====================================================${NC}"
echo -e "${BLUE}   Google Cloud Run Deployment: Telegram Bot        ${NC}"
echo -e "${BLUE}====================================================${NC}"

# 1. Check for gcloud CLI
if ! command -v gcloud &> /dev/null; then
    echo -e "${RED}Error: 'gcloud' CLI is not installed.${NC}"
    echo "Please install the Google Cloud SDK and try again."
    echo "See: https://cloud.google.com/sdk/docs/install"
    exit 1
fi

# 2. Check active gcloud account/auth
ACTIVE_ACCOUNT=$(gcloud auth list --filter=status:ACTIVE --format="value(account)" 2>/dev/null || true)
if [ -z "$ACTIVE_ACCOUNT" ]; then
    echo -e "${RED}Error: No active Google Cloud account found.${NC}"
    echo "Please run: gcloud auth login"
    exit 1
fi

# 3. Detect / Prompt for GCP Project
DEFAULT_PROJECT=${GCP_PROJECT_ID:-$(gcloud config get-value project 2>/dev/null || true)}
if [ -n "${DEFAULT_PROJECT}" ]; then
    echo -e "${GREEN}✔ Using GCP Project: $DEFAULT_PROJECT${NC}"
    GCP_PROJECT="$DEFAULT_PROJECT"
else
    echo -n "Enter GCP Project ID: "
    read -r GCP_PROJECT
fi

if [ -z "$GCP_PROJECT" ]; then
    echo -e "${RED}Error: GCP Project ID is required.${NC}"
    exit 1
fi

# Set active project
gcloud config set project "$GCP_PROJECT" &> /dev/null

# 4. Configure Service Parameters
DEFAULT_SERVICE=${SERVICE_NAME:-"telegram-integration"}
if [ -n "${SERVICE_NAME:-}" ]; then
    echo -e "${GREEN}✔ Using Cloud Run Service Name: $SERVICE_NAME${NC}"
else
    echo -n "Enter Cloud Run Service Name [Default: $DEFAULT_SERVICE]: "
    read -r SERVICE_NAME
    SERVICE_NAME=${SERVICE_NAME:-$DEFAULT_SERVICE}
fi

DEFAULT_REGION=${REGION:-"us-central1"}
if [ -n "${REGION:-}" ]; then
    echo -e "${GREEN}✔ Using Cloud Run Region: $REGION${NC}"
else
    echo -n "Enter Cloud Run Region [Default: $DEFAULT_REGION]: "
    read -r REGION
    REGION=${REGION:-$DEFAULT_REGION}
fi

DEFAULT_ADK_APP=${ADK_APP_NAME:-"restaurant_agent"}
if [ -n "${ADK_APP_NAME:-}" ]; then
    echo -e "${GREEN}✔ Using ADK App Name: $ADK_APP_NAME${NC}"
    ADK_APP="$ADK_APP_NAME"
else
    echo -n "Enter ADK App Name [Default: $DEFAULT_ADK_APP]: "
    read -r ADK_APP
    ADK_APP=${ADK_APP:-$DEFAULT_ADK_APP}
fi

# 5. Retrieve/Prompt for Telegram Bot Token
if [ -n "${TELEGRAM_BOT_TOKEN:-}" ]; then
    echo -e "${GREEN}✔ Found TELEGRAM_BOT_TOKEN in environment.${NC}"
    BOT_TOKEN="$TELEGRAM_BOT_TOKEN"
else
    echo -e "${YELLOW}TELEGRAM_BOT_TOKEN is not set in your environment.${NC}"
    echo -n "Enter your Telegram Bot Token (input will be hidden): "
    read -s -r BOT_TOKEN
    echo ""
fi

if [ -z "$BOT_TOKEN" ]; then
    echo -e "${RED}Error: Telegram Bot Token is required.${NC}"
    exit 1
fi

# 6. Retrieve/Prompt for ADK Server URL
DEFAULT_ADK_URL="http://localhost:8000"
if [ -n "${ADK_SERVER_URL:-}" ]; then
    echo -e "${GREEN}✔ Found ADK_SERVER_URL in environment: $ADK_SERVER_URL${NC}"
    ADK_URL="$ADK_SERVER_URL"
else
    echo -n "Enter your ADK Server URL [Default: $DEFAULT_ADK_URL]: "
    read -r ADK_URL
    ADK_URL=${ADK_URL:-$DEFAULT_ADK_URL}
fi

# Enable required GCP services
echo -e "\n${YELLOW}Checking and enabling required GCP services...${NC}"
gcloud services enable run.googleapis.com cloudbuild.googleapis.com artifactregistry.googleapis.com --project "$GCP_PROJECT"

# Determine source directory dynamically
SOURCE_DIR="."
if [ -d "telegram-integration" ]; then
    SOURCE_DIR="telegram-integration"
    echo -e "${GREEN}✔ Found source directory: telegram-integration${NC}"
elif [ -f "Dockerfile" ]; then
    SOURCE_DIR="."
    echo -e "${GREEN}✔ Dockerfile found in current directory. Using current directory as source.${NC}"
else
    echo -e "${RED}Error: Could not find source directory 'telegram-integration' or Dockerfile in current directory.${NC}"
    exit 1
fi

# 7. First-pass Deployment with placeholder SERVICE_URL
# This boots the container in Webhook mode (so health check binds to port)
# but uses a high-reliability placeholder URL (google.com) to pass DNS verification checks.
echo -e "\n${YELLOW}Deploying to Cloud Run (Step 1/2: Initial Deploy)...${NC}"
gcloud run deploy "$SERVICE_NAME" \
  --source "$SOURCE_DIR" \
  --region "$REGION" \
  --allow-unauthenticated \
  --set-env-vars "TELEGRAM_BOT_TOKEN=$BOT_TOKEN,ADK_SERVER_URL=$ADK_URL,ADK_APP_NAME=$ADK_APP,SERVICE_URL=https://google.com" \
  --project "$GCP_PROJECT"

# 8. Retrieve the actual service URL
echo -e "\n${YELLOW}Retrieving service URL...${NC}"
SERVICE_URL=$(gcloud run services describe "$SERVICE_NAME" --region "$REGION" --project "$GCP_PROJECT" --format 'value(status.url)')
echo -e "${GREEN}✔ Service URL is: $SERVICE_URL${NC}"

# 9. Update service environment variables with the real SERVICE_URL
# This triggers a rolling update and registers the correct webhook with Telegram automatically!
echo -e "\n${YELLOW}Updating configuration with final Webhook URL (Step 2/2)...${NC}"
gcloud run services update "$SERVICE_NAME" \
  --region "$REGION" \
  --set-env-vars "TELEGRAM_BOT_TOKEN=$BOT_TOKEN,ADK_SERVER_URL=$ADK_URL,ADK_APP_NAME=$ADK_APP,SERVICE_URL=$SERVICE_URL" \
  --project "$GCP_PROJECT"

echo -e "\n${GREEN}====================================================${NC}"
echo -e "${GREEN}   Deployment Completed Successfully! 🎉            ${NC}"
echo -e "${GREEN}====================================================${NC}"
echo -e "Service Name:   ${BLUE}$SERVICE_NAME${NC}"
echo -e "Region:         ${BLUE}$REGION${NC}"
echo -e "Active URL:     ${BLUE}$SERVICE_URL${NC}"
echo -e "Webhook Path:   ${BLUE}$SERVICE_URL/<bot-token>${NC}"
echo -e "ADK Backend:    ${BLUE}$ADK_URL${NC}"
echo -e "ADK App Name:   ${BLUE}$ADK_APP${NC}"
echo -e "${GREEN}====================================================${NC}"
echo "Your Telegram Bot has been configured to use webhooks."
echo "Any message sent to your bot will now trigger this Cloud Run instance."

สคริปต์การติดตั้งใช้งาน 2 ครั้ง (deploy.sh)

เมื่อติดตั้งใช้งานใน Google Cloud Run บ็อตต้องระบุ URL ของตัวเอง (SERVICE_URL) ในสภาพแวดล้อมเพื่อให้ลงทะเบียน URL เป็นเป้าหมายเว็บฮุคกับ Telegram ได้ หากต้องการแก้ไขการอ้างอิงแบบวงกลมนี้ (ไม่ทราบ URL จนกว่าจะติดตั้งใช้งาน แต่บริการต้องใช้ URL เพื่อบูตขึ้นโดยไม่มีข้อผิดพลาดในการตรวจสอบประสิทธิภาพการทำงาน) deploy.sh จะทำการติดตั้งใช้งาน 2 ขั้นตอนดังนี้

ขั้นตอนที่ 1: การติดตั้งใช้งานครั้งแรก: บูตคอนเทนเนอร์ด้วย DNS ตัวยึดตำแหน่ง (https://google.com) เพื่อให้บริการเริ่มต้นขึ้นได้สำเร็จ ผูกกับพอร์ตในเครื่อง และผ่านการตรวจสอบประสิทธิภาพการทำงานเริ่มต้นของ Cloud Run
ขั้นตอนที่ 2: ดึงข้อมูล URL: แยกปลายทาง Cloud Run ที่สร้างขึ้นใหม่โดยใช้ gcloud run services describe ผ่านโปรแกรม
ขั้นตอนที่ 3: อัปเดตการกำหนดค่า: อัปเดตตัวแปรสภาพแวดล้อมด้วย URL ของบริการจริงที่ใช้งานอยู่ ซึ่งจะทริกเกอร์การอัปเดตทีละส่วนที่สะอาดใน Cloud Run และลงทะเบียนเป้าหมายเว็บฮุคที่ถูกต้องกับ Telegram API อย่างปลอดภัย

ติดตั้งใช้งานใน Cloud Run

สคริปต์การติดตั้งใช้งานจะพิมพ์ URL ของ Agent เปิด URL ในเบราว์เซอร์เพื่อเข้าถึง UI สำหรับนักพัฒนาซอฟต์แวร์ของ ADK เดียวกันที่ทำงานใน Cloud Run

cd ~/build-agent-adk-telegram
bash ./telegram-integration/deploy.sh

หากทุกอย่างเป็นไปด้วยดี ตอนนี้คุณสามารถเริ่มแชทกับบ็อตได้โดยตรงจากแอปพลิเคชันแชท Telegram ค้นหาบ็อตที่คุณเพิ่งสร้างและเริ่มโต้ตอบกับบ็อตโดยทำดังนี้

What Italian dishes do you have?

หรือ

I want something spicy and creamy

ดูบ็อตส่งสถานะ "...กำลังพิมพ์" จากนั้นบ็อตจะแสดงข้อความจาก ADK ที่คุณสร้างไว้ก่อนหน้านี้ในไม่ช้า

7. ยินดีด้วย

คุณได้สร้าง ติดตั้งใช้งาน และผสานรวมผู้ช่วยเมนูร้านอาหารอัจฉริยะ ซึ่งเป็น AI Agent ที่ใช้ ADK กับ Telegram อย่างสมบูรณ์ผ่านการสื่อสารระหว่างไคลเอ็นต์ HTTP กับเซิร์ฟเวอร์ และอนุญาตให้ผู้คนค้นหาเมนูโปรดและจองร้านอาหารได้

สิ่งที่คุณได้เรียนรู้

ติดตั้งใช้งานและกำหนดค่าเจ้าหน้าที่บริการด้านร้านอาหาร, Agent ที่ใช้ ADK และ MCP Toolbox ใน Cloud Run
วิธีตั้งค่าบ็อต Telegram โดยใช้ BotFather
วิธีเขียนสคริปต์ Python เพื่อรับฟังเว็บฮุค Telegram และโต้ตอบกับ Agent ADK เพื่อส่งต่อข้อความค้นหาของผู้ใช้และตอบกลับตามความเหมาะสม
วิธีใช้ "... typing" ใน Telegram เพื่อส่งสัญญาณว่าระบบกำลังประมวลผลข้อความเป็นการตอบกลับแบบเรียลไทม์ให้กับผู้ใช้ขณะรอให้ Agent ADK ตอบกลับ
วิธีติดตั้งใช้งานสคริปต์ Python ใน Cloud Run และโต้ตอบกับสคริปต์ได้

ล้างข้อมูล

โปรดลบทรัพยากรที่สร้างขึ้นใน Codelab นี้เพื่อหลีกเลี่ยงการเรียกเก็บเงินกับบัญชี Google Cloud

ตัวเลือกที่ 1: ลบโปรเจ็กต์ (แนะนำ)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

ตัวเลือกที่ 2: ลบทรัพยากรแต่ละรายการ

# If you follow from previous A2A Agent Runtime codelab
# Delete the Agent Runtime deployment (skip if not found)
uv run python -c "
import vertexai
from google.genai import types
vertexai.init(project='$GOOGLE_CLOUD_PROJECT', location='$REGION')
client = vertexai.Client(
    project='$GOOGLE_CLOUD_PROJECT', location='$REGION',
    http_options=types.HttpOptions(api_version='v1beta1'),
)
try:
    agent = client.agent_engines.get(name='$RESERVATION_AGENT_RESOURCE_NAME')
    agent.delete(force=True)
    print('Agent Runtime deployment deleted.')
except Exception as e:
    print(f'No agent deployment found or already deleted, skipping. ({e})')
"

# Delete GCS staging bucket (skip if STAGING_BUCKET is not set)
if [ -n "$STAGING_BUCKET" ]; then
  gsutil rm -r gs://$STAGING_BUCKET
else
  echo "STAGING_BUCKET not set, skipping bucket deletion."
fi

# Delete Cloud Run services
gcloud run services delete restaurant-agent --region=$REGION --quiet
gcloud run services delete toolbox-service --region=$REGION --quiet
gcloud run services delete telegram-integration --region=$REGION --quiet

# Delete Cloud SQL instance
gcloud sql instances delete $DB_INSTANCE --quiet