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 端点而是 ADKAgent Development KitFastAPI 应用暴露的SSEServer-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 locust2.31.1这里把 Locust 固定在2.31.1版本保证压测脚本的 Locust APIHttpUser、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 \ --csvtests/load_test/.results/results \ --htmltests/load_test/.results/report.html各参数含义如下参数含义-f tests/load_test/load_test.py指定 Locust 压测脚本文件-H http://127.0.0.1:8000压测目标基地址--headless无 UI 模式适合终端/CI 直接执行-t 30s压测总时长 30 秒-u 10模拟用户总数Locust 标准语义并发用户上限为 10-r 2每秒新增用户数spawn rate10 个用户约 5 秒内全部拉起--csvtests/load_test/.results/results按results前缀输出系列 CSVstat 汇总、per-request、per-user、percentile、histogram、failures 等--htmltests/load_test/.results/report.html同时生成单文件 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每个虚拟用户在两次任务之间随机等待 13 秒load_test.py#L32-L37模拟人类点击节奏而非恒定满速打点使 CPU/模型侧的负载更接近真实线上形态。3.2 两段式请求先建会话再发流式消息任务函数chat_streamload_test.py#L37-L54首先为每个虚拟用户生成随机user_iduser_{uuid4}然后调用 ADK 的会话创建接口session_response self.client.post( f/apps/{{cookiecutter.agent_directory}}/users/{user_id}/sessions, headersheaders, jsonsession_data, # {state: {preferred_language: English, visit_count: 1}} ) session_id session_response.json()[id]随后以session_id发送流式对话消息load_test.py#L56-L77data { 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 namef{ENDPOINT} message, headersheaders, jsondata, catch_responseTrue, # 手动标记成功/失败 streamTrue, # 流式读取 SSE 响应 params{alt: sse}, ) as response:这段请求体的 schema 与同目录 E2E 测试 test_server_e2e.py 中test_adk_run_sse的用例完全一致——后者先 POST 创建会话、再带streamTrue读取data:前缀的 SSE 行并断言存在带文本的content.parts事件。两者相互印证了 ADK HTTP 面的标准调用契约new_messageparts结构仓库的部署测试技能文档中也特别提示把消息体写成{message: ...}会收到422 Field required且会话必须先创建。3.3 双重错误检测限流与业务错误码SSE 场景下“HTTP 200 不代表请求成功”脚本因此实现了两层错误识别load_test.py#L78-L128限流检测逐行扫描 SSE 流若行内出现429 Too Many Requests额外触发一条名为{ENDPOINT} rate_limited 429s的统计事件response_time0在报告中可以单独看到限流命中次数业务错误码检测尝试把每一行解析为 JSON若事件体包含code字段且code 400则通过response.failure(...)把整条请求标记为失败并记录message到日志成功口径只有整条流处理完毕且未发现任何错误才会触发{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-L42headers {Content-Type: application/json} if os.environ.get(_ID_TOKEN): headers[Authorization] fBearer {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_targetnone、Python/ADK 模板本地可用uv运行项目依赖Python 3 环境可创建 venv 安装 Locust服务需监听0.0.0.0:8000或相应调整-H参数。调整负载模型三个旋钮即可覆盖大多数场景——-u并发用户数、-r爬坡速率、-t时长需要观察交互式图表时去掉--headlessLocust 会启动 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
