1. 简介
本教程将指导您在 Google Cloud Run 上使用代理开发套件 (ADK) 构建强大的代理，并对其进行部署、管理和监控。借助 ADK，您可以创建能够处理复杂多智能体工作流的智能体。借助 Cloud Run（一种全代管式无服务器平台），您可以将代理部署为可扩缩的容器化应用，而无需担心底层基础架构。借助这两者的强大组合，您可以专注于代理的核心逻辑，同时受益于 Google Cloud 强大且可伸缩的环境。
在本教程中，我们将探索如何将 ADK 与 Cloud Run 无缝集成。您将了解如何部署代理，然后深入了解在类似生产环境的设置中管理应用的实际方面。我们将介绍如何通过管理流量安全地发布代理的新版本，以便您在全面发布之前，先让部分用户测试新功能。
此外，您还将获得监控客服人员表现的实践经验。我们将通过进行负载测试来模拟真实场景，以便观察 Cloud Run 的自动扩缩功能的运作方式。为了更深入地了解您的代理的行为和性能，我们将使用 Cloud Trace 启用跟踪功能。这样，您就可以详细了解请求在代理中传输的端到端情况，从而找出并解决任何性能瓶颈。在本教程结束时，您将全面了解如何在 Cloud Run 上有效部署、管理和监控由 ADK 提供支持的代理。
在本 Codelab 中，您将采用分步方法，具体步骤如下：
- 在 CloudSQL 上创建一个 PostgreSQL 数据库，以供 ADK Agent 数据库会话服务使用
- 设置基本 ADK 代理
- 设置供 ADK 运行程序使用的数据库会话服务
- 将代理初始部署到 Cloud Run
- 对 Cloud Run 自动扩缩进行负载测试和检查
- 部署新的代理修订版本，并逐步将流量增加到新修订版本
- 设置云跟踪并检查代理运行跟踪
架构概览
前提条件
- 熟练使用 Python
- 了解使用 HTTP 服务的基本全栈架构
学习内容
- ADK 结构和本地实用程序
- 使用数据库会话服务设置 ADK 代理
- 在 CloudSQL 中设置 PostgreSQL，以供数据库会话服务使用
- 使用 Dockerfile 将应用部署到 Cloud Run 并设置初始环境变量
- 使用负载测试配置和测试 Cloud Run 自动扩缩
- 使用 Cloud Run 进行逐步发布的策略
- 将 ADK Agent 跟踪设置为 Cloud Trace
所需条件
- Chrome 网络浏览器
- Gmail 账号
- 启用了结算功能的 Cloud 项目
此 Codelab 面向各种级别（包括新手）的开发者，其示例应用中使用了 Python。不过，您无需了解 Python 知识即可理解所介绍的概念。
2. 准备工作
在 Cloud 控制台中选择“有效项目”
本 Codelab 假定您已有一个启用了结算功能的 Google Cloud 项目。如果您还没有，可以按照以下说明开始使用。
- 在 Google Cloud Console 的项目选择器页面上，选择或创建一个 Google Cloud 项目。
- 确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。
准备 Cloud SQL 数据库
我们需要一个数据库，以供 ADK 代理稍后使用。我们来在 Cloud SQL 上创建一个 PostgreSQL 数据库。首先，前往 Cloud 控制台顶部的搜索栏，然后输入“Cloud SQL”。然后点击 Cloud SQL 产品
之后，我们需要创建一个新的数据库实例，点击创建实例，然后选择 PostgreSQL
如果您从新项目开始，可能还需要启用 Compute Engine API，如果系统显示此提示，只需点击启用 API 即可
接下来，我们将选择数据库的规格，选择企业版（使用沙盒版预设）
然后，在此处为用户 postgres 设置实例名称和默认密码。您可以使用任何凭据进行设置，但在本教程中，我们将实例名称和密码都设置为“adk-deployment”
在本教程中，我们使用单个可用区 us-central1，然后点击创建实例按钮，完成数据库创建并让系统完成所有必要的设置
等待此操作完成期间，我们可以继续下一部分
在 Cloud Shell 终端中设置 Cloud 项目
- 您将使用 Cloud Shell，它是在 Google Cloud 中运行的命令行环境。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。
- 连接到 Cloud Shell 后，您可以使用以下命令检查自己是否已通过身份验证，以及项目是否已设置为您的项目 ID：
gcloud auth list
- 在 Cloud Shell 中运行以下命令，以确认 gcloud 命令了解您的项目。
gcloud config list project
- 如果项目未设置，请使用以下命令进行设置：
gcloud config set project <YOUR_PROJECT_ID>
或者，您也可以在控制台中查看 PROJECT_ID ID
点击它，您会在右侧看到您的所有项目和项目 ID
- 通过以下命令启用所需的 API。此过程可能需要几分钟的时间，请耐心等待。
gcloud services enable aiplatform.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
cloudresourcemanager.googleapis.com \
sqladmin.googleapis.com
成功执行该命令后，您应该会看到如下所示的消息：
Operation "operations/..." finished successfully.
您可以通过控制台搜索各个产品或使用此链接，以替代 gcloud 命令。
如果缺少任何 API，您随时可以在实现过程中启用它。
如需了解 gcloud 命令和用法，请参阅文档。
前往 Cloud Shell 编辑器并设置应用工作目录
现在，我们可以设置代码编辑器来进行一些编码操作了。我们将使用 Cloud Shell Editor 来完成此操作
- 点击“打开编辑器”按钮，这将打开 Cloud Shell 编辑器，我们可以在此处编写代码
- 确保在 Cloud Shell 编辑器的左下角（状态栏）中设置 Cloud Code 项目（如下图所示），并将其设置为已启用结算功能的有效 Google Cloud 项目。在系统提示时授权。如果您已按照上一个命令操作，该按钮可能也会直接指向已启用的项目，而不是登录按钮
- 接下来，我们从 GitHub 克隆此 Codelab 的模板工作目录，运行以下命令。它将在 deploy_and_manage_adk 目录中创建工作目录
git clone https://github.com/alphinside/deploy-and-manage-adk-service.git deploy_and_manage_adk
- 然后，前往 Cloud Shell 编辑器的顶部，依次点击文件->打开文件夹，找到您的用户名目录，找到 deploy_and_manage_adk 目录，然后点击“确定”按钮。这会将所选目录设为主工作目录。在此示例中，用户名为 alvinprayuda，因此目录路径如下所示
现在，您的 Cloud Shell 编辑器应如下所示
接下来，我们可以配置 Python 环境设置
环境设置
准备 Python 虚拟环境
下一步是准备开发环境。您当前使用的终端工作目录应位于 deploy_and_manage_adk 工作目录中。在本 Codelab 中，我们将使用 Python 3.12，并使用 uv Python 项目管理器来简化创建和管理 Python 版本和虚拟环境的流程
- 如果您尚未打开终端，请依次点击 Terminal（终端）-> New Terminal（新建终端），或使用 Ctrl + Shift + C 打开终端，它会在浏览器底部打开一个终端窗口
- 使用以下命令下载
uv并安装 Python 3.12
curl -LsSf https://astral.sh/uv/0.6.16/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
- 现在，我们使用
uv初始化虚拟环境，运行以下命令
uv sync --frozen
这会创建 .venv 目录并安装依赖项。快速浏览 pyproject.toml 即可了解依赖项的相关信息，如下所示
dependencies = [
"google-adk==1.3.0",
"locust==2.37.10",
"pg8000==1.31.2",
"python-dotenv==1.1.0",
]
- 如需测试虚拟环境，请创建新文件 main.py 并复制以下代码
def main():
print("Hello from deploy_and_manage_adk!")
if __name__ == "__main__":
main()
- 然后，运行以下命令
uv run main.py
您将会看到如下所示的输出
Using CPython 3.12 Creating virtual environment at: .venv Hello from deploy_and_manage_adk!
这表明 Python 项目已正确设置。
设置配置文件
现在，我们需要为此项目设置配置文件。
将 .env.example 文件重命名为 .env，它将显示以下值。将 GOOGLE_CLOUD_PROJECT 值更新为您的项目 ID
# Google Cloud and Vertex AI configuration GOOGLE_CLOUD_PROJECT=your-project-id GOOGLE_CLOUD_LOCATION=global GOOGLE_GENAI_USE_VERTEXAI=True # Database connection for session service # SESSION_SERVICE_URI=postgresql+pg8000://<username>:<password>@/<database>?unix_sock=/cloudsql/<instance_connection_name>/.s.PGSQL.5432
在本 Codelab 中，我们将使用为 GOOGLE_CLOUD_LOCATION 和 GOOGLE_GENAI_USE_VERTEXAI. 预配置的值。目前，我们将 SESSION_SERVICE_URI 注释掉。
现在，我们可以进入下一步，检查代理逻辑并进行部署
3. 使用 ADK 和 Gemini 2.5 构建天气智能体
ADK 目录结构简介
首先，我们来探索 ADK 提供的功能以及如何构建代理。您可以访问此网址，查看 ADK 完整文档。ADK 在其 CLI 命令执行中提供了许多实用程序。其中一些包括：
- 设置代理目录结构
- 快速尝试通过 CLI 输入输出进行互动
- 快速设置本地开发界面 Web 界面
现在，我们来检查 weather_agent 目录中的代理结构
weather_agent/ ├── __init__.py ├── agent.py
如果您检查 init.py 和 agent.py，会看到以下代码
# __init__.py
from weather_agent.agent import root_agent
__all__ = ["root_agent"]
# agent.py
import os
from pathlib import Path
import google.auth
from dotenv import load_dotenv
from google.adk.agents import Agent
from google.cloud import logging as google_cloud_logging
# Load environment variables from .env file in root directory
root_dir = Path(__file__).parent.parent
dotenv_path = root_dir / ".env"
load_dotenv(dotenv_path=dotenv_path)
# Use default project from credentials if not in .env
_, project_id = google.auth.default()
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", project_id)
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "global")
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "True")
logging_client = google_cloud_logging.Client()
logger = logging_client.logger("weather-agent")
def get_weather(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Args:
city (str): The name of the city (e.g., "New York", "London", "Tokyo").
Returns:
dict: A dictionary containing the weather information.
Includes a 'status' key ('success' or 'error').
If 'success', includes a 'report' key with weather details.
If 'error', includes an 'error_message' key.
"""
logger.log_text(
f"--- Tool: get_weather called for city: {city} ---", severity="INFO"
) # Log tool execution
city_normalized = city.lower().replace(" ", "") # Basic normalization
# Mock weather data
mock_weather_db = {
"newyork": {
"status": "success",
"report": "The weather in New York is sunny with a temperature of 25°C.",
},
"london": {
"status": "success",
"report": "It's cloudy in London with a temperature of 15°C.",
},
"tokyo": {
"status": "success",
"report": "Tokyo is experiencing light rain and a temperature of 18°C.",
},
}
if city_normalized in mock_weather_db:
return mock_weather_db[city_normalized]
else:
return {
"status": "error",
"error_message": f"Sorry, I don't have weather information for '{city}'.",
}
root_agent = Agent(
name="weather_agent",
model="gemini-2.5-flash",
instruction="You are a helpful AI assistant designed to provide accurate and useful information.",
tools=[get_weather],
)
ADK 代码说明
此脚本包含代理初始化，其中我们会初始化以下内容：
- 将要使用的模型设置为
gemini-2.5-flash - 提供工具
get_weather，以支持作为天气代理的代理功能
运行 Web 界面
现在，我们可以在本地与代理进行交互并检查其行为。借助 ADK，我们可以使用开发 Web 界面进行互动，并检查互动期间发生的情况。运行以下命令以启动本地开发界面服务器
uv run adk web --port 8080
它将生成类似于以下示例的输出，表示我们已经可以访问 Web 界面
INFO: Started server process [xxxx] INFO: Waiting for application startup. +-----------------------------------------------------------------------------+ | ADK Web Server started | | | | For local testing, access at http://localhost:8080. | +-----------------------------------------------------------------------------+ INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
现在，如需进行检查，请点击 Cloud Shell 编辑器顶部区域的 Web Preview 按钮，然后选择 Preview on port 8080
您会看到以下网页，您可以在左上角的下拉按钮中选择可用的代理（在本例中，应该是 weather_agent），并与聊天机器人互动。您会在左侧窗口中看到有关代理运行时日志详情的许多信息
现在，尝试与它互动。在左侧栏中，我们可以检查每个输入的轨迹，以了解代理执行每项操作并形成最终答案所需的时间。
这是 ADK 中内置的可观测性功能之一，目前我们在本地检查它。稍后，我们将了解如何将其集成到 Cloud Tracing，以便集中跟踪所有请求
4. 后端服务器脚本
为了以服务的形式访问代理，我们将代理封装在 FastAPI 应用中。我们可以在此处配置必要的服务来支持代理，例如准备会话、内存或工件服务以供生产环境使用。以下是将要使用的 server.py 代码
import os
from dotenv import load_dotenv
from fastapi import FastAPI
from google.adk.cli.fast_api import get_fast_api_app
from pydantic import BaseModel
from typing import Literal
from google.cloud import logging as google_cloud_logging
from tracing import CloudTraceLoggingSpanExporter
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, export
# Load environment variables from .env file
load_dotenv()
logging_client = google_cloud_logging.Client()
logger = logging_client.logger(__name__)
AGENT_DIR = os.path.dirname(os.path.abspath(__file__))
# Get session service URI from environment variables
session_uri = os.getenv("SESSION_SERVICE_URI", None)
# Prepare arguments for get_fast_api_app
app_args = {"agents_dir": AGENT_DIR, "web": True}
# Only include session_service_uri if it's provided
if session_uri:
app_args["session_service_uri"] = session_uri
else:
logger.log_text(
"SESSION_SERVICE_URI not provided. Using in-memory session service instead. "
"All sessions will be lost when the server restarts.",
severity="WARNING",
)
provider = TracerProvider()
processor = export.BatchSpanProcessor(CloudTraceLoggingSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
# Create FastAPI app with appropriate arguments
app: FastAPI = get_fast_api_app(**app_args)
app.title = "weather-agent"
app.description = "API for interacting with the Agent weather-agent"
class Feedback(BaseModel):
"""Represents feedback for a conversation."""
score: int | float
text: str | None = ""
invocation_id: str
log_type: Literal["feedback"] = "feedback"
service_name: Literal["weather-agent"] = "weather-agent"
user_id: str = ""
@app.post("/feedback")
def collect_feedback(feedback: Feedback) -> dict[str, str]:
"""Collect and log feedback.
Args:
feedback: The feedback data to log
Returns:
Success message
"""
logger.log_struct(feedback.model_dump(), severity="INFO")
return {"status": "success"}
# Main execution
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8080)
服务器代码说明
以下是在 server.py 脚本中定义的内容：
- 使用
get_fast_api_app方法将我们的代理转换为 FastAPI 应用。这样，我们将继承用于 Web 开发界面的相同路由定义。 - 通过向
get_fast_api_app方法添加关键字参数，配置必要的会话、内存或工件服务。在本教程中，如果我们配置SESSION_SERVICE_URI环境变量，则会话服务将使用该变量；否则，它将使用内存中会话 - 我们可以添加自定义路由来支持其他后端业务逻辑，在脚本中，我们添加了反馈功能路由示例
- 启用 Cloud 跟踪，以将轨迹发送到 Google Cloud Trace
5. 部署到 Cloud Run
现在，我们将此代理服务部署到 Cloud Run。在本演示中，此服务将作为可供他人访问的公有服务公开。不过，请注意，这不是最佳做法，因为它不安全
在此 Codelab 中，我们将使用 Dockerfile 将代理部署到 Cloud Run。以下是将要使用的 Dockerfile 内容
FROM python:3.12-slim
RUN pip install --no-cache-dir uv==0.7.13
WORKDIR /app
COPY . .
RUN uv sync --frozen
EXPOSE 8080
CMD ["uv", "run", "uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]
至此，我们已经有了将应用部署到 Cloud Run 所需的所有文件，接下来就开始部署吧。前往 Cloud Shell 终端，并确保当前项目已配置为您的有效项目，如果没有，请使用 gcloud configure 命令设置项目 ID：
gcloud config set project [PROJECT_ID]
然后，运行以下命令将其部署到 Cloud Run。
gcloud run deploy weather-agent \
--source . \
--port 8080 \
--project {YOUR_PROJECT_ID} \
--allow-unauthenticated \
--add-cloudsql-instances {YOUR_DB_CONNECTION_NAME} \
--update-env-vars SESSION_SERVICE_URI="postgresql+pg8000://postgres:{YOUR_DEFAULT_USER_PASS}@postgres/?unix_sock=/cloudsql/{YOUR_DB_CONNECTION_NAME}/.s.PGSQL.5432",GOOGLE_CLOUD_PROJECT={YOUR_PROJECT_ID} \
--region us-central1
如需获取 {YOUR_DB_CONNECTION_NAME} 值，您可以再次前往 Cloud SQL，然后点击您创建的实例。在实例页面中，向下滚动到“连接到此实例”部分，然后复制连接名称以替换 {YOUR_DB_CONNECTION_NAME} 值。例如，请参阅下图
如果系统提示您确认为 Docker 仓库创建了工件注册库，请直接回答 Y。请注意，由于这是演示版应用，因此我们允许未经身份验证的访问。建议为企业应用和生产应用使用适当的身份验证。
部署完成后，您应该会收到类似于以下内容的链接：
https://weather-agent-*******.us-central1.run.app
接下来，您可以通过无痕式窗口或移动设备使用您的应用。它应该已经上线。
6. 使用负载测试检查 Cloud Run 自动扩缩
现在，我们来检查 Cloud Run 的自动扩缩功能。对于此场景，我们将部署新修订版本，同时启用每个实例的最大并发数。运行以下命令
gcloud run deploy weather-agent \
--source . \
--port 8080 \
--project {YOUR_PROJECT_ID} \
--allow-unauthenticated \
--region us-central1 \
--concurrency 10
然后，我们来检查 load_test.py 文件。我们将使用此脚本通过 locust 框架进行负载测试。此脚本将执行以下操作：
- 随机化的 user_id 和 session_id
- 为 user_id 创建 session_id
- 使用创建的 user_id 和 session_id 命中端点“/run_sse”
如果您错过了，我们需要知道已部署的服务网址。前往 Cloud Run 控制台，然后点击您的 weather-agent 服务
然后，找到您的 weather-agent 服务并点击它
服务网址将显示在“区域”信息旁边。例如，
然后，运行以下命令进行负载测试
uv run locust -f load_test.py \
-H {YOUR_SERVICE_URL} \
-u 60 \
-r 5 \
-t 120 \
--headless
运行此脚本后，您会看到如下所示的指标。（在此示例中，所有请求均成功）
Type Name # reqs # fails | Avg Min Max Med | req/s failures/s
--------|------------------------------------|-------|-------------|-------|-------|-------|-------|--------|-----------
POST /run_sse end 813 0(0.00%) | 5817 2217 26421 5000 | 6.79 0.00
POST /run_sse message 813 0(0.00%) | 2678 1107 17195 2200 | 6.79 0.00
--------|------------------------------------|-------|-------------|-------|-------|-------|-------|--------|-----------
Aggregated 1626 0(0.00%) | 4247 1107 26421 3500 | 13.59 0.00
接下来，我们来看看 Cloud Run 中发生了什么，再次前往已部署的服务，然后查看信息中心。这将显示 Cloud Run 如何自动扩缩实例以处理传入请求。由于我们将每个实例的并发数上限限制为 10，因此 Cloud Run 实例会尝试自动调整容器数量以满足此条件。
7. 逐步发布新修订版
现在，我们假设以下场景。我们希望将客服人员的提示更新为以下内容：
# agent.py
...
root_agent = Agent(
name="weather_agent",
model="gemini-2.5-flash-preview-05-20",
instruction="You are a helpful AI assistant designed to provide accurate and useful information. You only answer inquiries about the weather. Refuse all other user query",
tools=[get_weather],
)
然后，您想发布新修订版，但不想将所有请求流量直接转到新版本。我们可以使用 Cloud Run 进行逐步发布。首先，我们需要部署一个新修订版本，但要添加 –no-traffic 标志。保存上一个代理脚本，然后运行以下命令
gcloud run deploy weather-agent \
--source . \
--port 8080 \
--project {YOUR_PROJECT_ID} \
--allow-unauthenticated \
--region us-central1 \
--no-traffic
完成后，您会收到与上一个部署流程类似的日志，只不过其中的流量数有所不同。系统会显示0% 的流量投放量。
接下来，我们前往 Cloud Run 产品页面，找到您部署的实例。在搜索栏中输入“Cloud Run”，然后点击 Cloud Run 产品
然后，找到您的 weather-agent 服务并点击它
前往修订版本标签页，您会在其中看到已部署的修订版本列表
您会看到新部署的修订版本的广告投放量为 0%，然后点击三线状按钮 (⋮) 并选择管理流量
在新弹出的窗口中，您可以修改分配给各个修订版本的流量百分比。
等待一段时间后，系统会根据百分比配置按比例分配流量。这样一来，如果新版本出现问题，我们就可以轻松回滚到之前的修订版本
8. ADK 跟踪
使用 ADK 构建的代理已经支持使用嵌入其中的 OpenTelemetry 进行跟踪。我们有 Cloud Trace 来捕获这些跟踪记录并直观呈现它们。我们来检查 server.py，了解如何在之前部署的服务中启用它
# server.py
from tracing import CloudTraceLoggingSpanExporter
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, export
...
provider = TracerProvider()
processor = export.BatchSpanProcessor(CloudTraceLoggingSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
...
在这里，我们初始化跟踪器和导出器。您可以在 tracing.py 中检查导出器的详细信息。在这里，我们创建了一个自定义导出程序，因为可导出到云轨迹的轨迹数据存在上限。我们将使用 https://googlecloudplatform.github.io/agent-starter-pack/guide/observability.html 中的实现来实现此跟踪功能。
尝试访问您的服务 Web 开发者界面，并与客服人员聊天。然后，前往 Cloud 控制台搜索栏，输入“Trace 探索器”，然后选择 Trace 探索器产品
在轨迹浏览器页面上，您会看到我们与代理的对话已提交轨迹。您可以从代码段名称部分查看并过滤出特定于我们客服人员的代码段（名为 agent_run [weather_agent]）
过滤完跨度后，您还可以直接检查每个轨迹。它会显示客服人员执行的每项操作的详细时长。例如，请查看以下图片
在每个部分中，您都可以检查属性中的详细信息，如下所示
现在，我们可以很好地观察代理与用户的每一次互动，并获取相关信息，以帮助调试问题。欢迎随意尝试各种工具或工作流！
9. 挑战
尝试多智能体或智能体工作流，了解它们在负载下的表现以及轨迹的显示方式
10. 清理
为避免系统因本 Codelab 中使用的资源向您的 Google Cloud 账号收取费用，请按照以下步骤操作：