首页
/ Dify SSE 工作流压测实战：基于 Locust 的流式性能基准测试套件

Dify SSE 工作流压测实战：基于 Locust 的流式性能基准测试套件

2026-09-04 22:24:50作者：韦蓉瑛
dify
Build Agentic workflows, RAG pipelines, with rich AI model and tool support on one collaborative workspace. Deploy on cloud, VPC, or self-hosted, so teams move from prototype to production without rebuilding the stack.
项目地址：https://gitcode.com/GitHub_Trending/di/dify

本文以 Dify 仓库自带的压测套件（scripts/stress-test/）为主体，完整讲解如何用 Locust 对 Dify 工作流的 Server-Sent Events（SSE）流式执行接口 /v1/workflows/run 做性能基准测试。读完你能掌握：一键完成环境预置（管理员、插件、工作流、API Key）的全流程、四个核心 SSE 指标（活跃连接数、建连速率、TTFE、事件吞吐）的含义与判定标准，以及如何用 Gunicorn、PostgreSQL 与系统参数对服务端做针对性调优。

一、为什么专门测 SSE 流式性能

Dify 的工作流执行接口在 response_mode=streaming 下返回的不是一个一次性 JSON，而是一条持续的 Server-Sent Events 事件流。压测这类接口和普通 REST 接口有本质区别：连接在收到最后一个事件前不能关闭，性能瓶颈更多体现在首事件延迟和单位时间事件投递速率上，而不是单纯的请求-响应往返时间。

因此 sse_benchmark.py 没有简单复用 Locust 的默认统计，而是自己维护了一套面向流的指标采集器 MetricsTracker，并实现了符合 W3C 规范的 SSEParser。整个套件的四个核心观测指标为：

  1. Active SSE Connections（活跃 SSE 连接数）——任意时刻仍处于打开状态的 SSE 连接数量。
  2. New Connection Rate（建连速率，conn/sec）——每秒新建的 SSE 连接数。
  3. Time to First Event (TTFE)（首事件时间）——从发出请求到收到第一条 SSE 事件的延迟。
  4. Event Throughput（事件吞吐，events/sec）——所有连接上每秒投递的 SSE 事件数。

说明：普通 Locust 统计里的 req/s、Avg/Min/Max/Med 仍然会保留，但它们对 SSE 场景的参考价值有限，真正的判定要看下面这套流式指标。

二、被测端点与请求形态

压测集中打的是单一端点 /v1/workflows/run，这一定义在 sse_benchmark.py 顶部通过环境变量给出，并附带若干可调项：

WORKFLOW_PATH = os.getenv("WORKFLOW_PATH", "/v1/workflows/run")
CONNECT_TIMEOUT = float(os.getenv("CONNECT_TIMEOUT", "10"))
READ_TIMEOUT = float(os.getenv("READ_TIMEOUT", "60"))
TERMINAL_EVENTS = [e.strip() for e in os.getenv("TERMINAL_EVENTS", "workflow_finished,error").split(",") if e.strip()]
QUESTIONS_FILE = os.getenv("QUESTIONS_FILE", "")
  • WORKFLOW_PATH 默认为 /v1/workflows/run。
  • CONNECT_TIMEOUT / READ_TIMEOUT 分别是连接与读取超时（秒），默认 10s / 60s。
  • TERMINAL_EVENTS 是判定"一条流正常结束"的终止事件集合，默认 workflow_finished,error。
  • QUESTIONS_FILE 允许从外部文件加载自定义问题池，缺省时使用代码内置的默认问题。

每个虚拟用户 DifyWorkflowUser 发起的请求体形如下（见 sse_benchmark.py 的 test_workflow_stream 任务）：

headers = {
    "Authorization": f"Bearer {self.api_token}",
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
    "Cache-Control": "no-cache",
}

data = WorkflowRequestData(
    inputs=WorkflowInputs(question=question),
    response_mode="streaming",
    user=f"user_{self.user_counter}",
)

要点：

  • 认证走 Bearer <api_token>，这个 token 由前置 setup 流程创建并写进状态文件（下文详述）。
  • response_mode="streaming" 触发 SSE 流式响应。
  • user 字段用一个递增计数器保证每用户不同，便于在服务端日志里区分。
  • 请求通过 self.client.request(..., stream=True, catch_response=True) 发出，stream=True 是关键——它让 Locust 不一次性读满响应，而是按行迭代，从而真实模拟流式消费。

被测的工作流本身非常简单，是一个 Start → LLM → End 的三节点 DSL（见 workflow_llm.yml），LLM 节点使用 gpt-4o（provider 为 langgenius/openai/openai），prompt 直接引用开始节点的 question 变量。正因为工作流足够简单，压出来的指标主要反映平台与基础设施的流式承载能力，而非复杂编排逻辑。

三、环境预置：setup_all.py 一键装好依赖

压测前置依赖一个"能真正跑通流式响应"的 Dify 实例。setup_all.py 负责把这件事自动化，其执行顺序（见 setup_all.py）为：

login_admin.py            -> 登录拿到 access token
install_openai_plugin.py  -> 安装 OpenAI 插件
configure_openai_plugin.py-> 用 Mock 服务器配置 OpenAI 插件
import_workflow_app.py    -> 从 DSL 导入工作流应用
create_api_key.py         -> 为应用创建 API Key
publish_workflow.py       -> 发布工作流

如果检测到 /console/api/setup 返回的 step 不是 finished（即全新实例），会在最前面插入 setup_admin.py 创建首个管理员账号。

3.1 管理员账号的两种情形

全新实例会用默认值创建第一个管理员，可以通过环境变量覆盖（见 setup_all.py 的 build_admin_config，默认 test@dify.ai / dify / password123）：

STRESS_TEST_ADMIN_EMAIL='your-admin@example.com' \
STRESS_TEST_ADMIN_USERNAME='dify' \
STRESS_TEST_ADMIN_PASSWORD='your-password' \
python scripts/stress-test/setup_all.py

对于已初始化、已有管理员账号的实例，只需提供现有账号登录信息：

STRESS_TEST_ADMIN_EMAIL='your-admin@example.com' \
STRESS_TEST_ADMIN_PASSWORD='your-password' \
python scripts/stress-test/setup_all.py

STRESS_TEST_ADMIN_USERNAME 仅在"全新实例走 /console/api/setup 创建首个管理员"时才会用到。

3.2 状态文件 stress_test_state.json

setup 各步骤的产物统一写入 common/config_helper.py 管理的状态文件 setup/config/stress_test_state.json。它包含四个分区：admin、auth、app、api_key。其中压测真正要用的是 api_key.token——run_locust_stress_test.sh 会读取它来做可用性校验（见 run_locust_stress_test.sh），而 sse_benchmark.py 里每个用户通过 ConfigHelper().get_api_key() 取出（见 sse_benchmark.py）：

config_helper = ConfigHelper()
self.api_token = config_helper.get_api_key()

if not self.api_token:
    raise ValueError("API key not found. Please run setup_all.py first.")

问题池的加载也在这里：若指定了 QUESTIONS_FILE 且文件存在，则逐行读取非空行；否则回退到内置的 5 个默认问题（见 sse_benchmark.py）。

3.3 Mock OpenAI 服务器

为了让压测不依赖真实 OpenAI 配额与网络，套件自带一个 Mock 服务器 mock_openai_server.py。它监听 http://localhost:5004，提供与 OpenAI 兼容的端点：

GET  /v1/models
POST /v1/chat/completions
POST /v1/completions
POST /v1/embeddings
GET  /v1/models/<model_id>
GET  /health

其中 /v1/chat/completions 在 stream=True 时会按 OpenAI 的 chunk 格式逐词吐出 data: {json}\n\n，每个词之间 time.sleep(0.05) 模拟真实流式延迟，最后以 data: [DONE]\n\n 收尾（见 mock_openai_server.py）。这套"可控、可复现、无外部依赖"的 mock 是压测结果可比性的关键。

四、服务端启动：必须用 Gunicorn 生产模式

压测结果的准确性高度依赖服务端的启动方式。 README 明确强调：不要用 Flask debug 模式，而要用 Gunicorn + gevent worker 生产模式（README 中亦被 run_locust_stress_test.sh 检测到 werkzeug/flask 监听 5001 端口时主动告警拦截）。

# Run from the api directory
cd api
uv run gunicorn \
  --bind 0.0.0.0:5001 \
  --workers 4 \
  --worker-class gevent \
  --timeout 120 \
  --keep-alive 5 \
  --log-level info \
  --access-logfile - \
  --error-logfile - \
  app:app

各参数含义（README 给出的解释）：

  • --workers 4：worker 进程数，按 CPU 核心数调整。
  • --worker-class gevent：异步 worker，用于处理并发连接。
  • --timeout 120：长耗时请求的 worker 超时。
  • --keep-alive 5：保持连接存活以支撑 SSE 流式。

这里有一个值得注意的实现细节：Dify 的 Gunicorn 配置 api/gunicorn.conf.py 会在 gevent worker 里做 monkey-patching，把 psycopg2（psycogreen）和 gRPC 一并 patch 成 gevent 协程，从而让数据库调用与 gRPC 调用都不阻塞事件循环。这正是"高并发下用 gevent worker 能扛住 SSE 长连接"的底层原因——从源码结构看，若换成 sync worker，每条 SSE 流都会独占一个线程，并发承载会显著下降。

不推荐用于压测的方式：

# Debug mode - DO NOT use for stress testing (slow performance)
./dev/start-api  # 运行 Flask debug 单线程模式

Mock 服务器同样需要启动：

python scripts/stress-test/setup/mock_openai_server.py

五、运行压测

5.1 推荐方式：封装脚本

# 默认配置（headless）
./scripts/stress-test/run_locust_stress_test.sh

# 直接 uvx 运行
uvx --from locust locust -f scripts/stress-test/sse_benchmark.py --host http://localhost:5001

# 带 Web UI（访问 http://localhost:8089）
uvx --from locust locust -f scripts/stress-test/sse_benchmark.py --host http://localhost:5001 --web-port 8089

run_locust_stress_test.sh 会自动完成四件事：

  1. 校验 Dify API（http://localhost:5001/health）与 Mock 服务器（http://localhost:5004/v1/models）在运行，并在检测到 debug 模式时告警；
  2. 从 stress_test_state.json 读取并校验 API token；
  3. 交互式选择 headless 或 Web UI 模式，然后用 uvx --from locust locust 执行 sse_benchmark.py；
  4. 在 reports/YYYYMMDD_HHMMSS/ 目录生成报告并回显关键指标。

注意：Locust 通过 uvx --from locust 运行，不装在 API 项目环境里（README 的 Troubleshooting 也据此解释 ModuleNotFoundError: No module named 'locust' 并非问题）；而 sseclient-py 属于 API 项目依赖。

5.2 配置文件 locust.conf

压测参数集中在 locust.conf：

host = http://localhost:5001      # 目标地址
users = 10                        # 并发用户数
spawn-rate = 2                    # 每秒生成用户数
run-time = 1m                     # 测试时长（30s / 5m / 1h）
locustfile = scripts/stress-test/sse_benchmark.py
headless = true                   # 无 Web UI
print-stats = true
loglevel = INFO
# csv = reports/locust_results    # 取消注释启用 CSV
# html = reports/locust_report.html # 取消注释启用 HTML 报告

脚本会用 grep 解析其中的 users、spawn-rate、run-time 再拼进 --users --spawn-rate --run-time 命令行参数。

5.3 自定义问题池

直接改 sse_benchmark.py 里的 self.questions：

self.questions = [
    "Your custom question 1",
    "Your custom question 2",
    # Add more questions...
]

或者更推荐的方式：准备一个每行一个问题的文本文件，然后用环境变量 QUESTIONS_FILE=/path/to/questions.txt 运行，sse_benchmark.py 会优先读取该文件（见 sse_benchmark.py），无需改动源码。

5.4 高级用法

# 指定用户数与生成速率
uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \
  --host http://localhost:5001 --users 50 --spawn-rate 5

# 生成 CSV 报告
uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \
  --host http://localhost:5001 --csv reports/results

# 固定运行时长 + headless
uvx --from locust locust -f scripts/stress-test/sse_benchmark.py \
  --host http://localhost:5001 --run-time 5m --headless

多次迭代对比：

for i in {1..3}; do
    echo "Run $i of 3"
    ./scripts/stress-test/run_locust_stress_test.sh
    sleep 60
done

六、报告结构与指标解读

6.1 报告目录

每次运行在 reports/ 下生成一个按时间戳命名的子目录：

  • YYYYMMDD_HHMMSS/locust_summary.txt —— 完整控制台输出（含指标）
  • YYYYMMDD_HHMMSS/locust_report.html —— 带图表的交互式 HTML 报告
  • YYYYMMDD_HHMMSS/locust_stats.csv —— 详细统计 CSV
  • YYYYMMDD_HHMMSS/locust_stats_history.csv —— 时序数据
  • YYYYMMDD_HHMMSS/sse_metrics_YYYYMMDD_HHMMSS.json —— 自定义 SSE 指标（机器可读）

其中 JSON 报告由 on_test_stop 钩子调用 export_json_report 写出（见 sse_benchmark.py），结构为 { timestamp, duration_seconds, metrics, locust_stats }，metrics 即 MetricsSnapshot 的全部字段，方便 CI 做回归分析。

6.2 核心指标与健康阈值

指标 含义 健康参考值
Active Connections 任意时刻打开的 SSE 连接数 负载下应保持稳定、不塌落
Connection Rate (conn/sec) 每秒新建连接数 轻载 5–10；中载 20–50；重载 100+
TTFE (ms) 首事件延迟 优秀 <50ms；良好 50–100；可接受 100–500；差 >500
Event Throughput (events/sec) 全连接每秒事件数 单连接 10–20；10 连接 50–100；100 连接 200–500
RPS 每秒请求数 优秀 >50；良好 20–50；可接受 10–20；待改进 <10

分位数（P50/P95/P99）含义：P50 表示 50% 请求在该时间内完成，以此类推。成功率方面，生产就绪应 >99%，偏低通常意味着错误或超时。

6.3 示例输出

============================================================
DIFY SSE STRESS TEST
============================================================

[2025-09-12 15:45:44,468] Starting test run with 10 users at 2 users/sec

============================================================
SSE Metrics | Active:   8 | Total Conn:   142 | Events:   2841
Rates: 2.4 conn/s | 47.3 events/s | TTFE: 43ms
============================================================

Type     Name                          # reqs  # fails |    Avg     Min     Max    Med | req/s  failures/s
POST     /v1/workflows/run                  142   0(0.00%) |     41      18     192     38 |   2.37        0.00
         Aggregated                         142   0(0.00%) |     41      18     192     38 |   2.37        0.00

============================================================
FINAL RESULTS
============================================================
Total Connections: 142
Total Events:      2841
Average TTFE:      43 ms
============================================================

实时指标框每 5 秒刷新一次（on_test_start 里 report_stats 线程 time.sleep(5)，见 sse_benchmark.py）。各字段含义：

  • Active：当前打开的 SSE 连接数；Total Conn：累计建立连接数；Events：累计收到事件数。
  • conn/s：建连速率；events/s：事件投递速率；TTFE：平均首事件时间。

注意 README 正文有两处口径：一处说实时报告每 5 秒一次，一处把实时框写作"Updates every 10 seconds"。从源码看，刷新频率是 5 秒，而"10 秒"指的是速率计算用的滑动窗口长度（MetricsTracker.get_stats 中 time_window = 10.0，见 sse_benchmark.py）。也就是说：指标每 5 秒重算一次，但 conn/s、events/s 是基于最近 10 秒窗口内的事件数除窗口时长得到的速率。

6.4 结果判读

良好表现：

  • 零失败（0.00%）
  • TTFE < 100ms
  • 活跃连接稳定
  • 事件吞吐一致

预警信号：

  • 失败率 > 1%
  • TTFE > 500ms
  • 活跃连接下降
  • 事件速率随时间走低

七、测试场景与性能调优

7.1 四档负载场景

# 轻载
concurrency: 10
iterations: 100

# 正常
concurrency: 100
iterations: 1000

# 重载
concurrency: 500
iterations: 5000

# 极限
concurrency: 1000
iterations: 10000

7.2 Gunicorn 按负载分档调参

# 轻载（10-50 并发）
uv run gunicorn --bind 0.0.0.0:5001 --workers 2 --worker-class gevent app:app

# 中载（50-200 并发）
uv run gunicorn --bind 0.0.0.0:5001 --workers 4 --worker-class gevent --worker-connections 1000 app:app

# 重载（200-1000 并发）
uv run gunicorn --bind 0.0.0.0:5001 --workers 8 --worker-class gevent --worker-connections 2000 --max-requests 1000 app:app

worker 数经验公式：Workers = (2 × CPU 核心数) + 1；SSE/WebSocket 场景用 gevent worker；CPU 密集型任务用 sync worker。

7.3 PostgreSQL 连接池

高并发压测时，可上调 docker/middleware.env 里的 POSTGRES_MAX_CONNECTIONS（默认 100）：

# Edit docker/middleware.env
POSTGRES_MAX_CONNECTIONS=200  # 默认 100

# 分档参考
# 轻载（10-50 用户）: 100
# 中载（50-200 用户）: 200
# 重载（200-1000 用户）: 500

改完重启数据库容器：

docker compose -f docker/docker-compose.middleware.yaml down db
docker compose -f docker/docker-compose.middleware.yaml up -d db

从仓库配置可以印证这个链路：docker-compose.middleware.yaml 中 PostgreSQL 启动命令为 postgres -c 'max_connections=${POSTGRES_MAX_CONNECTIONS:-100}'，即最终传给 postgres 的 max_connections 就是该环境变量，缺省 100；docker/.env.example 则给了 POSTGRES_MAX_CONNECTIONS=200 的示例值。

内存占用经验：每个连接约 10MB RAM。100 连接约 1GB、200 连接约 2GB、500 连接约 5GB，需确保数据库服务器内存充足。

7.4 系统层优化

  1. 提高文件描述符上限：

    ulimit -n 65536

  2. Linux TCP 调优：

    sudo sysctl -w net.core.rmem_max=134217728
sudo sysctl -w net.core.wmem_max=134217728
sudo sysctl -w net.ipv4.tcp_fastopen=3

  3. macOS 提高最大连接数：

    sudo sysctl -w kern.ipc.somaxconn=2048

八、排障速查

  • ModuleNotFoundError: No module named 'locust'：Locust 本就通过 uvx --from locust 在 API 项目环境之外运行，属正常。可用 uvx --from locust locust --version 验证。
  • API key configuration not found：先跑 python scripts/stress-test/setup_all.py 生成 stress_test_state.json。
  • 服务未运行：按上文用 Gunicorn 启动 Dify API（5001），并启动 Mock 服务器（5004）。
  • 错误率偏高：降低并发、检查 CPU/内存、查看 API 服务端日志、必要时增大超时。
  • 脚本无执行权限：chmod +x run_benchmark.sh（实际入口是 run_locust_stress_test.sh）。

性能问题定位

  • 响应时间高：查数据库查询性能、外部 API 延迟、服务器资源、网络拥塞。
  • 吞吐低（RPS < 10）：查 CPU 瓶颈、内存约束、数据库连接池、API 限流。
  • 错误率高：查服务端错误日志、资源耗尽、超时配置、连接数上限。

九、为什么选 Locust

README 给出选型理由（相较 Drill）：

  1. 正确的 SSE 支持：能处理流式响应而不过早关闭连接；
  2. 自定义指标：可跟踪 TTFE、流时长等 SSE 专属指标；
  3. Web UI：实时可视化监控与控制；
  4. Python 集成：与现有 Python setup 代码无缝衔接；
  5. 可扩展：便于按具体测试场景定制。

从 sse_benchmark.py 的 SSEParser 可以看出这种"可扩展"落地得很具体：它按 W3C 规范逐行解析 data / event / id 字段，空行判定一条事件结束，: 开头的行按注释忽略，多行 data 会用换行拼接后再触发一次回调。正是这种"边解析边统计"的能力，让 TTFE、事件间隔、流时长这些指标能在压测过程中被实时、逐事件地采集下来。

十、如何改进这套压测套件

  • 配置类改动：调整 locust.conf 的 users / spawn-rate / run-time。
  • 流程改进：修改 run_locust_stress_test.sh 的校验与报告逻辑。
  • 问题覆盖：用 QUESTIONS_FILE 环境变量指向更大规模的问题池，或直接改 sse_benchmark.py 的默认 self.questions。
  • 指标扩展：在 sse_benchmark.py 的 MetricsTracker 与 on_test_start/on_test_stop 钩子中新增采集项与导出字段。
  • 工作流调整：更换 workflow_llm.yml 以压测不同复杂度的编排。

适用前提与限制：本套件面向本地自托管 Dify，默认目标为 http://localhost:5001，依赖 Mock OpenAI 服务器提供可控的流式响应，因此测得的是"平台 + 基础设施"在模拟 LLM 下的 SSE 承载能力，而非接入真实模型服务时的端到端性能。指标健康阈值（如 TTFE、RPS 分档）是 README 给出的经验参考值，实际目标仍应结合自身 SLA 与硬件规格校准。

dify
Build Agentic workflows, RAG pipelines, with rich AI model and tool support on one collaborative workspace. Deploy on cloud, VPC, or self-hosted, so teams move from prototype to production without rebuilding the stack.
项目地址：https://gitcode.com/GitHub_Trending/di/dify
登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.12 K
2.72 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
527
590
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
904
1.82 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
854
1.34 K
docsdocs
暂无描述
Markdown
889
5.78 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.52 K
1.01 K
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库，实现网络在NPU上加速计算。
C++
1.33 K
1.45 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
982
502
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR（Multi-Level Intermediate Representation）构建的，面向昇腾亲和算子编译时使用的中间表示，提供昇腾完备表达能力，通过编译优化提升昇腾AI处理器计算效率，支持通过生态框架使能昇腾AI处理器与深度调优
C++
540
384