如何部署呼叫後端 ADK 代理程式的 Gradio 前端應用程式,兩者皆在 Cloud Run 上執行

1. 簡介

總覽

在本程式碼研究室中,您會將 ADK 代理程式部署至 Cloud Run 做為後端服務,然後將 ADK 代理程式的 gradio 前端部署為第二項 Cloud Run 服務。本程式碼研究室說明如何要求驗證 ADK 代理程式服務,並從 gradio 前端服務對其進行通過驗證的呼叫。

課程內容

  • 如何將 ADK 代理程式部署至 Cloud Run
  • 如何將 Gradio 應用程式部署至 Cloud Run
  • 如何在 Cloud Run 中進行經過驗證的服務對服務呼叫

2. 啟用 API

首先,請設定 Google Cloud 專案。

gcloud config set project <YOUR_PROJECT_ID>

您可以執行下列指令,確認 Google Cloud 專案:

gcloud config get-value project

本程式碼研究室需要啟用下列 API:

gcloud services enable run.googleapis.com \
    compute.googleapis.com \
    run.googleapis.com \
    cloudbuild.googleapis.com \
    artifactregistry.googleapis.com \
    aiplatform.googleapis.com

3. 設定和需求

在本節中,您將建立幾個服務帳戶,並授予適當的 IAM 角色。每個 Cloud Run 服務都會有自己的服務帳戶。

首先,請為本程式碼研究室設定環境變數,這些變數會在整個程式碼研究室中使用。

export PROJECT_ID=<YOUR_PROJECT_ID>
export REGION=<YOUR_REGION>

export SERVICE_ACCOUNT_ADK="adk-agent-cr"
export SERVICE_ACCOUNT_ADDRESS_ADK=$SERVICE_ACCOUNT_ADK@$PROJECT_ID.iam.gserviceaccount.com

export SERVICE_ACCOUNT_GRADIO="adk-agent-gradio"
export SERVICE_ACCOUNT_ADDRESS_GRADIO=$SERVICE_ACCOUNT_GRADIO@$PROJECT_ID.iam.gserviceaccount.com

export AGENT_APP_NAME="multi_tool_agent"

接著,請為 ADK 代理程式建立服務帳戶。

gcloud iam service-accounts create $SERVICE_ACCOUNT_ADK \
--display-name="Service account for adk agent on cloud run"

並授予 ADK 服務帳戶「Vertex AI 使用者」角色

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$SERVICE_ACCOUNT_ADDRESS_ADK" \
  --role="roles/aiplatform.user"

現在,請為 Gradio 前端建立服務帳戶

gcloud iam service-accounts create $SERVICE_ACCOUNT_GRADIO \
  --display-name="Service account for gradio frontend cloud run"

並授予 Gradio 前端 Cloud Run 叫用者角色,允許呼叫 Cloud Run 上代管的 ADK 代理程式。

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$SERVICE_ACCOUNT_ADDRESS_GRADIO" \
  --role="roles/run.invoker"

4. 建立 ADK 應用程式

在下一個步驟中,您將建立 ADK 快速入門應用程式的程式碼。

注意:實驗室結束時,您的檔案結構應如下所示:

- codelab-gradio-adk  <-- you'll deploy the ADK agent from here
  - gradio-frontend
    - app.py
    - requirements.txt
  - multi_tool_agent  <-- you'll deploy the gradio app from here
    - __init__.py
    - agent.py
    - requirements.txt

首先,請為整個程式碼研究室建立目錄

mkdir codelab-gradio-adk
cd codelab-gradio-adk

現在,請為 ADK 代理程式服務建立目錄。

mkdir multi_tool_agent && cd multi_tool_agent

使用以下內容建立 __init__.py 檔案:

from . import agent

建立 requirements.txt 檔案:

google-adk

建立名為 agent.py 的檔案

import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city for which to retrieve the weather report.

    Returns:
        dict: status and result or error msg.
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "The weather in New York is sunny with a temperature of 25 degrees"
                " Celsius (77 degrees Fahrenheit)."
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available.",
        }


def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city.

    Args:
        city (str): The name of the city for which to retrieve the current time.

    Returns:
        dict: status and result or error msg.
    """

    if city.lower() == "new york":
        tz_identifier = "America/New_York"
    else:
        return {
            "status": "error",
            "error_message": (
                f"Sorry, I don't have timezone information for {city}."
            ),
        }

    tz = ZoneInfo(tz_identifier)
    now = datetime.datetime.now(tz)
    report = (
        f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
    )
    return {"status": "success", "report": report}


root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.5-flash",
    description=(
        "Agent to answer questions about the time and weather in a city."
    ),
    instruction=(
        "You are a helpful agent who can answer user questions about the time and weather in a city."
    ),
    tools=[get_weather, get_current_time],
)

5. 部署 ADK 代理程式

在本節中,您將 ADK 代理程式部署至 Cloud Run。接著,您可以使用 ADK 提供的開發人員網頁 UI,確認部署作業是否成功。最後,您需要對這項服務進行已驗證的呼叫。

前往上層資料夾。

注意:ADK 代理程式碼必須包含 multi_tool_agent 資料夾做為根資料夾。

cd ..

首先,請建立 Cloud Run 服務:

注意:如要使用開發人員使用者介面進行測試,--with_ui 為選用項目,如後續步驟所示:

注意:-- 指令可讓您將指令列標記傳遞至基礎 gcloud run deploy 指令。

注意:uvx --from 會執行 google-adk 套件中的指令。uvx 會建立臨時虛擬環境、在其中安裝 google-adk、執行指定指令,然後終止環境。

uvx --from google-adk \
adk deploy cloud_run \
    --project=$PROJECT_ID \
    --region=$REGION \
    --service_name=adk-agent-cr \
    --with_ui \
    ./multi_tool_agent \
    -- \
    --service-account=$SERVICE_ACCOUNT_ADDRESS_ADK \
    --allow-unauthenticated

接著,請將網址儲存為環境變數,以便在本程式碼研究室的第二部分中使用

AGENT_SERVICE_URL=$(gcloud run services describe adk-agent-cr --region $REGION --format 'value(status.url)')

現在試用代理程式

在網路瀏覽器中開啟服務網址,然後詢問 tell me about the weather in new york。畫面會顯示類似「紐約天氣晴朗,氣溫為攝氏 25 度 (華氏 77 度)。」的回應。

最後,保護代理程式

現在,我們要保護代理程式的存取權。在下一個部分中,您將部署 Cloud Run 服務,對這個後端服務進行經過驗證的呼叫。

gcloud run services remove-iam-policy-binding adk-agent-cr \
  --member="allUsers" \
  --role="roles/run.invoker" \
  --region=$REGION

6. 部署 gradio 前端

在這個步驟中,您將為 ADK 代理程式建立 gradio 前端

注意:您可以在與 ADK 代理程式相同的服務中,使用 gradio 應用程式。本程式碼研究室提供 2 項獨立服務,說明如何在 Cloud Run 中進行經過驗證的服務對服務呼叫。

首先,請在 multi_tool_agent 資料夾旁建立應用程式

mkdir gradio-frontend && cd gradio-frontend

接著,建立包含下列內容的 requirements.txt 檔案:

gradio
requests
google-auth

現在,請建立 app.py 檔案

import gradio as gr
import requests
import json
import uuid
import os
import google.auth.transport.requests
import google.oauth2.id_token

# https://weather-time-service2-392295011265.us-west4.run.app
BASE_URL = os.environ.get("AGENT_SERVICE_URL")

# multi_tool_agent
APP_NAME = os.environ.get("AGENT_APP_NAME")

# Generate a unique user ID for each session of the Gradio app
USER_ID = f"gradio-user-{uuid.uuid4()}"

# API Endpoints
CREATE_SESSION_URL = f"{BASE_URL}/apps/{APP_NAME}/users/{USER_ID}/sessions"
RUN_SSE_URL = f"{BASE_URL}/run_sse"

def get_id_token():
    """Get an ID token to authenticate with the other Cloud Run service."""
    audience = BASE_URL
    request = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(request, audience)
    return id_token

def create_session() -> str | None:
    """Creates a new session and returns the session ID."""
    try:
        id_token = get_id_token()
        headers = {"Authorization": f"Bearer {id_token}"}
        response = requests.post(CREATE_SESSION_URL, headers=headers)
        response.raise_for_status()
        return response.json().get("id")
    except Exception as e:
        print(f"Error creating session: {e}")
        return None

def query_agent(prompt: str):
    """Sends a prompt to the agent and returns the streamed response."""
    session_id = create_session()
    if not session_id:
        return "Error: Could not create a session."

    id_token = get_id_token()
    headers = {
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
        "Authorization": f"Bearer {id_token}",
    }
    payload = {
        "app_name": APP_NAME,
        "user_id": USER_ID,
        "session_id": session_id,
        "new_message": {"role": "user", "parts": [{"text": prompt}]},
        "streaming": True
    }

    full_response = ""
    try:
        with requests.post(RUN_SSE_URL, headers=headers, json=payload, stream=True) as response:
            response.raise_for_status()
            for chunk in response.iter_lines():
                if chunk and chunk.decode('utf-8').startswith('data:'):
                    json_data = chunk.decode('utf-8')[len('data:'):].strip()
                    try:
                        data = json.loads(json_data)
                        text = data.get("content", {}).get("parts", [{}])[0].get("text", "")
                        if text:
                            full_response = text
                    except json.JSONDecodeError:
                        pass # Ignore chunks that are not valid JSON
        return full_response
    except requests.exceptions.RequestException as e:
        return f"An error occurred: {e}"

iface = gr.Interface(
    fn=query_agent,
    inputs=gr.Textbox(lines=2, placeholder="e.g., What's the weather in new york?"),
    outputs="text",
    title="Weather and Time Agent",
    description="Ask a question about the weather or time in a specific location.",
)

if __name__ == "__main__":
    iface.launch()

7. 部署及測試 Gradio 應用程式

在這個步驟中,您會將前端 Gradio 應用程式部署至 Cloud Run。

確認您位於 Gradio 應用程式目錄。

pwd

畫面上會顯示 codelab-gradio-adk/gradio-frontend

現在部署 Gradio 應用程式。

注意:雖然這個 gradio 前端服務是公開網站,但後端服務需要驗證。舉例來說,您可以在這個前端服務中新增使用者驗證 (例如 Firebase 驗證),然後只允許已登入的使用者呼叫後端服務,藉此說明您可能想這麼做的原因。

gcloud run deploy my-adk-gradio-frontend \
--source . \
--region $REGION \
--allow-unauthenticated \
--set-env-vars AGENT_SERVICE_URL=$AGENT_SERVICE_URL,AGENT_APP_NAME=$AGENT_APP_NAME \
--service-account=$SERVICE_ACCOUNT_ADDRESS_GRADIO

部署完成後,請詢問 what's the weather in new york?,您應該會收到類似 The weather in New York is sunny with a temperature of 25 degrees Celsius (77 degrees Fahrenheit). 的回覆

8. 恭喜!

恭喜您完成本程式碼研究室!

建議您參閱這份說明文件,瞭解如何代管 AI 應用程式和代理程式。

涵蓋內容

  • 如何將 ADK 代理程式部署至 Cloud Run
  • 如何將 Gradio 應用程式部署至 Cloud Run
  • 如何在 Cloud Run 中進行經過驗證的服務對服務呼叫

9. 清除

為避免產生意外費用 (例如 Cloud Run 服務的叫用次數不慎超過免費層級的每月 Cloud Run 叫用次數配額),您可以刪除步驟 6 中建立的 Cloud Run 服務。

如要刪除 Cloud Run 服務,請前往 Cloud Run Cloud 控制台 (https://console.cloud.google.com/run),然後刪除 my-adk-gradio-frontendadk-agent-cr 服務。

如要刪除整個專案,請前往「管理資源」,選取您在步驟 2 中建立的專案,然後選擇「刪除」。刪除專案後,您必須在 Cloud SDK 中變更專案。如要查看所有可用專案的清單,請執行 gcloud projects list