程式碼研究室簡介
1. 學習目標
歡迎！今天我們將踏上精彩的旅程。首先，我們來思考熱門的社群活動平台 InstaVibe。雖然這項功能很成功，但我們知道，對部分使用者來說，實際規劃團體活動可能會覺得麻煩。試想一下，你要先想辦法瞭解所有好友的興趣所在，然後篩選無窮無盡的活動或場地選項，最後再進行協調。這可要跳超多下！這正是我們可以導入 AI，特別是智慧代理程式，以實現真正差異的領域。
這個概念是建立一個系統，讓這些服務方能處理繁重的工作，例如聰明地「傾聽」使用者和好友的偏好，然後主動推薦精彩的客製化活動。我們的目標是讓 InstaVibe 的社群媒體企劃功能變得更流暢、更實用。如要開始建構這些智慧助理，我們需要使用合適的工具奠定穩固的基礎。
您會看到以下概念：
使用 Google ADK 建立基礎知識：掌握使用 Google 代理開發套件 (ADK) 建構第一個智慧代理的基本知識。瞭解必要元件、代理程式生命週期，以及如何有效運用架構內建的工具。
透過 Model Context Protocol (MCP) 擴充代理功能：瞭解如何為代理提供自訂工具和上下文，讓代理執行專門工作並存取特定資訊。介紹 Model Context Protocol (MCP) 概念。您將瞭解如何設定 MCP 伺服器，以提供此內容。
設計代理互動與調度：除了單一代理外，還能瞭解代理調度。設計互動模式，從簡單的順序工作流程，到涉及迴圈、條件式邏輯和平行處理的複雜情境。介紹 ADK 架構中的子代理程式概念，以便管理模組化工作。
建構協作式多代理系統：瞭解如何建構系統，讓多個代理合作達成複雜目標。學習並實作代理與代理 (A2A) 通訊協定，為分散式代理 (可能在不同機器或服務上執行) 建立標準化方式，以便可靠地互動。
在 Google Cloud 上將代理程式推上生產環境：將代理程式應用程式從開發環境遷移至雲端。瞭解在 Google Cloud Platform (GCP) 上，架構及部署可擴充且穩健的多代理系統的最佳做法。瞭解如何運用 Cloud Run 等 GCP 服務，並探索最新版 Google Agent Engine 的功能，以便代管及管理您的代理程式。
2. 架構
運用 InstaVibe 進行 AI 輔助的社群媒體規劃
什麼是社群媒體觀測？
社群聆聽是指監控社群媒體、論壇和新聞網站等平台上的數位對話，瞭解使用者對某個主題、品牌或產業的看法。可提供有關大眾情緒、趨勢和使用者需求的寶貴洞察資料。在本工作坊中，我們會在以代理程式為基礎的系統中運用這項概念。
你是 InstaVibe 團隊的一員
假設您在「InstaVibe」工作，這是一家成功的新創公司，提供熱門的社群活動平台，鎖定年輕成人為目標對象。一切都很順利，但和許多科技公司一樣，您的團隊也面臨來自投資人的壓力，必須運用 AI 技術進行創新。您也發現內部有部分使用者參與度不如其他人，可能是因為他們不太願意發起團體活動，或是覺得規劃活動很困難。對貴公司而言，這表示這個重要使用者群組的平台黏著度降低。
你的團隊研究指出，AI 輔助功能可大幅改善這些使用者的體驗。這項功能的用意是根據使用者和朋友的興趣，主動建議相關活動，簡化規劃聚會活動的流程。你和同事面臨的問題是：AI 代理如何自動執行通常耗時的任務，例如發現興趣、研究活動，以及可能的初步協調？
以代理程式為基礎的解決方案 (原型概念)
您建議開發由多代理系統提供支援的原型功能。以下是概念分解：
- 社群剖析代理程式：這個代理程式採用社群聆聽技術，分析使用者的連結、互動，以及與使用者偏好相關的可能更廣泛的公開趨勢。目的是找出共同興趣和適合的活動特徵 (例如偏好安靜的聚會、特定興趣)。
- 活動規劃代理程式：這個代理程式會使用社群分析代理程式提供的洞察資料，搜尋符合指定條件 (例如地點、興趣) 的特定活動、場地或想法。
- 平台互動代理程式 (使用 MCP)：這個代理程式會從活動規劃代理程式取得最終計畫。其主要功能是利用預先定義的 MCP (Model Context Protocol) 工具，直接與 InstaVibe 平台互動。這項工具可讓服務專員草擬事件建議，並建立概述計畫的貼文。
- 調度器代理程式：此代理程式可做為中央調度器。接收 InstaVibe 平台的初始使用者要求，瞭解整體目標 (例如並依照邏輯順序將特定工作委派給適當的專業服務專員。它會管理服務專員之間的資訊流程，並確保最終結果會傳回給使用者。
主要架構元素和技術
Google Cloud Platform (GCP)：
- Vertex AI：
- Gemini 模型：提供 Gemini 等 Google 最先進的大型語言模型 (LLM)，強化 Google 服務機器人推理和決策能力。
- Vertex AI Agent Engine：這項代管服務可用於部署、代管及調度我們的調度器代理，簡化正式環境的複雜性，並抽象化基礎架構。
- Cloud Run：用於部署容器化應用程式的無伺服器平台。我們會用來：
- 主機代管 InstaVibe 的主要網頁應用程式。
- 將支援 A2A 的個別代理程式 (Planner、Social Profiling、Platform Interaction) 部署為獨立的微服務。
- 執行 MCP 工具伺服器，讓服務專員可以使用 InstaVibe 的內部 API。
- Spanner：全代管、全球分散式且具高度一致性的關聯資料庫。在本工作坊中，我們將利用 GRAPH DDL 和查詢功能，將其作為圖形資料庫使用，以便：
- 建立模型並儲存複雜的社交關係 (使用者、友誼、活動出席、貼文)。
- 讓 Social Profiling 服務專員能有效查詢這些關係。
- Artifact Registry：用於儲存、管理及保護容器映像檔的全代管服務。
- Cloud Build：可在 Google Cloud 執行建構作業的服務。我們會使用它自動從代理程式和應用程式原始碼建構 Docker 容器映像檔。
- Cloud Storage：Cloud Build 等服務會使用這項服務來儲存建構構件，而 Agent Engine 會根據其運作需求使用這項服務。
- 核心代理程式架構和通訊協定：
- Google 的 Agent Development Kit (ADK)：主要適用於下列項目：
- 為個別智慧代理程式定義核心邏輯、行為和指令集。
- 管理介面代理程式生命週期、狀態和記憶體 (短期工作階段狀態和可能的長期知識)。
- 整合代理人可用於與外界互動的工具 (例如 Google 搜尋或自訂工具)。
- 自動化調度管理多代理工作流程，包括依序、迴圈和平行執行子代理程式。
- 代理程式對代理程式 (A2A) 通訊協定：開放標準，可實現以下功能：
- 不同 AI 服務代理程式之間的直接、標準化通訊和協同作業，即使這些服務代理程式是以獨立服務或在不同機器上執行，也能達到這項效果。
- 代理可透過「代理卡」探索彼此的功能，並委派工作。這對於我們的調度器服務專員與專門的企劃書、社群媒體和平台服務專員互動至關重要。
- A2A Python 程式庫 (a2a-python)：用於讓 ADK 代理程式使用 A2A 通訊協定的具體程式庫。提供執行下列操作所需的伺服器端元件：
- 將我們的服務代理程式公開為符合 A2A 規範的伺服器。
- 自動處理「Agent Card」的探索功能。
- 接收並管理其他代理程式 (例如 Orchestrator) 傳入的工作要求。
- Model Context Protocol (MCP)：開放式標準，可讓代理程式執行以下操作：
- 以標準化方式連結及使用外部工具、資料來源和系統。
- 我們的平台互動代理程式會使用 MCP 用戶端與 MCP 伺服器進行通訊，進而公開工具，以便與 InstaVibe 平台的現有 API 互動。
- Google 的 Agent Development Kit (ADK)：主要適用於下列項目：
- 偵錯工具：
- A2A Inspector：A2A Inspector 是本工作坊中使用的網路偵錯工具，可用來連線至支援 A2A 的代理程式、檢查代理程式，以及與代理程式互動。雖然這不是最終的正式版架構，但卻是開發工作流程中不可或缺的一部分。提供以下功能：
- Agent Card Viewer：擷取並驗證介面的公開功能。
- 即時通訊介面：直接傳送訊息給已部署的服務專員，以便立即進行測試。
- 偵錯主控台：查看檢查器與代理程式之間交換的原始 JSON-RPC 訊息。
- A2A Inspector：A2A Inspector 是本工作坊中使用的網路偵錯工具，可用來連線至支援 A2A 的代理程式、檢查代理程式，以及與代理程式互動。雖然這不是最終的正式版架構，但卻是開發工作流程中不可或缺的一部分。提供以下功能：
- 語言模型 (LLM)：系統的「大腦」：
- Google 的 Gemini 模型：具體來說，我們會使用 gemini-2.0-flash 等版本。這些型號適用於：
- 進階推理與遵循指示：這類模型能夠理解複雜提示、遵循詳細指示，並推斷任務內容，因此適合用於協助服務專員做出決策。
- 使用工具 (函式呼叫)：Gemini 模型擅長判斷何時及如何使用透過 ADK 提供的工具，讓代理程式收集資訊或執行動作。
- 效率 (Flash 模型)：「flash」變化版本可兼顧效能和成本效益，適合用於許多需要快速回應的互動式服務工作。
- Google 的 Gemini 模型：具體來說，我們會使用 gemini-2.0-flash 等版本。這些型號適用於：
需要 Google Cloud 抵免額嗎？
3. 事前準備
👉 按一下 Google Cloud 控制台頂端的「啟用 Cloud Shell」圖示 (Cloud Shell 窗格頂端的終端形狀圖示)，
👉按一下「Open Editor」按鈕 (圖示為開啟的資料夾，內含鉛筆)。程式碼編輯器會在新視窗中開啟。左側會顯示檔案總管。
👉按一下底部狀態列中的「Cloud Code Sign-in」按鈕，如下圖所示。按照指示授權外掛程式。如果狀態列顯示「Cloud Code - no project」，請選取該項目，然後在下拉式選單中選取「Select a Google Cloud Project」，接著從您建立的專案清單中選取特定 Google Cloud 專案。
👉 找出您的 Google Cloud 專案 ID：
- 開啟 Google Cloud 控制台：https://console.cloud.google.com
- 在頁面頂端的專案下拉式選單中，選取要用於本工作坊的專案。
- 專案 ID 會顯示在資訊主頁的「專案資訊」資訊卡中
👉在雲端 IDE 中開啟終端機，
👉💻 在終端機中，使用下列指令確認您已完成驗證，且專案已設為專案 ID：
gcloud auth list
👉💻 從 GitHub 複製 instavibe-bootstrap 專案：
git clone -b adk-1.2.1-a2a-0.2.7 https://github.com/weimeilin79/instavibe-bootstrap.git
chmod +x ~/instavibe-bootstrap/init.sh
chmod +x ~/instavibe-bootstrap/set_env.sh
瞭解專案結構
開始建構前，請先花點時間瞭解您剛剛複製的 instavibe-bootstrap 專案版面配置。這樣您就能在工作坊期間找到檔案並進行編輯。
instavibe-bootstrap/
├── agents/
│ ├── orchestrate/
│ ├── planner/
│ ├── platform_mcp_client/
│ └── social/
├── instavibe/
│ ├── static/
│ └── templates/
├── tools/
│ └── instavibe/
├── utils/
├── init.sh
└── set_env.sh
以下是主要目錄的詳細說明：
agents/：這是 AI 系統的核心。每個子目錄 (planner/、social/ 等) 都包含特定智慧型代理程式的原始碼。agent.py：在每個代理程式資料夾中，這是代理程式邏輯的主要檔案。a2a_server.py：這個檔案會將 ADK 代理程式與代理程式對代理程式 (A2A) 伺服器包裝在一起。Dockerfile：定義如何建構容器映像檔，以便將代理程式部署至 Cloud Run 或 Agent Engine。
instavibe/：這個目錄包含 InstaVibe 網路應用程式的完整原始碼。tools/：這個目錄用於建構服務專員可使用的外部工具。instavibe/包含 Model Context Protocol (MCP) 伺服器。
這個模組結構可將網頁應用程式與各種 AI 元件分開，讓整個系統更容易管理、測試及部署。
👉💻 執行初始化指令碼：
這個指令碼會提示您輸入 Google Cloud 專案 ID。
系統在 init.sh 指令碼提示時，請輸入上一個步驟中找到的 Google Cloud 專案 ID：
cd ~/instavibe-bootstrap
./init.sh
👉💻 設定所需的專案 ID：
gcloud config set project $(cat ~/project_id.txt) --quiet
👉💻 執行下列指令，啟用必要的 Google Cloud API：
gcloud services enable run.googleapis.com \
cloudfunctions.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
spanner.googleapis.com \
apikeys.googleapis.com \
iam.googleapis.com \
compute.googleapis.com \
aiplatform.googleapis.com \
cloudresourcemanager.googleapis.com \
maps-backend.googleapis.com
👉💻 設定所有必要的環境變數：
export PROJECT_ID=$(gcloud config get project)
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export SERVICE_ACCOUNT_NAME=$(gcloud compute project-info describe --format="value(defaultServiceAccount)")
export SPANNER_INSTANCE_ID="instavibe-graph-instance"
export SPANNER_DATABASE_ID="graphdb"
export GOOGLE_CLOUD_PROJECT=$(gcloud config get project)
export GOOGLE_GENAI_USE_VERTEXAI=TRUE
export GOOGLE_CLOUD_LOCATION="us-central1"
設定權限
👉💻 授予權限。在終端機中執行：
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.admin"
# Spanner Database User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.databaseUser"
# Artifact Registry Admin
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/artifactregistry.admin"
# Cloud Build Editor
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/cloudbuild.builds.editor"
# Cloud Run Admin
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/run.admin"
# IAM Service Account User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/iam.serviceAccountUser"
# Vertex AI User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/aiplatform.user"
# Logging Writer (to allow writing logs)
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.logWriter"
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.viewer"
👉 在 IAM 主控台
中驗證結果
👉💻 在終端機中執行下列指令，建立 Artifact Registry 存放區。在部署至 Cloud Run 或 Agent Engine 之前，所有代理程式、MCP 伺服器和 InstaVibe 應用程式的 Docker 映像檔都會儲存在此處。
export REPO_NAME="introveally-repo"
gcloud artifacts repositories create $REPO_NAME \
--repository-format=docker \
--location=us-central1 \
--description="Docker repository for InstaVibe workshop"
設定地圖平台的 API 金鑰
如要在 InstaVibe 應用程式中使用 Google 地圖服務，您必須建立 API 金鑰並套用適當限制。
👉 在新的分頁中，依序前往「API 和服務」>「憑證」。在「憑證」頁面上，按一下頂端的「+ 建立憑證」按鈕。從下拉式選單中選取 API 金鑰。
👉 畫面上會顯示一個對話方塊，其中包含新建立的 API 金鑰。稍後會用到這項資訊來設定應用程式。
👉 在「建立的 API 金鑰」對話方塊中按一下「關閉」。
👉 您會看到新的 API 金鑰清單 (例如「API key 1」)。按一下右側的三橫線圖示，然後選取「編輯 API 金鑰」，即可開啟「限制並重新命名 API 金鑰」頁面。
👉 在頂端的「名稱」欄位中，將預設名稱變更為「Google 地圖平台 API 金鑰」 (🚨🚨重要事項🚨🚨：請使用這個名稱！)
Maps Platform API Key
👉 確認「應用程式限制」部分已選取「無」。
👉 在「API 限制」專區下方，選取「限制金鑰」圓形按鈕。
👉 點選「選取 API」下拉式選單。在隨即顯示的搜尋框中，輸入 Maps JavaScript API，然後從清單中選取。
👉 按一下「確定」。
👉 按一下頁面底部的「儲存」按鈕。
您現在已成功建立名為「Maps Platform API Key」的 API 金鑰，並限制只能使用「Maps JavaScript API」，確保專案已啟用該 API。
4. 設定圖譜資料庫
在建立智慧代理程式之前，我們需要一種方法來儲存及瞭解 InstaVibe 社群網路中的豐富連結。這時就需要圖形資料庫的協助。與傳統關聯資料庫不同，圖譜資料庫是專門用來以節點 (例如人物、事件或貼文) 和連結這些節點的關係 (例如友誼、活動出席或提及) 表示及查詢資料。這類結構對於社群媒體應用程式來說非常實用，因為它反映了現實世界中社群網路的結構，讓您能直觀地瞭解不同實體的連結方式。
我們正在使用 Google Cloud Spanner 實作這個圖形資料庫。雖然 Spanner 主要被視為全球分散式且具高度一致性的關聯資料庫，但我們也可以直接在關聯資料表上定義及查詢圖形結構。
這項功能結合了 Spanner 的彈性、交易一致性和熟悉的 SQL 介面，以及圖表查詢的強大功能，可用於分析 AI 技術輔助功能所需的複雜社交動態。
👉💻 在 Cloud Shell IDE 終端機中。在 Google Cloud 上佈建必要的基礎架構。我們會先建立 Spanner 執行個體，做為資料庫的專屬容器。執行個體準備就緒後，我們會在其中建立實際的 Spanner 資料庫，用於儲存 InstaVibe 的所有資料表和圖形資料：
. ~/instavibe-bootstrap/set_env.sh
gcloud spanner instances create $SPANNER_INSTANCE_ID \
--config=regional-us-central1 \
--description="GraphDB Instance InstaVibe" \
--processing-units=100 \
--edition=ENTERPRISE
gcloud spanner databases create $SPANNER_DATABASE_ID \
--instance=$SPANNER_INSTANCE_ID \
--database-dialect=GOOGLE_STANDARD_SQL
👉💻 將 Spanner 的讀取/寫入權限授予預設服務帳戶
echo "Granting Spanner read/write access to ${SERVICE_ACCOUNT_NAME} for database ${SPANNER_DATABASE_ID}..."
gcloud spanner databases add-iam-policy-binding ${SPANNER_DATABASE_ID} \
--instance=${SPANNER_INSTANCE_ID} \
--member="serviceAccount:${SERVICE_ACCOUNT_NAME}" \
--role="roles/spanner.databaseUser" \
--project=${PROJECT_ID}
👉💻 目前。我們會設定 Python 虛擬環境、安裝必要的 Python 套件，然後在 Spanner 中設定圖資料庫結構定義，並使用初始資料載入結構定義，然後執行 setup.py 指令碼。
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap
python -m venv env
source env/bin/activate
pip install -r requirements.txt
cd instavibe
python setup.py
👉 在新的瀏覽器分頁中前往 Google Cloud 控制台，然後前往「Spanner」，您應該會看到 Spanner 執行個體的清單。按一下 instavibe-graph-instance。
👉 在執行個體總覽頁面中，您會看到該執行個體內的資料庫清單。按一下 graphdb
👉 在資料庫的左側導覽窗格中，按一下 Spanner Studio 
👉 在查詢編輯器 (「未命名的查詢」分頁) 中，貼上下列 Graph SQL 查詢。這項查詢會找出所有 Person 節點，以及這些節點與其他 Person 節點的直接友誼關係。然後按一下「執行」即可查看結果。
Graph SocialGraph
MATCH result_paths = ((p:Person)-[f:Friendship]-(friend:Person))
RETURN SAFE_TO_JSON(result_paths) AS result_paths
👉 在同一個查詢編輯器中，取代先前的 DDL，找出參加同一活動的使用者，這表示透過共同活動間接連結。
Graph SocialGraph
MATCH result_paths = (p1:Person)-[:Attended]->(e:Event)<-[:Attended]-(p2:Person)
WHERE p1.person_id < p2.person_id
RETURN SAFE_TO_JSON(result_paths) AS result_paths
👉 這個查詢會探索不同類型的連結，也就是在特定使用者朋友撰寫的貼文中提及的使用者，請在查詢編輯器中執行下列查詢。
Graph SocialGraph
MATCH result_paths = (user:Person {name: "Alice"})-[:Friendship]-(friend:Person)-[:Wrote]->(post:Post)-[:Mentioned]->(mentioned_person:Person)
WHERE user <> mentioned_person AND friend <> mentioned_person -- Avoid self-mentions or friend mentioning themselves in their own post if not intended
RETURN SAFE_TO_JSON(result_paths) AS result_paths
這些查詢只是讓您一窺使用 Spanner 做為 InstaVibe 應用程式圖形資料庫的強大功能。我們將社群資料建模為相互連結的圖表，進而深入分析關係和活動，這對 AI 服務機器人來說至關重要，因為這有助於瞭解使用者情境、發現興趣，並最終提供智慧型社群活動規劃協助。
基礎資料結構已就位並通過測試，現在讓我們將注意力轉移到服務專員將與之互動的現有 InstaVibe 應用程式。
5. InstaVibe 的目前狀態
為了瞭解 AI 代理程式可用於何處，我們必須先部署並執行現有的 InstaVibe 網路應用程式。這個應用程式提供使用者介面和基本功能，可連結至我們已設定的 Spanner 圖表資料庫。
InstaVibe 應用程式會使用 Google 地圖，在活動詳細資料頁面上以視覺化方式顯示活動地點。如要啟用這項功能，應用程式需要先前建立的 API 金鑰。以下指令碼會使用我們指定的顯示名稱 (「Maps Platform API Key」) 擷取實際的金鑰字串。
👉💻 回到 Cloud Shell IDE。執行下列指令碼。接著，請仔細檢查輸出內容，確認顯示的 GOOGLE_MAPS_API_KEY 與先前從 Google Cloud 控制台建立及複製的金鑰相符。
. ~/instavibe-bootstrap/set_env.sh
export KEY_DISPLAY_NAME="Maps Platform API Key"
GOOGLE_MAPS_KEY_ID=$(gcloud services api-keys list \
--project="${PROJECT_ID}" \
--filter="displayName='${KEY_DISPLAY_NAME}'" \
--format="value(uid)" \
--limit=1)
GOOGLE_MAPS_API_KEY=$(gcloud services api-keys get-key-string "${GOOGLE_MAPS_KEY_ID}" \
--project="${PROJECT_ID}" \
--format="value(keyString)")
echo "${GOOGLE_MAPS_API_KEY}" > ~/mapkey.txt
echo "Retrieved GOOGLE_MAPS_API_KEY: ${GOOGLE_MAPS_API_KEY}"
👉💻 接下來，我們將為 InstaVibe 網頁應用程式建構容器映像檔，並推送至 Artifact Registry 存放區。
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
👉💻 將新建構的 InstaVibe 網頁應用程式映像檔部署至 Cloud Run
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="SPANNER_INSTANCE_ID=${SPANNER_INSTANCE_ID}" \
--set-env-vars="SPANNER_DATABASE_ID=${SPANNER_DATABASE_ID}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_MAPS_API_KEY=${GOOGLE_MAPS_API_KEY}" \
--project=${PROJECT_ID} \
--min-instances=1
部署作業完成後，Cloud Run 記錄應會顯示執行中的 InstaVibe 應用程式公開網址。
您也可以前往 Google Cloud 控制台的「Cloud Run」專區，然後選取 instavibe 服務，即可找到這個網址。
請在網路瀏覽器中開啟該網址，探索 InstaVibe 的基本平台。查看由我們設定的圖形資料庫所提供的貼文、活動和使用者連結。
目標應用程式已開始執行，現在我們要開始建構第一個智慧型代理，以強化應用程式功能。
6. 基本代理人、Event Planner 與 ADK
ADK 架構
Google ADK 架構簡介：現在基礎架構 (InstaVibe 應用程式和資料庫) 已設定完成，我們可以開始使用 Google 的Agent Development Kit (ADK) 建構第一個智慧代理程式。
Agent Development Kit (ADK) 是一個彈性十足的模組化框架，專門用於開發及部署 AI 代理。其設計原則是讓代理程式開發作業更像傳統軟體開發，讓開發人員能更輕鬆地建立、部署及調度代理架構，處理從簡單的單一用途工作到複雜的多代理工作流程。
ADK 的核心概念是 Agent，它封裝指令和設定 (例如所選的語言模型，例如Gemini)，以及一組可用於執行動作或收集資訊的 Tools。
我們的初始代理人會是「活動企劃人員」。其主要目的是接收使用者提出的社交活動要求 (指定地點、日期和興趣)，並產生創意且符合個人需求的建議。為確保建議內容與時俱進，並根據目前的資訊 (例如週末發生的特定事件) 提供相關資訊，我們會利用 ADK 內建的工具：Google 搜尋。這可讓服務機器人根據即時網頁搜尋結果回覆，擷取符合使用者條件的地點、活動和活動最新詳細資料。
👉📝 回到 Cloud Shell IDE，在 ~/instavibe-bootstrap/agents/planner/agent.py 中新增下列提示和指示，以建立服務項目
from google.adk.agents import Agent
from google.adk.tools import google_search
root_agent = Agent(
name="planner_agent",
model="gemini-2.0-flash",
description="Agent tasked with generating creative and fun dating plan suggestions",
instruction="""
You are a specialized AI assistant tasked with generating creative and fun plan suggestions.
Request:
For the upcoming weekend, specifically from **[START_DATE_YYYY-MM-DD]** to **[END_DATE_YYYY-MM-DD]**, in the location specified as **[TARGET_LOCATION_NAME_OR_CITY_STATE]** (if latitude/longitude are provided, use these: Lat: **[TARGET_LATITUDE]**, Lon: **[TARGET_LONGITUDE]**), please generate a distinct dating plan suggestions.
Constraints and Guidelines for Suggestions:
1. Creativity & Fun: Plans should be engaging, memorable, and offer a good experience for a date.
2. Budget: All generated plans should aim for a moderate budget (conceptually "$$"), meaning they should be affordable yet offer good value, without being overly cheap or extravagant. This budget level should be *reflected in the choice of activities and venues*, but **do not** explicitly state "Budget: $$" in the `plan_description`.
3. Interest Alignment:
Consider the following user interests: **[COMMA_SEPARATED_LIST_OF_INTERESTS, e.g., outdoors, arts & culture, foodie, nightlife, unique local events, live music, active/sports]**. Tailor suggestions specifically to these where possible. The plan should *embody* these interests.
Fallback: If specific events or venues perfectly matching all listed user interests cannot be found for the specified weekend, you should create a creative and fun generic dating plan that is still appealing, suitable for the location, and adheres to the moderate budget. This plan should still sound exciting and fun, even if it's more general.
4. Current & Specific: Prioritize finding specific, current events, festivals, pop-ups, or unique local venues operating or happening during the specified weekend dates. If exact current events cannot be found, suggest appealing evergreen options or implement the fallback generic plan.
5. Location Details: For each place or event mentioned within a plan, you MUST provide its name, precise latitude, precise longitude, and a brief, helpful description.
6. Maximum Activities: The plan must contain a maximum of 3 distinct activities.
RETURN PLAN in MARKDOWN FORMAT
""",
tools=[google_search]
)
這就是我們定義的第一個代理程式！ADK 最棒的優點之一，就是其直覺性和提供的實用工具。其中 ADK 開發人員 UI 特別實用，可讓您以互動方式測試代理程式，並即時查看回應。
👉💻 讓我們開始吧。下列指令會啟動 ADK DEV UI：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/planner/.env
adk web
執行指令後，終端機應會顯示類似以下的輸出內容，指出 ADK Web Server 已啟動：
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
👉 接下來，如要透過瀏覽器存取 ADK Dev UI，請按照下列步驟操作：
在 Cloud Shell 工具列 (通常位於右上方) 的「Web Preview」圖示 (通常看起來像眼睛或帶有箭頭的方塊) 中，選取「Change port」。在彈出式視窗中，將通訊埠設為 8000，然後按一下「變更並預覽」。接著，Cloud Shell 會開啟新的瀏覽器分頁或視窗，顯示 ADK Dev UI。
在瀏覽器中開啟 ADK Dev UI 後，請依序選取 UI 右上方的下拉式選單，然後選取「規劃工具」做為要互動的服務。接著，請在右側的即時通訊對話方塊中，嘗試指派工作給服務專員。舉例來說，您可以與服務專員進行對話：
Search and plan something in Seattle for me this weekend
This weekend and I enjoy food and anime
建議日期 (您的偏好設定)
July 12 2025
你應該會看到服務專員處理你的要求，並根據 Google 搜尋結果提供方案。
與服務專員互動是一回事，但我們要如何知道服務專員是否能持續正常運作，尤其是在我們進行變更時？
由於 AI 代理程式具有生成式和非決定性特性，因此傳統的軟體測試方法往往無法勝任。為了讓酷炫的示範影片轉變為可靠的正式版，必須採用完善的評估策略。與單純檢查生成式模型的最終輸出結果不同，評估代理程式通常會評估其決策過程，以及在各種情境下正確使用工具或遵循指示的能力。ADK 提供相關功能，可協助您完成這項工作。
👉 在 ADK 開發人員 UI 中，按一下左側導覽面板中的「Eval」分頁。您應該會看到預先載入的測試檔案，名稱為 plan_eval。這個檔案包含預先定義的輸入內容和測試企劃書代理程式的條件。
👉 選取情境 (例如「波士頓」)，然後按一下「執行評估」按鈕。在隨即顯示的彈出式視窗中，將比對分數調低至 0.3，然後按一下「開始」。
這麼做可讓服務專員執行測試輸入內容，並檢查輸出內容是否符合定義的預期結果。這樣一來，您就能有系統地測試代理程式的效能。
👉 接下來，我們來看看更嚴格的門檻會發生什麼情況。選取「紐約市」情境，然後再次按一下「Run Evaluation」。這次請將比對分數保留為預設值 (回應比對分數：0.7)，然後按一下「Start」(開始)。您會發現結果為「失敗」。這是預期結果，因為代理程式的創意輸出內容與預先定義的「黃金」答案不完全相符。
👉 如要瞭解失敗原因，請按一下「nyc」列中的失敗圖示。使用者介面現在會並排顯示代理人的實際回應和測試案例的預期回應。這個檢視畫面對於偵錯作業至關重要，可讓您準確查看代理程式輸出內容的差異，並據此調整指示。
完成探索 UI 和評估後，請返回 Cloud Shell 編輯器終端機，然後按下 Ctrl+C 停止 ADK Dev UI。
雖然自由格式文字輸出是個不錯的起點，但對於 InstaVibe 這類應用程式而言，要輕鬆使用 Google 助理建議，結構化資料 (例如 JSON) 會更實用。讓我們修改代理程式，以一致的 JSON 格式傳回其計畫。
👉📝 在 ~/instavibe-bootstrap/agents/planner/agent.py 中，找出代理程式指令字串中目前顯示 RETURN PLAN in MARKDOWN FORMAT 的行。將該行替換為以下詳細 JSON 結構：
Return your response *exclusively* as a single JSON object. This object should contain a top-level key, "fun_plans", which holds a plan objects. Each plan object in the list must strictly adhere to the following structure:
--json--
{
"plan_description": "A summary of the overall plan, consisting of **exactly three sentences**. Craft these sentences in a friendly, enthusiastic, and conversational tone, as if you're suggesting this awesome idea to a close friend. Make it sound exciting and personal, highlighting the positive aspects and appeal of the plan without explicitly mentioning budget or listing interest categories.",
"locations_and_activities": [
{
"name": "Name of the specific place or event",
"latitude": 0.000000, // Replace with actual latitude
"longitude": 0.000000, // Replace with actual longitude
"description": "A brief description of this place/event, why it's suitable for the date, and any specific details for the weekend (e.g., opening hours, event time)."
}
// Add more location/activity objects here if the plan involves multiple stops/parts
]
}
您已更新代理程式指示，明確要求 JSON 輸出內容，現在讓我們驗證這項變更。
👉💻 使用與先前相同的指令重新啟動 ADK Dev UI：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
adk web
如果您已開啟分頁，請重新整理分頁。或者，請按照先前的步驟在瀏覽器中開啟 ADK Dev UI (透過 Cloud Shell 的網頁預覽功能，在通訊埠 8000 上執行)。載入使用者介面後，請確認已選取企劃書代理程式。
👉 這次，我們來嘗試其他要求。在即時通訊對話方塊中輸入：
Plan an event Boston this weekend with art and coffee
仔細查看服務專員的回覆。您現在應該會看到回應的格式嚴格符合我們在操作說明中定義的結構 (包含 fun_plans、plan_description、locations_and_activities 等)，而非純粹的對話文字回覆。這表示代理程式現在可以產生結構化輸出內容，適合由 InstaVibe 應用程式以程式輔助方式使用。
確認 JSON 輸出內容後，請返回 Cloud Shell 終端機，然後按下 Ctrl+C 停止 ADK Dev UI。
ADK 元件
雖然 ADK 開發人員介面非常適合進行互動式測試，但我們通常需要透過程式碼執行我們的代理程式，例如作為大型應用程式或後端服務的一部分。為瞭解這項功能的運作方式，我們將介紹一些與執行階段和內容管理相關的核心 ADK 概念。
為了進行有意義的多回合對話，服務專員必須瞭解脈絡，也就是回想先前說過的話和所做的事，以便維持對話連貫性。ADK 提供結構化方式，可透過工作階段、狀態和記憶體管理此背景資訊：
- 工作階段：當使用者開始與服務專員互動時，系統就會建立工作階段。可將其視為單一特定聊天串流的容器。它會保留專屬 ID、互動記錄 (事件)、目前的工作資料 (狀態)，以及上次更新時間等中繼資料。
- 狀態：這是在單一工作階段中，代理程式短期的運作記憶體。這是可變動的字典，可讓服務機器人儲存完成目前工作所需的臨時資訊，例如目前收集到的使用者偏好設定，以及工具呼叫的中間結果。
- 記憶：這項指標代表服務機器人是否能在不同工作階段中長期喚回資訊，或存取外部知識庫。工作階段和狀態會處理即時對話，而記憶 (通常由 MemoryService 管理) 則可讓服務機器人從過往互動或結構化資料來源擷取資訊，提供更廣泛的知識脈絡。(注意：為了簡單起見，我們的簡易用戶端會使用記憶體服務，也就是說，記憶體/狀態只會在指令碼執行時保留)。
- 事件：工作階段中的每項互動 (使用者訊息、服務專員回應、工具使用要求、工具結果、狀態變更、錯誤) 都會記錄為不可變更的事件。這會建立按時間排序的記錄，基本上就是對話的轉錄稿和操作記錄。
那麼，在代理程式執行時，這些項目會如何管理？這就是執行程式的工作。
- Runner：Runner 是 ADK 提供的核心執行引擎。您定義代理程式和所用工具，Runner 就會協調執行使用者要求的程序。它會管理工作階段、處理事件流程、更新狀態、叫用基礎語言模型、協調工具呼叫，並可能與 MemoryService 互動。您可以將其視為指揮家，確保所有不同部分都能正確運作。
我們可以使用 Runner 將代理程式當成獨立的 Python 應用程式執行，完全不受開發人員使用者介面影響。
我們來建立簡單的用戶端指令碼，以程式設計方式叫用企劃書代理程式。
👉📝 在檔案 ~/instavibe-bootstrap/agents/planner/planner_client.py 中，在現有匯入項目下方新增下列 Python 程式碼。在 planner_client.py 的匯入內容下方新增以下內容：
async def async_main():
session_service = InMemorySessionService()
session = await session_service.create_session(
state={}, app_name='planner_app', user_id='user_dc'
)
query = "Plan Something for me in San Francisco this weekend on wine and fashion "
print(f"User Query: '{query}'")
content = types.Content(role='user', parts=[types.Part(text=query)])
root_agent = agent.root_agent
runner = Runner(
app_name='planner_app',
agent=root_agent,
session_service=session_service,
)
print("Running agent...")
events_async = runner.run_async(
session_id=session.id, user_id=session.user_id, new_message=content
)
async for event in events_async:
print(f"Event received: {event}")
if __name__ == '__main__':
try:
asyncio.run(async_main())
except Exception as e:
print(f"An error occurred: {e}")
這段程式碼會設定用於管理工作階段和構件內容的記憶體服務 (為了簡化範例，我們不採用記憶體服務)，建立工作階段、定義使用者查詢，並使用我們的代理程式設定 Runner，然後以非同步方式執行代理程式，並列印執行期間產生的每個事件。
👉💻 接著，請從終端機執行這個用戶端指令碼：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
python -m planner.planner_client
👀 查看輸出內容。您會看到代理程式執行流程中產生的每個事件物件詳細結構，而非僅顯示最終 JSON 企劃書。包括初始使用者訊息事件、與工具呼叫 (例如 Google 搜尋) 相關的潛在事件，以及包含 JSON 計畫的模型回應事件。這個詳細的事件串流非常適合用於偵錯，以及瞭解 ADK 執行階段內的逐步處理程序。
Running agent...
Event received: content=Content(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='```json\n{\n "fun_plans": [\n {\n "plan_description": "Embark on a stylish adventure through Hayes Valley,
...(turncated)
, offering a variety of fashion styles to browse and enjoy."\n }\n ]\n }\n ]\n}\n```')], role='model') grounding_metadata=GroundingMetadata(grounding_chunks=[GroundingChunk(retrieved_context=None, web=GroundingChunkWeb(domain='islands.com', title='islands.com', uri='http
...(turncated)
QyTpPV7jS6wUt-Ix7GuP2mC9J4eY_8Km6Vv44liF9cb2VSs='))], grounding_supports=[GroundingSupport(confide
...(turncated)
>\n', sdk_blob=None), web_search_queries=['..e']) partial=None turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None invocation_id='e-04d97b8b-9021-47a5-ab41-17b5cbb4bf03' author='location_search_agent' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=None branch=None id='CInHdkKw' timestamp=1746978846.232674
如果指令碼持續執行或停止運作，您可能需要按下 Ctrl+C 手動停止。
7. 平台互動代理程式 - 與 MCP 伺服器互動
雖然 ADK 可協助我們建構服務機器人，但服務機器人通常需要與外部系統或 API 互動，才能執行實際動作。
Model Context Protocol (MCP)
Model Context Protocol (MCP) 是開放標準，旨在將 AI 應用程式 (例如代理程式) 連結至外部資料來源、工具和系統的做法標準化。其目的是提供通用介面，解決每個 AI 應用程式和資料來源組合都需要自訂整合的情況。MCP 採用用戶端-伺服器架構，其中 MCP 用戶端位於 AI 應用程式 (主機) 中，負責管理與 MCP 伺服器的連線。這些伺服器是外部程式，可提供特定功能，例如存取本機資料、透過 API 與遠端服務互動，或提供預先定義的提示，讓 AI 模型存取目前資訊，並執行初始訓練以外的任務。這個結構可讓 AI 模型以標準化方式探索及與外部功能互動，讓整合作業更簡單且更具擴充性。
建構及部署 InstaVibe MCP 伺服器
我們的服務專員最終需要與 InstaVibe 平台本身互動，具體來說，就是使用平台現有的 API 建立貼文和註冊事件。InstaVibe 應用程式已透過標準 HTTP 端點公開這些功能：
Enpoint | 網址 | HTTP 方法 | 說明 |
建立訊息 | api/posts | POST | 新增貼文的 API 端點。預期 JSON 主體： |
建立活動 | api/events | POST | 新增活動和參與者的 API 端點 (簡化版結構定義)。 |
為了透過 MCP 為我們的服務機器人提供這些功能，我們必須先建立簡單的 Python 函式，做為這些 API 呼叫的包裝函式。這些函式會處理 HTTP 要求邏輯。
👉 首先，我們要實作用於建立貼文的包裝函式。開啟檔案 ~/instavibe-bootstrap/tools/instavibe/instavibe.py，並將 #REPLACE ME CREATE POST 註解替換為以下 Python 程式碼：
def create_post(author_name: str, text: str, sentiment: str, base_url: str = BASE_URL):
"""
Sends a POST request to the /posts endpoint to create a new post.
Args:
author_name (str): The name of the post's author.
text (str): The content of the post.
sentiment (str): The sentiment associated with the post (e.g., 'positive', 'negative', 'neutral').
base_url (str, optional): The base URL of the API. Defaults to BASE_URL.
Returns:
dict: The JSON response from the API if the request is successful.
Returns None if an error occurs.
Raises:
requests.exceptions.RequestException: If there's an issue with the network request (e.g., connection error, timeout).
"""
url = f"{base_url}/posts"
headers = {"Content-Type": "application/json"}
payload = {
"author_name": author_name,
"text": text,
"sentiment": sentiment
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
print(f"Successfully created post. Status Code: {response.status_code}")
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error creating post: {e}")
# Optionally re-raise the exception if the caller needs to handle it
# raise e
return None
except json.JSONDecodeError:
print(f"Error decoding JSON response from {url}. Response text: {response.text}")
return None
👉📝 接下來，我們會為事件建立 API 建立包裝函式。在同一個 ~/instavibe-bootstrap/tools/instavibe/instavibe.py 檔案中，將 #REPLACE ME CREATE EVENTS 註解替換為以下程式碼：
def create_event(event_name: str, description: str, event_date: str, locations: list, attendee_names: list[str], base_url: str = BASE_URL):
"""
Sends a POST request to the /events endpoint to create a new event registration.
Args:
event_name (str): The name of the event.
description (str): The detailed description of the event.
event_date (str): The date and time of the event (ISO 8601 format recommended, e.g., "2025-06-10T09:00:00Z").
locations (list): A list of location dictionaries. Each dictionary should contain:
'name' (str), 'description' (str, optional),
'latitude' (float), 'longitude' (float),
'address' (str, optional).
attendee_names (list[str]): A list of names of the people attending the event.
base_url (str, optional): The base URL of the API. Defaults to BASE_URL.
Returns:
dict: The JSON response from the API if the request is successful.
Returns None if an error occurs.
Raises:
requests.exceptions.RequestException: If there's an issue with the network request (e.g., connection error, timeout).
"""
url = f"{base_url}/events"
headers = {"Content-Type": "application/json"}
payload = {
"event_name": event_name,
"description": description,
"event_date": event_date,
"locations": locations,
"attendee_names": attendee_names,
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
print(f"Successfully created event registration. Status Code: {response.status_code}")
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error creating event registration: {e}")
# Optionally re-raise the exception if the caller needs to handle it
# raise e
return None
except json.JSONDecodeError:
print(f"Error decoding JSON response from {url}. Response text: {response.text}")
return None
如您所見，這些函式是圍繞現有 InstaVibe API 的簡單包裝函式。這個模式很實用，如果您已有服務的 API，只要建立這類包裝函式，就能輕鬆將這些功能公開為服務機器人的工具。
MCP 伺服器導入作業
有了執行動作的 Python 函式 (呼叫 InstaVibe API)，我們現在需要建構 MCP 伺服器元件。這項伺服器會根據 MCP 標準將這些函式公開為「工具」，讓 MCP 用戶端 (例如我們的代理程式) 能夠發現及叫用這些函式。
MCP 伺服器通常會實作兩項重要功能：
- list_tools：負責讓用戶端在伺服器上探索可用的工具，提供名稱、說明和必要參數等中繼資料，通常會使用 JSON 結構定義
- call_tool：處理執行用戶端要求的特定工具，接收工具名稱和引數，並執行對應動作，例如與 API 互動
MCP 伺服器可讓 AI 模型存取實際資料和動作，執行傳送電子郵件、在專案管理系統中建立工作、搜尋資料庫或與各種軟體和網路服務互動等工作。雖然為了簡化操作，初始導入作業通常會著重於透過標準 輸入/輸出 (stdio) 進行通訊的本機伺服器，尤其是在開發或「工作室」環境中，但如果要廣泛採用及用於企業用途，則應改用採用 HTTP 與伺服器推送事件 (SSE) 等通訊協定的遠端伺服器。
雖然遠端架構會新增網路通訊層，但仍有許多優點：可讓多個 AI 用戶端共用單一伺服器的存取權、集中管理工具和更新，並將機密資料和 API 金鑰保留在伺服器端，而非分散在多台用戶端電腦上，藉此提升安全性，此外，AI 模型與外部系統整合的具體細節會分開，因此整個生態系統可擴充、安全且可維護，不必要求每個 AI 例項管理自己的直接整合作業。
我們會使用 HTTP 和伺服器傳送事件 (SSE) 實作 MCP 伺服器，以便進行通訊，這非常適合用於可能長時間執行的工具執行作業和企業情境。
👉📝 首先，讓我們實作 list_tools 端點。開啟檔案 ~/instavibe-bootstrap/tools/instavibe/mcp_server.py，並以下列程式碼取代 #REPLACE ME - LIST TOOLS 註解。:
@app.list_tools()
async def list_tools() -> list[mcp_types.Tool]:
"""MCP handler to list available tools."""
# Convert the ADK tool's definition to MCP format
mcp_tool_schema_event = adk_to_mcp_tool_type(event_tool)
mcp_tool_schema_post = adk_to_mcp_tool_type(post_tool)
print(f"MCP Server: Received list_tools request. \n MCP Server: Advertising tool: {mcp_tool_schema_event.name} and {mcp_tool_schema_post}")
return [mcp_tool_schema_event,mcp_tool_schema_post]
這個函式會定義工具 (create_event、create_post)，並告知連線的用戶端。
👉📝 接下來，請實作 call_tool 端點，以便處理來自用戶端的實際執行要求。在同一個 ~/instavibe-bootstrap/tools/instavibe/mcp_server.py 檔案中，將 #REPLACE ME - CALL TOOLS 註解替換為以下程式碼。
@app.call_tool()
async def call_tool(
name: str, arguments: dict
) -> list[mcp_types.TextContent | mcp_types.ImageContent | mcp_types.EmbeddedResource]:
"""MCP handler to execute a tool call."""
print(f"MCP Server: Received call_tool request for '{name}' with args: {arguments}")
# Look up the tool by name in our dictionary
tool_to_call = available_tools.get(name)
if tool_to_call:
try:
adk_response = await tool_to_call.run_async(
args=arguments,
tool_context=None, # No ADK context available here
)
print(f"MCP Server: ADK tool '{name}' executed successfully.")
response_text = json.dumps(adk_response, indent=2)
return [mcp_types.TextContent(type="text", text=response_text)]
except Exception as e:
print(f"MCP Server: Error executing ADK tool '{name}': {e}")
# Creating a proper MCP error response might be more robust
error_text = json.dumps({"error": f"Failed to execute tool '{name}': {str(e)}"})
return [mcp_types.TextContent(type="text", text=error_text)]
else:
# Handle calls to unknown tools
print(f"MCP Server: Tool '{name}' not found.")
error_text = json.dumps({"error": f"Tool '{name}' not implemented."})
return [mcp_types.TextContent(type="text", text=error_text)]
這個函式會接收工具名稱和引數，找出我們先前定義的對應 Python 包裝函式，執行該函式，然後傳回結果
👉💻 定義 MCP 伺服器邏輯後，現在需要將其封裝為容器，請在終端機中執行下列指令碼，使用 Cloud Build 建構 Docker 映像檔：
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/tools/instavibe
export IMAGE_TAG="latest"
export MCP_IMAGE_NAME="mcp-tool-server"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${MCP_IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="mcp-tool-server"
export INSTAVIBE_BASE_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe)/api
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
👉💻 並將映像檔部署為 Google Cloud Run 上的服務。
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/tools/instavibe
export IMAGE_TAG="latest"
export MCP_IMAGE_NAME="mcp-tool-server"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${MCP_IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="mcp-tool-server"
export INSTAVIBE_BASE_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe)/api
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="INSTAVIBE_BASE_URL=${INSTAVIBE_BASE_URL}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=TRUE" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--project=${PROJECT_ID} \
--min-instances=1
👉💻 部署作業完成後，MCP 伺服器就會開始執行，並可透過公開網址存取。我們需要擷取這個網址，讓服務專員 (扮演 MCP 用戶端) 知道要連線至哪裡。
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
您現在應該也能在 Google Cloud 控制台的「Cloud Run」專區中，看到「Running」狀態的 mcp-tool-server 服務。
部署 MCP 伺服器並擷取其網址後，我們現在可以實作代理程式，讓其充當 MCP 用戶端，並利用此伺服器公開的工具。
8. 平台互動代理程式 (使用 MCP)
MCP 用戶端：MCP 用戶端是 AI 應用程式或代理程式中的元件，可做為 AI 模型與一或多個 MCP 伺服器之間的介面；在我們的實作中，這個用戶端會直接整合至代理程式。這個用戶端的主要功能是與 MCP 伺服器通訊，透過 list_tools 函式探索可用的工具，然後使用 call_tool 函式要求執行特定工具，並傳遞 AI 模型或協調呼叫的代理程式提供的必要引數。
接下來，我們將建構做為 MCP 用戶端的代理程式。這個在 ADK 架構中執行的服務專員，會負責與我們剛部署的 mcp-tool-server 進行通訊。
👉 首先，我們需要修改代理程式定義，以便動態擷取執行中的 MCP 伺服器工具。在 agents/platform_mcp_client/agent.py 中，將 #REPLACE ME - FETCH TOOLS 替換為以下內容：
"""Gets tools from the File System MCP Server."""
tools = MCPToolset(
connection_params=SseServerParams(url=MCP_SERVER_URL, headers={})
)
這個程式碼會使用 MCPToolset.from_server 方法連線至 MCP_SERVER_URL (我們先前已將其設為環境變數)，並擷取可用工具清單。
接下來，我們需要告訴 ADK 代理程式定義，實際使用這些動態擷取的工具。
👉 在 agents/platform_mcp_client/agent.py 中，將 #REPLACE ME - SET TOOLs 替換為以下內容：
tools=[tools],
👉💻 接下來，我們將使用 ADK 開發人員 UI 在本機測試這個代理程式，看看是否能正確連線至 MCP 伺服器，並使用工具與執行中的 InstaVibe 應用程式互動。
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/platform_mcp_client/.env
sed -i "s|^\(O\?MCP_SERVER_URL\)=.*|MCP_SERVER_URL=${MCP_SERVER_URL}|" ~/instavibe-bootstrap/agents/platform_mcp_client/.env
adk web
再次在瀏覽器中開啟 ADK Dev UI (使用 Cloud Shell 的網頁預覽功能，透過通訊埠 8000 連線)。這次請在右上方的下拉式選單中選取 platform_mcp_client 代理程式。
我們來測試 create_post 工具。在對話方塊中輸入以下要求：
Create a post saying "Y'all I just got the cutest lil void baby 😭✨ Naming him Abyss bc he's deep, mysterious, and lowkey chaotic 🔥🖤 #VoidCat #NewRoomie" I'm Julia
代理程式應處理這項作業，找出是否需要使用 create_post 工具，並與 MCP 伺服器進行通訊，而 MCP 伺服器會反過來呼叫 InstaVibe API。
👉 驗證步驟：服務專員確認動作後，請開啟 InstaVibe 應用程式執行所在的分頁 (或重新整理分頁)。你應該會在主動態消息中看到「Julia」發布的新貼文！
👉💻 如有需要，請在獨立的終端機中執行此指令碼，取得 Instavibe 連結：
gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe
👉📝 現在，讓我們測試 create_event 工具。在對話方塊中輸入以下多行要求：
Hey, can you set up an event for Hannah and George and me, and I'm Julia? Let's call it 'Mexico City Culinary & Art Day'.
here are more info
{"event_name": "Mexico City Culinary & Art Day",
"description": "A vibrant day in Mexico City for Hannah and George, starting with lunch at one of the city's best taco spots in the hip Condesa neighborhood, followed by an inspiring afternoon exploring the Museo Soumaya's stunning art collection.",
"event_date": "2025-10-17T12:00:00-06:00",
"locations": [
{
"name": "El Tizoncito",
"description": "Considered one of the original creators of tacos al pastor, El Tizoncito offers a legendary taco experience in the heart of Condesa. Their flavorful meats, house salsas, and casual vibe make it a must-visit for foodies.",
"latitude": 19.412179,
"longitude": -99.171308,
"address": "Av. Tamaulipas 122, Hipódromo, Cuauhtémoc, 06100 Ciudad de México, CDMX, Mexico"
},
{
"name": "Museo Soumaya",
"description": "An architectural icon in Mexico City, Museo Soumaya houses over 66,000 works of art, including pieces by Rodin, Dalí, and Rivera. The striking silver structure is a cultural landmark and a visual feast inside and out.",
"latitude": 19.440056,
"longitude": -99.204281,
"address": "Plaza Carso, Blvd. Miguel de Cervantes Saavedra 303, Granada, Miguel Hidalgo, 11529 Ciudad de México, CDMX, Mexico"
}
],
"attendee_names": ["Hannah", "George", Julia],
}
代理人應透過 MCP 伺服器使用適當的工具。您可以按一下「事件」分頁中的個別事件，查看執行作業的詳細步驟追蹤記錄。
👉 驗證步驟：返回執行中的 InstaVibe 應用程式，然後前往「Events」(或同等) 部分。您現在應該會看到新建立的「墨西哥城美食與藝術日」活動。
這成功證明瞭 MCP 如何讓我們的服務專員以標準化方式運用外部工具 (在本例中為 InstaVibe 的 API)。
確認這兩項操作後，請返回 Cloud Shell 終端機，然後按下 Ctrl+C 停止 ADK Dev UI。
9. ADK 中的 workflow agent 和多代理
目前我們的服務專員可以規劃外出活動，並與平台互動。不過，要想提供真正個人化的規劃，就必須瞭解使用者的社交圈。對於可能不會密切追蹤好友動態的忙碌使用者而言，手動收集這類背景資訊相當困難。為解決這個問題，我們將建構社交分析代理程式，利用 Spanner 圖譜資料庫分析好友的活動和興趣，提供更貼近使用者需求的建議。
首先，我們需要提供工具，讓這個代理程式能夠存取圖表資料。
👉📝 將下列 Python 函式新增至檔案 ~/instavibe-bootstrap/agents/social/instavibe.py 的結尾：
def get_person_attended_events(person_id: str)-> list[dict]:
"""
Fetches events attended by a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns: list[dict] or None.
"""
if not db_instance: return None
graph_sql = """
Graph SocialGraph
MATCH (p:Person)-[att:Attended]->(e:Event)
WHERE p.person_id = @person_id
RETURN e.event_id, e.name, e.event_date, att.attendance_time
ORDER BY e.event_date DESC
"""
params = {"person_id": person_id}
param_types_map = {"person_id": param_types.STRING}
fields = ["event_id", "name", "event_date", "attendance_time"]
results = run_graph_query( graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
if results is None: return None
for event in results:
if isinstance(event.get('event_date'), datetime):
event['event_date'] = event['event_date'].isoformat()
if isinstance(event.get('attendance_time'), datetime):
event['attendance_time'] = event['attendance_time'].isoformat()
return results
def get_person_id_by_name( name: str) -> str:
"""
Fetches the person_id for a given name using SQL.
Args:
name (str): The name of the person to search for.
Returns:
str or None: The person_id if found, otherwise None.
Returns the ID of the *first* match if names are duplicated.
"""
if not db_instance: return None
sql = """
SELECT person_id
FROM Person
WHERE name = @name
LIMIT 1 -- Return only the first match in case of duplicate names
"""
params = {"name": name}
param_types_map = {"name": param_types.STRING}
fields = ["person_id"]
# Use the standard SQL query helper
results = run_sql_query( sql, params=params, param_types=param_types_map, expected_fields=fields)
if results: # Check if the list is not empty
return results[0].get('person_id') # Return the ID from the first dictionary
else:
return None # Name not found
def get_person_posts( person_id: str)-> list[dict]:
"""
Fetches posts written by a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns:
list[dict] or None: List of post dictionaries with ISO date strings,
or None if an error occurs.
"""
if not db_instance: return None
# Graph Query: Find the specific Person node, follow 'Wrote' edge to Post nodes
graph_sql = """
Graph SocialGraph
MATCH (author:Person)-[w:Wrote]->(post:Post)
WHERE author.person_id = @person_id
RETURN post.post_id, post.author_id, post.text, post.sentiment, post.post_timestamp, author.name AS author_name
ORDER BY post.post_timestamp DESC
"""
# Parameters now include person_id and limit
params = {
"person_id": person_id
}
param_types_map = {
"person_id": param_types.STRING
}
# Fields returned remain the same
fields = ["post_id", "author_id", "text", "sentiment", "post_timestamp", "author_name"]
results = run_graph_query(graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
if results is None:
return None
# Convert datetime objects to ISO format strings
for post in results:
if isinstance(post.get('post_timestamp'), datetime):
post['post_timestamp'] = post['post_timestamp'].isoformat()
return results
def get_person_friends( person_id: str)-> list[dict]:
"""
Fetches friends for a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns: list[dict] or None.
"""
if not db_instance: return None
graph_sql = """
Graph SocialGraph
MATCH (p:Person {person_id: @person_id})-[f:Friendship]-(friend:Person)
RETURN DISTINCT friend.person_id, friend.name
ORDER BY friend.name
"""
params = {"person_id": person_id}
param_types_map = {"person_id": param_types.STRING}
fields = ["person_id", "name"]
results = run_graph_query( graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
return results
接下來，我們來討論如何建構服務代理。分析多位好友的個人資料，然後總結所得的結果，需要完成幾個步驟。這是使用 ADK 多代理功能 (特別是工作流程代理程式) 的理想情境。
在 Google ADK 中，工作流程代理程式不會自行執行工作，而是會協調其他代理程式 (稱為「子代理程式」)。這可實現模組化設計，將複雜問題分解為專門的元件。ADK 提供內建的工作流程類型，例如：
- 依序 (逐步)
- 並行 (同時執行)
- 和 Loop (重複執行)
在社群媒體分析任務中，我們的設計會使用迴圈代理程式建立迭代式工作流程。這項操作的目的是一一處理每位使用者：profile_agent 會收集資料、summary_agent 會更新分析結果，而 check_agent 會判斷是否應再次執行迴圈。
讓我們定義這個工作流程所需的子代理程式。
👉📝 在 ~/instavibe-bootstrap/agents/social/agent.py 中，將 #REPLACE FOR profile_agent 取代為以下內容：
profile_agent = LlmAgent(
name="profile_agent",
model="gemini-2.5-flash",
description=(
"Agent to answer questions about the this person social profile. Provide the person's profile using their name, make sure to fetch the id before getting other data."
),
instruction=(
"You are a helpful agent to answer questions about the this person social profile. You'll be given a list of names, provide the person's profile using their name, make sure to fetch the id before getting other data. Get one person at a time, start with the first one on the list, and skip if already provided. return this person's result"
),
tools=[get_person_posts,get_person_friends,get_person_id_by_name,get_person_attended_events],
)
接著，代理程式會採用收集到的個人資料資訊 (在迴圈迭代期間累積)，並產生最終摘要，找出多位使用者分析結果的共同點。
👉📝 在同一個 ~/instavibe-bootstrap/agents/social/agent.py 中，將 #REPLACE FOR summary_agent 替換為以下內容：
summary_agent = LlmAgent(
name="summary_agent",
model="gemini-2.5-flash",
description=(
"Generate a comprehensive social summary as a single, cohesive paragraph. This summary should cover the activities, posts, friend networks, and event participation of one or more individuals. If multiple profiles are analyzed, the paragraph must also identify and integrate any common ground found between them."
),
instruction=(
"""
Your primary task is to synthesize social profile information into a single, comprehensive paragraph.
**Input Scope & Default Behavior:**
* If specific individuals are named by the user, focus your analysis on them.
* **If no individuals are specified, or if the request is general, assume the user wants an analysis of *all relevant profiles available in the current dataset/context*.**
**For each profile (whether specified or determined by default), you must analyze:**
1. **Post Analysis:**
* Systematically review their posts (e.g., content, topics, frequency, engagement).
* Identify recurring themes, primary interests, and expressed sentiments.
2. **Friendship Relationship Analysis:**
* Examine their connections/friends list.
* Identify key relationships, mutual friends (especially if comparing multiple profiles), and the general structure of their social network.
3. **Event Participation Analysis:**
* Investigate their past (and if available, upcoming) event participation.
* Note the types of events, frequency of attendance, and any notable roles (e.g., organizer, speaker).
**Output Generation (Single Paragraph):**
* **Your entire output must be a single, cohesive summary paragraph.**
* **If analyzing a single profile:** This paragraph will detail their activities, interests, and social connections based on the post, friend, and event analysis.
* **If analyzing multiple profiles:** This paragraph will synthesize the key findings regarding posts, friends, and events for each individual. Crucially, it must then seamlessly integrate or conclude with an identification and description of the common ground found between them (e.g., shared interests from posts, overlapping event attendance, mutual friends). The aim is a unified narrative within this single paragraph.
**Key Considerations:**
* Base your summary strictly on the available data.
* If data for a specific category (posts, friends, events) is missing or sparse for a profile, you may briefly acknowledge this within the narrative if relevant.
"""
),
output_key="summary"
)
我們需要一種方法來判斷迴圈何時應停止 (也就是所有要求的設定檔都已匯總完成)
👉📝 在同一個 ~/instavibe-bootstrap/agents/social/agent.py 中，將 #REPLACE FOR check_agent 替換為以下內容：
check_agent = LlmAgent(
name="check_agent",
model="gemini-2.5-flash",
description=(
"Check if everyone's social profile are summarized and has been generated. Output 'completed' or 'pending'."
),
output_key="summary_status"
)
我們新增了簡單的程式輔助檢查 (CheckCondition)，明確查看 check_agent 傳回的 State 中儲存的 summary_status，並告知 Loop Agent 是否要繼續 (escalate=False) 或停止 (escalate=True)。
👉📝 在同一個 ~/instavibe-bootstrap/agents/social/agent.py 中，將檔案頂端的 #REPLACE FOR CheckCondition 替換為以下內容：
class CheckCondition(BaseAgent):
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
#log.info(f"Checking status: {ctx.session.state.get("summary_status", "fail")}")
log.info(f"Summary: {ctx.session.state.get("summary")}")
status = ctx.session.state.get("summary_status", "fail").strip()
is_done = (status == "completed")
yield Event(author=self.name, actions=EventActions(escalate=is_done))
迴圈結果的狀態和回呼
在 Google ADK 中，狀態是重要的概念，代表執行期間代理程式的記憶體或工作資料。這項資訊本質上是持續性的內容，可儲存代理需要在不同步驟、工具呼叫或互動中維持的資訊。這個狀態可儲存中間結果、使用者資訊、後續動作的參數，或任何其他在執行工作時需要記住的代理程式資料。
在本情境中，當迴圈代理程式執行迴迭時，summary_agent 和 check_agent 會將其輸出內容 (summary 和 summary_status) 儲存在代理程式的 State 中。這樣一來，資訊就能在各個疊代中保留。不過，Loop Agent 本身不會在完成時自動傳回狀態的最終摘要。
ADK 中的回呼可讓我們在代理程生命週期中的特定時間點，或在回應特定事件 (例如工具呼叫完成或代理程執行完畢前) 時，插入自訂邏輯。可用於動態自訂代理程式的行為和處理結果。
我們會使用 after_agent_callback，在迴圈結束時執行 (因為 CheckCondition 已升級)。這個回呼 modify_output_after_agent 會從狀態中擷取最終摘要，並將其格式化為服務機器人的最終輸出訊息。
👉📝 在同一個 ~/instavibe-bootstrap/agents/social/agent.py 中，將 #REPLACE FOR modify_output_after_agent 替換為以下內容：
def modify_output_after_agent(callback_context: CallbackContext) -> Optional[types.Content]:
agent_name = callback_context.agent_name
invocation_id = callback_context.invocation_id
current_state = callback_context.state.to_dict()
current_user_content = callback_context.user_content
print(f"[Callback] Exiting agent: {agent_name} (Inv: {invocation_id})")
print(f"[Callback] Current summary_status: {current_state.get("summary_status")}")
print(f"[Callback] Current Content: {current_user_content}")
status = current_state.get("summary_status").strip()
is_done = (status == "completed")
# Retrieve the final summary from the state
final_summary = current_state.get("summary")
print(f"[Callback] final_summary: {final_summary}")
if final_summary and is_done and isinstance(final_summary, str):
log.info(f"[Callback] Found final summary, constructing output Content.")
# Construct the final output Content object to be sent back
return types.Content(role="model", parts=[types.Part(text=final_summary.strip())])
else:
log.warning("[Callback] No final summary found in state or it's not a string.")
# Optionally return a default message or None if no summary was generated
return None
定義 Root Loop 代理程式
最後，我們定義主要的 LoopAgent。在每個迴圈疊代中，它會依序協調子代理程式 (profile_agent -> summary_agent -> check_agent -> CheckCondition)。這個序列會重複執行至 max_iterations 次，或直到 CheckCondition 傳送完成信號為止。after_agent_callback 可確保傳回最終摘要。
👉📝 在同一個 ~/instavibe-bootstrap/agents/social/agent.py 中，將 #REPLACE FOR root_agent 替換為以下內容：
root_agent = LoopAgent(
name="InteractivePipeline",
sub_agents=[
profile_agent,
summary_agent,
check_agent,
CheckCondition(name="Checker")
],
description="Find everyone's social profile on events, post and friends",
max_iterations=10,
after_agent_callback=modify_output_after_agent
)
讓我們使用 ADK 開發人員 UI 測試這個多代理工作流程。
👉💻 啟動 ADK 網路伺服器：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/social/.env
adk web
開啟 ADK 開發人員 UI (透過網頁預覽開啟的通訊埠 8000)。在「服務專員」下拉式選單 (右上方) 中，選取「社群媒體」服務專員。
👉 接著，請將這項工作交給它，讓它分析多位使用者的個人資料。在即時通訊對話方塊中輸入：
Tell me about Mike and Bob
當服務專員回覆後 (由於循環和多次 LLM 呼叫，可能會花費較長時間)，請不要只查看即時通訊的最終輸出內容。前往 ADK Dev UI 左側窗格的「事件」分頁。
👉 驗證步驟：在「事件」分頁中，您會看到執行作業的詳細逐步追蹤記錄。
觀察服務代理程式如何叫用各個子服務代理程式後，您預期的流程會從 profile_agent -> summary_agent -> check_agent，並在每次迭代中執行檢查器。但實際上，我們發現這項技術能發揮強大的「自我最佳化」功能。
因為基礎模型會看到整個要求 (例如’)，通常會選擇最有效率的路徑，在單一整合回合中收集所有必要資料，而非重複執行多次。您可以查看每個步驟的輸入內容、輸出內容和狀態，包括 profile_agent 發出的工具呼叫
以及來自 check_agent 和 CheckCondition 的狀態更新。
這項視覺化追蹤功能非常實用，可協助您瞭解及偵錯多代理程工作流程的運作方式，直到回呼產生並傳回最終摘要為止。
查看即時通訊回應和事件追蹤後，請返回 Cloud Shell 終端機，然後按下 Ctrl+C 停止 ADK Dev UI。
10. 代理程式對代理程式 (A2A) 通訊
我們目前已建構專屬的代理程式，但這些代理程式會在同一部機器上獨立運作，或在預先定義的工作流程中運作。為了建構真正分散式且協同合作的多代理人系統，我們需要讓代理人 (可能以個別服務執行) 能夠互相探索並有效溝通。這時就需要使用代理程式對代理程式 (A2A) 通訊協定。
A2A 通訊協定是專為 AI 代理程式之間的互通通訊所設計的開放標準。MCP 著重於服務專員與工具的互動，而 A2A 則著重於服務專員與服務專員的互動。這項功能可讓服務專員：
- 探索：透過標準化代理人資訊卡，尋找其他代理人並瞭解其功能。
- 通訊：安全地交換訊息和資料。
- 協作：委派工作並協調行動，以達成複雜的目標。
A2A 通訊協定會透過「代理卡」等機制促進這類通訊，代理商可利用這些機制宣傳自己的功能和連線資訊。
A2A 採用常見的網路標準 (HTTP、SSE、JSON-RPC)，並經常採用用戶端-伺服器模型，其中一個代理程式 (用戶端) 會將工作傳送至另一個代理程式/遠端伺服器。這種標準化做法是建構模組化、可擴充的系統的關鍵，可讓獨立開發的代理程式協同運作。
為 InstaVibe 代理程式啟用 A2A
為了讓現有的 Planner、Platform Interaction 和 Social 代理程式可透過 A2A 與其他代理程式互動，我們需要為每個代理程式套用 A2A 伺服器元件。這個伺服器會執行以下操作：
- 公開服務卡：透過 HTTP 端點提供服務代理程式功能的標準說明。
- Listen for Tasks(Request Messages)：根據 A2A 通訊協定，接受其他代理程式 (A2A 用戶端) 傳入的工作要求。
- 管理工作(要求訊息) 執行作業：將收到的工作交由基礎 ADK 代理程式邏輯處理。
Planner Agent (已啟用 A2A)
首先，我們要將 A2A 伺服器層新增至 Planner Agent。
定義 A2A 伺服器啟動邏輯。這段程式碼會定義 AgentCard (代理程式的公開說明)、設定 A2AServer 並啟動，然後將其連結至 PlatformAgentExecutor。
👉📝 將下列程式碼新增至 ~/instavibe-bootstrap/agents/planner/a2a_server.py 結尾：
class PlannerAgent:
"""An agent to help user planning a event with its desire location."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="event_planner",
name="Event planner",
description="""
This agent generates multiple fun plan suggestions tailored to your specified location, dates, and interests,
all designed for a moderate budget. It delivers detailed itineraries,
including precise venue information (name, latitude, longitude, and description), in a structured JSON format.
""",
tags=["instavibe"],
examples=["What about Bostona MA this weekend?"],
)
self.agent_card = AgentCard(
name="Event Planner Agent",
description="""
This agent generates multiple fun plan suggestions tailored to your specified location, dates, and interests,
all designed for a moderate budget. It delivers detailed itineraries,
including precise venue information (name, latitude, longitude, and description), in a structured JSON format.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=PlannerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=PlannerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the planning request..."
def _build_agent(self) -> LlmAgent:
"""Builds the LLM agent for the night out planning agent."""
return agent.root_agent
if __name__ == '__main__':
try:
plannerAgent = PlannerAgent()
request_handler = DefaultRequestHandler(
agent_executor=PlannerAgentExecutor(plannerAgent.runner,plannerAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=plannerAgent.agent_card,
http_handler=request_handler,
)
logger.info(f"Attempting to start server with Agent Card: {plannerAgent.agent_card.name}")
logger.info(f"Server object created: {server}")
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
👉💻 讓我們快速測試 A2A 伺服器是否在本機上正確啟動，並提供其服務卡。在第一個終端機中執行下列指令：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents/
python -m planner.a2a_server
👉 接著，請開啟另一個終端機視窗。(按一下終端機面板中的「+」號) 
👉💻 使用 curl 從本機執行的伺服器要求介面卡：
curl http://localhost:10003/.well-known/agent.json | jq
您應該會看到我們定義的 AgentCard 的 JSON 表示法，確認伺服器正在執行並宣傳 Planner 代理程式。
返回第一個終端機 (伺服器執行所在位置)，然後按下 Ctrl+C 停止伺服器。
👉💻 加入 A2A 伺服器邏輯後，我們現在可以建構容器映像檔。
建構及部署 Planner 代理程式
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
# Set variables specific to the PLANNER agent
export IMAGE_TAG="latest"
export AGENT_NAME="planner"
export IMAGE_NAME="planner-agent"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="planner-agent"
export PUBLIC_URL="https://planner-agent-${PROJECT_NUMBER}.${REGION}.run.app"
echo "Building ${AGENT_NAME} agent..."
gcloud builds submit . \
--config=cloudbuild-build.yaml \
--project=${PROJECT_ID} \
--region=${REGION} \
--substitutions=_AGENT_NAME=${AGENT_NAME},_IMAGE_PATH=${IMAGE_PATH}
echo "Image built and pushed to: ${IMAGE_PATH}"
👉💻 並在 Cloud Run 上部署 Planner Agent。
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
# Set variables specific to the PLANNER agent
export IMAGE_TAG="latest"
export AGENT_NAME="planner"
export IMAGE_NAME="planner-agent"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="planner-agent"
export PUBLIC_URL="https://planner-agent-${PROJECT_NUMBER}.${REGION}.run.app"
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--set-env-vars="A2A_HOST=0.0.0.0" \
--set-env-vars="A2A_PORT=8080" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=TRUE" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="PUBLIC_URL=${PUBLIC_URL}" \
--allow-unauthenticated \
--project=${PROJECT_ID} \
--min-instances=1
請使用 A2A 檢查器，確認已部署的服務是否正在執行，並正確地從雲端提供其代理程式資訊卡。
👉 在 Cloud Shell 工具列的網頁預覽圖示中，選取「變更通訊埠」。將通訊埠設為 8081，然後按一下「Change and Preview」(變更並預覽)。系統會開啟新的瀏覽器分頁，並顯示 A2A Inspector 介面。
👉💻 在終端機中取得已部署的企劃書代理程式網址：
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
echo ${PLANNER_AGENT_URL}
👉💻 複製輸出網址。
👉 在 A2A Inspector UI 中，將網址貼到「Agent URL」欄位，然後按一下「Connect」。
👀 代理卡片詳細資料和 JSON 應顯示在「代理卡片」分頁中，確認連線成功。
👉 按一下 A2A 檢查器中的「Chat」分頁。您可以直接與已部署的代理程式互動，傳送訊息來測試其規劃功能。例如：
Plan something for me in Boston MA this weekend, and I enjoy classical music
👀 如要查看原始通訊內容，請在即時通訊視窗中依序按一下訊息泡泡和客服專員的回覆泡泡。點選每個項目時，系統會顯示已傳送或接收的完整 JSON-RPC 2.0 訊息，這對於偵錯非常有用。
請將 A2A Inspector 分頁保留在手邊。請勿關閉！我們稍後會再次使用這個方法，測試其他兩個代理程式。
平台互動代理程式 (已啟用 A2A)
接下來，我們會重複執行平台互動代理程式 (使用 MCP 的代理程式) 的程序。
👉📝 在 ~/instavibe-bootstrap/agents/platform_mcp_client/a2a_server.py 結尾定義 A2A 伺服器設定，包括其專屬的 AgentCard：
class PlatformAgent:
"""An agent that post event and post to instavibe."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="instavibe_posting",
name="Post social post and events on instavibe",
description="""
This "Instavibe" agent helps you create posts (identifying author, text, and sentiment – inferred if unspecified) and register
for events (gathering name, date, attendee). It efficiently collects required information and utilizes dedicated tools
to perform these actions on your behalf, ensuring a smooth sharing experience.
""",
tags=["instavibe"],
examples=["Create a post for me, the post is about my cute cat and make it positive, and I'm Alice"],
)
self.agent_card = AgentCard(
name="Instavibe Posting Agent",
description="""
This "Instavibe" agent helps you create posts (identifying author, text, and sentiment – inferred if unspecified) and register
for events (gathering name, date, attendee). It efficiently collects required information and utilizes dedicated tools
to perform these actions on your behalf, ensuring a smooth sharing experience.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=PlatformAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=PlatformAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the social post and event request..."
def _build_agent(self) -> LlmAgent:
"""Builds the LLM agent for the Processing the social post and event request."""
return agent.root_agent
if __name__ == '__main__':
try:
platformAgent = PlatformAgent()
request_handler = DefaultRequestHandler(
agent_executor=PlatformAgentExecutor(platformAgent.runner,platformAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=platformAgent.agent_card,
http_handler=request_handler,
)
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
社群媒體服務專員 (已啟用 A2A)
最後，我們為社交媒體分析代理程式啟用 A2A。
👉📝 在 ~/instavibe-bootstrap/agents/social/a2a_server.py 結尾定義 A2A 伺服器設定和 AgentCard：
class SocialAgent:
"""An agent that handles social profile analysis."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="social_profile_analysis",
name="Analyze Instavibe social profile",
description="""
Using a provided list of names, this agent synthesizes Instavibe social profile information by analyzing posts, friends, and events.
It delivers a comprehensive single-paragraph summary for individuals, and for groups, identifies commonalities in their social activities
and connections based on profile data.
""",
tags=["instavibe"],
examples=["Can you tell me about Bob and Alice?"],
)
self.agent_card = AgentCard(
name="Social Profile Agent",
description="""
Using a provided list of names, this agent synthesizes Instavibe social profile information by analyzing posts, friends, and events.
It delivers a comprehensive single-paragraph summary for individuals, and for groups, identifies commonalities in their social activities
and connections based on profile data.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=self.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=self.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the social profile analysis request..."
def _build_agent(self) -> LoopAgent:
"""Builds the LLM agent for the social profile analysis agent."""
return agent.root_agent
if __name__ == '__main__':
try:
socialAgent = SocialAgent()
request_handler = DefaultRequestHandler(
agent_executor=SocialAgentExecutor(socialAgent.runner,socialAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=socialAgent.agent_card,
http_handler=request_handler,
)
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
建構及部署平台互動和社交代理程式
這些代理程式需要存取 Spanner，因此請確保在部署期間正確傳遞 SPANNER_INSTANCE_ID、SPANNER_DATABASE_ID 和 MCP_SERVER_URL 環境變數。
👉💻 使用 Cloud Build 建構及部署至 Cloud Run：
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
gcloud builds submit . \
--config=cloudbuild.yaml \
--project="${PROJECT_ID}" \
--region="${REGION}" \
--substitutions=\
_PROJECT_ID="${PROJECT_ID}",\
_PROJECT_NUMBER="${PROJECT_NUMBER}",\
_REGION="${REGION}",\
_REPO_NAME="${REPO_NAME}",\
_SPANNER_INSTANCE_ID="${SPANNER_INSTANCE_ID}",\
_SPANNER_DATABASE_ID="${SPANNER_DATABASE_ID}",\
_MCP_SERVER_URL="${MCP_SERVER_URL}"
👉💻 在終端機中取得已部署平台代理程式的網址：
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
echo $PLATFORM_MPC_CLIENT_URL
👉💻 複製輸出網址。
👉 在 A2A Inspector UI 中，將網址貼到「Agent URL」欄位，然後按一下「Connect」。
👀 代理卡片詳細資料和 JSON 應顯示在「代理卡片」分頁中，確認連線成功。
👉 按一下 A2A 檢查器中的「Chat」分頁。您可以在這裡直接與已部署的代理程式互動，傳送訊息測試代理程式建立貼文的功能：
Create a post for me, the post says 'Paws, purrs, and ocean views 🐾☕🌊. Spent my morning at the Morning Seaside Cat Café, where every sip comes with a side of snuggles and sea breeze.' and make it positive, and I'm Oscar.
👀 如要查看原始通訊內容，請在即時通訊視窗中依序按一下訊息泡泡和客服專員的回覆泡泡。點選每個項目時，系統會顯示已傳送或接收的完整 JSON-RPC 2.0 訊息，這對於偵錯非常有用。
👉💻 在終端機中取得已部署的社群媒體代理程式的網址：
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
echo $SOCIAL_AGENT_URL
👉💻 複製輸出網址。
👉 在 A2A Inspector UI 中，將網址貼到「Agent URL」欄位，然後按一下「Connect」。
👀 代理卡片詳細資料和 JSON 應顯示在「代理卡片」分頁中，確認連線成功。
👉 按一下 A2A 檢查器中的「Chat」分頁。您可以直接與已部署的代理程式互動，傳送訊息給代理程式，以便分析資料庫中的使用者個人資料：
Can you tell me about both Ian and Kevin's profile, what are their common interests?
👀 如要查看原始通訊內容，請在即時通訊視窗中依序按一下訊息泡泡和客服專員的回覆泡泡。點選每個項目時，系統會顯示已傳送或接收的完整 JSON-RPC 2.0 訊息，這對於偵錯非常有用。
👉 太好了，我們已完成所有服務專員的檢查。您現在可以關閉 A2A Inspector 分頁。
11. 自動調度管理工具代理程式 (A2A 用戶端)
我們現在有三個專門的服務代理程式 (Planner、Platform、Social)，可在 Cloud Run 上做為獨立的 A2A 服務執行。最後一個部分是自動化調度管理工具代理程式。這個代理程式會充當中央協調器或 A2A 用戶端。它會接收使用者要求，找出需要哪些遠端代理程式來執行要求 (可能會依序執行)，然後使用 A2A 通訊協定將工作委派給這些遠端代理程式。在本工作坊中，我們會使用 ADK Dev UI 在本機執行 Orchestrator 代理程式。
首先，讓我們強化 Orchestrator 的邏輯，以便處理所發現的遠端代理程式註冊作業。在初始化期間，儲存從擷取的介面卡中擷取的連線詳細資料。
👉📝 在 ~/instavibe-bootstrap/agents/orchestrate/agent.py 中，將 #REPLACE ME REG AGENT CARD 替換為：
async with httpx.AsyncClient(timeout=30) as client:
for i, address in enumerate(REMOTE_AGENT_ADDRESSES):
log.info(f"--- STEP 3.{i}: Attempting connection to: {address} ---")
try:
card_resolver = A2ACardResolver(client, address)
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(agent_card=card, agent_url=address)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
log.info(f"--- STEP 5.{i}: Successfully stored connection for {card.name} ---")
except Exception as e:
log.error(f"--- CRITICAL FAILURE at STEP 4.{i} for address: {address} ---")
log.error(f"--- The hidden exception type is: {type(e).__name__} ---")
log.error(f"--- Full exception details and traceback: ---", exc_info=True)
接著，請在 ADK 中定義自動化調度管理工具的代理程式。
send_message(用於委派工作的 A2A 函式)。
👉📝 將 ~/instavibe-bootstrap/agents/orchestrate/agent.py 中的 #REPLACE ME CREATE AGENT 替換為：
def create_agent(self) -> Agent:
"""Synchronously creates the ADK Agent object."""
return Agent(
model="gemini-2.5-flash",
name="orchestrate_agent",
instruction=self.root_instruction,
before_agent_callback=self.before_agent_callback,
description=("Orchestrates tasks for child agents."),
tools=[self.send_message],
)
指令是指揮家的核心邏輯，可告知指揮家如何使用 A2A。
👉📝 請將 ~/instavibe-bootstrap/agents/orchestrate/agent.py 中的 #REPLACE ME INSTRUCTIONS 替換為以下指令產生方法：
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""
You are an expert AI Orchestrator. Your primary responsibility is to intelligently interpret user requests, break them down into a logical plan of discrete actions, and delegate each action to the most appropriate specialized remote agent using the send_message function. You do not perform the tasks yourself but manage their assignment, sequence, and critically, their outcomes.
**Core Directives & Decision Making:**
* **Understand User Intent & Complexity:**
* Carefully analyze the user's request to determine the core task(s) they want to achieve. Pay close attention to keywords and the overall goal.
* Identify if the request requires a single agent or a sequence of actions from multiple agents. For example, "Analyze John Doe's profile and then create a positive post about his recent event attendance" would require two agents in sequence.
* **Task Planning & Sequencing (for Multi-Step Requests):**
* Before delegating, outline the clear sequence of agent tasks.
* Identify dependencies. If Task B requires output from Task A, execute them sequentially. If tasks are independent (like creating a post and then creating an event), execute them one after the other as separate delegations.
* Agent Reusability: An agent's completion of one task does not make it unavailable. If a user's plan involves multiple, distinct actions that fall under the same agent's expertise (e.g., create a post, then create an event), you must call that same agent again for the subsequent task.
* **Task Delegation & Management (using `send_message`):**
* **Delegation:** Use `send_message` to assign actionable tasks to the selected remote agent. Your `send_message` call MUST include:
* The `remote_agent_name` you've selected.
* The `user_request` or all necessary parameters extracted from the user's input, formatted in a way the target agent will understand.
* **Contextual Awareness for Remote Agents:** If a remote agent repeatedly requests user confirmation or seems to lack context, assume it lacks access to the full conversation history. In such cases, enrich your `send_message` with all necessary contextual information relevant to that specific agent from the conversation history.
* **Sequential Task Execution:**
* After a preceding task completes (indicated by the agent's response or a success signal), gather any necessary output from it.
* Then, use `send_message` for the next agent in the sequence, providing it with the user's original relevant intent and any necessary data obtained from the previous agent's task.
* **Active Agent Prioritization:** If an active agent is already engaged and the user's request is related to its current task, route subsequent related requests directly to that agent by providing updated context via `send_message`.
**Critical Success Verification:**
* You **MUST** wait for the tool_output after every send_message call before taking any further action.
* Your decision to proceed to the next task in a sequence **MUST** be based entirely on a confirmation of success from the tool_output of the previous task.
* If a tool call fails, returns an error, or the tool_output is ambiguous, you MUST STOP the sequence. Your next action is to report the exact failure or ambiguity to the user.
* DO NOT assume a task was successful. Do not invent success messages like "The event has been created." Only state that a task is complete if the tool's response explicitly says so.
**Communication with User:**
* **Transparent Communication:** Always present the complete and detailed response from the remote agent to the user. Do not summarize or filter unless explicitly instructed.
* When you delegate a task (or the first task in a sequence), clearly inform the user which remote agent is handling it.
* For multi-step requests, you can optionally inform the user of the planned sequence (e.g., "Okay, first I'll ask the 'Social Profile Agent' to analyze the profile, and then I'll have the 'Instavibe Posting Agent' create the post.").
* If waiting for a task in a sequence to complete, you can inform the user (e.g., "The 'Social Profile Agent' is currently processing. I'll proceed with the post once that's done.").
* **User Confirmation Relay:** If a remote agent asks for confirmation, and the user has not already provided it, just make up something.
* If the user's request is ambiguous, if necessary information is missing for any agent in the sequence, or if you are unsure about the plan, just make up something.
**Important Reminders:**
* **Autonomous Agent Engagement:** Never seek user permission before engaging with remote agents. If multiple agents are required to fulfill a request, connect with them directly without requesting user preference or confirmation.
* **Focused Information Sharing:** Provide remote agents with only relevant contextual information. Avoid extraneous details that are not directly pertinent to their task.
* **No Redundant Confirmations:** Do not ask remote agents for confirmation of information or actions they have already processed or committed to.
* **Tool Reliance:** Strictly rely on your available tools, primarily `send_message`, to address user requests. Do not generate responses based on assumptions. If information is insufficient, request clarification from the user.
* **Prioritize Recent Interaction:** Focus primarily on the most recent parts of the conversation when processing requests, while maintaining awareness of the overall goal for multi-step tasks.
* Always prioritize selecting the correct agent(s) based on their documented purpose.
* Ensure all information required by the chosen remote agent is included in the `send_message` call, including outputs from previous agents if it's a sequential task.
Agents:
{self.agents}
Current agent: {current_agent['active_agent']}`
"""
測試自動調度器和完整的 A2A 系統
接下來，我們來測試整個系統。我們會使用 ADK Dev UI 在本機執行 Orchestrator，並與在 Cloud Run 上遠端執行的 Planner、Platform 和 Social 代理程式進行通訊。
👉💻 首先，請確認環境變數 REMOTE_AGENT_ADDRESSES 包含以半形逗號分隔的已部署 A2A 的代理程式網址。接著，為調度器代理程式設定必要的環境變數，並啟動 ADK Dev UI：
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
export REMOTE_AGENT_ADDRESSES=${PLANNER_AGENT_URL},${PLATFORM_MPC_CLIENT_URL},${SOCIAL_AGENT_URL}
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?REMOTE_AGENT_ADDRESSES\)=.*|REMOTE_AGENT_ADDRESSES=${REMOTE_AGENT_ADDRESSES}|" ~/instavibe-bootstrap/agents/orchestrate/.env
adk web
👉 開啟 ADK 開發人員使用者介面 (透過網頁預覽功能將通訊埠改回 8000)。
👉 在代理程式下拉式選單中，選取「orchestrate」代理程式。
👉 接著，請為它提供需要協調多個遠端服務專員的複雜任務。請試試這個第一個範例，其中應包含 Social Agent 和 Planner Agent：
You are an expert event planner for a user named Diana.
Your task is to design a fun and personalized event.
Here are the details for the plan:
- Friends to invite: Ian, Nora
- Desired date: "2025-10-15"
- Location idea or general preference: "Chicago"
Your process should be:
1. Analyze the provided friend names. If you have access to a tool to get their InstaVibe profiles or summarized interests, please use it.
2. Based on their potential interests (or general good taste if profiles are unavailable), create a tailored plan for the outing, check if you have access to any event planner tools.
3. Ensure the plan includes the original `planned_date`.
The user wants a comprehensive plan that includes:
- The list of invited friends.
- A catchy and descriptive name for the event.
- The exact planned date for the event.
- A summary of what the group will do.
- Specific recommended spots (e.g., restaurants, bars, activity venues) with their names, (if possible, approximate latitude/longitude for mapping, and address), and a brief description of why it fits the plan.
- A short, exciting message that {Diana} can send to {Ian, Nora} to get them excited about the event.
觀察 ADK 開發人員 UI 即時通訊視窗中的互動情形。請仔細留意 Orchestrator 的回應，其中應會指出要將工作委派給哪個遠端代理程式 (例如「好的，我會先向社群個人資料專員詢問 Ian 和 Nora 的相關問題...」)。
此外，請查看 UI 中的「事件」分頁，瞭解向遠端代理程式網址發出的基礎工具呼叫 (send_message)。
👉 現在，請試試第二個範例，這個範例應直接涉及平台整合服務代理程式：
Hey, can you register an event on Instavibe for Laura and Charlie? Let's call it 'Vienna Concert & Castles Day'.
here are more info
"event_name": "Vienna Concert & Castles Day",
"description": "A refined and unforgettable day in Vienna with Laura and Charlie. The day begins with a guided tour of the magnificent Schönbrunn Palace, showcasing imperial architecture and history. In the evening, enjoy a classical music concert in one of Vienna's most iconic concert halls.",
"event_date": "2025-10-14T10:00:00+02:00",
"locations": [
{
"name": "Schönbrunn Palace",
"description": "A UNESCO World Heritage Site and former imperial summer residence, Schönbrunn Palace offers opulent rooms, beautiful baroque gardens, and a glimpse into the life of the Habsburg monarchy. Visitors can stroll the grounds or take a guided historical tour.",
"latitude": 48.184516,
"longitude": 16.312222,
"address": "Schönbrunner Schloßstraße 47, 1130 Wien, Austria"
},
{
"name": "Musikverein Vienna",
"description": "Home to the world-renowned Vienna Philharmonic, the Musikverein is one of the finest concert halls in the world. Its 'Golden Hall' is famous for its acoustics and ornate design. Attendees can enjoy a powerful classical concert in an unforgettable setting.",
"latitude": 48.200132,
"longitude": 16.373777,
"address": "Musikvereinsplatz 1, 1010 Wien, Austria"
}
],
"attendee_names": ["Laura", "Charlie", "Oscar"] And I am Oscar
再次監控即時通訊和「事件」分頁。編排器應判斷是否需要建立事件，並將工作 (連同所有提供的詳細資料) 委派給「平台整合服務」。您也可以按一下「追蹤」Trace按鈕，查看追蹤記錄，以便分析查詢回應時間和執行的作業。
接著，您可以確認事件是否顯示在 InstaVibe 網頁應用程式中。
這項實驗說明如何使用 ADK 和 A2A 通訊協定成功實作多代理系統，其中中央協調器會將工作委派給專門的遠端代理程式。
測試完成後，請務必停止 ADK Dev UI (在終端機中為 Ctrl+C)。
12. Agent Engine 和 InstaVibe 的遠端通話
到目前為止，我們已在 Cloud Run 上執行專屬的代理程式，並使用 ADK 開發人員 UI 在本機測試了 Orchestrator。在實際工作情境中，我們需要穩固、可擴充且受控管的環境來主控服務機器人。這時 Google Vertex AI Agent Engine 就能派上用場。
Agent Engine 是 Vertex AI 上的全代管服務，專門用於部署及調度 AI 代理。它可讓開發人員 (尤其是不熟悉複雜雲端環境的開發人員) 專注於代理程式邏輯和功能，而非管理伺服器，進而省去基礎架構管理、安全防護和營運額外負擔的麻煩。提供專屬的執行階段，可針對代理式工作負載進行最佳化。
我們現在會將 Orchestrator 代理程式部署至 Agent Engine。(注意：下方顯示的部署機制會使用工作坊資料提供的自訂指令碼 (agent_engine_app.py)，因為官方直接 ADK 到 Agent-Engine 部署工具可能仍在開發中。這個指令碼會處理包裝及部署 Orchestrator 代理程式，並設定必要的遠端代理程式位址。
執行下列指令，將 Orchestrator 代理程式部署至 Agent Engine。請確認 REMOTE_AGENT_ADDRESSES 環境變數 (包含 Cloud Run 中的 Planner、Platform 和 Social 代理程式的網址) 是否仍正確設定，如上一個部分所述。
👉💻 我們會將 Orchestrate 代理程式部署至 Agent Engine (注意：這是我自己實作部署作業，ADT 有可協助部署的 CLI，我會在 BYO-SA 實作後更新這項資訊)
cd ~/instavibe-bootstrap/agents/
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
export REMOTE_AGENT_ADDRESSES=${PLANNER_AGENT_URL},${PLATFORM_MPC_CLIENT_URL},${SOCIAL_AGENT_URL}
sed -i "s|^\(O\?REMOTE_AGENT_ADDRESSES\)=.*|REMOTE_AGENT_ADDRESSES=${REMOTE_AGENT_ADDRESSES}|" ~/instavibe-bootstrap/agents/orchestrate/.env
adk deploy agent_engine \
--display_name "orchestrate-agent" \
--project $GOOGLE_CLOUD_PROJECT \
--region $GOOGLE_CLOUD_LOCATION \
--staging_bucket gs://$GOOGLE_CLOUD_PROJECT-agent-engine \
--trace_to_cloud \
--requirements_file orchestrate/requirements.txt \
orchestrate
由於 Orchestrator 已託管在受管理的 Agent Engine 平台上，因此 InstaVibe 網路應用程式需要與其通訊。網頁應用程式會對 Agent Engine 端點發出遠端呼叫，而非透過 ADK Dev UI 互動。
首先，我們需要修改 InstaVibe 應用程式程式碼，以便使用已部署的 Orchestrator 代理程式的專屬 ID 初始化 Agent Engine 用戶端。您必須使用這個 ID，才能在平台上指定正確的代理程式執行個體。
👉📝 開啟 ~/instavibe-bootstrap/instavibe/introvertally.py，並使用以下程式碼取代 #REPLACE ME initiate agent_engine。這會從環境變數 (我們稍後會設定) 擷取 Agent Engine ID，並取得用戶端物件：
ORCHESTRATE_AGENT_ID = os.environ.get('ORCHESTRATE_AGENT_ID')
agent_engine = agent_engines.get(ORCHESTRATE_AGENT_ID)
我們在 InstaVibe 中規劃的使用者流程包含兩個與服務專員的互動：首先，產生建議的企劃書；其次，在服務專員實際將活動發布至平台前，請使用者確認。
由於 InstaVibe 網頁應用程式 (在 Cloud Run 上執行) 和調度器代理程式 (在 Agent Engine 上執行) 現已是獨立的服務，因此網頁應用程式需要向 Agent Engine 端點發出遠端呼叫，才能與代理程式互動。
👉📝 讓我們更新用來進行初始呼叫的程式碼，以產生方案建議。在同一個 introvertally.py 檔案中，將 #REPLACE ME Query remote agent get plan 替換為以下程式碼片段，該程式碼片段會使用 agent_engine 用戶端傳送使用者的要求：
agent_engine.stream_query(
user_id=user_id,
message=prompt_message,
)
👉📝 接下來，請更新處理使用者確認作業的程式碼 (例如使用者按一下「確認方案」時)。這會將後續訊息傳送至 Agent Engine 上的相同對話，指示 Orchestrator 繼續發布事件 (會委派給平台整合服務代理程式)。將 introvertally.py 中的 #REPLACE ME Query remote agent for confirmation 替換為以下內容：
agent_engine.stream_query(
user_id=agent_session_user_id,
message=prompt_message,
)
網頁應用程式的路徑需要存取這些函式。請確認 Flask 路徑檔案中已匯入 introvertally.py 中的必要函式。
👉📝 在 cd ~/instavibe-bootstrap/instavibe/ally_routes.py 中，我們會先將 # REPLACE ME TO ADD IMPORT 指向以下的例項：
from introvertally import call_agent_for_plan, post_plan_event
👉📝 將原型功能新增至 InstaVibe，在 ~/instavibe-bootstrap/instavibe/templates/base.html 中將 <!–REPLACE_ME_LINK_TO_INTROVERT_ALLY–> 替換為以下內容：
<li class="nav-item">
<a class="nav-link" href="{{ url_for('ally.introvert_ally_page') }}">Introvert Ally</a>
</li>
我們需要部署至 Agent Engine 的編排器代理程式特定 Resource ID，才能重新部署 InstaVibe 應用程式。
目前，透過程式輔助函式 gcloud 擷取此 ID 可能會受到限制，因此我們會使用輔助 Python 指令碼 (工作坊提供的 temp-endpoint.py) 擷取 ID，並將其儲存在環境變數中。
👉💻 執行下列指令來執行指令碼。這個指令碼會擷取 Agent Engine 端點 ID，並授予代理程式引擎的預設服務帳戶必要權限 (注意：這個指令碼會使用預設服務帳戶，因為目前無法由使用者修改)。
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
source ~/instavibe-bootstrap/env/bin/activate
python temp-endpoint.py
export ORCHESTRATE_AGENT_ID=$(cat temp_endpoint.txt)
echo "ORCHESTRATE_AGENT_ID set to: ${ORCHESTRATE_AGENT_ID}"
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-$PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com" \
--role="roles/viewer"
最後，我們需要使用更新的程式碼和新的 ORCHESTRATE_AGENT_ID 環境變數重新部署 InstaVibe 網頁應用程式，讓應用程式知道如何連線至在 Agent Engine 上執行的代理程式。
👉💻 下列指令會重新建構 InstaVibe 應用程式映像檔，並將新版本部署至 Cloud Run：
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
echo "Building ${APP_FOLDER_NAME} webapp image..."
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
echo "Deploying ${SERVICE_NAME} to Cloud Run..."
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="SPANNER_INSTANCE_ID=${SPANNER_INSTANCE_ID}" \
--set-env-vars="SPANNER_DATABASE_ID=${SPANNER_DATABASE_ID}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_MAPS_API_KEY=${GOOGLE_MAPS_API_KEY}" \
--set-env-vars="ORCHESTRATE_AGENT_ID=${ORCHESTRATE_AGENT_ID}" \
--project=${PROJECT_ID} \
--min-instances=1 \
--cpu=2 \
--memory=2Gi
最終部署作業完成後，請在其他瀏覽器分頁中前往 InstaVibe 應用程式網址。
測試 AI 輔助的完整 InstaVibe 體驗
「InstaVibe Ally」功能現已上線，這項功能由多代理系統提供支援，並透過 Vertex AI Agent Engine 自動調度及透過 A2A 進行通訊。
點選「InstaVibe Ally」，並要求它規劃活動。
在服務機器人運作期間，請觀察右側的活動記錄 (可能需要 90 到 120 秒)。企劃書顯示後，請詳閱內容，然後按一下「確認此企劃書」，繼續發布。
編排器現在會指示平台代理程式在 InstaVibe 中建立貼文和活動。
請前往 InstaVibe 首頁查看最新貼文和活動。
事件頁面會顯示代理程式產生的詳細資料。
使用 Cloud Trace 分析效能
您可能會發現這項程序需要一些時間。Vertex AI Agent Engine 與 Cloud Trace 整合，可讓我們分析多代理系統的延遲時間。
前往 Google Cloud 控制台的「Traces」，選取 Span 中的 agent_run[orchestrate_agent]，您應該會看到幾個 Span，請點選其中一個
您可以在追蹤記錄詳細資料中，找出哪些部分耗費較多時間。舉例來說，由於搜尋基礎和複雜產生作業，因此對企劃書代理程式的呼叫可能會顯示較長的延遲時間。
同樣地，在建立貼文和事件時，您可能會看到 Orchestrator 處理資料和為 Platform 代理人準備工具呼叫所花費的時間。
探索這些追蹤記錄有助於瞭解並改善服務機器人系統的效能。
恭喜！您已成功使用 Google ADK、A2A、MCP 和 Google Cloud 服務，建構、部署及測試複雜的多代理 AI 系統。您已處理了代理程式調度、工具使用、狀態管理和雲端部署作業，為 InstaVibe 建立了功能強大的 AI 技術輔助功能。恭喜您完成工作坊！
13. 清除
為避免 Google Cloud 帳戶持續產生費用，請務必刪除我們在工作坊中建立的資源。下列指令可協助您移除 Spanner 執行個體、Cloud Run 服務、Artifact Registry 存放區、API 金鑰、Vertex AI Agent Engine 和相關 IAM 權限。
重要事項：
- 請務必在工作坊使用的 Google Cloud 專案中執行這些指令。
- 如果您已關閉 Cloud Shell 終端機，則可能未設定 $PROJECT_ID、$SPANNER_INSTANCE_ID 等部分環境變數。您必須按照工作坊設定時的做法重新匯出，或是將下方指令中的變數替換為實際值。
- 這些指令會永久刪除資源。如果這個專案中還有其他重要資料，請在執行前仔細檢查。
👉💻 執行以下指令碼來清除。
重設環境變數
. ~/instavibe-bootstrap/set_env.sh
刪除代理引擎：
cd ~/instavibe-bootstrap/utils
source ~/instavibe-bootstrap/env/bin/activate
export ORCHESTRATE_AGENT_ID=$(cat ~/instavibe-bootstrap/instavibe/temp_endpoint.txt)
echo "ORCHESTRATE_AGENT_ID set to: ${ORCHESTRATE_AGENT_ID}"
python remote_delete.py
deactivate
echo "Vertex AI Agent Engine deletion initiated."
刪除 Cloud Run 服務：
# InstaVibe Web Application
gcloud run services delete instavibe --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# MCP Tool Server
gcloud run services delete mcp-tool-server --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Planner Agent (A2A Server)
gcloud run services delete planner-agent --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Platform MCP Client Agent (A2A Server)
gcloud run services delete platform-mcp-client --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Social Agent (A2A Server)
gcloud run services delete social-agent --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
echo "Cloud Run services deletion initiated."
停止及移除 A2A Inspector Docker 容器
docker rm --force a2a-inspector
刪除 Spanner 執行個體：
echo "Deleting Spanner instance: ${SPANNER_INSTANCE_ID}..."
gcloud spanner instances delete ${SPANNER_INSTANCE_ID} --project=${PROJECT_ID} --quiet
echo "Spanner instance deletion initiated."
刪除 Artifact Registry 存放區：
echo "Deleting Artifact Registry repository: ${REPO_NAME}..."
gcloud artifacts repositories delete ${REPO_NAME} --location=${REGION} --project=${PROJECT_ID} --quiet
echo "Artifact Registry repository deletion initiated."
從服務帳戶中移除角色：
echo "Removing roles from service account: $SERVICE_ACCOUNT_NAME in project $PROJECT_ID"
# Remove Project-level roles for default service account
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.databaseUser"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/artifactregistry.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/cloudbuild.builds.editor"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/run.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/iam.serviceAccountUser"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/aiplatform.user"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.logWriter"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.viewer"
echo "All specified roles have been removed."
刪除本機工作坊檔案：
echo "Removing local workshop directory ~/instavibe-bootstrap..."
rm -rf ~/instavibe-bootstrap
rm -rf ~/a2a-inspector
rm -f ~/mapkey.txt
rm -f ~/project_id.txt
echo "Local directory removed."
