「快與準」的完美交織：基於 Gemini Enterprise Agent Platform 的雙路徑旅遊搜尋系統實戰

# 1. 啟用 Google Cloud 必要 API
gcloud services enable \
  aiplatform.googleapis.com \
  storage.googleapis.com \
  documentai.googleapis.com

# 2. 建立儲存貯體並上傳旅遊文件
export PROJECT_ID="your-project-id"
export REGION="asia-northeast1"

gsutil mb -l $REGION gs://$PROJECT_ID-travel-docs
gsutil cp ./japan_travel_guide.pdf  gs://$PROJECT_ID-travel-docs/
gsutil cp ./ticket_prices_2026.pdf  gs://$PROJECT_ID-travel-docs/

import vertexai
from vertexai.preview import rag

# 初始化 Vertex AI 環境
vertexai.init(project="your-project-id", location="asia-northeast1")

# 設定 Embedding 模型 (使用 Google 最新 text-embedding-005)
embedding_model_config = rag.RagEmbeddingModelConfig(
    vertex_prediction_endpoint=rag.VertexPredictionEndpoint(
        publisher_model="publishers/google/models/text-embedding-005"
    )
)

# 建立全託管 Corpus
corpus = rag.create_corpus(
    display_name="japan-travel-knowledge-base",
    backend_config=rag.RagVectorDbConfig(
        rag_embedding_model_config=embedding_model_config
    ),
)
print(f"✓ 成功建立 Corpus: {corpus.name}")

# 從 GCS 匯入所有旅遊指南 (設定 chunk 大小與 overlap 確保上下文不中斷)
operation = rag.import_files(
    corpus.name,
    paths=[f"gs://your-project-id-travel-docs/"],
    chunk_size=512,
    chunk_overlap=100,
)
print(f"✓ 檔案匯入並索引完成。狀態: {operation.metadata}")

# 測試 Corpus 檢索效果
test_query = "東京迪士尼門票價格"
response = rag.retrieval_query(
    rag_resources=[rag.RagResource(rag_corpus=corpus.name)],
    text=test_query,
    similarity_top_k=3,
)

for i, chunk in enumerate(response.contexts.contexts):
    print(f"\n--- Chunk {i+1} (檢索分數: {chunk.score:.3f}) ---")
    print(chunk.text[:200] + "...")

import json
import asyncio
from google.adk.agents import LlmAgent
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.genai import types

# 制定 Intent Router 的 Prompt 規範，要求其輸出 JSON 格式
ROUTER_INSTRUCTION = """
你是旅遊查詢的分流器。分析使用者的問題，回傳 JSON（只回傳 JSON，不要加任何 markdown 標記或說明）：

{"route": "cag", "score": 0.0-0.3, "reason": "說明原因"} 
→ 適用於靜態知識查詢：單一票價、時刻表、地理距離、開放時間等固定資訊。

{"route": "agentic", "score": 0.7-1.0, "reason": "說明原因"} 
→ 適用於需要規劃、個人化、多步驟推理：行程規劃、綜合預算分配、排雷推薦等。
"""

intent_router = LlmAgent(
    model="gemini-2.0-flash-latest", # 使用 Flash 降低路由延遲至 50ms 內
    name="travel_intent_router",
    instruction=ROUTER_INSTRUCTION.strip(),
)

app = App(name="travel-router-app", root_agent=intent_router)
runner = InMemoryRunner(agent=intent_router, app_name="travel-router-app")

async def test_routing():
    queries = [
        "東京迪士尼一日門票多少錢？",
        "幫我規劃 5 天日本行程，預算 8 萬，我帶了 7 歲小孩，已去過京都"
    ]
    for q in queries:
        session = await runner.session_service.create_session(
            app_name="travel-router-app", user_id="test_user"
        )
        content = types.Content(role="user", parts=[types.Part(text=q)])
        
        # 串流讀取分流結果
        text_res = ""
        async for event in runner.run_async(
            user_id="test_user",
            session_id=session.id,
            new_message=content,
        ):
            if event.is_final_response():
                text_res = event.content.parts[0].text
                break
        
        result_data = json.loads(text_res.strip())
        print(f"\n問題: {q}")
        print(f"路由決策: {result_data['route']} (分數: {result_data['score']})")
        print(f"分流原因: {result_data['reason']}")

asyncio.run(test_routing())

from google.adk.agents import LlmAgent
from google.adk.apps import App
from google.adk.agents.context_cache_config import ContextCacheConfig
import time
import asyncio
from google.adk.runners import InMemoryRunner

# 建立 Path A 專屬的 Flash Agent
cag_agent = LlmAgent(
    model="gemini-2.0-flash-latest",
    name="travel_cag_agent",
    instruction="你是旅遊資訊查詢助手，專門回答靜態旅遊資訊。請利用內建 Corpus 回答。限制 100 字內，簡明扼要。",
)

# 載入 Context Caching 配置
cag_app = App(
    name="travel-cag-app",
    root_agent=cag_agent,
    context_cache_config=ContextCacheConfig(
        min_tokens=1024,      # 超過此門檻自動啟用快取
        ttl_seconds=3600,     # 快取生命週期 (1小時)
        cache_intervals=5,    # 每 5 次請求自動重新整理
    )
)

runner = InMemoryRunner(agent=cag_agent, app_name="travel-cag-app")

async def benchmark_latency():
    session = await runner.session_service.create_session(
        app_name="travel-cag-app", user_id="bench_user"
    )
    
    # 進行連續 3 次相同靜態查詢，觀察延遲變化
    for idx in range(3):
        t0 = time.time()
        result = await runner.run_async(
            user_id="bench_user",
            session_id=session.id,
            new_message="東京迪士尼樂園門票一日票多少錢？"
        )
        elapsed = (time.time() - t0) * 1000
        print(f"第 {idx+1} 次查詢耗時: {elapsed:.0f} ms")
        if idx == 0:
            print(f"回覆內容: {result.content.parts[0].text.strip()}\n")

asyncio.run(benchmark_latency())

from google.adk.agents import LlmAgent
from google.adk.memory import PreloadMemory
from google.adk.runners import Runner
from google.adk.memory import VertexAiMemoryBankService
from google.adk.sessions import VertexAiSessionService
from vertexai.preview import rag
import asyncio

# 1. 宣告 Agent 的 Tool 函數 (ADK 會自動進行 Schema 包裝)
def search_travel_knowledge(query: str, top_k: int = 6) -> str:
    """搜尋旅遊知識庫，取得相關景點與費用片段。"""
    response = rag.retrieval_query(
        rag_resources=[rag.RagResource(rag_corpus="japan-travel-knowledge-base")],
        text=query,
        similarity_top_k=top_k,
    )
    chunks = [ctx.text for ctx in response.contexts.contexts]
    return "\n---\n".join(chunks) if chunks else "未找到相關資料"

def calculate_budget(items: list[dict]) -> str:
    """計算行程總費用，並以 JSON 結構回傳明細與總額。
    items 結構範例: [{'name': '迪士尼門票', 'cost_twd': 2500}]
    """
    total = sum(item['cost_twd'] for item in items)
    breakdown = "\n".join([f"- {i['name']}: NT$ {i['cost_twd']}" for i in items])
    return f"預算驗證結果:\n{breakdown}\n\n總計費用: NT$ {total}"

# 2. 建立具備推理、驗證、記憶感知的 Agentic RAG Agent
agentic_agent = LlmAgent(
    model="gemini-2.0-pro-latest",
    name="travel_agentic_planner",
    instruction="""
你是專業的旅遊規劃師。規劃行程時必須嚴格執行以下流程：
1. 查看系統預載的使用者旅遊記憶 (PreloadMemory)，主動排除已造訪的目的地。
2. 呼叫 search_travel_knowledge 檢索景點與房價。
3. 呼叫 calculate_budget 驗證所有消費是否在使用者預算限制內。若超支，必須重新調整。
4. 最終回覆需包含：每日行程路線、預算花費明細、貼心提示。
    """.strip(),
    tools=[
        search_travel_knowledge,
        calculate_budget,
        PreloadMemory() # 自動載入長期記憶 Profile
    ]
)

# 3. 初始化長期記憶體服務 (Memory Bank)
# 注意：AGENT_ENGINE_ID 將於 Lab 5 部署後取得，此處使用 mock_id
ENGINE_ID = "mock-agent-engine-id"
memory_service = VertexAiMemoryBankService(
    project="your-project-id", location="asia-northeast1", agent_engine_id=ENGINE_ID
)
session_service = VertexAiSessionService(
    project="your-project-id", location="asia-northeast1", agent_engine_id=ENGINE_ID
)

# 建立連接長期記憶體與狀態機的 Runner
runner = Runner(
    agent=agentic_agent,
    app_name="travel-agentic-app",
    session_service=session_service,
    memory_service=memory_service,
)

async def seed_memory_and_run():
    user_id = "user_vip_99"
    
    # 模擬注入「過去對話所沉澱的長期記憶」
    await memory_service.add_fact_to_profile(
        user_id=user_id,
        fact="使用者在 2025/11 去過京都，且偏好自然與文化景點，討厭擁擠，孩子目前 7 歲。"
    )
    
    # 開啟新 Session 進行複雜查詢
    session = await session_service.create_session(
        app_name="travel-agentic-app", user_id=user_id
    )
    
    query = "幫我規劃 5 天日本行程，預算 8 萬台幣，我帶小孩去玩。請避開我之前去過的地方。"
    from google.genai import types
    content = types.Content(role="user", parts=[types.Part(text=query)])
    
    print("AI 開始進行多步思考、檢索與計算...")
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session.id,
        new_message=content,
    ):
        if event.is_final_response():
            print("\n=== 最終行程推薦 ===")
            print(event.content.parts[0].text)
            
    # 對話結束後，觸發非同步 Memory Bank 萃取
    await memory_service.add_session_to_memory(
        app_name="travel-agentic-app", user_id=user_id, session_id=session.id
    )
    print("\n✓ 對話結束，新事實已非同步沉澱至長期記憶體。")

asyncio.run(seed_memory_and_run())

import vertexai
from vertexai.Client import Client
from vertexai.agent_engines import AdkApp
from google.adk.apps import App

# 封裝 App
travel_app = App(
    name="travel-prod-app",
    root_agent=intent_router, # 以 Router 作為系統根節點
)
adk_app = AdkApp(agent=intent_router)

# 初始化託管平台客戶端並部署
client = Client(project="your-project-id", location="asia-northeast1")
remote_app = client.agent_engines.create(
    agent=adk_app,
    config={
        "requirements": [
            "google-cloud-aiplatform[agent_engines,adk]",
            "google-cloud-storage",
        ],
        "staging_bucket": "gs://your-project-id-staging",
        "display_name": "travel-search-prod",
    }
)
print(f"✓ 成功部署！資源名稱: {remote_app.resource_name}")

# 啟用 Gateway 與安全模組
gcloud vertex-ai agent-runtimes update travel-search-prod \
  --region=asia-northeast1 \
  --enable-agent-gateway \
  --model-armor-template=travel-safety-template