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 代理跟踪到 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 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 编辑器来完成此
- 点击“打开编辑器”按钮,系统会打开 Cloud Shell 编辑器,我们可以在这里编写代码
- 确保 Cloud Code 项目已在 Cloud Shell 编辑器的左下角(状态栏)中设置,如下图中突出显示的那样,并且已设置为已启用结算的有效 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 版本和虚拟环境的需求
- 如果您尚未打开终端,请依次点击终端 -> 新建终端,或使用 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 输入/输出快速尝试互动
- 快速设置本地开发界面网页界面
现在,我们来检查 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
以支持作为天气智能体的代理功能
运行网页界面
现在,我们可以在本地与代理互动并检查其行为。借助 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 编辑器顶部区域的网页预览按钮,然后选择在端口 8080 上预览
您将看到以下网页,您可以在左上角的下拉按钮中选择可用的代理(在本例中应为 weather_agent),然后与机器人互动。您将在左侧窗口中看到有关代理运行时日志详情的许多信息
现在,尝试与之互动。在左侧栏中,我们可以检查每个输入的轨迹,从而了解代理在形成最终答案之前执行每个操作所花费的时间。
这是 ADK 中内置的可观测性功能之一,目前我们在本地检查它。稍后,我们将了解如何将其集成到 Cloud Trace 中,以便集中跟踪所有请求
4. 后端服务器脚本
为了使代理可以作为服务访问,我们将代理封装在 FastAPI 应用中。我们可以在此处配置必要的服务来支持代理,例如为生产目的准备 Session、Memory 或 Artifact 服务。以下是将使用的 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% 的流量,您可以在此处点击 Kebab 按钮 (⋮),然后选择管理流量
在新弹出的窗口中,您可以修改流向各个修订版本的流量百分比。
等待一段时间后,流量将根据百分比配置按比例分配。这样一来,如果新版本出现问题,我们就可以轻松回滚到之前的修订版本
8. ADK 跟踪
使用 ADK 构建的代理已支持使用嵌入其中的开放遥测技术进行跟踪。我们有 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 中检查导出器的详细信息。在此处,我们创建了一个自定义导出器,因为可以导出到 Cloud Trace 的轨迹数据量有限。我们使用 https://googlecloudplatform.github.io/agent-starter-pack/guide/observability.html 中的实现来实现此跟踪功能。
尝试访问服务 Web 开发界面,并与代理进行对话。之后,前往 Cloud 控制台搜索栏,输入“trace 探索器”,然后选择其中的 Trace 探索器产品
在跟踪探索器页面上,您会看到与代理的对话跟踪记录已提交。您可以从 Span name 部分中看到,并过滤掉特定于我们代理的 span(命名为 agent_run [weather_agent]
)
如果已过滤 span,您还可以直接检查每个轨迹。它会显示代理采取的每项操作的详细时长。例如,请参阅下图
在每个部分中,您都可以检查属性中的详细信息,如下所示
这样一来,我们就能够很好地观测到代理与用户的每次互动,并获取相关信息,从而帮助我们调试问题。欢迎随意尝试各种工具或工作流!
9. 挑战
尝试多智能体或智能体工作流,看看它们在负载下的表现如何,以及轨迹是什么样的
10. 清理
为避免系统因本 Codelab 中使用的资源向您的 Google Cloud 账号收取费用,请按照以下步骤操作: