January 22nd, 2026
| 圖片 A | 圖片 B |
|---|---|
 |
 |
前情提要
在維護 linebot-helper-python 專案時，我一直面臨一個架構問題：隨著功能增加，main.py 中的 if/elif 判斷式越來越長，程式碼越來越難維護。
原本的訊息路由邏輯：
# ❌ 舊版 - if/elif 地獄
async def handle_text_message(event):
msg = event.message.text
if msg.startswith('/clear'):
# 處理清除指令
elif msg.startswith('/help'):
# 處理說明指令
elif msg == '@g':
# 處理 GitHub 摘要
elif is_youtube_url(msg):
# 處理 YouTube 摘要
elif extract_url(msg):
# 處理一般 URL 摘要
else:
# 處理一般對話
這樣的架構有幾個明顯問題：
❌ 難以維護 - 新增功能就要加 elif，檔案越來越肥 ❌ 難以測試 - 所有邏輯混在一起，單元測試困難 ❌ 難以擴展 - 想要並行處理多個意圖？改起來很痛苦 ❌ 職責不清 - 對話、摘要、位置搜尋全部混在一起
在 2025 年初，Google 發布了 Agent Development Kit (ADK)，提供了建構 Multi-Agent 系統的框架。我決定用 ADK 重構整個專案，實現 Agent-to-Agent (A2A) 架構。
這篇文章記錄了完整的遷移過程，從規劃到實作的 5 個階段。
什麼是 ADK 和 A2A？
Google Agent Development Kit (ADK)
ADK 是 Google 提供的 Python 框架，用於建構基於 LLM 的 Agent 系統：
from google.adk.agents import Agent
chat_agent = Agent(
name="chat_agent",
model="gemini-2.5-flash",
description="處理一般對話",
instruction="你是一個智能助手...",
tools=[google_search_tool],
)
Agent-to-Agent (A2A) 通訊
A2A 是一種架構模式，讓多個專業 Agent 協作處理複雜任務：
User: "幫我找台北好吃的拉麵，然後摘要這篇文章 https://..."
│
▼
┌─────────────────────┐
│ Orchestrator Agent │
│ (意圖解析 + 路由) │
└─────────────────────┘
│
┌─────────────┴─────────────┐
│ Parallel Dispatch (A2A) │
▼ ▼
┌──────────────┐ ┌──────────────┐
│LocationAgent │ │ContentAgent │
│ Task: 找拉麵 │ │ Task: 摘要URL│
└──────────────┘ └──────────────┘
│ │
└──────────────┬───────────┘
▼
彙整回覆給用戶
遷移前的架構分析
現有架構摘要
┌──────────────┬────────────────────────────────────────────────────┐
│ 項目 │ 現況 │
├──────────────┼────────────────────────────────────────────────────┤
│ 框架 │ FastAPI + LINE Bot SDK │
├──────────────┼────────────────────────────────────────────────────┤
│ AI 引擎 │ Google Vertex AI (google-genai SDK) │
├──────────────┼────────────────────────────────────────────────────┤
│ Session 管理 │ 自建 ChatSessionManager (記憶體內) │
├──────────────┼────────────────────────────────────────────────────┤
│ 訊息路由 │ if/elif 條件判斷式 │
├──────────────┼────────────────────────────────────────────────────┤
│ 工具整合 │ Grounding (Google Search)、Maps、YouTube、網頁抓取 │
└──────────────┴────────────────────────────────────────────────────┘
可行性評估結論
✅ 完全可行 - 專案已經使用 Google Vertex AI 和 google-genai SDK，這正是 ADK 的基礎。轉換需要的是重構架構而非重寫邏輯。
目標架構設計
Agent 拆分規劃
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ (接收 LINE 訊息，決定路由到哪個專家 Agent) │
└─────────────────────────────────────────────────────────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Chat Agent │ │ Content Agent │ │ Location Agent │
│ (對話式問答) │ │ (內容摘要) │ │ (地點搜尋) │
│ │ │ │ │ │
│ Tools: │ │ Tools: │ │ Tools: │
│ - GoogleSearch │ │ - URLLoader │ │ - MapsGrounding│
│ - Grounding │ │ - YouTubeFetch │ │ - PlaceSearch │
└─────────────────┘ │ - PDFLoader │ └─────────────────┘
│ - Summarizer │
└─────────────────┘
│
┌────────┴────────┐
▼ ▼
┌───────────┐ ┌───────────┐
│ Vision │ │ GitHub │
│ Agent │ │ Agent │
└───────────┘ └───────────┘
Agent 職責對照表
┌───────────────────────────────────┬──────────────────────────┬───────────────────────────────────────┐
│ 現有模組 │ 轉換後 Agent │ 職責 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_text_message() │ ChatAgent │ 處理一般對話、Google Search Grounding │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_url_message() + load_url() │ ContentAgent │ URL 內容抓取與摘要 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ load_transcript_from_youtube() │ ContentAgent │ YouTube 影片摘要 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_location_message() │ LocationAgent │ 地點搜尋與推薦 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_image_message() │ VisionAgent │ 圖片分析 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_github_summary() │ GitHubAgent │ GitHub Issues 摘要 │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ ChatSessionManager │ SessionManager │ 跨 Agent 的對話記憶 │
└───────────────────────────────────┴──────────────────────────┴───────────────────────────────────────┘
目標專案結構
linebot-helper-python/
├── main.py # FastAPI + LINE webhook (保留)
├── agents/
│ ├── __init__.py
│ ├── orchestrator.py # 主控 Agent (路由決策)
│ ├── chat_agent.py # 對話 Agent
│ ├── content_agent.py # 內容摘要 Agent
│ ├── location_agent.py # 地點搜尋 Agent
│ ├── vision_agent.py # 圖片分析 Agent
│ └── github_agent.py # GitHub Agent
├── tools/
│ ├── __init__.py
│ ├── url_loader.py # 從 loader/url.py 重構
│ ├── youtube_tool.py # 從 loader/youtube_gcp.py 重構
│ ├── maps_tool.py # 從 loader/maps_grounding.py 重構
│ ├── summarizer.py # 從 loader/langtools.py 重構
│ └── pdf_tool.py # 從 loader/pdf.py 重構
├── services/
│ ├── session_manager.py # Session 管理服務
│ └── line_service.py # LINE API 封裝
├── config/
│ └── agent_config.py # Agent 設定與 Model 選擇
└── loader/ # 保留舊版相容
遷移計畫：5 個階段
┌─────────┬─────────────────────────────────────────────────────────────────┬──────────┐
│ 階段 │ 工作內容 │ 預估風險 │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 1 │ 安裝 ADK、建立 tools/ 目錄，將 loader/*.py 重構為 ADK Tool 格式 │ 低風險 │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 2 │ 建立 ChatAgent，驗證與 LINE webhook 整合 │ 中風險 │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 3 │ 建立其他專業 Agent (Content, Location, Vision, GitHub) │ 中風險 │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 4 │ 建立 Orchestrator，實作 A2A 路由邏輯 │ 高風險 │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 5 │ 遷移 Session 管理、測試完整流程 │ 中風險 │
└─────────┴─────────────────────────────────────────────────────────────────┴──────────┘
Phase 1: 安裝 ADK 與重構 Tools
目標
將現有的 loader/ 模組重構為 ADK 相容的 Tool 格式。
實作內容
1. 安裝 ADK 依賴
# requirements.txt
google-adk>=1.0.0
2. 建立 tools/ 目錄結構
# tools/summarizer.py
"""ADK-compatible summarization tools"""
def summarize_text(text: str, mode: str = "normal") -> dict:
"""
Summarize text content using Gemini.
Args:
text: Content to summarize
mode: "normal", "detailed", or "twitter"
Returns:
dict with 'status' and 'summary' keys
"""
# 實作摘要邏輯...
return {"status": "success", "summary": result}
def analyze_image(image_data: bytes, prompt: str) -> dict:
"""
Analyze image using Gemini multimodal.
Args:
image_data: Image bytes
prompt: Analysis prompt
Returns:
dict with 'status' and 'analysis' keys
"""
# 實作圖片分析邏輯...
return {"status": "success", "analysis": result}
3. 重構的工具模組
| 原始模組 | 新工具模組 | 功能 |
|---|---|---|
loader/langtools.py |
tools/summarizer.py |
文字摘要、圖片分析 |
loader/youtube_gcp.py |
tools/youtube_tool.py |
YouTube 影片摘要 |
loader/maps_grounding.py |
tools/maps_tool.py |
地點搜尋 |
loader/pdf.py |
tools/pdf_tool.py |
PDF 處理 |
loader/url.py |
tools/url_loader.py |
URL 內容抓取 |
成果
- PR #10 merged
- Release v0.3.0
Phase 2: 建立 ChatAgent 與 LINE 整合
目標
建立第一個 ADK Agent，處理一般對話並整合 Google Search Grounding。
實作內容
1. 建立 Agent 配置
# config/agent_config.py
@dataclass
class AgentConfig:
"""Configuration for ADK agents"""
# Vertex AI settings
project_id: str
location: str
# Model settings
chat_model: str = "gemini-2.5-flash"
orchestrator_model: str = "gemini-2.5-pro"
fast_model: str = "gemini-2.5-flash-lite"
# Session settings
session_timeout_minutes: int = 30
max_history_length: int = 20
# Feature flags
enable_grounding: bool = True
enable_maps_grounding: bool = True
2. 建立 ChatAgent
# agents/chat_agent.py
from google.adk.agents import Agent
CHAT_AGENT_INSTRUCTION = """你是一個智能助手，專門回答用戶的問題。
## 回應原則
1. 使用台灣用語的繁體中文回答
2. 如果需要最新資訊，請搜尋網路並提供準確的答案
3. 回答要簡潔但完整，適合在 LINE 訊息中閱讀
"""
class ChatAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.sessions: Dict[str, dict] = {}
# Initialize ADK agent
self.adk_agent = Agent(
name="chat_agent",
model=config.chat_model,
description="對話式問答 Agent",
instruction=CHAT_AGENT_INSTRUCTION,
tools=[],
)
async def chat(self, user_id: str, message: str) -> dict:
"""Process a chat message and return response"""
chat, history = self.get_or_create_session(user_id)
response = chat.send_message(message)
return {
'status': 'success',
'answer': response.text,
'sources': self._extract_sources(response),
'has_history': len(history) > 0
}
3. 建立 LINE Service 封裝
# services/line_service.py
class LineService:
"""Wrapper for LINE Bot API operations"""
@staticmethod
def format_error_message(error: Exception, action: str) -> str:
"""Format user-friendly error message"""
error_str = str(error)
if "quota" in error_str.lower():
return f"⚠️ {action}時 API 額度不足，請稍後再試"
elif "timeout" in error_str.lower():
return f"⚠️ {action}時連線逾時，請稍後再試"
else:
return f"⚠️ {action}時發生錯誤，請稍後再試"
成果
- PR #11 merged
- Release v0.4.0
Phase 3: 建立專業 Agents
目標
建立 ContentAgent、LocationAgent、VisionAgent、GitHubAgent。
實作內容
1. ContentAgent - 內容摘要
# agents/content_agent.py
class ContentAgent:
"""Agent for URL content summarization"""
async def process_url(self, url: str, mode: str = "normal") -> dict:
"""Process any URL and return summary"""
if self._is_youtube_url(url):
return await self.summarize_youtube(url, mode)
elif self._is_pdf_url(url):
return await self.summarize_pdf(url, mode)
else:
return await self.summarize_webpage(url, mode)
async def summarize_youtube(self, url: str, mode: str) -> dict:
"""Summarize YouTube video"""
transcript = get_youtube_transcript(url)
summary = summarize_text(transcript, mode)
return {"status": "success", "summary": summary}
2. LocationAgent - 地點搜尋
# agents/location_agent.py
class LocationAgent:
"""Agent for location-based searches"""
async def search(
self,
latitude: float,
longitude: float,
place_type: Literal["gas_station", "parking", "restaurant"]
) -> dict:
"""Search for nearby places"""
result = search_nearby_places(
latitude=latitude,
longitude=longitude,
place_type=place_type,
language_code="zh-TW"
)
return result
3. VisionAgent - 圖片分析
# agents/vision_agent.py
class VisionAgent:
"""Agent for image analysis"""
async def analyze(self, image_data: bytes, prompt: str = None) -> dict:
"""Analyze an image"""
analysis_prompt = prompt or DEFAULT_IMAGE_PROMPT
result = analyze_image(image_data, analysis_prompt)
return result
4. GitHubAgent - GitHub 整合
# agents/github_agent.py
class GitHubAgent:
"""Agent for GitHub operations"""
def get_issues_summary(self) -> dict:
"""Get summary of open GitHub issues"""
issues = fetch_github_issues()
summary = format_issues_summary(issues)
return {"status": "success", "summary": summary}
成果
- PR #12 merged
- Release v0.5.0
Phase 4: 建立 Orchestrator 與 A2A 路由
目標
建立主控 Orchestrator，實現意圖偵測和 Agent 路由。
實作內容
1. 定義意圖類型
# agents/orchestrator.py
class IntentType(Enum):
"""Types of user intents"""
CHAT = "chat" # 一般對話
URL_SUMMARY = "url_summary" # URL 摘要
YOUTUBE_SUMMARY = "youtube" # YouTube 摘要
IMAGE_ANALYSIS = "image" # 圖片分析
LOCATION_SEARCH = "location" # 地點搜尋
GITHUB_SUMMARY = "github" # GitHub 操作
COMMAND = "command" # 系統指令
UNKNOWN = "unknown"
2. Orchestrator 主類別
class Orchestrator:
"""Main controller for A2A routing"""
def __init__(self, config: AgentConfig):
# Initialize all specialized agents
self.chat_agent = create_chat_agent(config)
self.content_agent = create_content_agent(config)
self.location_agent = create_location_agent(config)
self.vision_agent = create_vision_agent(config)
self.github_agent = create_github_agent(config)
# URL patterns for intent detection
self._url_pattern = re.compile(r'https?://[^\s]+')
self._youtube_pattern = re.compile(r'https?://(?:www\.)?(?:youtube\.com|youtu\.be)')
def detect_intents(self, message: str) -> List[Intent]:
"""Detect user intents from message"""
intents = []
# Check for commands
if message.lower() in ['/clear', '/help', '/status']:
return [Intent(IntentType.COMMAND, 1.0, {'command': message})]
# Check for GitHub command
if message == '@g':
return [Intent(IntentType.GITHUB_SUMMARY, 1.0, {})]
# Check for URLs
urls = self._url_pattern.findall(message)
for url in urls:
if self._youtube_pattern.match(url):
intents.append(Intent(IntentType.YOUTUBE_SUMMARY, 0.95, {'url': url}))
else:
intents.append(Intent(IntentType.URL_SUMMARY, 0.95, {'url': url}))
# Default to chat
if not intents:
intents.append(Intent(IntentType.CHAT, 0.9, {'message': message}))
return intents
3. A2A 路由邏輯
async def process_text(self, user_id: str, message: str) -> OrchestratorResult:
"""Process text message with A2A routing"""
intents = self.detect_intents(message)
# Single intent
if len(intents) == 1:
result = await self._route_intent(user_id, intents[0])
return OrchestratorResult(success=True, responses=[result], intents=intents)
# Multiple intents - parallel execution!
tasks = [self._route_intent(user_id, intent) for intent in intents]
results = await asyncio.gather(*tasks, return_exceptions=True)
return OrchestratorResult(success=True, responses=results, intents=intents)
async def _route_intent(self, user_id: str, intent: Intent) -> dict:
"""Route single intent to appropriate agent"""
if intent.type == IntentType.CHAT:
return await self.chat_agent.chat(user_id, intent.data['message'])
elif intent.type == IntentType.YOUTUBE_SUMMARY:
return await self.content_agent.summarize_youtube(intent.data['url'])
elif intent.type == IntentType.URL_SUMMARY:
return await self.content_agent.process_url(intent.data['url'])
# ... 其他路由
4. 更新 main.py 使用 Orchestrator
# main.py
from agents import create_orchestrator, format_orchestrator_response
# Initialize single orchestrator (manages all agents)
orchestrator = create_orchestrator()
async def handle_text_message(event: MessageEvent, user_id: str):
"""Simplified handler using Orchestrator"""
message = event.message.text
# Single line routing - Orchestrator handles everything!
result = await orchestrator.process_text(user_id=user_id, message=message)
response_text = format_orchestrator_response(result)
await line_bot_api.reply_message(event.reply_token, TextSendMessage(text=response_text))
架構對比
Before (if/elif 地獄):
if is_command(msg):
handle_command()
elif msg == '@g':
handle_github()
elif is_youtube_url(msg):
handle_youtube()
elif has_url(msg):
handle_url()
else:
handle_chat()
After (A2A Orchestration):
result = await orchestrator.process_text(user_id, message)
response = format_orchestrator_response(result)
成果
- PR #13 merged
- Release v0.6.0
Phase 5: Session 管理遷移
目標
建立集中式 SessionManager，支援 TTL 自動過期和背景清理。
實作內容
1. SessionManager 類別
# services/session_manager.py
@dataclass
class SessionData:
"""Data structure for a single user session"""
user_id: str
chat: Any # Gemini chat instance
history: List[dict]
created_at: datetime
last_active: datetime
class SessionManager:
"""Centralized session manager with TTL and auto-cleanup"""
def __init__(
self,
timeout_minutes: int = 30,
max_history_length: int = 20,
cleanup_interval_seconds: int = 300
):
self.timeout = timedelta(minutes=timeout_minutes)
self.max_history_length = max_history_length
self._sessions: Dict[str, SessionData] = {}
self._lock = Lock() # Thread-safe
self._cleanup_task = None
def get_or_create_session(self, user_id: str, chat_factory: Callable) -> SessionData:
"""Get existing session or create new one"""
with self._lock:
session = self._sessions.get(user_id)
if session and not self.is_expired(session):
session.last_active = datetime.now()
return session
# Create new session
new_session = SessionData(
user_id=user_id,
chat=chat_factory(),
history=[],
created_at=datetime.now(),
last_active=datetime.now()
)
self._sessions[user_id] = new_session
return new_session
async def start_cleanup_task(self):
"""Start background cleanup task"""
self._running = True
self._cleanup_task = asyncio.create_task(self._cleanup_loop())
async def _cleanup_loop(self):
"""Background loop for periodic cleanup"""
while self._running:
await asyncio.sleep(self.cleanup_interval)
self.cleanup_expired_sessions()
2. FastAPI 生命週期整合
# main.py
from services.session_manager import get_session_manager
session_manager = get_session_manager()
@app.on_event("startup")
async def startup_event():
"""Start background tasks on startup"""
await session_manager.start_cleanup_task()
logger.info("Session cleanup background task started")
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup on shutdown"""
await session_manager.stop_cleanup_task()
await session.close()
logger.info("Application shutdown complete")
3. 新增 /stats 指令
# 在 Orchestrator 中處理
elif command in ['/session-stats', '/stats']:
stats = self.chat_agent.session_manager.get_stats()
return {
'status': 'success',
'response': f"""📈 Session 統計資訊
👥 活躍對話數：{stats.active_sessions}
💬 總訊息數：{stats.total_messages}
⏱️ 最舊對話：{stats.oldest_session_age_minutes:.1f} 分鐘
🧹 清理次數：{stats.cleanup_runs}
🗑️ 已清理對話：{stats.sessions_cleaned}"""
}
成果
- PR #14 merged
- Release v0.7.0
最終架構
完整的 A2A 流程
LINE Message
│
▼
┌─────────────────────────────────────────────────────────────┐
│ FastAPI Webhook │
│ (main.py) │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ │
│ 1. detect_intents() - 分析用戶意圖 │
│ 2. _route_intent() - 路由到專業 Agent │
│ 3. format_response() - 彙整回覆 │
└─────────────────────────────────────────────────────────────┘
│
├─── CHAT ────────▶ ChatAgent (Google Search Grounding)
│
├─── URL ─────────▶ ContentAgent (Summarization)
│
├─── YOUTUBE ─────▶ ContentAgent (YouTube API)
│
├─── IMAGE ───────▶ VisionAgent (Gemini Multimodal)
│
├─── LOCATION ────▶ LocationAgent (Maps Grounding)
│
└─── GITHUB ──────▶ GitHubAgent (GitHub API)
│
▼
┌─────────────────┐
│ SessionManager │
│ (背景清理 Task) │
└─────────────────┘
支援的意圖與指令
| 意圖 | 觸發條件 | 處理 Agent |
|---|---|---|
| CHAT | 一般文字訊息 | ChatAgent |
| URL_SUMMARY | 包含 HTTP URL | ContentAgent |
| YOUTUBE_SUMMARY | YouTube URL | ContentAgent |
| IMAGE_ANALYSIS | 圖片訊息 | VisionAgent |
| LOCATION_SEARCH | 位置訊息 | LocationAgent |
| GITHUB_SUMMARY | @g 指令 |
GitHubAgent |
| COMMAND | /clear, /help, /status, /stats |
Orchestrator 直接處理 |
開發心得
1. 分階段遷移降低風險
這次遷移最重要的決策是分 5 個階段進行，每個階段都是獨立的 PR：
Phase 1 (低風險) → Phase 2 (中風險) → Phase 3 (中風險) → Phase 4 (高風險) → Phase 5 (中風險)
每個階段完成後都可以正常運作，不會有「改到一半系統壞掉」的情況。
2. 保留舊版相容
我刻意保留了 loader/ 目錄，讓舊的程式碼仍然可用：
# 新版 tools/ 內部呼叫舊版 loader/
from loader.url import load_url as legacy_load_url
def load_url(url: str) -> dict:
"""ADK-compatible wrapper"""
result = legacy_load_url(url)
return {"status": "success", "content": result}
這樣做的好處是：
- ✅ 可以漸進式遷移
- ✅ 出問題可以快速回滾
- ✅ 不需要一次重寫所有邏輯
3. Intent Detection vs LLM Routing
在設計 Orchestrator 時，我考慮過兩種方案：
方案 A: 規則式 Intent Detection（選擇此方案）
def detect_intents(self, message: str) -> List[Intent]:
if self._youtube_pattern.match(message):
return [Intent(IntentType.YOUTUBE_SUMMARY, ...)]
方案 B: LLM Routing
async def detect_intents(self, message: str) -> List[Intent]:
prompt = f"分析以下訊息的意圖: {message}"
result = await self.llm.generate(prompt)
return parse_intents(result)
我選擇方案 A 的原因：
- ✅ 速度更快（不需要額外 API 呼叫）
- ✅ 成本更低
- ✅ 行為可預測
- ✅ 對於 LINE Bot 這種明確的輸入格式已經足夠
4. Parallel Execution 的威力
A2A 架構最強大的地方是並行執行：
# 用戶訊息: "幫我摘要 https://... 然後找附近餐廳"
intents = [
Intent(IntentType.URL_SUMMARY, ...),
Intent(IntentType.LOCATION_SEARCH, ...)
]
# 並行執行！
tasks = [self._route_intent(user_id, intent) for intent in intents]
results = await asyncio.gather(*tasks)
舊版架構需要順序執行，新版可以並行處理，大幅縮短回應時間。
5. Session 管理的重要性
將 Session 管理獨立成 SessionManager 服務帶來很多好處：
- ✅ 集中管理 - 所有 Agent 共用同一個 Session 狀態
- ✅ 自動清理 - 背景 Task 定期清理過期 Session
- ✅ Thread-safe - 使用 Lock 確保並發安全
- ✅ 可監控 -
/stats指令查看 Session 統計
版本發布記錄
| 版本 | 階段 | 主要內容 |
|---|---|---|
| v0.3.0 | Phase 1 | 安裝 ADK、重構 Tools |
| v0.4.0 | Phase 2 | ChatAgent + LINE 整合 |
| v0.5.0 | Phase 3 | 專業 Agents (Content, Location, Vision, GitHub) |
| v0.6.0 | Phase 4 | Orchestrator + A2A 路由 |
| v0.7.0 | Phase 5 | SessionManager + 背景清理 |
結論
從 if/elif 地獄到 Multi-Agent Orchestration，這次重構帶來的改變：
| 面向 | 改造前 | 改造後 |
|---|---|---|
| 程式碼組織 | 全部在 main.py | 分散到 agents/, tools/, services/ |
| 訊息路由 | if/elif 條件式 | Intent Detection + A2A |
| 並行處理 | 不支援 | asyncio.gather 並行執行 |
| Session 管理 | 散落各處 | 集中式 SessionManager |
| 可測試性 | 困難 | 每個 Agent 可獨立測試 |
| 可擴展性 | 加 elif | 加新 Agent |
如果你也在維護一個功能越來越多的 Bot，強烈建議考慮 ADK + A2A 架構。雖然初期投入較大，但長期維護成本會大幅降低。
參考資料
- Google Agent Development Kit (ADK) - 官方文件
- Vertex AI Gemini API - Gemini 模型文件
- linebot-helper-python Repository - 專案原始碼
- LINE Messaging API - LINE Bot 文件