企业官网建设流程全解析

agents-cli 本地压测模板实战：用 Locust 对 ADK 智能体 /run_sse 流式接口做端到端负载测试

【免费下载链接】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 脚手架中none部署目标（纯本地运行）自带的负载测试模板展开，完整讲解其 README 中定义的三步本地压测流程（启动 FastAPI 服务、隔离安装 Locust、执行无头压测并产出 CSV/HTML 报告），并结合同目录的 load_test.py 源码剖析压测脚本如何模拟真实的“建会话 + SSE 流式对话”链路、如何识别限流与业务级错误，使读者能够直接复用该模板为自己的生成式 AI 智能体建立可重复的本地压测基线。

一、模板定位：none部署目标下的本地压测目录

agents-cli 通过 cookiecutter 模板生成智能体项目，其中deployment_target取none表示项目只运行在本地（不部署到 Cloud Run、GKE 或 Agent Runtime）。该目标生成的项目中会包含如下测试结构：

tests/ ├── integration/ │ └── test_server_e2e.py # 服务端 E2E 测试（pytest + uvicorn 子进程） └── load_test/ ├── README.md # 本文章节的原始文档 └── load_test.py # Locust 压测脚本

原始文档 README.md 开篇即说明：该目录提供了一套面向生成式 AI 应用的完整负载测试框架，底层使用开源压测工具 Locust。其核心思路是：压测对象不是简单的 HTTP 端点，而是 ADK（Agent Development Kit）FastAPI 应用暴露的SSE（Server-Sent Events）流式对话接口/run_sse——即真实用户与智能体交互所走的完整路径。

模板中的{{cookiecutter.agent_directory}}是 cookiecutter 占位符，scaffold 生成项目时会被替换为实际的智能体目录名，下文命令中的该写法在生成的项目中会呈现为真实包名。

二、三步完成本地负载测试（继承原文档操作）

第 1 步：启动 FastAPI 服务

在独立终端中启动生成的智能体服务：

uv run uvicorn {{cookiecutter.agent_directory}}.fast_api_app:app --host 0.0.0.0 --port 8000 --reload

• {{cookiecutter.agent_directory}}.fast_api_app:app：指向生成项目的 FastAPI 应用入口。从 fast_api_app.py 的模板源码看，该app由 ADK 的get_fast_api_app()构建，因此/run_sse、会话管理（/apps/{app}/users/{user}/sessions）等 HTTP 路由均由 ADK 的 fast_api 模块提供，而非项目手写路由；
• --host 0.0.0.0：允许本机任意回环地址访问（压测端使用127.0.0.1即可命中）；
• --reload：开发期热重载，正式压测时可去掉以避免重载干扰指标。

第 2 步：为 Locust 创建独立虚拟环境

在另一个终端标签页中执行（原文档建议这样做，以避免与应用自身的 Python 环境产生依赖冲突）：

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

这里把 Locust 固定在2.31.1版本，保证压测脚本的 Locust API（HttpUser、between、task、catch_response等）行为可复现。

第 3 步：执行无头（headless）压测

locust -f tests/load_test/load_test.py \ -H http://127.0.0.1:8000 \ --headless \ -t 30s -u 10 -r 2 \ --csv=tests/load_test/.results/results \ --html=tests/load_test/.results/report.html

各参数含义如下：

原文档对负载形态的描述为：“发起一次 30 秒的负载测试，按每秒 2 个用户的速率拉起用户，达到最多 60 个并发用户”（即 2 users/s × 30 s 的计算）。需要注意按 Locust 的标准 flag 语义，-u 10决定的是最终并发用户总数，-r 2是爬坡速率；如需更大的并发规模，直接调大-u与-r即可，例如-u 100 -r 10。

结果产物：综合的 CSV 与 HTML 性能报告会生成在tests/load_test/.results目录中，包含响应时间分布、吞吐量、失败数等指标；HTML 报告可直接在浏览器打开查看。

三、压测脚本源码剖析：它到底在压什么

load_test.py 定义了一个ChatStreamUser用户类，完整模拟“真实用户”的一次对话交互，而不是一次裸请求。

3.1 用户模型与请求节奏

class ChatStreamUser(HttpUser): """Simulates a user interacting with the chat stream API.""" wait_time = between(1, 3) # Wait 1-3 seconds between tasks

每个虚拟用户在两次任务之间随机等待 1～3 秒（load_test.py#L32-L37），模拟人类点击节奏而非恒定满速打点，使 CPU/模型侧的负载更接近真实线上形态。

3.2 两段式请求：先建会话，再发流式消息

任务函数chat_stream（load_test.py#L37-L54）首先为每个虚拟用户生成随机user_id（user_{uuid4}），然后调用 ADK 的会话创建接口：

session_response = self.client.post( f"/apps/{{cookiecutter.agent_directory}}/users/{user_id}/sessions", headers=headers, json=session_data, # {"state": {"preferred_language": "English", "visit_count": 1}} ) session_id = session_response.json()["id"]

随后以session_id发送流式对话消息（load_test.py#L56-L77）：

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( ENDPOINT, # "/run_sse" name=f"{ENDPOINT} message", headers=headers, json=data, catch_response=True, # 手动标记成功/失败 stream=True, # 流式读取 SSE 响应 params={"alt": "sse"}, ) as response:

这段请求体的 schema 与同目录 E2E 测试 test_server_e2e.py 中test_adk_run_sse的用例完全一致——后者先 POST 创建会话、再带stream=True读取data:前缀的 SSE 行并断言存在带文本的content.parts事件。两者相互印证了 ADK HTTP 面的标准调用契约：new_message+parts结构（仓库的部署测试技能文档中也特别提示：把消息体写成{"message": ...}会收到422 Field required，且会话必须先创建）。

3.3 双重错误检测：限流与业务错误码

SSE 场景下“HTTP 200 不代表请求成功”，脚本因此实现了两层错误识别（load_test.py#L78-L128）：

1. 限流检测：逐行扫描 SSE 流，若行内出现429 Too Many Requests，额外触发一条名为{ENDPOINT} rate_limited 429s的统计事件（response_time=0），在报告中可以单独看到限流命中次数；
2. 业务错误码检测：尝试把每一行解析为 JSON，若事件体包含code字段且code >= 400，则通过response.failure(...)把整条请求标记为失败，并记录message到日志；
3. 成功口径：只有整条流处理完毕且未发现任何错误，才会触发{ENDPOINT} end成功事件，其中response_time为整条 SSE 流的总耗时（毫秒）、response_length为收到的事件行数。

也就是说，Locust 报告里会出现三类自定义名称的统计项：/run_sse message（基础请求计数）、/run_sse end（流完整成功的耗时分布）、/run_sse rate_limited 429s（限流计数）。对 LLM 类服务而言，“流完整走完且无错误事件”的 P95/P99 时长比单点响应时间更有参考价值。

3.4 对鉴权部署的兼容：_ID_TOKEN

脚本开头对请求头的处理（load_test.py#L40-L42）：

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

本地none目标下服务不鉴权，该分支不生效；但从仓库结构看，cloud_run、gke、agent_runtime等部署目标下携带完全相同的 load_test.py。由于 Cloud Run 默认开启 IAM 鉴权，同一脚本可配合_ID_TOKEN（如gcloud auth print-identity-token的返回值）直接压测已部署环境——这与 testing-deployed-agents.md 技能文档中“负载测试配置与 CI/CD 集成参见tests/load_test/README.md，负载测试会在 staging CD 阶段自动运行”的说明相互呼应。

四、运行与扩展建议

运行前提：项目通过 agents-cli scaffold 生成（deployment_target=none、Python/ADK 模板），本地可用uv运行项目依赖，Python 3 环境可创建 venv 安装 Locust；服务需监听0.0.0.0:8000或相应调整-H参数。

调整负载模型：三个旋钮即可覆盖大多数场景——-u（并发用户数）、-r（爬坡速率）、-t（时长）；需要观察交互式图表时去掉--headless，Locust 会启动 Web UI（默认http://localhost:8089）。

扩展压测维度：修改load_test.py中的new_message文本即可切换不同复杂度的对话负载；若需要模拟多轮会话，可将user_id/session_id提到类属性级别并在任务间复用，而不是每次任务都新建会话。

解读报告：CSV 中的request列对应脚本内name=指定的三类统计项；重点观察/run_sse end行的 95/99 分位耗时与failure列比例，/run_sse rate_limited 429s行非零则说明模型后端或网关限流已被触达，应作为容量上限信号。

小结：该模板把“压测生成式 AI 智能体”沉淀成了三步可复制的操作：本地起服务、隔离装 Locust、无头跑脚本出报告；其脚本层面真正有价值的设计是——按真实用户路径（建会话 → SSE 流式对话）施压，并把“流内错误”与“限流”纳入失败统计口径，这正好弥补了普通 HTTP 压测对 LLM 流式服务不适用的短板。

【免费下载链接】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