Day13 行程管家進化中：讓 sub_Agent 各司其職的智慧管家

2025 iThome 鐵人賽

DAY 13

Mobile Development

結合AI Agent技術打造自己的行程管家系列第 13 篇

17th鐵人賽

wenlin1227

團隊nutc imac

2025-09-27 14:44:08

72 瀏覽

分享至

在上一篇文章中，我們已經完整完成了 Directions Agent 的設計與實作，讓行程管家具備了靈活而精準的路線規劃能力。
今天，我們將更進一步，帶大家認識並撰寫一個具備管理與協調能力的核心代理 —— sub_Agent 的總管 (itinerary_root_agent)。

這個「總管代理」不僅能依照使用者輸入的問題，即時判斷需求、智慧分流，還能自動安排底下的子代理（destination_agent、places_agent、geocode_agent）來回答。
換句話說，它就像是一個高效率的指揮中心，能根據不同場景快速派遣最適合的助手出馬，確保整個行程規劃過程有條不紊、準確高效、智慧全面。

sub_agent 的概念

在我們的行程管家架構中，每個 sub_agent（子代理）都是專精於某個領域的小助手，它們各自專注於單一任務，由 itinerary_root_agent（總管代理）來進行調度。

destination_agent 負責路線與目的地的規劃
places_agent 專精於搜尋與推薦附近的景點
geocode_agent 處理地理位置查詢，取得經緯度資訊

你可以把 sub_agent 想成行程管家底下的小組長，各自擁有不同專業；而總管代理就像是指揮中心，能根據使用者的問題，快速判斷並派遣最合適的子代理來處理。

這樣的設計有幾個關鍵優點：

模組化：每個子代理能獨立維護與升級，不會互相干擾。
可擴充：未來要新增新的子代理（例如天氣查詢、美食推薦）非常容易。
高效率：總管代理只需做決策與分派，讓整體系統更有彈性與智慧。

main 的程式碼

import os
import requests


from fastapi import FastAPI
from dotenv import load_dotenv
from typing import Optional
from pydantic import BaseModel
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from root.itinerary_root_agent import itinerary_root_agent
import json
from google.genai import types

app = FastAPI()
load_dotenv()
GOOGLE_MAPS_API_KEY = os.getenv("GOOGLE_MAPS_API_KEY")

session_service = InMemorySessionService()
artifacts_service = InMemoryArtifactService()



class LoginRequest(BaseModel):
    email: Optional[str] = None
    password: Optional[str] = None


@app.post("/itinerary_login")
def itinerary_login(login_request: LoginRequest):
    email = login_request.email
    password = login_request.password

    if email == "user@example.com" and password == "password":
        return {"message": "登入成功"}
    elif email != "user@example.com" or password != "password":
        return {"message": "帳號或密碼不正確"}
    elif not email or not password:
        return {"message": "請輸入帳號密碼"}
    else:
        return {"message": "登入失敗"}
    
    

class registerRequest(BaseModel):
    email: Optional[str] = None
    password: Optional[str] = None
    confirm_password: Optional[str] = None
    phone: Optional[str] = None

@app.post("/itinerary_register")
def itinerary_register(register_request: registerRequest):
    email = register_request.email
    password = register_request.password
    confirm_password = register_request.confirm_password
    phone = register_request.phone

    if not email or not password or not confirm_password or not phone:
        return {"message": "請輸入完整資料"}
    elif password != confirm_password:
        return {"message": "密碼與確認密碼不符"}
    elif email == "user@example.com":
        return {"message": "帳號已存在"}
    else:
        return {"message": "註冊成功"}
    

@app.post("/itinerary_googlemap_geocode")
async def itinerary_googlemap_geocode(city: str):
    # 使用 Google Maps Geocoding API 取得經緯度
    url = f"https://maps.googleapis.com/maps/api/geocode/json?address={city}&key={GOOGLE_MAPS_API_KEY}"
    response = requests.get(url)
    data = response.json()
    
    # 檢查 API 回傳狀態，並取得經緯度
    if data['status'] == 'OK':
        location = data['results'][0]['geometry']['location']
        lat = location['lat']
        lng = location['lng']

        # 呼叫 itinerary_root_agent
        query = f"請提供關於 {city} 的旅遊建議。"
        parts = [
            types.Part(text=query),
            types.Part(text=json.dumps({"city": city, "latitude": lat, "longitude": lng}))
        ]

        runner = Runner(
            app_name="itinerary_housekeeper",
            agent=itinerary_root_agent,
            artifact_service=artifacts_service,
            session_service=session_service,
        )

        session = await session_service.create_session(
            state={}, app_name="itinerary_housekeeper", user_id="user"
        )

        content = types.Content(role="user", parts=parts)
        # 執行 Agent
        events_async = runner.run_async(
            session_id=session.id, user_id="user", new_message=content
        )

        response = []
        async for event in events_async:
            if event.content:
                for part in event.content.parts:
                    if part.text:
                        response.append(part.text)
        result = "\n".join(response)

        return {
            "latitude": lat,
            "longitude": lng,
            "ai_result": result
        }
    else:
        return {"error": "無法取得地理位置"}


@app.post("/itinerary_googlemap_places")
async def itinerary_googlemap_places(lat: float, lng: float, place_type: str):
    # 使用 Google Maps Places API 取得附近景點
    url = f"https://maps.googleapis.com/maps/api/place/nearbysearch/json?location={lat},{lng}&radius=5000&type={place_type}&key={GOOGLE_MAPS_API_KEY}"
    response = requests.get(url)
    data = response.json()
    
    # 檢查 API 回傳狀態，並取得景點列表
    if data['status'] == 'OK':
        places = []
        for place in data['results']:
            places.append({
                "name": place['name'],
                "address": place.get('vicinity', '無地址資訊'),
                "rating": place.get('rating', '無評分資訊')
            })


    # 呼叫 itinerary_root_agent
        query = f"請提供關於經緯度 ({lat}, {lng}) 附近的 {place_type} 景點建議。"
        parts = [
            types.Part(text=query),
            types.Part(text=json.dumps({"latitude": lat, "longitude": lng, "place_type": place_type}))
        ]

        runner = Runner(
            app_name="itinerary_housekeeper",
            agent=itinerary_root_agent,
            artifact_service=artifacts_service,
            session_service=session_service,
        )

        session = await session_service.create_session(
            state={}, app_name="itinerary_housekeeper", user_id="user"
        )

        content = types.Content(role="user", parts=parts)
        # 執行 Agent
        events_async = runner.run_async(
            session_id=session.id, user_id="user", new_message=content
        )

        response = []
        async for event in events_async:
            if event.content:
                for part in event.content.parts:
                    if part.text:
                        response.append(part.text)
        result = "\n".join(response)

        return {"places": places, 
                "result": result}
    else:
        return {"error": "無法取得附近景點"}

@app.post("/itinerary_googlemap_directions")
async def itinerary_googlemap_directions(origin: str, destination: str):
    # 使用 Google Maps Directions API 取得路線規劃
    url = f"https://maps.googleapis.com/maps/api/directions/json?origin={origin}&destination={destination}&key={GOOGLE_MAPS_API_KEY}"
    response = requests.get(url)
    data = response.json()
    
    # 檢查 API 回傳狀態，並取得路線資訊
    if data['status'] == 'OK':
        route = data['routes'][0]['legs'][0]
        directions = {
            "start_address": route['start_address'],
            "end_address": route['end_address'],
            "distance": route['distance']['text'],
            "duration": route['duration']['text'],
            "steps": []
        }
        for step in route['steps']:
            directions['steps'].append({
                "instruction": step['html_instructions'],
                "distance": step['distance']['text'],
                "duration": step['duration']['text']
            })
            query = f"請提供從 {origin} 到 {destination} 的路線建議。"
            parts = [types.Part(text=query)]
            runner = Runner(
                app_name="itinerary_housekeeper",
                agent=itinerary_root_agent,
                artifact_service=artifacts_service,
                session_service=session_service,
            )
            session = await session_service.create_session(
                state={}, app_name="itinerary_housekeeper", user_id="user"
            )
            content = types.Content(role="user", parts=parts)
            # 執行 Agent
            events_async = runner.run_async(
                session_id=session.id, user_id="user", new_message=content
            )
            response = []
            async for event in events_async:
                if event.content:
                    for part in event.content.parts:
                        if part.text:
                            response.append(part.text)
            result = "\n".join(response)
        return {"directions": directions,
                "result": result}
    else:
        return {"error": "無法取得路線資訊"}

sub Agent的程式碼

import os

from dotenv import load_dotenv
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from googlemaps_destination.destination_agent import destination_agent
from googlemaps_places.places_agent import places_agent
from googleMaps_geocode.geocode_agent import geocode_agent

from . import prompt

# 載入.env檔
load_dotenv()


itinerary_root_agent = Agent(
    model=LiteLlm(
        model=os.getenv("GEMINI_API_MODEL"),  
        api_key=os.getenv("GOOGLE_API_KEY")
    ),
    description=prompt.ITINERARY_ROOT_AGENT_DESCRIPTION,
    instruction=prompt.ITINERARY_ROOT_AGENT_INSTRUCTION,
    sub_agents=[
        destination_agent,
        places_agent,
        geocode_agent,
    ]
)

prompt

ITINERARY_ROOT_AGENT_DESCRIPTION = """
    "這是一個行程管家代理。"
    "當用戶詢問行程地點規劃路線時，使用 destination_agent"
    "當用戶詢問搜尋景點或是推薦景點時，使用 places_agent。"
    "當用戶詢問地理位置相關問題時，使用 geocode_agent。"
    """
ITINERARY_ROOT_AGENT_INSTRUCTION = """
    你是一個行程管家代理，專門協助用戶規劃旅遊行程。你有三個子代理可以使用：
    1. destination_agent：用於規劃旅遊路線和目的地。
    2. places_agent：用於搜尋和推薦旅遊景點。
    3. geocode_agent：用於處理地理位置相關的查詢。
    當用戶提出問題時，請根據問題的性質選擇最合適的子代理來處理。
    你不能直接回答用戶的問題，必須將問題委派給適當的代理。
"""