如何部署一个调用后端 ADK 代理的 Gradio 前端应用,这两个应用都在 Cloud Run 上运行

1. 简介

概览

在此 Codelab 中,您将 ADK 代理部署到 Cloud Run 作为后端服务,然后将 ADK 代理的 Gradio 前端部署为第二个 Cloud Run 服务。此 Codelab 演示了如何要求对 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

本 Codelab 需要启用以下 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 服务都有自己的服务账号。

首先,为本 Codelab 设置环境变量,这些变量将在整个 Codelab 中使用。

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 User”角色

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 Invoker 角色,以便其调用托管在 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

首先,为整个 Codelab 创建一个目录

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 提供的开发者网页界面验证部署是否成功。最后,您需要对该服务进行经过身份验证的调用。

前往父文件夹。

注意: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

接下来,将该网址保存为环境变量,以便在本 Codelab 的第二部分中使用

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 前端

注意:您可以将 Gradio 应用与 ADK 代理放在同一服务中。此 Codelab 提供了 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 Auth),然后仅允许已登录的用户调用后端服务。

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. 恭喜!

恭喜您完成此 Codelab!

建议您查看有关托管 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 查看所有可用项目的列表。