Evan Lin for Google Developer Experts

Posted on Jan 24 • Originally published at evanlin.com on Jan 24

LINE Bot Architecture Overhaul: From if/elif Hell to Multi-Agent Orchestration

#agents #architecture #python #softwareengineering

Image A	Image B

Background

When maintaining the linebot-helper-python project, I have been facing an architectural problem: as the features increase, the if/elif conditional statements in main.py become longer and longer, and the code becomes more and more difficult to maintain.

Original message routing logic:

# ❌ Old version - if/elif hell
async def handle_text_message(event):
    msg = event.message.text

    if msg.startswith('/clear'):
        # Handle clear command
    elif msg.startswith('/help'):
        # Handle help command
    elif msg == '@g':
        # Handle GitHub summary
    elif is_youtube_url(msg):
        # Handle YouTube summary
    elif extract_url(msg):
        # Handle general URL summary
    else:
        # Handle general conversation

This architecture has several obvious problems:

❌ Difficult to maintain - Adding a feature requires adding an elif, and the file becomes fatter.

❌ Difficult to test - All the logic is mixed together, making unit testing difficult.

❌ Difficult to extend - Want to process multiple intents in parallel? It's painful to change.

❌ Unclear responsibilities - Conversation, summary, and location search are all mixed together.

In early 2025, Google released the Agent Development Kit (ADK), which provides a framework for building Multi-Agent systems. I decided to refactor the entire project with ADK to implement an Agent-to-Agent (A2A) architecture.

This article records the complete migration process, from planning to implementation, in 5 stages.

What is ADK and A2A?

Google Agent Development Kit (ADK)

ADK is a Python framework provided by Google for building LLM-based Agent systems:

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) Communication

A2A is an architectural pattern that allows multiple specialized Agents to collaborate on complex tasks:

User: "幫我找台北好吃的拉麵，然後摘要這篇文章 https://..."
                    │
                    ▼
          ┌─────────────────────┐
          │ Orchestrator Agent │
          │ (意圖解析 + 路由) │
          └─────────────────────┘
                    │
      ┌─────────────┴─────────────┐
      │ Parallel Dispatch (A2A) │
      ▼ ▼
┌──────────────┐ ┌──────────────┐
│LocationAgent │ │ContentAgent │
│ Task: 找拉麵 │ │ Task: 摘要URL│
└──────────────┘ └──────────────┘
      │ │
      └──────────────┬───────────┘
                     ▼
               彙整回覆給用戶

Architecture Analysis Before Migration

Summary of Existing Architecture

┌──────────────┬────────────────────────────────────────────────────┐
│ Item │ Current Status │
├──────────────┼────────────────────────────────────────────────────┤
│ Framework │ FastAPI + LINE Bot SDK │
├──────────────┼────────────────────────────────────────────────────┤
│ AI Engine │ Google Vertex AI (google-genai SDK) │
├──────────────┼────────────────────────────────────────────────────┤
│ Session Management │ Self-built ChatSessionManager (in memory) │
├──────────────┼────────────────────────────────────────────────────┤
│ Message Routing │ if/elif conditional statements │
├──────────────┼────────────────────────────────────────────────────┤
│ Tool Integration │ Grounding (Google Search), Maps, YouTube, Web Scraping │
└──────────────┴────────────────────────────────────────────────────┘

Feasibility Assessment Conclusion

✅ Fully feasible - The project already uses Google Vertex AI and google-genai SDK, which is the foundation of ADK. The conversion requires refactoring the architecture rather than rewriting the logic.

Target Architecture Design

Agent Splitting Planning

┌─────────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ (Receives LINE messages, decides which expert Agent to route to) │
└─────────────────────────────────────────────────────────────┘
                              │
         ┌────────────────────┼────────────────────┐
         │ │ │
         ▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Chat Agent │ │ Content Agent │ │ Location Agent │
│ (Conversational Q&A) │ │ (Content Summary) │ │ (Location Search) │
│ │ │ │ │ │
│ Tools: │ │ Tools: │ │ Tools: │
│ - GoogleSearch │ │ - URLLoader │ │ - MapsGrounding│
│ - Grounding │ │ - YouTubeFetch │ │ - PlaceSearch │
└─────────────────┘ │ - PDFLoader │ └─────────────────┘
                     │ - Summarizer │
                     └─────────────────┘
                              │
                     ┌────────┴────────┐
                     ▼ ▼
              ┌───────────┐ ┌───────────┐
              │ Vision │ │ GitHub │
              │ Agent │ │ Agent │
              └───────────┘ └───────────┘

Agent Responsibility Comparison Table

┌───────────────────────────────────┬──────────────────────────┬───────────────────────────────────────┐
│ Existing Module │ Transformed Agent │ Responsibility │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_text_message() │ ChatAgent │ Handle general conversation, Google Search Grounding │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_url_message() + load_url() │ ContentAgent │ URL content retrieval and summary │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ load_transcript_from_youtube() │ ContentAgent │ YouTube video summary │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_location_message() │ LocationAgent │ Location search and recommendation │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_image_message() │ VisionAgent │ Image analysis │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ handle_github_summary() │ GitHubAgent │ GitHub Issues summary │
├───────────────────────────────────┼──────────────────────────┼───────────────────────────────────────┤
│ ChatSessionManager │ SessionManager │ Cross-Agent conversation memory │
└───────────────────────────────────┴──────────────────────────┴───────────────────────────────────────┘

Target Project Structure

linebot-helper-python/
├── main.py # FastAPI + LINE webhook (保留)
├── agents/
│ ├── __init__.py
│ ├── orchestrator.py # Main control Agent (routing decision)
│ ├── chat_agent.py # Conversation Agent
│ ├── content_agent.py # Content summary Agent
│ ├── location_agent.py # Location search Agent
│ ├── vision_agent.py # Image analysis Agent
│ └── github_agent.py # GitHub Agent
├── tools/
│ ├── __init__.py
│ ├── url_loader.py # Refactored from loader/url.py
│ ├── youtube_tool.py # Refactored from loader/youtube_gcp.py
│ ├── maps_tool.py # Refactored from loader/maps_grounding.py
│ ├── summarizer.py # Refactored from loader/langtools.py
│ └── pdf_tool.py # Refactored from loader/pdf.py
├── services/
│ ├── session_manager.py # Session management service
│ └── line_service.py # LINE API encapsulation
├── config/
│ └── agent_config.py # Agent settings and Model selection
└── loader/ # Retain the old version for compatibility

Migration Plan: 5 Stages

┌─────────┬─────────────────────────────────────────────────────────────────┬──────────┐
│ Stage │ Work Content │ Estimated Risk │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 1 │ Install ADK, create tools/ directory, refactor loader/*.py to ADK Tool format │ Low Risk │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 2 │ Create ChatAgent, verify integration with LINE webhook │ Medium Risk │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 3 │ Create other specialized Agents (Content, Location, Vision, GitHub) │ Medium Risk │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 4 │ Create Orchestrator, implement A2A routing logic │ High Risk │
├─────────┼─────────────────────────────────────────────────────────────────┼──────────┤
│ Phase 5 │ Migrate Session management, test complete process │ Medium Risk │
└─────────┴─────────────────────────────────────────────────────────────────┴──────────┘

Phase 1: Install ADK and Refactor Tools

Goal

Refactor the existing loader/ module to the ADK-compatible Tool format.

Implementation Content

1. Install ADK Dependencies

# requirements.txt
google-adk>=1.0.0

2. Create tools/ Directory Structure

# 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
    """
    # Implement summarization logic...
    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
    """
    # Implement image analysis logic...
    return {"status": "success", "analysis": result}

3. Refactored Tool Modules

Original Module	New Tool Module	Function
`loader/langtools.py`	`tools/summarizer.py`	Text summary, image analysis
`loader/youtube_gcp.py`	`tools/youtube_tool.py`	YouTube video summary
`loader/maps_grounding.py`	`tools/maps_tool.py`	Location search
`loader/pdf.py`	`tools/pdf_tool.py`	PDF processing
`loader/url.py`	`tools/url_loader.py`	URL content retrieval

Results

PR #10 merged
Release v0.3.0

Phase 2: Create ChatAgent and LINE Integration

Goal

Create the first ADK Agent, handle general conversation, and integrate Google Search Grounding.

Implementation Content

1. Create Agent Configuration

# 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. Create 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. Create LINE Service Encapsulation

# 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}時發生錯誤，請稍後再試"

Results

PR #11 merged
Release v0.4.0

Phase 3: Create Specialized Agents

Goal

Create ContentAgent, LocationAgent, VisionAgent, and GitHubAgent.

Implementation Content

1. ContentAgent - Content Summary

# 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 - Location Search

# 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 - Image Analysis

# 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 Integration

# 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}

Results

PR #12 merged
Release v0.5.0

Phase 4: Create Orchestrator and A2A Routing

Goal

Create the main Orchestrator, implement intent detection, and Agent routing.

Implementation Content

1. Define Intent Types

# agents/orchestrator.py
class IntentType(Enum):
    """Types of user intents"""
    CHAT = "chat" # General conversation
    URL_SUMMARY = "url_summary" # URL summary
    YOUTUBE_SUMMARY = "youtube" # YouTube summary
    IMAGE_ANALYSIS = "image" # Image analysis
    LOCATION_SEARCH = "location" # Location search
    GITHUB_SUMMARY = "github" # GitHub operation
    COMMAND = "command" # System command
    UNKNOWN = "unknown"

2. Orchestrator Main Class

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 Routing Logic

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. Update main.py to Use 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))

Architecture Comparison

Before (if/elif hell):

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)

Results

PR #13 merged
Release v0.6.0

Phase 5: Session Management Migration

Goal

Create a centralized SessionManager, supporting TTL auto-expiration and background cleanup.

Implementation Content

1. SessionManager Class

# 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 Lifecycle Integration

# 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. Add /stats Command

# 在 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}"""
    }

Results

PR #14 merged
Release v0.7.0

Final Architecture

Complete A2A Process

LINE Message
     │
     ▼
┌─────────────────────────────────────────────────────────────┐
│ FastAPI Webhook │
│ (main.py) │
└─────────────────────────────────────────────────────────────┘
     │
     ▼
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ │
│ 1. detect_intents() - Analyze user intent │
│ 2. _route_intent() - Route to specialized Agent │
│ 3. format_response() - Aggregate replies │
└─────────────────────────────────────────────────────────────┘
     │
     ├─── 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 │
                    │ (Background Cleanup Task) │
                    └─────────────────┘

Supported Intents and Commands

Intent	Trigger Condition	Processing Agent
CHAT	General text message	ChatAgent
URL_SUMMARY	Contains HTTP URL	ContentAgent
YOUTUBE_SUMMARY	YouTube URL	ContentAgent
IMAGE_ANALYSIS	Image message	VisionAgent
LOCATION_SEARCH	Location message	LocationAgent
GITHUB_SUMMARY	`@g` command	GitHubAgent
COMMAND	`/clear`, `/help`, `/status`, `/stats`	Orchestrator directly handles

Development Experience

1. Phased Migration Reduces Risk

The most important decision in this migration is to proceed in 5 stages, and each stage is an independent PR:

Phase 1 (Low Risk) → Phase 2 (Medium Risk) → Phase 3 (Medium Risk) → Phase 4 (High Risk) → Phase 5 (Medium Risk)

Each stage can function normally after completion, and there will be no situation where "the system breaks down halfway through the changes."

2. Retain Compatibility with the Old Version

I deliberately retained the loader/ directory, so that the old code can still be used:

# New version tools/ internal calls to the old version 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}

The advantages of doing this are:

✅ Can migrate incrementally
✅ Can quickly roll back if problems arise
✅ No need to rewrite all the logic at once

3. Intent Detection vs LLM Routing

When designing the Orchestrator, I considered two options:

Option A: Rule-based Intent Detection (selected this option)

def detect_intents(self, message: str) -> List[Intent]:
    if self._youtube_pattern.match(message):
        return [Intent(IntentType.YOUTUBE_SUMMARY, ...)]

Option 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)

The reasons I chose Option A:

✅ Faster (no extra API calls required)
✅ Lower cost
✅ Predictable behavior
✅ Sufficient for LINE Bot's explicit input format

4. The Power of Parallel Execution

The most powerful aspect of the A2A architecture is parallel execution:

# User message: "Summarize https://... and find nearby restaurants"
intents = [
    Intent(IntentType.URL_SUMMARY, ...),
    Intent(IntentType.LOCATION_SEARCH, ...)
]

# Parallel execution!
tasks = [self._route_intent(user_id, intent) for intent in intents]
results = await asyncio.gather(*tasks)

The old architecture required sequential execution, while the new version can process in parallel, significantly reducing response time.

5. The Importance of Session Management

Making Session Management an independent SessionManager service brings many benefits:

✅ Centralized management - All Agents share the same Session state
✅ Automatic cleanup - Background Task periodically cleans up expired Sessions
✅ Thread-safe - Use Lock to ensure concurrency safety
✅ Monitorable - /stats command to view Session statistics

Version Release History

Version	Stage	Main Content
v0.3.0	Phase 1	Install ADK, refactor Tools
v0.4.0	Phase 2	ChatAgent + LINE Integration
v0.5.0	Phase 3	Specialized Agents (Content, Location, Vision, GitHub)
v0.6.0	Phase 4	Orchestrator + A2A Routing
v0.7.0	Phase 5	SessionManager + Background Cleanup

Conclusion

From if/elif hell to Multi-Agent Orchestration, the changes brought about by this refactoring:

Aspect	Before Refactoring	After Refactoring
Code Organization	All in main.py	Distributed to agents/, tools/, services/
Message Routing	if/elif conditional statements	Intent Detection + A2A
Parallel Processing	Not supported	asyncio.gather parallel execution
Session Management	Scattered everywhere	Centralized SessionManager
Testability	Difficult	Each Agent can be tested independently
Extensibility	Add elif	Add new Agent

If you are also maintaining a Bot with increasing features, it is strongly recommended to consider the ADK + A2A architecture. Although the initial investment is larger, the long-term maintenance cost will be significantly reduced.

References

Google Agent Development Kit (ADK) - Official Documentation
Vertex AI Gemini API - Gemini Model Documentation
linebot-helper-python Repository - Project Source Code
LINE Messaging API - LINE Bot Documentation