负载测试 Databricks Apps 代理

负载测试发现 Databricks Apps 代理在性能下降之前每秒可以维持的最大查询数（QPS）。 本页演示如何执行以下操作：

1. 部署代理的模拟版本，以隔离基础结构吞吐量与 LLM 延迟。
2. 使用 Locust 运行渐变饱和负载测试。
3. 使用交互式仪表板分析结果。

可以使用 Claude Code 技能遵循 AI 辅助路径，或手动设置每个步骤。

要求

• 启用了 Databricks Apps 的 Azure Databricks 工作区。
• 使用 OpenAI 代理 SDK、LangGraph 或自定义框架在 Databricks 应用上部署的代理应用（或已准备好部署）。 请参阅 “创作 AI 代理”并将其部署到 Databricks 应用。
• Databricks CLI 已安装并经过身份验证。 请参阅安装或更新 Databricks CLI。
• Python 3.10 及以上版本与uv 包管理器。
• （对于 AI 辅助路径），已安装 Claude Code。
• 对于负载测试超过约1小时的情况，需要具有M2M OAuth凭据（client_id 和 client_secret）的服务主体。 请参阅 使用 OAuth 授权服务主体访问 Azure Databricks。
  • 对于短时间的负载测试（少于约 1 小时），您现有的用户（U2M）OAuth 凭据来自 databricks auth login 可以正常工作。 对于较长的测试，将 M2M OAuth 与服务主体配合使用 — U2M 令牌在长时间运行期间过期，并导致中测试失败。 创建服务主体需要工作区管理员访问权限。

AI 辅助设置（建议）

如果使用 Claude Code，/load-testing 功能会自动化工作流。 它会读取代理代码、生成模拟、创建负载测试脚本，并指导你完成部署。

Clone https://github.com/databricks/app-templates and run the /load-testing skill against the {your-template} template.

步骤 1：克隆代理模板

技能 /load-testing 包含在 databricks/app-templates 存储库中，既作为顶级 agent-load-testing 技能，又预先同步到每个单独的代理模板中。 拥有 app-templates 的项目，就说明你拥有了相应的技能。

克隆存储库并将其更改为要加载测试的代理的模板目录：

git clone https://github.com/databricks/app-templates.git
cd app-templates/{your-template}

步骤 2：运行负载测试功能

在 Claude Code 中，运行：

/load-testing

该技能以交互方式引导你完成以下步骤。 可以跳过模拟来测试真实代理，或者跳过部署（如果应用已在运行）。

1. 收集参数：询问您的部署状态、计算大小、工作者配置和 OAuth 凭据。
2. 创建负载测试脚本：生成针对您的项目定制的locustfile.pyrun_load_test.py和dashboard_template.py。
3. 模拟 LLM：创建一个特定于 SDK（OpenAI Agents SDK、LangGraph 或自定义）的模拟客户端，该客户端将真实的 LLM 调用替换为可配置的流式延迟。
4. 部署测试应用：指导你部署具有不同计算大小和辅助角色计数的多个应用配置。
5. 运行测试：使用 M2M OAuth 身份验证和渐变饱和执行负载测试。
6. 生成结果：生成具有 QPS、延迟和失败指标的交互式 HTML 仪表板。

手动设置

请按照以下步骤设置和运行负载测试，而无需 AI 帮助。

步骤 1：模拟代理的 LLM 调用（可选）

如果希望包含实际 LLM 延迟的端到端结果，请跳过此步骤。 若要以隔离的方式测量 Databricks Apps 基础结构吞吐量，请模拟 LLM，使其每个请求延迟（通常为 1-30 秒）不会成为瓶颈。

模拟返回具有可配置流式延迟的预设响应，保留完整的请求/响应管道（SSE 流式传输、工具调度、SDK 运行器），仅替换 LLM。 这揭示了 Databricks Apps 平台在负载测试期间可以提供的最大 QPS，同时避免 Foundation Model API 令牌的成本。

模拟计时由两个环境变量控制：

使用默认值时，每个模拟响应大约需要 800 毫秒（10 毫秒 x 80 区块），明显快于实际 LLM 响应（3-15 秒）。 然后，吞吐量数字反映平台，而不是模型。

创建替换真实 LLM 客户端的模拟客户端。 代理代码的其余部分保持不变，方法取决于 SDK。 有关 OpenAI，请参阅 mock_openai_client.py参考实现 中的databricks/app-templates。 相同的模式适用于其他 SDK。

OpenAI 代理 SDK

创建 agent_server/mock_openai_client.py - 使用 MockAsyncOpenAI 流式处理实现的 chat.completions.create() 类。 它立即返回工具调用区块（模拟 LLM 决定调用工具），文本响应区块则根据环境变量MOCK_CHUNK_DELAY_MS和MOCK_CHUNK_COUNT具备可配置的延迟。

将其替换至你的代理中：

from agent_server.mock_openai_client import MockAsyncOpenAI
from agents import set_default_openai_client, set_default_openai_api

set_default_openai_client(MockAsyncOpenAI())
set_default_openai_api("chat_completions")

代理代码（处理程序、工具、流式处理逻辑）的其余部分保持不变。

LangGraph

将 ChatDatabricks 模型替换为返回预生成 AIMessage 对象的模拟：

# Before:
# model = ChatDatabricks(endpoint="databricks-claude-sonnet-4")

# After:
from agent_server.mock_llm import MockChatModel
model = MockChatModel()

模拟应在第一次调用时返回包含工具调用的 AIMessage 对象，在后续调用时返回文本内容，并具有可配置的流式延迟。

自定义代理

请使用模拟实现包裹代理进行的任何外部 API 调用（如大型语言模型、向量搜索、工具 API），这些模拟实现会返回具有可配置延迟的真实响应格式。

步骤 2：设置负载测试脚本

在项目中创建 load-test-scripts/ 目录。 负载测试框架由三个脚本组成，这些脚本与框架无关，适用于任何 Databricks Apps 代理。

<project-root>/
  agent_server/                  # Your existing agent code
  load-test-scripts/             # Load testing scripts (create this)
    run_load_test.py             #   CLI orchestrator
    locustfile.py                #   Locust test with SSE streaming + TTFT tracking
    dashboard_template.py        #   Interactive HTML dashboard generator
  load-test-runs/                # Results (auto-created per run)
    <run-name>/
      dashboard.html             #   Interactive dashboard
      test_config.json           #   Test parameters for reproducibility
      <label>/                   #   Per-config Locust CSV output

框架包括以下文件：

• locustfile.py：一个 Locust 负载测试，发送 POST /invocations 请求 stream: true，解析 SSE 流，跟踪 TTFT（时间到第一个令牌）作为自定义指标，使用 M2M OAuth 令牌交换自动刷新，并实现一个StepRampShape以便将用户数量从 step_size 增加到 max_users，每级保持 step_duration 秒。
• run_load_test.py：一个命令行编排器，它按配置顺序测试每个应用 URL，并提供独立的指标。 它处理 OAuth 令牌刷新，在每个测试之前运行健康检查和预热，并将结果保存到load-test-runs/<run-name>/<label>/。
• dashboard_template.py：Chart.js 生成一个独立的 HTML 仪表板，包含 KPI 卡、条形图（按配置的 QPS、延迟、TTFT）、QPS 渐变进度折线图以及完整的结果表。 可以独立运行： uv run dashboard_template.py ../load-test-runs/<run-name>/.

安装依赖项

负载测试脚本在内部使用自己的 pyproject.toml 脚本 load-test-scripts/ ，以避免污染代理的生产依赖项。 创建 load-test-scripts/pyproject.toml：

[project]
name = "load-test-scripts"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
  "locust>=2.32,<2.40",
  "urllib3<2.3",
  "requests",
]

从 load-test-scripts/ 目录中安装：

cd load-test-scripts/
uv sync

步骤 3：部署具有不同配置的测试应用

部署具有不同计算大小和工作器计数的多个 Databricks 应用，以查找工作负荷的最佳配置。

建议的测试矩阵

以下配置侧重于先前测试中识别的最佳点。 如果需要更广泛的覆盖范围，请在任一端（例如， medium-w1 或 large-w12）添加一个配置，但下面的六个配置通常足够。

配置计算大小

使用 Databricks CLI 在创建或更新应用时设置计算大小：

# Create a new app with Medium compute
databricks apps create <app-name> --compute-size MEDIUM

# Update an existing app to Large compute
databricks apps update <app-name> --compute-size LARGE

使用声明性自动化捆绑配置工作进程数量

start-server （通过 AgentServer.run()） --workers 直接接受标志。 使用 DAB 变量将工作器计数传递到 command 数组中：

variables:
  app_name:
    default: 'my-agent-medium-w2'
  workers:
    default: '2'

resources:
  apps:
    load_test_app:
      name: ${var.app_name}
      source_code_path: .
      config:
        command: ['uv', 'run', 'start-server', '--workers', '${var.workers}']
        env:
          - name: MOCK_CHUNK_DELAY_MS
            value: '10'
          - name: MOCK_CHUNK_COUNT
            value: '80'

targets:
  medium-w2:
    default: true
    variables:
      app_name: 'my-agent-medium-w2'
      workers: '2'
  large-w8:
    variables:
      app_name: 'my-agent-large-w8'
      workers: '8'

部署和验证

使用 Databricks CLI 部署每个目标：

databricks bundle deploy --target medium-w2
databricks bundle run load_test_app --target medium-w2

在运行负载测试之前验证应用是否处于活动状态：

databricks apps get <app-name> --output json | jq '{app_status, compute_status, url}'

步骤 4：运行负载测试

设置身份验证

根据您计划运行的持续时间选择您的身份验证方式：

• 短测试（不到 1 小时）：使用来自databricks auth login的已存在用户登录凭据。 无需额外的设置。
• 长测试（超过 1 小时（例如隔夜运行）：将 M2M OAuth 与服务主体配合使用。 U2M 令牌过期，导致测试在中途中断。 创建服务主体需要工作区管理员访问权限。

对于 M2M OAuth，在运行测试之前导出服务主体凭据：

export DATABRICKS_HOST=https://your-workspace.cloud.databricks.com
export DATABRICKS_CLIENT_ID=<your-client-id>
export DATABRICKS_CLIENT_SECRET=<your-client-secret>

参数参考

示例命令

快速进行单个应用程序测试（短时间运行 - 使用 databricks auth login 会话）：

cd load-test-scripts/

uv run run_load_test.py \
    --app-url https://my-app.aws.databricksapps.com \
    --dashboard --run-name quick-test

建议的 6 个配置（长期运行 — 传递 M2M 凭据）的完整矩阵。 按--app-url相同的顺序传递--compute-size标志：

uv run run_load_test.py \
    --app-url https://my-app-medium-w2.aws.databricksapps.com \
    --app-url https://my-app-medium-w3.aws.databricksapps.com \
    --app-url https://my-app-medium-w4.aws.databricksapps.com \
    --app-url https://my-app-large-w6.aws.databricksapps.com \
    --app-url https://my-app-large-w8.aws.databricksapps.com \
    --app-url https://my-app-large-w10.aws.databricksapps.com \
    --compute-size medium --compute-size medium --compute-size medium \
    --compute-size large --compute-size large --compute-size large \
    --client-id $DATABRICKS_CLIENT_ID \
    --client-secret $DATABRICKS_CLIENT_SECRET \
    --dashboard --run-name overnight-sweep

用于确保统计一致性的多次运行：

for RUN in r1 r2 r3 r4 r5; do
  uv run run_load_test.py \
      --app-url https://my-app.aws.databricksapps.com \
      --client-id $DATABRICKS_CLIENT_ID \
      --client-secret $DATABRICKS_CLIENT_SECRET \
      --max-users 1000 --step-size 20 --step-duration 10 \
      --run-name my_test_${RUN} --dashboard || break
done

运行期间会发生什么

1. 运行状况检查：验证应用程序的流功能是否正常（接收 [DONE]）。
2. 预热：发送顺序请求来预热应用。
3. 逐步达到饱和：每 step_duration 秒增加并发用户数。
4. 饱和检测：当增加更多用户后，QPS 仍然持平时，表明已达到吞吐量上限。

估计持续时间

测试中的每个应用都有自己的加载阶段，因此总运行时间会随着矩阵中配置数的增加而增长。 使用以下公式规划运行窗口。

每个应用的持续时间： (max_users / step_size) * step_duration 秒。

默认值（--max-users 300 --step-size 20 --step-duration 30）：

• 15 个步骤 x 30 秒 = 每个应用大约 7.5 分钟
• 对于建议的 6 配置矩阵：每次运行大约 45 分钟

步骤 5：查看和解释结果

1. 打开仪表板：

   open load-test-runs/<run-name>/dashboard.html
   

2. （可选）从现有数据重新生成仪表板，例如更新模板后：

   cd load-test-scripts/
   uv run dashboard_template.py ../load-test-runs/<run-name>/
   

仪表板部分

交互式仪表板包括：

• KPI 卡：最佳配置（按峰值成功 QPS）、整体峰值 QPS、最低延迟，以及总服务请求数。
• QPS by Config：分组条形图显示了每个配置的 QPS 中值、排除故障的峰值 QPS，以及并排显示的峰值 QPS。
• 按配置划分的延迟：显示 p50 和 p95 延迟的分组条。
• TTFT按配置：到第一个令牌的时间（p50 和 p95）。
• 服务的请求总数：每个配置的请求计数。
• QPS 渐变进度：包含 QPS、QPS（不包括故障）、延迟和故障的选项卡的折线图。 包括一个最大用户数滑块，用于聚焦到较低的并发范围。 图表按计算大小（中和大并排）分组。
• 完整结果表：具有峰值 QPS、高峰用户、延迟百分位数和故障率的所有配置。
• 测试参数：用于重现性的配置摘要。

如何解释结果

• 峰值 QPS：在任何渐变步骤中实现的最大 QPS。 这是该配置的吞吐量上限。
• 达到峰值的用户数：达到峰值 QPS 时的并发用户数。 在此点之外添加更多用户不会增加吞吐量。
• 失败率：应为 0% 或非常低。 高故障率意味着应用程序在该并发级别下承受过多负载。
• QPS 渐变图表：查找线条平展的位置。 这就是饱和点：添加更多用户不会增加吞吐量。

Troubleshooting

后续步骤

• 使用实际 LLM 调用进行测试：跳过模拟步骤并部署实际代理以测量端到端延迟，包括 LLM 响应时间。
• 优化工作器数量：使用测试矩阵结果查找适合您的计算能力的最佳工作器数量。
• 教程：评估和改进 GenAI 应用程序 ，以衡量吞吐量的准确性、相关性和安全性。