agents-cli 实战：用 Locust 对部署在 Agent Runtime 上的生成式 AI Agent 做负载测试

agents-cli 实战：用 Locust 对部署在 Agent Runtime 上的生成式 AI Agent 做负载测试

【免费下载链接】agents-cliThe CLI and skills that turn any coding assistant into an expert at creating, evaluating, and deploying AI agents on Google Cloud.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-cli

本篇指南以 agents-cli 脚手架中 Agent Runtime 部署目标的负载测试模板为核心，讲解如何先通过agents-cli deploy把后端部署到 Google Cloud Agent Runtime，再搭建隔离的 Locust 环境并对部署实例的/run_sse流式接口发起无头（headless）负载测试；读完你将掌握该模板的完整操作步骤、Locust 命令各参数含义，以及load_test.py如何通过deployment_metadata.json自动解析部署地址、模拟会话创建与 SSE 流式对话、并识别限流与错误响应的源码级实现。

一、这套负载测试框架在脚手架中的位置

agents-cli 是一个把任意编码助手变成"在 Google Cloud 上创建、评估、部署 AI Agent 专家"的 CLI 与技能集合。它的scaffold子系统在创建 Agent 项目时，会根据所选部署目标（Agent Runtime、Cloud Run、GKE 或 none）把对应的测试模板渲染到用户项目中。Agent Runtime 的 Python 模板位于：

• 负载测试说明文档
• Locust 负载测试脚本

同一套 Locust 负载测试框架在其他部署目标下也有对应模板，例如 Cloud Run 版本的负载测试说明 和 GKE 版本的负载测试说明。从技能文档 Agent Runtime 参考 中的 CI/CD 对比表可以看到两者的定位差异：Agent Runtime 的负载测试是"通过 locust 打向 Agent Runtime 端点"，而 Cloud Run 是"直接 HTTP 打向 Cloud Run URL"。此外，部署后测试指南 还指出：负载测试会在 staging 阶段的 CD 流水线中自动运行——也就是说，本文介绍的这套脚本不仅是本地手动工具，也是 CI/CD 流水线的组成部分。

二、前置条件：先把后端部署到远端

Agent Runtime 上的 Agent 没有独立的公网服务 URL，Agent Engine 会把容器自身的 HTTP 路由（如/run_sse）通过 reasoningEngines 资源的/api/<route>透传路径对外暴露。因此负载测试的前提是部署必须已经完成。按模板文档的步骤操作：

1. 选择项目并部署后端

gcloud config set project <your-dev-project-id> agents-cli deploy

agents-cli deploy会打包项目文件（遵循根目录的.gcloudignore或.gitignore），由 Agent Engine 构建容器镜像并创建或更新 Agent Runtime 实例。部署成功后，CLI 会在项目根目录写入deployment_metadata.json。这一点在源码中得到印证：deploy/agent_runtime.py 中的write_deployment_metadata函数会写出如下字段：

{ "remote_agent_runtime_id": "projects/PROJECT/locations/LOCATION/reasoningEngines/ENGINE_ID", "deployment_target": "agent_runtime", "is_a2a": true, "agent_directory": "app", "deployment_timestamp": "2025-02-25T10:30:00.000+00:00" }

脚手架模板中预置的 deployment_metadata.json 值均为占位符"None"，真正部署完成后会被覆盖为真实的资源名（完整字段说明见 Agent Runtime 参考）。后续的负载测试脚本正是依赖这个文件来定位部署实例。

三、为 Locust 创建隔离的虚拟环境

Locust 是开源的分布式负载测试工具。模板文档建议在另一个终端标签页中为 Locust 创建独立虚拟环境，避免与应用自身的 Python 环境产生依赖冲突：

python3 -m venv .locust_env && source .locust_env/bin/activate && pip install locust==2.31.1

这里有两个值得注意的细节：

• 版本被固定为locust==2.31.1，保证不同开发者与 CI 环境下压测行为一致；
• 脚手架 Python 模板的 pyproject.toml 在 lint 配置中也跳过了./locust_env/*目录，说明该虚拟环境就是预期放在项目根目录、且不纳入应用环境管理的。

四、执行无头负载测试

激活 Locust 环境后，执行模板文档给出的完整命令：

export _AUTH_TOKEN=$(gcloud auth print-access-token -q) locust -f tests/load_test/load_test.py \ --headless \ -t 30s -u 5 -r 2 \ --csv=tests/load_test/.results/results \ --html=tests/load_test/.results/report.html

各参数含义如下：

模板文档将这一默认配置描述为：发起一次 30 秒的负载测试，按每秒 2 个用户的速度 spawn，达到最大并发。实际压测压力完全由-t/-u/-r三个参数控制，可按部署实例的容量按需放大（例如-u 100 -r 10），无需改动脚本。

关于_AUTH_TOKEN：Agent Runtime 端点使用 Google Cloud 凭证鉴权，脚本在发请求时读取该环境变量并附加Authorization: Bearer头。gcloud auth print-access-token -q会输出当前登录身份的有效访问令牌（-q表示静默）。

五、源码剖析：load_test.py 如何打向 Agent Runtime

load_test.py 只有约 150 行，但完整覆盖了"解析部署地址 → 创建会话 → 流式对话 → 错误/限流判定"的压测闭环，值得逐段理解。

1. 从部署元数据解析 Agent Runtime 地址

脚本开头（L23-L39）读取项目根目录的deployment_metadata.json：

with open("deployment_metadata.json", encoding="utf-8") as f: remote_agent_runtime_id = json.load(f)["remote_agent_runtime_id"] # Format: projects/{project_number}/locations/{location}/reasoningEngines/{id} parts = remote_agent_runtime_id.split("/") project_number = parts[1] location = parts[3] engine_id = parts[5] BASE_HOST = f"https://{location}-aiplatform.googleapis.com" API_PREFIX = ( f"/reasoningEngines/v1/projects/{project_number}" f"/locations/{location}/reasoningEngines/{engine_id}/api" )

这段逻辑与 Agent Runtime 参考文档 描述的/apiHTTP 透传机制严格对应：Agent Engine 对外暴露的完整路径是

https://{location}-aiplatform.googleapis.com/reasoningEngines/v1/{resource}/api/{container_path}

脚本把容器内的 ADK 路由（/run_sse、会话创建路由）挂到/api前缀之后，从而无需公网 Cloud Run URL 就能压测部署实例。这也解释了为什么必须先完成agents-cli deploy——没有deployment_metadata.json里的真实资源名，脚本无法构造任何请求地址。

2. 用户模型：模拟真实对话的两步操作

核心用户类ChatStreamUser（L48-L52）基于locust.HttpUser：

class ChatStreamUser(HttpUser): """Simulates a user interacting with the agent via the ADK /run_sse route.""" wait_time = between(1, 3) # Wait 1-3 seconds between tasks host = BASE_HOST

wait_time = between(1, 3)让每个虚拟用户在两次任务之间随机等待 1~3 秒，模拟人类用户的不规则到达节奏。

每个任务的chat_stream方法分两步：

第一步：创建会话。生成一个基于uuid的user_id（user_{uuid.uuid4()}），POST 到会话创建路由：

session_response = self.client.post( f"{API_PREFIX}/apps/{{cookiecutter.agent_directory}}/users/{user_id}/sessions", name="/api/apps/.../sessions", headers=headers, json={"state": {"preferred_language": "English", "visit_count": 1}}, )

这里有两点说明：

• {{cookiecutter.agent_directory}}是 cookiecutter 模板占位符，脚手架生成项目时会被替换为真实的 agent 目录名（通常与deployment_metadata.json中的agent_directory字段一致），即压测路径最终形如/api/apps/app/users/.../sessions；
• 若会话创建未返回 200，脚本调用session_response.failure(...)将其记为一次失败请求后直接返回，不再继续发对话——与真实客户端"没有会话就不发消息"的行为一致。成功后从响应体取id作为session_id。

第二步：流式发送对话消息。向/run_sse发起 SSE 流式请求：

data = { "app_name": "{{cookiecutter.agent_directory}}", "user_id": user_id, "session_id": session_id, "new_message": { "role": "user", "parts": [{"text": "Hello! Weather in New york?"}], }, "streaming": True, } with self.client.post( f"{API_PREFIX}/run_sse", name="/api/run_sse message", headers=headers, json=data, catch_response=True, stream=True, params={"alt": "sse"}, ) as response: ...

请求体遵循 ADK HTTP 服务的new_message/parts消息模式；params={"alt": "sse"}让 Agent Runtime 以 SSE 方式返回事件流；catch_response=True把成功/失败的判定权交给脚本自己——这对流式接口很重要，因为"HTTP 状态码 200"并不代表内容正确。

鉴权头的组装（L57-L59）：

headers = {"Content-Type": "application/json"} if os.environ.get("_AUTH_TOKEN"): headers["Authorization"] = f"Bearer {os.environ['_AUTH_TOKEN']}"

即未导出_AUTH_TOKEN时请求不带鉴权头——Agent Runtime 端点会拒绝这类请求，这正是文档把export _AUTH_TOKEN=...列为执行步骤的原因。

3. 流式响应中的错误与限流判定

SSE 响应逐行解析（L98-L148），脚本做三层判定：

1. 限流（429）单独记账。若某行包含"429 Too Many Requests"，通过self.environment.events.request.fire(...)额外上报一条名为/api/run_sse rate_limited 429s的请求事件。这样在 Locust 统计报表中，限流会作为独立指标出现，便于观察部署实例的配额与容量水位，而不是被淹没在普通请求耗时里。

2. 事件级业务错误。对每行尝试json.loads，若解析出的字典带有code字段且code >= 400，则标记has_error = True并调用response.failure(...)把该请求计为失败，同时用logger.error打印具体 code 与 message。这意味着即使 HTTP 层是 200，SSE 事件体内携带的错误也会被如实计入失败率。

3. 端到端耗时上报。仅当整条流没有发现错误时，才通过environment.events.request.fire上报一条/api/run_sse end事件，response_time为从发起请求到读完整个流的总耗时（毫秒），response_length为接收到的事件行数。因此最终报表里/api/run_sse end指标反映的是完整流式响应的端到端延迟，这对评估生成式应用的流式体验（首字之后的整体吞吐感受）比单请求 RT 更贴近真实场景。

若 HTTP 状态码不是 200，则直接response.failure(f"Unexpected status code: ...")。

六、实操检查清单

把模板文档与源码结合起来，执行前可以按以下清单自检：

1. 部署完成：项目根目录存在且deployment_metadata.json中remote_agent_runtime_id为真实的projects/.../reasoningEngines/...资源名（而非占位符"None"），可参考 Agent Runtime 参考 中"若部署超时但引擎已创建，需手工补写该文件"的提示；
2. 凭证有效：gcloud auth print-access-token能正常输出令牌；脚本以_AUTH_TOKEN环境变量读取它；
3. 环境隔离：Locust 装在独立 venv（locust==2.31.1），未污染应用环境；
4. 产物归档：CSV（.results/results*）与 HTML（.results/report.html）落在tests/load_test/.results/下，.results目录名以点开头，配合脚手架的 lint/忽略配置不会干扰项目本身；
5. 手动验证先行：压测前可先用agents-cli run --url对同一部署实例做单发冒烟验证，方法见 部署后测试指南。

七、小结

agents-cli 为 Agent Runtime 部署目标预置的这套负载测试模板，把"部署后压测"收敛为三步：agents-cli deploy写入部署元数据 → 隔离 venv 安装固定版本 Locust → headless 执行对/run_sse的流式压测。其设计上有三个可借鉴的工程点：用deployment_metadata.json作为部署脚本与压测脚本之间的"契约"，免去手工填 URL；用独立事件名把 429 限流从普通请求中剥离出来单独观测；用"事件体内code >= 400也计失败"的判定，让流式接口的成功率统计真正反映业务正确性而非仅看 HTTP 状态码。若你的项目选择的是其他部署目标，对应的 Locust 模板位于各自的deployment_targets/<target>/python/tests/load_test/目录下，可按同样思路复用。

【免费下载链接】agents-cliThe CLI and skills that turn any coding assistant into an expert at creating, evaluating, and deploying AI agents on Google Cloud.项目地址: https://gitcode.com/GitHub_Trending/ag/agents-cli