大家好！歡迎來到「Build on Google AI」工程挑戰的第 26 天。

各位觀眾，我們終於到了！經過漫長的鋪陳，為解決長文本、多智能體協作、防迴圈機制與可觀測性，我們一路過關斬將。今天，我真的帶著半糖少冰的珍奶來兌現這個系列最核心的承諾：Action Confirmations (人機協作確認機制)。

在企業級生成式 AI 的應用中，「完全自治 (Fully Autonomous)」往往是個假議題，真正的落地標準是「人機協作 (Human-in-the-loop, HITL)」。當 Agent 準備去刷公司的信用卡、發送電子報給十萬名會員，或是準備發布爭議文案時，系統必須懂得「停下來，問老闆」。

但實務上，老闆不會整天盯著終端機或 ADK Web UI。今天，我們要將 ADK 的「中斷點 (Breakpoints)」與 Telegram 的「Inline Keyboard (互動按鈕)」完美結合，讓你的主管在高鐵上也能「一鍵掌控全局」！

第一步：理解 ADK 的中斷點與狀態恢復機制
在非同步的多智能體圖結構 (Graph) 中，我們無法使用傳統的 input() 來暫停程式。Google ADK 提供了 interrupt_before 參數，告訴系統：「在進入這個節點之前，強制凍結整個工作流，並保存當下所有的記憶 (State Checkpointing)。」

當人類決定好下一步後，我們透過 app.update_state() 覆寫白板資料，再透過傳入 None 給 app.run() 來喚醒並恢復執行這條被凍結的工作流。

第二步：結合 Telegram 互動按鈕 (Inline Keyboard)
為了讓體驗達到極致，我們會使用 pyTelegramBotAPI 套件。當 ADK 暫停時，主程式會發送一則帶有 Approve 與 Reject 按鈕的訊息到 Telegram。當主管按下按鈕，Telegram 會發送一個 callback_query 給我們的程式，我們便能在這個 Callback 中喚醒 ADK 繼續工作。

第三步：完整實作程式碼 (main.py)
請確保你已安裝 pyTelegramBotAPI (pip install pyTelegramBotAPI) 與 google-adk。這份程式碼展示了極具商業價值的「行動化審核產線」。

import os
import telebot
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
import logging
from typing import List
from pydantic import BaseModel, Field

# 匯入 Google ADK 核心元件
from google.adk import Agent, Workflow

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
logger = logging.getLogger(__name__)

# ==========================================
# [前置設定] Telegram Bot Token
# ==========================================
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "請替換為你的_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "請替換為你的_CHAT_ID")
bot = telebot.TeleBot(TELEGRAM_BOT_TOKEN)

# ==========================================
# 1. 定義全局白板與 Agent (簡化版示意)
# ==========================================
class CampaignState(BaseModel):
    marketing_copy: str = ""
    is_approved_by_human: bool = False # 人類授權的終極開關

# 模擬一個負責寫作的 Agent
marketing_agent = Agent(
    name="marketing_agent",
    model="gemini-2.5-flash",
    instruction="請寫一篇 50 字以內的超強社群貼文，推廣我們的 AI 課程。",
    state_mapping={"outputs": ["marketing_copy"]}
)

def publish_node(state: CampaignState) -> CampaignState:
    """這是一個高風險操作節點，只有在人類授權後才能執行"""
    if state.is_approved_by_human:
        logger.info(f"🚀 【正式對外發布】貼文已成功上線！內容：{state.marketing_copy}")
    else:
        logger.warning("❌ 貼文遭到人類主管退回，專案終止。")
    return state

# ==========================================
# 2. 組裝 Workflow 並設定中斷點 (Breakpoint)
# ==========================================
workflow = Workflow(name="Human_Approval_Workflow", state_schema=CampaignState)
workflow.add_node("marketing_agent", marketing_agent)
workflow.add_node("publish_node", publish_node)

workflow.set_entry_point("marketing_agent")
workflow.add_edge("marketing_agent", "publish_node")

# 【核心魔法】：在進入 publish_node 之前強制暫停
app = workflow.compile(interrupt_before=["publish_node"])

# 定義一個唯一的執行緒 ID，確保中斷與恢復時找得到同一個 State
thread_config = {"configurable": {"thread_id": "campaign_promo_001"}}

# ==========================================
# 3. 觸發 Telegram 審核與回呼處理
# ==========================================
def request_human_approval(current_copy: str):
    """發送帶有 Inline Keyboard 的審核訊息給主管"""
    markup = InlineKeyboardMarkup()
    markup.row_width = 2
    markup.add(
        InlineKeyboardButton("👍 核准發布", callback_data="approve"),
        InlineKeyboardButton("👎 退回廢棄", callback_data="reject")
    )
    
    msg_text = f"🛑 *【需要您的授權】*\nAgent 已生成最新貼文，請確認是否對外發布：\n\n📝 內容：\n_{current_copy}_"
    bot.send_message(TELEGRAM_CHAT_ID, msg_text, parse_mode="Markdown", reply_markup=markup)
    logger.info("📱 已發送審核請求至 Telegram，等待主管點擊...")

@bot.callback_query_handler(func=lambda call: True)
def handle_approval_callback(call):
    """攔截 Telegram 上的按鈕點擊事件"""
    is_approved = (call.data == "approve")
    action_text = "已核准" if is_approved else "已退回"
    
    # 1. 更新 Telegram 上的按鈕，避免重複點擊
    bot.edit_message_text(
        chat_id=call.message.chat.id,
        message_id=call.message.message_id,
        text=f"{call.message.text}\n\n✅ *主管決策：{action_text}*",
        parse_mode="Markdown"
    )
    logger.info(f"👤 人類主管決策：{action_text}")
    
    # 2. 【喚醒 ADK 工作流】：覆寫被凍結的狀態
    app.update_state(
        thread_config,
        {"is_approved_by_human": is_approved}
    )
    
    # 3. 傳入 None 恢復執行
    logger.info("▶️ 系統解除凍結，恢復執行工作流...")
    app.run(None, config=thread_config)
    
    # 停止 Bot 的 Polling，結束程式 (實務上 Webhook 會持續監聽)
    bot.stop_polling()

# ==========================================
# 主程式執行入口
# ==========================================
if __name__ == "__main__":
    initial_state = CampaignState()
    
    logger.info("🚀 啟動工作流...")
    
    # 工作流會執行 marketing_agent，然後撞到 publish_node 前的 breakpoint 暫停
    frozen_state = app.run(initial_state, config=thread_config)
    
    # 檢查下一步是否被中斷在 publish_node
    pending_nodes = app.get_state(thread_config).next
    
    if "publish_node" in pending_nodes:
        # 讀取 Agent 剛寫好的文案
        draft_copy = frozen_state.marketing_copy
        # 發送 Telegram 給主管審核
        request_human_approval(draft_copy)
        
        # 啟動 Telegram 監聽 (模擬背景等待 Callback)
        bot.infinity_polling()

小結
「最強大的 AI 系統，永遠會為人類保留一個發號施令的座位。」

今天，我們完成了企業級 Agent 應用中最具價值的一塊拼圖。透過 ADK 的 Breakpoints 與 Telegram API，我們讓人類能用最優雅、最無縫的方式介入 AI 工作流。這個機制適用於所有高風險場景：寄發正式報價單、修改生產庫資料、動用高額 API 預算。